Files
ss-tools/backend/tests/api/test_git_release_routes.py
busya 81de959ef7 feat(037): verification pipeline automation + GET read-API (T080-T081)
Close the 037 pipeline-automation and read-API gaps found in the audit:
deploy/release hooks did not create VerificationRun, and GET endpoints for
history/detail were absent even though 039 UI and client call them.

T080 - _release_routes.py: create_release now fires best-effort
  _trigger_release_verification -> VerificationRun with trigger=release_create
  (metric+structure); verification scheduling failures never roll back the
  release transaction.
T081 - verification.py: add GET /verification/history (dashboard_id +
  environment_id filters, newest-first) and GET /verification/{run_id}
  (404 RUN_NOT_FOUND); reuse _record_to_response.
  - verification_run.py + alembic migration p2q3r4s5t6u7: nullable indexed
    dashboard_id populated from structure/visual/metric category_params.
  - verification_service.py: _derive_dashboard_id helper.

Verification: release routes (32) + verification API (8) + persistence (21)
= 53 passed; ruff clean for changed code (pre-existing RUF012/UP017 on old
lines left untouched).
2026-08-07 14:21:25 +07:00

772 lines
32 KiB
Python

# #region Test.Api.GitReleaseRoutes [C:3] [TYPE Module] [SEMANTICS test,git,release,approval,publication]
# @BRIEF Unit tests for Git release API routes — policy, create, approve, publish.
# @RELATION BINDS_TO -> [Api.ReleaseRoutes.GitReleaseRoutes]
# @TEST_EDGE: release_not_found -> 404
# @TEST_EDGE: approval_gate_error -> 409
# @TEST_EDGE: publish_gate_error -> 409
# @TEST_EDGE: non_admin_cannot_update_policy -> 403
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")
from datetime import datetime, timezone
from pathlib import Path
import sys
from unittest.mock import AsyncMock, MagicMock, patch
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)
# ── Shared test fixtures ──
def _make_mock_repository(**overrides) -> MagicMock:
"""Build a GitRepository mock with sensible defaults."""
repo = MagicMock()
repo.id = "repo-1"
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"
repo.release_policy = None
for k, v in overrides.items():
setattr(repo, k, v)
return repo
def _make_mock_release(**overrides) -> MagicMock:
"""Build a DashboardRelease mock with sensible defaults."""
release = MagicMock()
release.id = "release-1"
release.repository_id = "repo-1"
release.deployment_id = "deploy-42"
release.name = "v2.0"
release.version = "2.0.0"
release.notes = "Major release"
release.commit_hash = "abc123def456"
release.content_hash = "c0ffee42"
release.status = "awaiting_approval"
release.created_by = "admin"
release.created_at = datetime.now(timezone.utc)
release.approved_at = None
release.approved_by = None
release.approval_comment = None
release.published_at = None
release.published_by = None
for k, v in overrides.items():
setattr(release, k, v)
return release
def _make_mock_deployment(**overrides) -> MagicMock:
"""Build a DeploymentRecord mock."""
dep = MagicMock()
dep.id = "deploy-42"
dep.repository_id = "repo-1"
dep.environment_id = "preprod-1"
dep.status = "success"
dep.commit_hash = "abc123def456"
dep.content_hash = "c0ffee42"
dep.validation_status = "validated"
dep.deployed_at = datetime.now(timezone.utc)
dep.validated_at = datetime.now(timezone.utc)
dep.validated_by = "admin"
dep.resources_changed = {"source_branch": "dev"}
for k, v in overrides.items():
setattr(dep, k, v)
return dep
def _make_config_manager(overrides: dict | None = None) -> MagicMock:
"""Build a config_manager mock with release policy defaults."""
mgr = MagicMock()
cfg = MagicMock()
class _FakeReleaseSettings:
"""Duck-typed settings object with model_dump for _resolve_policy."""
require_prod_approval = True
approval_roles = ["Admin"]
require_approval_comment = False
approval_expires_hours = 0
block_publish_on_drift = True
def model_dump(self):
return {
"require_prod_approval": self.require_prod_approval,
"approval_roles": self.approval_roles,
"require_approval_comment": self.require_approval_comment,
"approval_expires_hours": self.approval_expires_hours,
"block_publish_on_drift": self.block_publish_on_drift,
}
release_settings = _FakeReleaseSettings()
cfg.settings.git_release = release_settings
mgr.get_config.return_value = cfg
if overrides:
for k, v in overrides.items():
setattr(mgr, k, v)
return mgr
def _make_mock_user(is_admin: bool = True) -> MagicMock:
"""Build a user mock with an admin or regular role."""
from datetime import datetime
from src.schemas.auth import RoleSchema, User as UserSchema
admin_role = RoleSchema(
id="r1",
name="Admin",
description="",
is_admin=is_admin,
permissions=[],
)
user = UserSchema(
id="admin-1",
username="admin" if is_admin else "user",
email="admin@x.com" if is_admin else "user@x.com",
auth_source="LOCAL",
created_at=datetime.now(),
roles=[admin_role] if is_admin else [],
)
return user
def _make_client(
db_mock: MagicMock | None = None,
config_manager_mock: MagicMock | None = None,
user_mock: MagicMock | None = None,
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_config_manager, get_current_user
app = FastAPI()
app.include_router(parent_router)
if user_mock is None:
user_mock = _make_mock_user(is_admin=True)
if config_manager_mock is None:
config_manager_mock = _make_config_manager()
if db_mock is None:
db_mock = MagicMock()
app.dependency_overrides[get_current_user] = lambda: user_mock
app.dependency_overrides[get_config_manager] = lambda: config_manager_mock
app.dependency_overrides[get_db] = lambda: db_mock
if overrides:
for dep, fn in overrides.items():
app.dependency_overrides[dep] = fn
return TestClient(app)
# ── get_release_policy ──
class TestGetReleasePolicy:
"""GET /repositories/{dashboard_ref}/release-policy"""
def test_success(self):
"""Returns the effective release policy for a repository."""
repo = _make_mock_repository()
db_mock = MagicMock()
db_mock.query.return_value.filter.return_value.first.return_value = repo
cm = _make_config_manager()
with patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)):
client = _make_client(db_mock=db_mock, config_manager_mock=cm)
resp = client.get("/repositories/test-dash/release-policy")
assert resp.status_code == 200
data = resp.json()
assert data["require_prod_approval"] is True
assert data["approval_roles"] == ["Admin"]
assert data["is_override"] is False
def test_with_repo_override(self):
"""Repository-level policy overrides the installation default."""
repo = _make_mock_repository(release_policy={"require_prod_approval": False, "block_publish_on_drift": False})
db_mock = MagicMock()
db_mock.query.return_value.filter.return_value.first.return_value = repo
cm = _make_config_manager()
with patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)):
client = _make_client(db_mock=db_mock, config_manager_mock=cm)
resp = client.get("/repositories/test-dash/release-policy")
assert resp.status_code == 200
data = resp.json()
assert data["require_prod_approval"] is False
assert data["block_publish_on_drift"] is False
assert data["is_override"] is True
def test_repo_not_found(self):
"""Returns 404 when repository is not initialized."""
db_mock = MagicMock()
db_mock.query.return_value.filter.return_value.first.return_value = None
with patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)):
client = _make_client(db_mock=db_mock)
resp = client.get("/repositories/test-dash/release-policy")
assert resp.status_code == 404
assert "not initialized" in resp.json()["detail"]
# Note: no test_unexpected_error here because get_release_policy
# lacks a try/except block — RuntimeErrors propagate uncaught.
# ── update_release_policy ──
class TestUpdateReleasePolicy:
"""PUT /repositories/{dashboard_ref}/release-policy"""
POLICY_PAYLOAD = {"require_prod_approval": False, "block_publish_on_drift": False}
def test_success(self):
"""Admin can update the release policy override."""
repo = _make_mock_repository()
db_mock = MagicMock()
db_mock.query.return_value.filter.return_value.first.return_value = repo
cm = _make_config_manager()
with patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)):
client = _make_client(db_mock=db_mock, config_manager_mock=cm)
resp = client.put("/repositories/test-dash/release-policy", json=self.POLICY_PAYLOAD)
assert resp.status_code == 200
data = resp.json()
assert data["require_prod_approval"] is False
db_mock.commit.assert_called_once()
def test_non_admin_forbidden(self):
"""Non-admin users receive 403."""
repo = _make_mock_repository()
db_mock = MagicMock()
db_mock.query.return_value.filter.return_value.first.return_value = repo
user_mock = _make_mock_user(is_admin=False)
with patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)):
client = _make_client(db_mock=db_mock, user_mock=user_mock)
resp = client.put("/repositories/test-dash/release-policy", json=self.POLICY_PAYLOAD)
assert resp.status_code == 403
def test_repo_not_found(self):
"""Returns 404 when repository is not initialized."""
db_mock = MagicMock()
db_mock.query.return_value.filter.return_value.first.return_value = None
with patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)):
client = _make_client(db_mock=db_mock)
resp = client.put("/repositories/test-dash/release-policy", json=self.POLICY_PAYLOAD)
assert resp.status_code == 404
# ── list_releases ──
class TestListReleases:
"""GET /repositories/{dashboard_ref}/releases"""
def test_success_empty(self):
"""Returns empty list when no releases exist."""
repo = _make_mock_repository()
db_mock = MagicMock()
db_mock.query.return_value.filter.return_value.first.return_value = repo
db_mock.query.return_value.filter.return_value.order_by.return_value.all.return_value = []
with patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)):
client = _make_client(db_mock=db_mock)
resp = client.get("/repositories/test-dash/releases")
assert resp.status_code == 200
assert resp.json() == []
def test_success_with_releases(self):
"""Returns releases ordered by created_at desc."""
repo = _make_mock_repository()
release = _make_mock_release()
releases = [release]
db_mock = MagicMock()
db_mock.query.return_value.filter.return_value.first.return_value = repo
db_mock.query.return_value.filter.return_value.order_by.return_value.all.return_value = releases
with patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)):
client = _make_client(db_mock=db_mock)
resp = client.get("/repositories/test-dash/releases")
assert resp.status_code == 200
data = resp.json()
assert len(data) == 1
assert data[0]["name"] == "v2.0"
assert data[0]["version"] == "2.0.0"
def test_repo_not_found(self):
"""Returns 404 when repository is not initialized."""
db_mock = MagicMock()
db_mock.query.return_value.filter.return_value.first.return_value = None
with patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)):
client = _make_client(db_mock=db_mock)
resp = client.get("/repositories/test-dash/releases")
assert resp.status_code == 404
# ── create_release ──
class TestCreateRelease:
"""POST /repositories/{dashboard_ref}/releases"""
CREATE_PAYLOAD = {"name": "v2.0", "version": "2.0.0", "notes": "Major release"}
def _make_query_chain(self, db_mock, first_result=None, candidate_result=None):
"""Configure db_mock.query chains for consistent create_release mocking.
create_release uses db.query() in three patterns:
1. repo: db.query(GitRepository).filter(...).first()
2. candidate: db.query(DeploymentRecord).filter(cond1,cond2,cond3).order_by(...).first()
3. dup: db.query(DashboardRelease).filter(...).first()
Since db.query() returns the same mock each time, we set up
the filter chain with side_effect for .first() on the base
filter, and a separate chain for the order_by().first().
"""
from datetime import datetime
q = MagicMock(name="query")
db_mock.query.return_value = q
# Pattern 1 & 3: q.filter().first()
# _get_repository calls this first (needs first_result = repo)
# duplicate check calls this third (needs None)
if first_result is not None:
q.filter.return_value.first.side_effect = [first_result, None]
# Pattern 2: q.filter(cond1,cond2,cond3).order_by(...).first()
# Note: filter() is called ONCE with 3 args, not 3 chained filter() calls
f1 = q.filter.return_value # q.filter(cond1, cond2, cond3)
f2 = f1.order_by.return_value # .order_by(desc, desc)
if candidate_result is not None:
f2.first.return_value = candidate_result
# Make db.refresh populate id and created_at on SQLAlchemy model instances
def _refresh_model(instance):
if not hasattr(instance, '_sa_instance_state'):
return
if not getattr(instance, 'id', None):
instance.id = "auto-id-42"
if not getattr(instance, 'created_at', None):
instance.created_at = datetime.now()
db_mock.refresh.side_effect = _refresh_model
return q
def test_success_without_approval(self):
"""Creates a release with ready_to_publish when approval is not required."""
repo = _make_mock_repository()
deployment = _make_mock_deployment()
cm = _make_config_manager()
cm.get_config.return_value.settings.git_release.require_prod_approval = False
db_mock = MagicMock()
self._make_query_chain(db_mock, first_result=repo, candidate_result=deployment)
with (
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
patch("src.api.routes.git._release_routes._resolve_stage_environment") as mock_resolve_env,
patch("src.api.routes.git._release_routes._probe_drift", AsyncMock(return_value=("in_sync", "c0ffee42"))),
):
mock_resolve_env.return_value = MagicMock(id="preprod-1")
client = _make_client(db_mock=db_mock, config_manager_mock=cm)
resp = client.post("/repositories/test-dash/releases", json=self.CREATE_PAYLOAD)
assert resp.status_code == 201
data = resp.json()
assert data["name"] == "v2.0"
assert data["status"] == "ready_to_publish"
assert data["created_by"] == "admin"
def test_create_release_triggers_verification_run(self):
"""037 T080: creating a release must spawn a VerificationRun with trigger=release_create."""
repo_id = "11111111-2222-3333-4444-555555555555"
repo = _make_mock_repository(id=repo_id)
deployment = _make_mock_deployment()
cm = _make_config_manager()
cm.get_config.return_value.settings.git_release.require_prod_approval = False
db_mock = MagicMock()
self._make_query_chain(db_mock, first_result=repo, candidate_result=deployment)
release_id = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
# Override the chain's refresh side-effect so the created release gets a valid UUID id
# (real DashboardRelease.id is a String(36) UUID; the default "auto-id-42" is not).
def _refresh_uuid(instance):
if hasattr(instance, "_sa_instance_state"):
if not getattr(instance, "id", None):
instance.id = release_id
if not getattr(instance, "created_at", None):
instance.created_at = datetime.now()
db_mock.refresh.side_effect = _refresh_uuid
create_run = AsyncMock(return_value=MagicMock())
with (
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
patch("src.api.routes.git._release_routes._resolve_stage_environment") as mock_resolve_env,
patch("src.api.routes.git._release_routes._probe_drift", AsyncMock(return_value=("in_sync", "c0ffee42"))),
patch("src.services.dashboard_testing.verification_service.create_verification_run_async", create_run),
):
mock_resolve_env.return_value = MagicMock(id="preprod-1")
client = _make_client(db_mock=db_mock, config_manager_mock=cm)
resp = client.post("/repositories/test-dash/releases", json=self.CREATE_PAYLOAD)
assert resp.status_code == 201
# Verification hook fired with the release_create trigger
create_run.assert_awaited_once()
request = create_run.await_args.args[1] # db, request, created_by
assert request.trigger == "release_create"
assert str(request.repository_id) == repo_id
def test_success_with_approval_required(self):
"""Creates a release with awaiting_approval when policy requires prod approval."""
repo = _make_mock_repository()
deployment = _make_mock_deployment()
cm = _make_config_manager()
cm.get_config.return_value.settings.git_release.require_prod_approval = True
db_mock = MagicMock()
self._make_query_chain(db_mock, first_result=repo, candidate_result=deployment)
with (
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
patch("src.api.routes.git._release_routes._resolve_stage_environment") as mock_resolve_env,
patch("src.api.routes.git._release_routes._probe_drift", AsyncMock(return_value=("in_sync", "c0ffee42"))),
):
mock_resolve_env.return_value = MagicMock(id="preprod-1")
client = _make_client(db_mock=db_mock, config_manager_mock=cm)
resp = client.post("/repositories/test-dash/releases", json=self.CREATE_PAYLOAD)
assert resp.status_code == 201
assert resp.json()["status"] == "awaiting_approval"
def test_no_validated_preprod(self):
"""Returns 409 when PREPROD deployment is not validated."""
repo = _make_mock_repository()
deployment = _make_mock_deployment(validation_status="pending")
cm = _make_config_manager()
db_mock = MagicMock()
self._make_query_chain(db_mock, first_result=repo, candidate_result=deployment)
with (
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
patch("src.api.routes.git._release_routes._resolve_stage_environment") as mock_resolve_env,
):
mock_resolve_env.return_value = MagicMock(id="preprod-1")
client = _make_client(db_mock=db_mock, config_manager_mock=cm)
resp = client.post("/repositories/test-dash/releases", json=self.CREATE_PAYLOAD)
assert resp.status_code == 409
assert "Validate the current PREPROD deployment" in resp.json()["detail"]
def test_no_preprod_deployment(self):
"""Returns 409 when no successful PREPROD deployment exists."""
repo = _make_mock_repository()
cm = _make_config_manager()
db_mock = MagicMock()
self._make_query_chain(db_mock, first_result=repo, candidate_result=None)
with (
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
patch("src.api.routes.git._release_routes._resolve_stage_environment") as mock_resolve_env,
):
mock_resolve_env.return_value = MagicMock(id="preprod-1")
client = _make_client(db_mock=db_mock, config_manager_mock=cm)
resp = client.post("/repositories/test-dash/releases", json=self.CREATE_PAYLOAD)
assert resp.status_code == 409
assert "Validate the current PREPROD deployment" in resp.json()["detail"]
def test_drift_blocked(self):
"""Returns 409 when PREPROD has drifted from the candidate."""
repo = _make_mock_repository()
deployment = _make_mock_deployment()
cm = _make_config_manager()
cm.get_config.return_value.settings.git_release.block_publish_on_drift = True
db_mock = MagicMock()
self._make_query_chain(db_mock, first_result=repo, candidate_result=deployment)
with (
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
patch("src.api.routes.git._release_routes._resolve_stage_environment") as mock_resolve_env,
patch("src.api.routes.git._release_routes._probe_drift", AsyncMock(return_value=("drifted", "different"))),
):
mock_resolve_env.return_value = MagicMock(id="preprod-1")
client = _make_client(db_mock=db_mock, config_manager_mock=cm)
resp = client.post("/repositories/test-dash/releases", json=self.CREATE_PAYLOAD)
assert resp.status_code == 409
assert "PREPROD differs" in resp.json()["detail"]
def test_duplicate_deployment(self):
"""Returns 409 when this deployment already has a named release."""
repo = _make_mock_repository()
deployment = _make_mock_deployment()
existing = _make_mock_release()
cm = _make_config_manager()
db_mock = MagicMock()
self._make_query_chain(db_mock, first_result=repo, candidate_result=deployment)
# Override side_effect: make duplicate check return existing
# _get_repository uses side_effect[0] (repo), duplicate check uses side_effect[1]
q = db_mock.query.return_value
q.filter.return_value.first.side_effect = [repo, existing]
with (
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
patch("src.api.routes.git._release_routes._resolve_stage_environment") as mock_resolve_env,
patch("src.api.routes.git._release_routes._probe_drift", AsyncMock(return_value=("in_sync", "c0ffee42"))),
):
mock_resolve_env.return_value = MagicMock(id="preprod-1")
client = _make_client(db_mock=db_mock, config_manager_mock=cm)
resp = client.post("/repositories/test-dash/releases", json=self.CREATE_PAYLOAD)
assert resp.status_code == 409
def test_duplicate_version(self):
"""Returns 409 when release version already exists (DB constraint)."""
repo = _make_mock_repository()
deployment = _make_mock_deployment()
cm = _make_config_manager()
db_mock = MagicMock()
self._make_query_chain(db_mock, first_result=repo, candidate_result=deployment)
db_mock.commit.side_effect = Exception("duplicate key")
with (
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
patch("src.api.routes.git._release_routes._resolve_stage_environment") as mock_resolve_env,
patch("src.api.routes.git._release_routes._probe_drift", AsyncMock(return_value=("in_sync", "c0ffee42"))),
):
mock_resolve_env.return_value = MagicMock(id="preprod-1")
client = _make_client(db_mock=db_mock, config_manager_mock=cm)
resp = client.post("/repositories/test-dash/releases", json=self.CREATE_PAYLOAD)
assert resp.status_code == 409
assert "Release version already exists" in resp.json()["detail"]
def test_repo_not_found(self):
"""Returns 404 when repository is not initialized."""
db_mock = MagicMock()
db_mock.query.return_value.filter.return_value.first.return_value = None
with patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)):
client = _make_client(db_mock=db_mock)
resp = client.post("/repositories/test-dash/releases", json=self.CREATE_PAYLOAD)
assert resp.status_code == 404
# ── approve_release ──
class TestApproveRelease:
"""POST /repositories/{dashboard_ref}/releases/{release_id}/approve"""
APPROVE_PAYLOAD = {"comment": "Looks good"}
def test_success(self):
"""Approves a release that is awaiting approval."""
repo = _make_mock_repository()
release = _make_mock_release(status="awaiting_approval")
cm = _make_config_manager()
db_mock = MagicMock()
db_mock.query.return_value.filter.return_value.first.side_effect = [repo, release]
with (
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
patch("src.api.routes.git._release_routes._enforce_approval_policy"),
):
client = _make_client(db_mock=db_mock, config_manager_mock=cm)
resp = client.post("/repositories/test-dash/releases/release-1/approve", json=self.APPROVE_PAYLOAD)
assert resp.status_code == 200
data = resp.json()
assert data["status"] == "ready_to_publish"
assert data["approved_by"] == "admin"
def test_release_not_found(self):
"""Returns 404 when release does not exist."""
repo = _make_mock_repository()
db_mock = MagicMock()
db_mock.query.return_value.filter.return_value.first.side_effect = [repo, None]
with patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)):
client = _make_client(db_mock=db_mock)
resp = client.post("/repositories/test-dash/releases/release-999/approve", json=self.APPROVE_PAYLOAD)
assert resp.status_code == 404
def test_not_awaiting_approval(self):
"""Returns 409 when release is not in awaiting_approval status."""
repo = _make_mock_repository()
release = _make_mock_release(status="published")
db_mock = MagicMock()
db_mock.query.return_value.filter.return_value.first.side_effect = [repo, release]
with patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)):
client = _make_client(db_mock=db_mock)
resp = client.post("/repositories/test-dash/releases/release-1/approve", json=self.APPROVE_PAYLOAD)
assert resp.status_code == 409
assert "not awaiting approval" in resp.json()["detail"]
def test_approval_policy_rejected(self):
"""Returns 403 when approval policy rejects the user."""
repo = _make_mock_repository()
release = _make_mock_release(status="awaiting_approval")
cm = _make_config_manager()
db_mock = MagicMock()
db_mock.query.return_value.filter.return_value.first.side_effect = [repo, release]
with (
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
patch("src.api.routes.git._release_routes._enforce_approval_policy", side_effect=HTTPException(status_code=403, detail="Your role cannot approve")),
):
client = _make_client(db_mock=db_mock, config_manager_mock=cm)
resp = client.post("/repositories/test-dash/releases/release-1/approve", json=self.APPROVE_PAYLOAD)
assert resp.status_code == 403
def test_repo_not_found(self):
"""Returns 404 when repository is not initialized."""
db_mock = MagicMock()
db_mock.query.return_value.filter.return_value.first.return_value = None
with patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)):
client = _make_client(db_mock=db_mock)
resp = client.post("/repositories/test-dash/releases/release-1/approve", json=self.APPROVE_PAYLOAD)
assert resp.status_code == 404
# ── publish_release ──
class TestPublishRelease:
"""POST /repositories/{dashboard_ref}/releases/{release_id}/publish"""
def test_success(self):
"""Publishes a release by delegating to deploy_dashboard."""
repo = _make_mock_repository()
release = _make_mock_release(status="ready_to_publish")
deploy_result = {"status": "deployed", "target": "prod"}
db_mock = MagicMock()
db_mock.query.return_value.filter.return_value.first.side_effect = [repo, release]
cm = _make_config_manager()
with (
patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)),
patch("src.api.routes.git._repo_lifecycle_routes.deploy_dashboard", AsyncMock(return_value=deploy_result)),
):
client = _make_client(db_mock=db_mock, config_manager_mock=cm)
resp = client.post("/repositories/test-dash/releases/release-1/publish")
assert resp.status_code == 200
assert resp.json() == deploy_result
def test_release_not_found(self):
"""Returns 404 when release does not exist."""
repo = _make_mock_repository()
db_mock = MagicMock()
db_mock.query.return_value.filter.return_value.first.side_effect = [repo, None]
with patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)):
client = _make_client(db_mock=db_mock)
resp = client.post("/repositories/test-dash/releases/release-999/publish")
assert resp.status_code == 404
def test_repo_not_found(self):
"""Returns 404 when repository is not initialized."""
db_mock = MagicMock()
db_mock.query.return_value.filter.return_value.first.return_value = None
with patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(return_value=42)):
client = _make_client(db_mock=db_mock)
resp = client.post("/repositories/test-dash/releases/release-1/publish")
assert resp.status_code == 404
# Note: no test_unexpected_error here because publish_release
# lacks a try/except block — RuntimeErrors propagate uncaught.
# ── HTTPException propagation ──
class TestHttpExceptionPropagation:
"""HTTPException from _resolve_dashboard_id_from_ref must propagate unchanged."""
HTTPERR = HTTPException(status_code=404, detail="Not found")
def test_get_release_policy_http_error(self):
with patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(side_effect=self.HTTPERR)):
client = _make_client()
resp = client.get("/repositories/42/release-policy")
assert resp.status_code == 404
def test_update_release_policy_http_error(self):
with patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(side_effect=self.HTTPERR)):
client = _make_client()
resp = client.put("/repositories/42/release-policy", json={"require_prod_approval": False})
assert resp.status_code == 404
def test_list_releases_http_error(self):
with patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(side_effect=self.HTTPERR)):
client = _make_client()
resp = client.get("/repositories/42/releases")
assert resp.status_code == 404
def test_create_release_http_error(self):
with patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(side_effect=self.HTTPERR)):
client = _make_client()
resp = client.post("/repositories/42/releases", json={"name": "v1", "version": "1.0", "notes": "x"})
assert resp.status_code == 404
def test_approve_release_http_error(self):
with patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(side_effect=self.HTTPERR)):
client = _make_client()
resp = client.post("/repositories/42/releases/r-1/approve", json={"comment": "ok"})
assert resp.status_code == 404
def test_publish_release_http_error(self):
with patch("src.api.routes.git._resolve_dashboard_id_from_ref", AsyncMock(side_effect=self.HTTPERR)):
client = _make_client()
resp = client.post("/repositories/42/releases/r-1/publish")
assert resp.status_code == 404
# #endregion Test.Api.GitReleaseRoutes