From 31b363edb0e6a35a42fe2749ec58e3ad1a9d9ac9 Mon Sep 17 00:00:00 2001 From: Fusion Date: Thu, 14 May 2026 08:27:31 +0200 Subject: [PATCH] =?UTF-8?q?test(FN-011):=20complete=20Step=207=20=E2=80=94?= =?UTF-8?q?=20tests=20for=20provider,=20credentials,=20and=20operations?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fusion-Task-Id: FN-011 Fusion-Task-Lineage: 4a9aca6f-9d91-43aa-8d2a-d59657c1541a --- apps/api/app/git/credentials.py | 10 +- apps/api/app/git/ssh_key.py | 14 +-- apps/api/tests/test_git_credentials.py | 77 +++++++++++++ apps/api/tests/test_git_operations.py | 148 +++++++++++++++++++++++++ apps/api/tests/test_git_provider.py | 66 +++++++++++ 5 files changed, 299 insertions(+), 16 deletions(-) create mode 100644 apps/api/tests/test_git_credentials.py create mode 100644 apps/api/tests/test_git_operations.py create mode 100644 apps/api/tests/test_git_provider.py diff --git a/apps/api/app/git/credentials.py b/apps/api/app/git/credentials.py index 55b561c..52c9eff 100644 --- a/apps/api/app/git/credentials.py +++ b/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( - default_factory=lambda: datetime.now(timezone.utc) - ) - updated_at: datetime = Field( - default_factory=lambda: datetime.now(timezone.utc) - ) + created_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) + updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) class AccessTokenCredential(GitCredential): diff --git a/apps/api/app/git/ssh_key.py b/apps/api/app/git/ssh_key.py index e918700..6b26367 100644 --- a/apps/api/app/git/ssh_key.py +++ b/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( - default_factory=lambda: datetime.now(timezone.utc) - ) - updated_at: datetime = Field( - default_factory=lambda: datetime.now(timezone.utc) - ) + created_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) + updated_at: datetime = Field(default_factory=lambda: datetime.now(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 diff --git a/apps/api/tests/test_git_credentials.py b/apps/api/tests/test_git_credentials.py new file mode 100644 index 0000000..c607303 --- /dev/null +++ b/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 diff --git a/apps/api/tests/test_git_operations.py b/apps/api/tests/test_git_operations.py new file mode 100644 index 0000000..d2b20ef --- /dev/null +++ b/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" diff --git a/apps/api/tests/test_git_provider.py b/apps/api/tests/test_git_provider.py new file mode 100644 index 0000000..5a5d325 --- /dev/null +++ b/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