310 lines
13 KiB
Python
310 lines
13 KiB
Python
# #region Test.Api.GitRepoLifecycleRoutes [C:3] [TYPE Module] [SEMANTICS test,git,lifecycle,sync,promote,deploy]
|
|
# @BRIEF Tests for Git lifecycle API routes — sync, promote, deploy.
|
|
# @RELATION BINDS_TO -> [GitRepoLifecycleRoutes]
|
|
|
|
import os
|
|
|
|
os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:")
|
|
os.environ.setdefault("AUTH_DATABASE_URL", "sqlite:///:memory:")
|
|
os.environ.setdefault("SECRET_KEY", "test-secret-key-for-tests")
|
|
os.environ.setdefault("DEV_MODE", "true")
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.testclient import TestClient
|
|
|
|
_src = str(Path(__file__).resolve().parent.parent.parent / "src")
|
|
if _src not in sys.path:
|
|
sys.path.insert(0, _src)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_db_repo():
|
|
repo = MagicMock()
|
|
repo.dashboard_id = 42
|
|
repo.config_id = "cfg-1"
|
|
repo.remote_url = "https://example.com/org/repo.git"
|
|
repo.local_path = "/tmp/repo"
|
|
repo.current_branch = "dev"
|
|
return repo
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_git_config():
|
|
config = MagicMock()
|
|
config.id = "cfg-1"
|
|
config.url = "https://gitea.example.com"
|
|
config.pat = "pat-123"
|
|
config.provider = "GITEA"
|
|
config.default_branch = "main"
|
|
return config
|
|
|
|
|
|
def _make_client(overrides: dict | None = None) -> TestClient:
|
|
from src.api.routes.git._router import router as parent_router
|
|
from src.core.database import get_db
|
|
from src.dependencies import get_current_user
|
|
from src.schemas.auth import User, RoleSchema
|
|
|
|
app = FastAPI()
|
|
app.include_router(parent_router)
|
|
|
|
mock_user = User(
|
|
id="admin-1", username="admin", email="admin@x.com",
|
|
auth_source="LOCAL",
|
|
created_at=__import__("datetime").datetime.now(),
|
|
roles=[RoleSchema(id="r1", name="Admin", description="", permissions=[])],
|
|
)
|
|
|
|
app.dependency_overrides[get_db] = lambda: MagicMock()
|
|
app.dependency_overrides[get_current_user] = lambda: mock_user
|
|
if overrides:
|
|
for dep, fn in overrides.items():
|
|
app.dependency_overrides[dep] = fn
|
|
return TestClient(app)
|
|
|
|
|
|
# ── sync_dashboard ──
|
|
|
|
class TestSyncDashboard:
|
|
"""POST /repositories/{dashboard_ref}/sync"""
|
|
|
|
def test_success(self):
|
|
mock_plugin = MagicMock()
|
|
mock_plugin.execute = AsyncMock(return_value={"status": "synced"})
|
|
|
|
with (
|
|
patch("src.plugins.git_plugin.GitPlugin", return_value=mock_plugin),
|
|
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
|
|
):
|
|
client = _make_client()
|
|
resp = client.post("/repositories/42/sync")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "synced"
|
|
|
|
def test_success_with_source_env(self):
|
|
mock_plugin = MagicMock()
|
|
mock_plugin.execute = AsyncMock(return_value={"status": "synced"})
|
|
|
|
with (
|
|
patch("src.plugins.git_plugin.GitPlugin", return_value=mock_plugin),
|
|
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
|
|
):
|
|
client = _make_client()
|
|
resp = client.post("/repositories/42/sync?source_env_id=staging")
|
|
assert resp.status_code == 200
|
|
|
|
def test_unexpected_error(self):
|
|
mock_plugin = MagicMock()
|
|
mock_plugin.execute = AsyncMock(side_effect=RuntimeError("boom"))
|
|
|
|
with (
|
|
patch("src.plugins.git_plugin.GitPlugin", return_value=mock_plugin),
|
|
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
|
|
):
|
|
client = _make_client()
|
|
resp = client.post("/repositories/42/sync")
|
|
assert resp.status_code == 500
|
|
|
|
def test_dashboard_not_found(self):
|
|
with (
|
|
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(side_effect=HTTPException(404))),
|
|
):
|
|
client = _make_client()
|
|
resp = client.post("/repositories/bad/42/sync")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
# ── promote_dashboard ──
|
|
|
|
class TestPromoteDashboard:
|
|
"""POST /repositories/{dashboard_ref}/promote"""
|
|
|
|
PROMOTE_PAYLOAD = {"from_branch": "dev", "to_branch": "main"}
|
|
|
|
def _setup_mock_db(self, mock_db_repo, mock_git_config):
|
|
mock_db = MagicMock()
|
|
mock_db.query.return_value.filter.return_value.first.side_effect = [mock_db_repo, mock_git_config]
|
|
return mock_db
|
|
|
|
def test_success_mr_gitea(self, mock_db_repo, mock_git_config):
|
|
mock_db = self._setup_mock_db(mock_db_repo, mock_git_config)
|
|
mock_gs = MagicMock()
|
|
mock_gs.create_gitea_pull_request = AsyncMock(return_value={
|
|
"status": "opened", "url": "https://pr.url", "id": 1,
|
|
})
|
|
|
|
from src.core.database import get_db
|
|
with (
|
|
patch("src.api.routes.git._repo_lifecycle_routes.get_git_service", return_value=mock_gs),
|
|
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
|
|
):
|
|
client = _make_client({get_db: lambda: mock_db})
|
|
resp = client.post("/repositories/42/promote", json=self.PROMOTE_PAYLOAD)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["mode"] == "mr"
|
|
assert data["status"] == "opened"
|
|
assert data["policy_violation"] is False
|
|
|
|
def test_success_mr_github(self, mock_db_repo, mock_git_config):
|
|
mock_git_config.provider = "GITHUB"
|
|
mock_db = self._setup_mock_db(mock_db_repo, mock_git_config)
|
|
mock_gs = MagicMock()
|
|
mock_gs.create_github_pull_request = AsyncMock(return_value={
|
|
"status": "opened", "url": "https://pr.url", "id": 2,
|
|
})
|
|
|
|
from src.core.database import get_db
|
|
with (
|
|
patch("src.api.routes.git._repo_lifecycle_routes.get_git_service", return_value=mock_gs),
|
|
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
|
|
):
|
|
client = _make_client({get_db: lambda: mock_db})
|
|
resp = client.post("/repositories/42/promote", json={**self.PROMOTE_PAYLOAD, "draft": True})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["mode"] == "mr"
|
|
|
|
def test_success_mr_gitlab(self, mock_db_repo, mock_git_config):
|
|
mock_git_config.provider = "GITLAB"
|
|
mock_db = self._setup_mock_db(mock_db_repo, mock_git_config)
|
|
mock_gs = MagicMock()
|
|
mock_gs.create_gitlab_merge_request = AsyncMock(return_value={
|
|
"status": "opened", "url": "https://pr.url", "id": 3,
|
|
})
|
|
|
|
from src.core.database import get_db
|
|
with (
|
|
patch("src.api.routes.git._repo_lifecycle_routes.get_git_service", return_value=mock_gs),
|
|
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
|
|
):
|
|
client = _make_client({get_db: lambda: mock_db})
|
|
resp = client.post("/repositories/42/promote", json={**self.PROMOTE_PAYLOAD, "remove_source_branch": True})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["mode"] == "mr"
|
|
|
|
def test_success_direct_mode(self, mock_db_repo, mock_git_config):
|
|
mock_db = self._setup_mock_db(mock_db_repo, mock_git_config)
|
|
mock_gs = MagicMock()
|
|
mock_gs.promote_direct_merge.return_value = {"status": "merged"}
|
|
|
|
from src.core.database import get_db
|
|
with (
|
|
patch("src.api.routes.git._repo_lifecycle_routes.get_git_service", return_value=mock_gs),
|
|
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
|
|
patch("src.api.routes.git._repo_lifecycle_routes._apply_git_identity_from_profile", AsyncMock()),
|
|
):
|
|
client = _make_client({get_db: lambda: mock_db})
|
|
resp = client.post("/repositories/42/promote", json={
|
|
"from_branch": "dev", "to_branch": "main", "mode": "direct", "reason": "Hotfix",
|
|
})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["mode"] == "direct"
|
|
assert data["policy_violation"] is True
|
|
|
|
def test_repo_not_initialized(self):
|
|
mock_db = MagicMock()
|
|
mock_db.query.return_value.filter.return_value.first.return_value = None
|
|
|
|
from src.core.database import get_db
|
|
with (
|
|
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
|
|
):
|
|
client = _make_client({get_db: lambda: mock_db})
|
|
resp = client.post("/repositories/42/promote", json=self.PROMOTE_PAYLOAD)
|
|
assert resp.status_code == 404
|
|
|
|
def test_unsupported_provider(self, mock_db_repo, mock_git_config):
|
|
mock_git_config.provider = "UNKNOWN"
|
|
mock_db = self._setup_mock_db(mock_db_repo, mock_git_config)
|
|
|
|
from src.core.database import get_db
|
|
with (
|
|
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
|
|
):
|
|
client = _make_client({get_db: lambda: mock_db})
|
|
resp = client.post("/repositories/42/promote", json=self.PROMOTE_PAYLOAD)
|
|
assert resp.status_code == 501
|
|
|
|
def test_empty_branch_after_strip_raises_400(self, mock_db_repo, mock_git_config):
|
|
"""Whitespace-only from_branch (passes Pydantic, stripped empty in handler) → 400 (line 102)."""
|
|
mock_db = self._setup_mock_db(mock_db_repo, mock_git_config)
|
|
from src.core.database import get_db
|
|
with (
|
|
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
|
|
):
|
|
client = _make_client({get_db: lambda: mock_db})
|
|
# Pydantic min_length=1 passes " ", but handler strips to ""
|
|
resp = client.post("/repositories/42/promote", json={"from_branch": " ", "to_branch": "main"})
|
|
assert resp.status_code == 400
|
|
|
|
def test_same_branch_raises_400(self, mock_db_repo, mock_git_config):
|
|
"""from_branch == to_branch → 400 (line 106)."""
|
|
mock_db = self._setup_mock_db(mock_db_repo, mock_git_config)
|
|
from src.core.database import get_db
|
|
with (
|
|
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
|
|
):
|
|
client = _make_client({get_db: lambda: mock_db})
|
|
resp = client.post("/repositories/42/promote", json={"from_branch": "dev", "to_branch": "dev"})
|
|
assert resp.status_code == 400
|
|
|
|
def test_direct_promote_no_reason_raises_400(self, mock_db_repo, mock_git_config):
|
|
"""Direct promote without reason → 400 (line 114)."""
|
|
mock_db = self._setup_mock_db(mock_db_repo, mock_git_config)
|
|
from src.core.database import get_db
|
|
with (
|
|
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
|
|
):
|
|
client = _make_client({get_db: lambda: mock_db})
|
|
resp = client.post("/repositories/42/promote", json={
|
|
"from_branch": "dev", "to_branch": "main", "mode": "direct", "reason": "",
|
|
})
|
|
assert resp.status_code == 400
|
|
|
|
|
|
# ── deploy_dashboard ──
|
|
|
|
class TestDeployDashboard:
|
|
"""POST /repositories/{dashboard_ref}/deploy"""
|
|
|
|
def test_success(self):
|
|
mock_plugin = MagicMock()
|
|
mock_plugin.execute = AsyncMock(return_value={"status": "deployed"})
|
|
|
|
with (
|
|
patch("src.plugins.git_plugin.GitPlugin", return_value=mock_plugin),
|
|
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
|
|
):
|
|
client = _make_client()
|
|
resp = client.post("/repositories/42/deploy", json={"environment_id": "prod"})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "deployed"
|
|
|
|
def test_unexpected_error(self):
|
|
mock_plugin = MagicMock()
|
|
mock_plugin.execute = AsyncMock(side_effect=RuntimeError("boom"))
|
|
|
|
with (
|
|
patch("src.plugins.git_plugin.GitPlugin", return_value=mock_plugin),
|
|
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
|
|
):
|
|
client = _make_client()
|
|
resp = client.post("/repositories/42/deploy", json={"environment_id": "prod"})
|
|
assert resp.status_code == 500
|
|
|
|
def test_dashboard_not_found(self):
|
|
"""HTTPException from _resolve_dashboard_id_from_ref propagates (line 220)."""
|
|
with (
|
|
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(side_effect=HTTPException(status_code=404, detail="Not found"))),
|
|
):
|
|
client = _make_client()
|
|
resp = client.post("/repositories/bad-ref/deploy", json={"environment_id": "prod"})
|
|
assert resp.status_code == 404
|
|
# #endregion Test.Api.GitRepoLifecycleRoutes
|