Enabled hybrid setups

git_remote_s3/lfs.py

@@ -52,6 +52,8 @@ def write_error_event(*, oid: str, error: str, flush=False):
         sys.stdout.flush()
 
 
+
+
 class LFSProcess:
     def __init__(self, s3uri: str):
         uri_scheme, profile, bucket, prefix = parse_git_url(s3uri)
@@ -152,6 +154,41 @@ def install():
     sys.stdout.write("git-lfs-s3 installed\n")
     sys.stdout.flush()
 
+def _lfs_only_url(remote: str) -> str | None:
+    """
+    Try to get LFS-specific URL (for hybrid setups like GitHub + S3).
+    Priority: .lfsconfig (tracked) > .git/config (local)
+
+    Args:
+        remote: The remote name to look up
+
+    Returns:
+        The LFS URL if found, None otherwise
+    """
+    # 1. Try .lfsconfig first (tracked file, shared with team)
+    result = subprocess.run(
+        ["git", "config", "--file", ".lfsconfig", "--get", f"remote.{remote}.lfsurl"],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+    if result.returncode == 0 and result.stdout.strip():
+        s3uri = result.stdout.decode("utf-8").strip()
+        logger.debug(f"Using lfsurl from .lfsconfig: {s3uri}")
+        return s3uri
+
+    # 2. Try local .git/config (allows per-developer override)
+    result = subprocess.run(
+        ["git", "config", "--get", f"remote.{remote}.lfsurl"],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+    if result.returncode == 0 and result.stdout.strip():
+        s3uri = result.stdout.decode("utf-8").strip()
+        logger.debug(f"Using lfsurl from .git/config: {s3uri}")
+        return s3uri
+
+    return None
+
 
 def main():  # noqa: C901
     if len(sys.argv) > 1:
@@ -196,23 +233,29 @@ def main():  # noqa: C901
                 sys.stdout.write("{}\n")
                 sys.stdout.flush()
                 sys.exit(1)
-            result = subprocess.run(
-                ["git", "remote", "get-url", event["remote"]],
-                stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE,
-            )
-            if result.returncode != 0:
-                logger.error(result.stderr.decode("utf-8").strip())
-                error_event = {
-                    "error": {
-                        "code": 2,
-                        "message": f"cannot resolve remote \"{event['remote']}\"",
+            # Try to get LFS-specific URL (for hybrid setups like GitHub + S3)
+            # Priority: .lfsconfig (tracked) > .git/config (local) > git remote url (fallback)
+            s3uri = _lfs_only_url(event["remote"])
+            if s3uri is None:
+                # 3. Fall back to regular remote URL
+                result = subprocess.run(
+                    ["git", "remote", "get-url", event["remote"]],
+                    stdout=subprocess.PIPE,
+                    stderr=subprocess.PIPE,
+                )
+                if result.returncode != 0:
+                    logger.error(result.stderr.decode("utf-8").strip())
+                    error_event = {
+                        "error": {
+                            "code": 2,
+                            "message": f"cannot resolve remote \"{event['remote']}\"",
+                        }
                     }
-                }
-                sys.stdout.write(f"{json.dumps(error_event)}")
-                sys.stdout.flush()
-                sys.exit(1)
-            s3uri = result.stdout.decode("utf-8").strip()
+                    sys.stdout.write(f"{json.dumps(error_event)}")
+                    sys.stdout.flush()
+                    sys.exit(1)
+                s3uri = result.stdout.decode("utf-8").strip()
+                logger.debug(f"Using remote URL: {s3uri}")
             lfs_process = LFSProcess(s3uri=s3uri)
 
         elif event["event"] == "upload":

test/lfs_test.py

@@ -0,0 +1,176 @@
+from mock import patch
+from git_remote_s3.lfs import _lfs_only_url
+import subprocess
+
+REMOTE_ORIGIN = "origin"
+REMOTE_UPSTREAM = "upstream"
+EXPECTED_URL = "s3://bucket-name/path/to"
+LFS_CONFIG_URL = "s3://lfsconfig-bucket/path"
+
+
+@patch("git_remote_s3.lfs.subprocess.run")
+def test_lfs_only_url_from_lfsconfig(mock_run):
+    """Test that _lfs_only_url returns URL from .lfsconfig when available"""
+    # Mock successful .lfsconfig lookup
+    mock_result = subprocess.CompletedProcess(
+        args=[],
+        returncode=0,
+        stdout=EXPECTED_URL.encode("utf-8"),
+        stderr=b""
+    )
+    mock_run.return_value = mock_result
+
+    result = _lfs_only_url(REMOTE_ORIGIN)
+
+    assert result == EXPECTED_URL
+    # Verify it was called with .lfsconfig first
+    assert mock_run.call_count == 1
+    call_args = mock_run.call_args[0][0]
+    assert "--file" in call_args
+    assert ".lfsconfig" in call_args
+    assert f"remote.{REMOTE_ORIGIN}.lfsurl" in call_args
+
+
+@patch("git_remote_s3.lfs.subprocess.run")
+def test_lfs_only_url_from_git_config(mock_run):
+    """Test that _lfs_only_url falls back to .git/config when .lfsconfig is not available"""
+    # Mock .lfsconfig lookup failing, then .git/config succeeding
+    def side_effect(*args, **kwargs):
+        if "--file" in args[0] and ".lfsconfig" in args[0]:
+            # First call: .lfsconfig fails
+            return subprocess.CompletedProcess(
+                args=args[0],
+                returncode=1,
+                stdout=b"",
+                stderr=b""
+            )
+        else:
+            # Second call: .git/config succeeds
+            return subprocess.CompletedProcess(
+                args=args[0],
+                returncode=0,
+                stdout=EXPECTED_URL.encode("utf-8"),
+                stderr=b""
+            )
+
+    mock_run.side_effect = side_effect
+
+    result = _lfs_only_url(REMOTE_ORIGIN)
+
+    assert result == EXPECTED_URL
+    # Verify both calls were made
+    assert mock_run.call_count == 2
+    # First call should be .lfsconfig
+    first_call = mock_run.call_args_list[0][0][0]
+    assert "--file" in first_call
+    assert ".lfsconfig" in first_call
+    # Second call should be .git/config
+    second_call = mock_run.call_args_list[1][0][0]
+    assert "--get" in second_call
+    assert f"remote.{REMOTE_ORIGIN}.lfsurl" in second_call
+
+
+@patch("git_remote_s3.lfs.subprocess.run")
+def test_lfs_only_url_returns_none_when_not_found(mock_run):
+    """Test that _lfs_only_url returns None when neither config has the URL"""
+    # Mock both lookups failing
+    mock_result = subprocess.CompletedProcess(
+        args=[],
+        returncode=1,
+        stdout=b"",
+        stderr=b""
+    )
+    mock_run.return_value = mock_result
+
+    result = _lfs_only_url(REMOTE_ORIGIN)
+
+    assert result is None
+    # Verify both calls were made
+    assert mock_run.call_count == 2
+
+
+@patch("git_remote_s3.lfs.subprocess.run")
+def test_lfs_only_url_handles_empty_stdout(mock_run):
+    """Test that _lfs_only_url handles empty stdout even with returncode 0"""
+    # Mock .lfsconfig returning success but empty stdout
+    def side_effect(*args, **kwargs):
+        if "--file" in args[0] and ".lfsconfig" in args[0]:
+            return subprocess.CompletedProcess(
+                args=args[0],
+                returncode=0,
+                stdout=b"",  # Empty stdout
+                stderr=b""
+            )
+        else:
+            # .git/config also returns empty
+            return subprocess.CompletedProcess(
+                args=args[0],
+                returncode=0,
+                stdout=b"",
+                stderr=b""
+            )
+
+    mock_run.side_effect = side_effect
+
+    result = _lfs_only_url(REMOTE_ORIGIN)
+
+    assert result is None
+    assert mock_run.call_count == 2
+
+
+@patch("git_remote_s3.lfs.subprocess.run")
+def test_lfs_only_url_handles_whitespace_in_url(mock_run):
+    """Test that _lfs_only_url properly strips whitespace from URLs"""
+    url_with_whitespace = "  s3://bucket-name/path/to  \n"
+    expected_url = "s3://bucket-name/path/to"
+
+    mock_result = subprocess.CompletedProcess(
+        args=[],
+        returncode=0,
+        stdout=url_with_whitespace.encode("utf-8"),
+        stderr=b""
+    )
+    mock_run.return_value = mock_result
+
+    result = _lfs_only_url(REMOTE_ORIGIN)
+
+    assert result == expected_url
+
+
+@patch("git_remote_s3.lfs.subprocess.run")
+def test_lfs_only_url_priority_lfsconfig_over_git_config(mock_run):
+    """Test that .lfsconfig takes priority over .git/config even if both exist"""
+    # Mock .lfsconfig succeeding (should return this, not .git/config)
+    mock_result = subprocess.CompletedProcess(
+        args=[],
+        returncode=0,
+        stdout=LFS_CONFIG_URL.encode("utf-8"),
+        stderr=b""
+    )
+    mock_run.return_value = mock_result
+
+    result = _lfs_only_url(REMOTE_ORIGIN)
+
+    assert result == LFS_CONFIG_URL
+    # Should only call .lfsconfig, not .git/config
+    assert mock_run.call_count == 1
+
+
+@patch("git_remote_s3.lfs.subprocess.run")
+def test_lfs_only_url_with_different_remote_name(mock_run):
+    """Test that _lfs_only_url works with different remote names"""
+    mock_result = subprocess.CompletedProcess(
+        args=[],
+        returncode=0,
+        stdout=EXPECTED_URL.encode("utf-8"),
+        stderr=b""
+    )
+    mock_run.return_value = mock_result
+
+    result = _lfs_only_url(REMOTE_UPSTREAM)
+
+    assert result == EXPECTED_URL
+    # Verify the remote name is used correctly in the config key
+    call_args = mock_run.call_args[0][0]
+    assert f"remote.{REMOTE_UPSTREAM}.lfsurl" in call_args
+