163 lines
6.2 KiB
Python
163 lines
6.2 KiB
Python
# #region Test.Core.DefensiveGuards [C:3] [TYPE Module] [SEMANTICS test,defensive,guards,git,superset]
|
|
# @BRIEF Tests verifying defensive null-guards and error handling in GitService and SupersetClient.
|
|
# @RELATION BINDS_TO -> [GitService]
|
|
# @RELATION BINDS_TO -> [SupersetClient]
|
|
# @TEST_EDGE: null_dashboard_id -> ValueError raised for None dashboard_id
|
|
# @TEST_EDGE: null_file_name -> ValueError raised for None file_name
|
|
# @TEST_EDGE: missing_base_dir -> Directory recreated on access
|
|
# @TEST_EDGE: invalid_git_repo -> Recloned when path is not a valid repo
|
|
from pathlib import Path
|
|
import pytest
|
|
import shutil
|
|
import sys
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
|
|
|
|
from src.core.config_models import Environment
|
|
from src.core.superset_client import SupersetClient
|
|
from src.services.git_service import GitService
|
|
|
|
|
|
# #region test_git_service_get_repo_path_guard [C:2] [TYPE Function]
|
|
@pytest.mark.asyncio
|
|
async 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"):
|
|
await service._get_repo_path(None)
|
|
|
|
|
|
# #endregion test_git_service_get_repo_path_guard
|
|
|
|
# #region test_git_service_get_repo_path_recreates_base_dir [C:2] [TYPE Function]
|
|
@pytest.mark.asyncio
|
|
async 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 = await service._get_repo_path(42)
|
|
|
|
assert Path(service.base_path).is_dir()
|
|
assert repo_path == str(Path(service.base_path) / "42")
|
|
|
|
# #endregion test_git_service_get_repo_path_recreates_base_dir
|
|
|
|
# #region test_superset_client_import_dashboard_guard [C:2] [TYPE Function]
|
|
@pytest.mark.asyncio
|
|
async 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"):
|
|
await client.import_dashboard(None)
|
|
|
|
|
|
# #endregion test_superset_client_import_dashboard_guard
|
|
|
|
# #region test_git_service_init_repo_reclones_when_path_is_not_a_git_repo [C:2] [TYPE Function]
|
|
@pytest.mark.asyncio
|
|
async 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_dir = Path(service.base_path)
|
|
if target_dir.exists():
|
|
shutil.rmtree(target_dir, ignore_errors=True)
|
|
target_path = target_dir / "covid"
|
|
target_path.mkdir(parents=True, exist_ok=True)
|
|
|
|
clone_result = MagicMock()
|
|
with patch.object(service, "_clone_with_auth", return_value=clone_result) as mock_clone:
|
|
result = await service.init_repo(10, "https://example.com/org/repo.git", "token", repo_key="covid")
|
|
|
|
assert result is clone_result
|
|
mock_clone.assert_called_once()
|
|
assert not target_path.exists()
|
|
|
|
|
|
# #endregion test_git_service_init_repo_reclones_when_path_is_not_a_git_repo
|
|
|
|
# #region test_git_service_ensure_gitflow_branches_creates_and_pushes_missing_defaults [C:2] [TYPE Function]
|
|
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
|
|
|
|
|
|
# #endregion test_git_service_ensure_gitflow_branches_creates_and_pushes_missing_defaults
|
|
|
|
# #region test_git_service_configure_identity_updates_repo_local_config [C:2] [TYPE Function]
|
|
@pytest.mark.asyncio
|
|
async def test_git_service_configure_identity_updates_repo_local_config():
|
|
"""Verify configure_identity writes repository-local user.name/user.email."""
|
|
service = GitService(base_path="test_repos_identity")
|
|
|
|
config_writer_context = MagicMock()
|
|
config_writer = config_writer_context.__enter__.return_value
|
|
fake_repo = MagicMock()
|
|
fake_repo.config_writer.return_value = config_writer_context
|
|
|
|
with patch.object(service, "get_repo", return_value=fake_repo):
|
|
await service.configure_identity(42, "user_1", "user1@mail.ru")
|
|
|
|
fake_repo.config_writer.assert_called_once_with(config_level="repository")
|
|
config_writer.set_value.assert_any_call("user", "name", "user_1")
|
|
config_writer.set_value.assert_any_call("user", "email", "user1@mail.ru")
|
|
# #endregion test_git_service_configure_identity_updates_repo_local_config
|
|
# #endregion Test.Core.DefensiveGuards
|