import sys from pathlib import Path import shutil import pytest from unittest.mock import MagicMock, patch from git.exc import InvalidGitRepositoryError sys.path.insert(0, str(Path(__file__).parent.parent.parent)) from src.services.git_service import GitService from src.core.superset_client import SupersetClient from src.core.config_models import Environment def test_git_service_get_repo_path_guard(): """Verify that _get_repo_path raises ValueError if dashboard_id is None.""" service = GitService(base_path="test_repos") with pytest.raises(ValueError, match="dashboard_id cannot be None"): service._get_repo_path(None) def test_git_service_get_repo_path_recreates_base_dir(): """Verify _get_repo_path recreates missing base directory before returning repo path.""" service = GitService(base_path="test_repos_runtime_recreate") shutil.rmtree(service.base_path, ignore_errors=True) repo_path = service._get_repo_path(42) assert Path(service.base_path).is_dir() assert repo_path == str(Path(service.base_path) / "42") def test_superset_client_import_dashboard_guard(): """Verify that import_dashboard raises ValueError if file_name is None.""" mock_env = Environment( id="test", name="test", url="http://localhost:8088", username="admin", password="admin" ) client = SupersetClient(mock_env) with pytest.raises(ValueError, match="file_name cannot be None"): client.import_dashboard(None) def test_git_service_init_repo_reclones_when_path_is_not_a_git_repo(): """Verify init_repo reclones when target path exists but is not a valid Git repository.""" service = GitService(base_path="test_repos_invalid_repo") target_path = Path(service.base_path) / "covid" target_path.mkdir(parents=True, exist_ok=True) (target_path / "placeholder.txt").write_text("not a git repo", encoding="utf-8") clone_result = MagicMock() with patch("src.services.git_service.Repo") as repo_ctor: repo_ctor.side_effect = InvalidGitRepositoryError("invalid repo") repo_ctor.clone_from.return_value = clone_result result = service.init_repo(10, "https://example.com/org/repo.git", "token", repo_key="covid") assert result is clone_result repo_ctor.assert_called_once_with(str(target_path)) repo_ctor.clone_from.assert_called_once() assert not target_path.exists() def test_git_service_ensure_gitflow_branches_creates_and_pushes_missing_defaults(): """Verify _ensure_gitflow_branches creates dev/preprod locally and pushes them to origin.""" service = GitService(base_path="test_repos_gitflow_defaults") class FakeRemoteRef: def __init__(self, remote_head): self.remote_head = remote_head class FakeHead: def __init__(self, name, commit): self.name = name self.commit = commit class FakeOrigin: def __init__(self): self.refs = [FakeRemoteRef("main")] self.pushed = [] def fetch(self): return [] def push(self, refspec=None): self.pushed.append(refspec) return [] class FakeHeadPointer: def __init__(self, commit): self.commit = commit class FakeRepo: def __init__(self): self.head = FakeHeadPointer("main-commit") self.heads = [FakeHead("main", "main-commit")] self.origin = FakeOrigin() def create_head(self, name, commit): head = FakeHead(name, commit) self.heads.append(head) return head def remote(self, name="origin"): if name != "origin": raise ValueError("unknown remote") return self.origin repo = FakeRepo() service._ensure_gitflow_branches(repo, dashboard_id=10) local_branch_names = {head.name for head in repo.heads} assert {"main", "dev", "preprod"}.issubset(local_branch_names) assert "dev:dev" in repo.origin.pushed assert "preprod:preprod" in repo.origin.pushed