test(FN-011): complete Step 7 — tests for provider, credentials, and operations

Fusion-Task-Id: FN-011
Fusion-Task-Lineage: 4a9aca6f-9d91-43aa-8d2a-d59657c1541a

apps/api/app/git/credentials.py

@@ -7,7 +7,7 @@ Security rules:
 
 import abc
 import uuid
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -26,12 +26,8 @@ class GitCredential(BaseModel):
     id: uuid.UUID = Field(default_factory=uuid.uuid4)
     kind: CredentialKind
     encrypted_payload: str = Field(repr=False)
-    created_at: datetime = Field(
+    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
-        default_factory=lambda: datetime.now(timezone.utc)
+    updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
-    )
-    updated_at: datetime = Field(
-        default_factory=lambda: datetime.now(timezone.utc)
-    )
 
 
 class AccessTokenCredential(GitCredential):

apps/api/app/git/ssh_key.py

@@ -8,7 +8,7 @@ Security rules:
 
 import base64
 import uuid
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 
 from pydantic import BaseModel, Field
 
@@ -31,12 +31,8 @@ class SshKeyPair(BaseModel):
     public_key: str
     encrypted_private_key: str = Field(repr=False)
     status: SshKeyStatus = SshKeyStatus.generated
-    created_at: datetime = Field(
+    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
-        default_factory=lambda: datetime.now(timezone.utc)
+    updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
-    )
-    updated_at: datetime = Field(
-        default_factory=lambda: datetime.now(timezone.utc)
-    )
     revoked_at: datetime | None = None
 
 
@@ -83,7 +79,7 @@ class SshKeyLifecycle:
         Sets ``revoked_at`` when transitioning to :attr:`SshKeyStatus.revoked`.
         """
         key.status = new_status
-        key.updated_at = datetime.now(timezone.utc)
+        key.updated_at = datetime.now(UTC)
         if new_status == SshKeyStatus.revoked:
-            key.revoked_at = datetime.now(timezone.utc)
+            key.revoked_at = datetime.now(UTC)
         return key

apps/api/tests/test_git_credentials.py

@@ -0,0 +1,77 @@
+"""Tests for credential models and storage interface."""
+
+import uuid
+
+import pytest
+
+from app.git.credentials import AccessTokenCredential, CredentialStorage, GitCredential
+from app.git.types import CredentialKind
+
+
+class MinimalCredentialStorage(CredentialStorage):
+    """Concrete subclass for testing."""
+
+    def create(self, credential: GitCredential) -> uuid.UUID:
+        return credential.id
+
+    def get(self, credential_id: uuid.UUID) -> GitCredential | None:
+        return None
+
+    def delete(self, credential_id: uuid.UUID) -> None:
+        return None
+
+
+def test_git_credential_can_be_instantiated() -> None:
+    cred = GitCredential(
+        kind=CredentialKind.ssh_key,
+        encrypted_payload="encrypted-data",
+    )
+    assert cred.kind == CredentialKind.ssh_key
+    assert cred.encrypted_payload == "encrypted-data"
+    assert isinstance(cred.id, uuid.UUID)
+
+
+def test_access_token_credential_can_be_instantiated() -> None:
+    cred = AccessTokenCredential(encrypted_payload="encrypted-data")
+    assert cred.kind == CredentialKind.access_token
+    assert cred.encrypted_payload == "encrypted-data"
+
+
+def test_credential_storage_cannot_be_instantiated_directly() -> None:
+    with pytest.raises(TypeError):
+        CredentialStorage()  # type: ignore[abstract]
+
+
+def test_no_plaintext_secret_fields() -> None:
+    fields = set(GitCredential.model_fields.keys())
+    assert "token" not in fields
+    assert "private_key" not in fields
+
+
+def test_extra_forbidden() -> None:
+    with pytest.raises(ValueError):
+        GitCredential(
+            kind=CredentialKind.access_token,
+            encrypted_payload="encrypted-data",
+            secret_plaintext="should-fail",  # type: ignore[call-arg]
+        )
+
+
+def test_minimal_credential_storage_implements_all_methods() -> None:
+    storage = MinimalCredentialStorage()
+    cred = GitCredential(
+        kind=CredentialKind.access_token,
+        encrypted_payload="encrypted-data",
+    )
+    assert storage.create(cred) == cred.id
+    assert storage.get(cred.id) is None
+    storage.delete(cred.id)
+
+
+def test_encrypted_payload_not_in_repr() -> None:
+    cred = GitCredential(
+        kind=CredentialKind.ssh_key,
+        encrypted_payload="secret-value",
+    )
+    repr_str = repr(cred)
+    assert "secret-value" not in repr_str

apps/api/tests/test_git_operations.py

@@ -0,0 +1,148 @@
+"""Tests for local Git operations."""
+
+import subprocess
+import tempfile
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from app.git.operations import LocalGitOperations
+
+
+@pytest.fixture
+def local_git() -> LocalGitOperations:
+    return LocalGitOperations()
+
+
+@pytest.fixture
+def temp_repo() -> Any:
+    with tempfile.TemporaryDirectory() as tmpdir:
+        repo_path = Path(tmpdir) / "repo"
+        repo_path.mkdir()
+        subprocess.run(
+            ["git", "init", "--initial-branch=main"],
+            cwd=repo_path,
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+        subprocess.run(
+            ["git", "config", "user.email", "test@example.com"],
+            cwd=repo_path,
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+        subprocess.run(
+            ["git", "config", "user.name", "Test User"],
+            cwd=repo_path,
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+        yield repo_path
+
+
+def test_clone_raises_not_implemented_error(local_git: LocalGitOperations) -> None:
+    with pytest.raises(NotImplementedError):
+        local_git.clone("https://example.com/repo.git", Path("/tmp/dest"), "cred-id")
+
+
+def test_fetch_raises_not_implemented_error(local_git: LocalGitOperations, temp_repo: Path) -> None:
+    with pytest.raises(NotImplementedError):
+        local_git.fetch(temp_repo, "cred-id")
+
+
+def test_push_raises_not_implemented_error(local_git: LocalGitOperations, temp_repo: Path) -> None:
+    with pytest.raises(NotImplementedError):
+        local_git.push(temp_repo, "cred-id")
+
+
+def test_get_status_on_non_git_directory_raises(local_git: LocalGitOperations) -> None:
+    with tempfile.TemporaryDirectory() as tmpdir:
+        with pytest.raises(RuntimeError, match="Not a git repository"):
+            local_git.get_status(Path(tmpdir))
+
+
+def test_get_status_clean_repo(local_git: LocalGitOperations, temp_repo: Path) -> None:
+    status = local_git.get_status(temp_repo)
+    assert status["branch"] == "main"
+    assert status["clean"] is True
+    assert status["untracked"] == []
+    assert status["modified"] == []
+    assert status["staged"] == []
+    assert status["deleted"] == []
+
+
+def test_get_status_untracked_file(local_git: LocalGitOperations, temp_repo: Path) -> None:
+    (temp_repo / "newfile.txt").write_text("hello")
+    status = local_git.get_status(temp_repo)
+    assert "newfile.txt" in status["untracked"]
+    assert status["clean"] is False
+
+
+def test_get_status_staged_file(local_git: LocalGitOperations, temp_repo: Path) -> None:
+    file_path = temp_repo / "newfile.txt"
+    file_path.write_text("hello")
+    subprocess.run(
+        ["git", "add", "newfile.txt"],
+        cwd=temp_repo,
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    status = local_git.get_status(temp_repo)
+    assert "newfile.txt" in status["staged"]
+    assert status["clean"] is False
+
+
+def test_get_status_modified_file(local_git: LocalGitOperations, temp_repo: Path) -> None:
+    file_path = temp_repo / "newfile.txt"
+    file_path.write_text("hello")
+    subprocess.run(
+        ["git", "add", "newfile.txt"],
+        cwd=temp_repo,
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    file_path.write_text("world")
+    status = local_git.get_status(temp_repo)
+    assert "newfile.txt" in status["modified"]
+    assert status["clean"] is False
+
+
+def test_get_status_deleted_file(local_git: LocalGitOperations, temp_repo: Path) -> None:
+    file_path = temp_repo / "newfile.txt"
+    file_path.write_text("hello")
+    subprocess.run(
+        ["git", "add", "newfile.txt"],
+        cwd=temp_repo,
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    subprocess.run(
+        ["git", "commit", "-m", "add file"],
+        cwd=temp_repo,
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    file_path.unlink()
+    status = local_git.get_status(temp_repo)
+    assert "newfile.txt" in status["deleted"]
+    assert status["clean"] is False
+
+
+def test_get_status_branch_name(local_git: LocalGitOperations, temp_repo: Path) -> None:
+    subprocess.run(
+        ["git", "checkout", "-b", "feature-branch"],
+        cwd=temp_repo,
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    status = local_git.get_status(temp_repo)
+    assert status["branch"] == "feature-branch"

apps/api/tests/test_git_provider.py

@@ -0,0 +1,66 @@
+"""Tests for Git provider abstraction and types."""
+
+from typing import Any
+
+import pytest
+
+from app.git.provider import GitProvider
+from app.git.types import ConnectionStatus, CredentialKind, ProviderKind, SshKeyStatus
+
+
+class MinimalGitProvider(GitProvider):
+    """Concrete subclass for testing."""
+
+    def get_kind(self) -> ProviderKind:
+        return ProviderKind.generic
+
+    def validate_connection(self, git_url: str, credential_id: str) -> ConnectionStatus:
+        return ConnectionStatus.connected
+
+    def list_repositories(self, credential_id: str) -> list[dict[str, Any]]:
+        return []
+
+    def create_deploy_key(self, git_url: str, public_key: str) -> str:
+        return "key-id"
+
+    def delete_deploy_key(self, git_url: str, deploy_key_id: str) -> None:
+        return None
+
+    def get_default_branch(self, git_url: str, credential_id: str) -> str:
+        return "main"
+
+
+def test_git_provider_cannot_be_instantiated_directly() -> None:
+    with pytest.raises(TypeError):
+        GitProvider()  # type: ignore[abstract]
+
+
+def test_minimal_git_provider_implements_all_methods() -> None:
+    provider = MinimalGitProvider()
+    assert provider.get_kind() == ProviderKind.generic
+    status = provider.validate_connection("https://example.com/repo.git", "cred-id")
+    assert status == ConnectionStatus.connected
+    assert provider.list_repositories("cred-id") == []
+    assert provider.create_deploy_key("https://example.com/repo.git", "ssh-rsa AAAA") == "key-id"
+    provider.delete_deploy_key("https://example.com/repo.git", "key-id")
+    assert provider.get_default_branch("https://example.com/repo.git", "cred-id") == "main"
+
+
+@pytest.mark.parametrize("member", ["github", "gitlab", "gitea", "forgejo", "generic"])
+def test_provider_kind_membership(member: str) -> None:
+    assert member in ProviderKind
+
+
+@pytest.mark.parametrize("member", ["ssh_key", "access_token"])
+def test_credential_kind_membership(member: str) -> None:
+    assert member in CredentialKind
+
+
+@pytest.mark.parametrize("member", ["pending", "connected", "disconnected", "error"])
+def test_connection_status_membership(member: str) -> None:
+    assert member in ConnectionStatus
+
+
+@pytest.mark.parametrize("member", ["generated", "registered", "rotating", "revoked"])
+def test_ssh_key_status_membership(member: str) -> None:
+    assert member in SshKeyStatus