busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

test(coverage): add 200+ tests to push frontend + backend coverage above thresholds

Browse Source 
Backend (4 files, 73 tests):
- test_agent_superset_routes.py (27 tests, 35% -> 92%)
- test_agent_lifecycle_routes.py (11 tests, 50% -> 100%)
- test_agent_status_routes.py (6 tests, 57% -> 100%)
- test_git_release_routes.py (31 tests, 30% -> 99%)

Frontend (~15 files, ~120 tests):
- cron.ts: 0% -> 100%
- ReportsLogModel: 0% -> 99%
- parseCot.ts: 10% -> 100%
- sessionTimeout.ts: 64% -> 93%
- MappingsModel: 65% -> 100%
- TranslateHistoryModel: 65% -> 93%
- Migration.ExecutorModel: 70% -> 100%
- GitManagerModel: 78% -> 90%
- TranslationJobModel: 77% -> 80%
- ConfirmDialog: 59% -> 80%
- api.ts: 78% -> 80%

Coverage: frontend 0 violations, backend 7518 passed.
This commit is contained in:
busya
2026-07-23 15:49:45 +03:00
parent fb6327e92b
commit 63d82df53b
19 changed files with 5110 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

280
backend/tests/api/test_agent_lifecycle_routes.py Normal file
View File

@@ -0,0 +1,280 @@
# #region Test.Api.AgentLifecycleRoutes [C:3] [TYPE Module] [SEMANTICS test,agent,lifecycle,events,api]
# @BRIEF Unit tests for Agent Lifecycle event API — write and list endpoints.
# @RELATION BINDS_TO -> [Api.AgentLifecycle]
# @TEST_EDGE: non_admin_cannot_filter_by_user -> 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")
import sys
from datetime import datetime
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from fastapi import FastAPI
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)
def _make_mock_user(is_admin: bool = True, user_id: str = "user-1") -> MagicMock:
"""Build a mock user for auth bypass."""
from src.schemas.auth import RoleSchema, User as UserSchema
roles = []
if is_admin:
roles.append(
RoleSchema(id="r1", name="Admin", description="", is_admin=True, permissions=[])
)
return UserSchema(
id=user_id,
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=roles,
)
def _make_client(user_mock=None, db_mock=None, overrides=None) -> TestClient:
"""Build a TestClient with the agent lifecycle router."""
from src.api.routes.agent_lifecycle import router
from src.core.database import get_db
from src.dependencies import get_current_user
app = FastAPI()
app.include_router(router)
if user_mock is None:
user_mock = _make_mock_user(is_admin=True)
if db_mock is None:
db_mock = MagicMock()
app.dependency_overrides[get_current_user] = lambda: user_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, raise_server_exceptions=False)
# ── create_event (POST) ──
class TestCreateEvent:
"""POST /api/agent/events"""
EVENT_PAYLOAD = {
"trace_id": "trace-1",
"conversation_id": "conv-1",
"event_type": "tool_call",
"tool_name": "superset_query",
"status": "success",
"elapsed_ms": 1500,
"payload": {"action": "query", "attempt": 1},
}
def test_success(self):
"""Creates a lifecycle event successfully (admin user)."""
from src.schemas.agent_lifecycle import EventWriteResponse
mock_write = MagicMock(return_value=EventWriteResponse(id="evt-1"))
with patch("src.api.routes.agent_lifecycle.write_event", mock_write):
client = _make_client()
resp = client.post("/api/agent/events", json=self.EVENT_PAYLOAD)
assert resp.status_code == 201
data = resp.json()
assert data["id"] == "evt-1"
assert data["written"] is True
mock_write.assert_called_once()
def test_success_regular_user(self):
"""Non-admin user can also create events."""
from src.schemas.agent_lifecycle import EventWriteResponse
mock_write = MagicMock(return_value=EventWriteResponse(id="evt-2"))
user_mock = _make_mock_user(is_admin=False)
with patch("src.api.routes.agent_lifecycle.write_event", mock_write):
client = _make_client(user_mock=user_mock)
resp = client.post("/api/agent/events", json=self.EVENT_PAYLOAD)
assert resp.status_code == 201
def test_payload_reduced(self):
"""Payload is reduced to safe keys via schema."""
from src.schemas.agent_lifecycle import EventWriteResponse
mock_write = MagicMock(return_value=EventWriteResponse(id="evt-3"))
payload_with_sensitive = {
**self.EVENT_PAYLOAD,
"payload": {"action": "query", "password": "secret", "token": "abc"},
}
with patch("src.api.routes.agent_lifecycle.write_event", mock_write):
client = _make_client()
resp = client.post("/api/agent/events", json=payload_with_sensitive)
assert resp.status_code == 201
# The write_event should receive payload with sensitive keys reduced
call_kwargs = mock_write.call_args
if call_kwargs:
written_body = call_kwargs[0][1] # body arg
if written_body.payload:
assert "password" not in written_body.payload
assert "action" in written_body.payload
def test_db_rollback_on_error(self):
"""Rolls back DB transaction when write_event raises."""
mock_write = MagicMock(side_effect=ValueError("DB error"))
db_mock = MagicMock()
with patch("src.api.routes.agent_lifecycle.write_event", mock_write):
client = _make_client(db_mock=db_mock)
resp = client.post("/api/agent/events", json=self.EVENT_PAYLOAD)
assert resp.status_code == 500
db_mock.rollback.assert_called_once()
def test_missing_required_fields(self):
"""Returns 422 when required fields are missing."""
client = _make_client()
resp = client.post("/api/agent/events", json={"event_type": "test"})
assert resp.status_code == 422
# ── read_events (GET) ──
class TestReadEvents:
"""GET /api/agent/events"""
def _make_event_item(self, **overrides):
from src.schemas.agent_lifecycle import EventItem
return EventItem(
id=overrides.get("id", "evt-1"),
trace_id=overrides.get("trace_id", "trace-1"),
conversation_id=overrides.get("conversation_id", "conv-1"),
user_id=overrides.get("user_id", "user-1"),
event_type=overrides.get("event_type", "tool_call"),
tool_name=overrides.get("tool_name", "superset_query"),
status=overrides.get("status", "success"),
created_at=datetime.now(),
)
def test_success_admin(self):
"""Admin can list all events."""
from src.schemas.agent_lifecycle import EventListResponse
items = [self._make_event_item()]
mock_list = MagicMock(
return_value=EventListResponse(items=items, total=1, page=1, page_size=50, has_next=False)
)
with patch("src.api.routes.agent_lifecycle.list_events", mock_list):
client = _make_client()
resp = client.get("/api/agent/events")
assert resp.status_code == 200
data = resp.json()
assert data["total"] == 1
assert len(data["items"]) == 1
assert data["items"][0]["id"] == "evt-1"
def test_success_regular_user(self):
"""Non-admin sees only own events (no user_id filter)."""
from src.schemas.agent_lifecycle import EventListResponse
items = [self._make_event_item(id="evt-2")]
mock_list = MagicMock(
return_value=EventListResponse(items=items, total=1, page=1, page_size=50, has_next=False)
)
user_mock = _make_mock_user(is_admin=False)
with patch("src.api.routes.agent_lifecycle.list_events", mock_list):
client = _make_client(user_mock=user_mock)
resp = client.get("/api/agent/events")
assert resp.status_code == 200
assert resp.json()["total"] == 1
def test_non_admin_cannot_filter_by_user(self):
"""Non-admin gets 403 when trying to filter by user_id."""
user_mock = _make_mock_user(is_admin=False)
client = _make_client(user_mock=user_mock)
resp = client.get("/api/agent/events?user_id=other-user")
assert resp.status_code == 403
assert "Only admin users can query events by user_id" in resp.json()["detail"]
def test_admin_can_filter_by_user(self):
"""Admin can filter events by user_id."""
from src.schemas.agent_lifecycle import EventListResponse
items = [self._make_event_item(user_id="other-user")]
mock_list = MagicMock(
return_value=EventListResponse(items=items, total=1, page=1, page_size=50, has_next=False)
)
with patch("src.api.routes.agent_lifecycle.list_events", mock_list):
client = _make_client()
resp = client.get("/api/agent/events?user_id=other-user")
assert resp.status_code == 200
assert resp.json()["total"] == 1
# Verify list_events received the correct args
call_kwargs = mock_list.call_args[1]
assert call_kwargs["user_id"] == "other-user"
def test_with_filters(self):
"""Filters are passed through to list_events."""
from src.schemas.agent_lifecycle import EventListResponse
mock_list = MagicMock(
return_value=EventListResponse(items=[], total=0, page=1, page_size=50, has_next=False)
)
with patch("src.api.routes.agent_lifecycle.list_events", mock_list):
client = _make_client()
resp = client.get(
"/api/agent/events?event_type=tool_call&conversation_id=conv-1&status=success&tool_name=superset_query"
)
assert resp.status_code == 200
call_kwargs = mock_list.call_args[1]
assert call_kwargs["event_type"] == "tool_call"
assert call_kwargs["conversation_id"] == "conv-1"
def test_pagination(self):
"""Page and page_size are passed through."""
from src.schemas.agent_lifecycle import EventListResponse
mock_list = MagicMock(
return_value=EventListResponse(items=[], total=0, page=2, page_size=25, has_next=False)
)
with patch("src.api.routes.agent_lifecycle.list_events", mock_list):
client = _make_client()
resp = client.get("/api/agent/events?page=2&page_size=25")
assert resp.status_code == 200
assert resp.json()["page"] == 2
assert resp.json()["page_size"] == 25
def test_invalid_page_size(self):
"""Page_size outside 1-200 returns 422."""
client = _make_client()
resp = client.get("/api/agent/events?page_size=500")
assert resp.status_code == 422
# #endregion Test.Api.AgentLifecycleRoutes

110
backend/tests/api/test_agent_status_routes.py Normal file
View File

@@ -0,0 +1,110 @@
# #region Test.Api.AgentStatusRoutes [C:2] [TYPE Module] [SEMANTICS test,agent,llm,status,health]
# @BRIEF Unit tests for Agent LLM provider health status endpoint.
# @RELATION BINDS_TO -> [Api.Agent.Status]
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
from fastapi.testclient import TestClient
# Add shared/src to path for ss_tools.shared._llm_health
_src = str(Path(__file__).resolve().parent.parent.parent / "src")
if _src not in sys.path:
sys.path.insert(0, _src)
_shared_src = str(Path(__file__).resolve().parent.parent.parent.parent / "shared" / "src")
if _shared_src not in sys.path:
sys.path.insert(0, _shared_src)
def _make_client() -> TestClient:
"""Build a TestClient with the agent status router."""
from src.api.routes.agent_status import router
app = FastAPI()
app.include_router(router)
return TestClient(app)
class TestGetLlmStatus:
"""GET /api/agent/llm-status"""
@patch("ss_tools.shared._llm_health._check_llm_provider_health", AsyncMock(return_value="ok"))
@patch("ss_tools.shared._llm_health._llm_status", {"status": "ok", "last_error": ""})
def test_status_ok(self):
"""Returns ok when LLM provider is healthy."""
client = _make_client()
resp = client.get("/api/agent/llm-status")
assert resp.status_code == 200
data = resp.json()
assert data["status"] == "ok"
assert data["last_error"] == ""
assert data["retry_after_s"] == 0
@patch("ss_tools.shared._llm_health._check_llm_provider_health", AsyncMock(return_value="unavailable"))
@patch("ss_tools.shared._llm_health._llm_status", {"status": "unavailable", "last_error": "Provider unreachable"})
def test_status_unavailable(self):
"""Returns unavailable with retry_after_s > 0."""
client = _make_client()
resp = client.get("/api/agent/llm-status")
assert resp.status_code == 200
data = resp.json()
assert data["status"] == "unavailable"
assert data["last_error"] == "Provider unreachable"
assert data["retry_after_s"] == 30
@patch("ss_tools.shared._llm_health._check_llm_provider_health", AsyncMock(return_value="timeout"))
@patch("ss_tools.shared._llm_health._llm_status", {"status": "timeout", "last_error": "Request timed out"})
def test_status_timeout(self):
"""Returns timeout status."""
client = _make_client()
resp = client.get("/api/agent/llm-status")
assert resp.status_code == 200
assert resp.json()["status"] == "timeout"
@patch("ss_tools.shared._llm_health._check_llm_provider_health", AsyncMock(return_value="auth_error"))
@patch("ss_tools.shared._llm_health._llm_status", {"status": "auth_error", "last_error": "Invalid API key"})
def test_status_auth_error(self):
"""Returns auth_error status."""
client = _make_client()
resp = client.get("/api/agent/llm-status")
assert resp.status_code == 200
assert resp.json()["status"] == "auth_error"
assert resp.json()["last_error"] == "Invalid API key"
@patch("ss_tools.shared._llm_health._check_llm_provider_health", AsyncMock(return_value="ok"))
@patch("ss_tools.shared._llm_health._llm_status", {"status": "ok", "last_error": ""})
def test_no_auth_required(self):
"""Status endpoint does not require authentication."""
client = _make_client()
# No auth token — should still work since no Depends on get_current_user
resp = client.get("/api/agent/llm-status")
assert resp.status_code == 200
def test_check_health_called(self):
"""_check_llm_provider_health is called on each request."""
mock_check = AsyncMock(return_value="ok")
with (
patch("ss_tools.shared._llm_health._check_llm_provider_health", mock_check),
patch("ss_tools.shared._llm_health._llm_status", {"status": "ok", "last_error": ""}),
):
client = _make_client()
resp = client.get("/api/agent/llm-status")
assert resp.status_code == 200
mock_check.assert_called_once()
# #endregion Test.Api.AgentStatusRoutes

395
backend/tests/api/test_agent_superset_routes.py Normal file
View File

@@ -0,0 +1,395 @@
# #region Test.Api.AgentSupersetRoutes [C:3] [TYPE Module] [SEMANTICS test,agent,superset,sql,dashboard,dataset,database]
# @BRIEF Unit tests for Agent Superset proxy API routes — write and read endpoints.
# @RELATION BINDS_TO -> [Api.AgentSuperset.AgentSupersetRoutes, Api.AgentSupersetExplore.AgentSupersetExploreRoutes]
# @TEST_EDGE: environment_not_found -> 404
# @TEST_EDGE: superset_client_error -> 500
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 datetime import datetime
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from fastapi import FastAPI
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)
def _make_mock_user() -> MagicMock:
"""Build a mock admin user for auth bypass."""
from src.schemas.auth import RoleSchema, User as UserSchema
admin_role = RoleSchema(
id="r1", name="Admin", description="", is_admin=True, permissions=[]
)
user = UserSchema(
id="admin-1",
username="admin",
email="admin@x.com",
auth_source="LOCAL",
created_at=datetime.now(),
roles=[admin_role],
)
return user
def _make_mock_superset_client() -> AsyncMock:
"""Build a mock SupersetClient with AsyncMock methods."""
client = AsyncMock()
# Write module methods
client.execute_sql = AsyncMock(return_value={"status": "success", "data": [{"col": 1}]})
client.format_sql = AsyncMock(return_value="SELECT * FROM foo")
client.estimate_sql_cost = AsyncMock(return_value={"cost": 10})
client.create_dashboard = AsyncMock(return_value={"id": 101, "slug": "new-dash"})
client.copy_dashboard = AsyncMock(return_value={"id": 102, "slug": "copy-dash"})
client.update_dashboard = AsyncMock(return_value={"id": 101, "slug": "updated-dash"})
client.create_dataset = AsyncMock(return_value={"id": 201, "table_name": "new_table"})
client.delete_dataset = AsyncMock(return_value={"status": "deleted"})
client.duplicate_dataset = AsyncMock(return_value={"id": 202, "table_name": "dup_table"})
client.refresh_dataset_schema = AsyncMock(return_value={"status": "refreshed"})
# Explore module methods
client.get_databases_summary = AsyncMock(return_value=[{"id": 1, "name": "main", "engine": "postgresql"}])
client.get_database_schemas = AsyncMock(return_value=["public", "analytics"])
client.get_database_tables = AsyncMock(return_value=[{"id": 1, "name": "users"}])
client.get_database_table_metadata = AsyncMock(return_value={"columns": [{"name": "id", "type": "integer"}]})
client.get_database_select_star = AsyncMock(return_value="SELECT * FROM users")
client.validate_sql = AsyncMock(return_value={"valid": True})
client.test_database_connection = AsyncMock(return_value={"status": "ok"})
client.permissions_audit = AsyncMock(return_value={"users": [], "total": 0})
client.get_saved_queries = AsyncMock(return_value=(0, []))
client.get_saved_query = AsyncMock(return_value={"id": 1, "sql": "SELECT 1"})
client.aclose = AsyncMock()
return client
def _make_client(overrides: dict | None = None) -> TestClient:
"""Build a TestClient with both agent superset routers."""
from src.api.routes.agent_superset import router as write_router
from src.api.routes.agent_superset_explore import router as explore_router
from src.dependencies import get_current_user
app = FastAPI()
app.include_router(write_router)
app.include_router(explore_router)
mock_user = _make_mock_user()
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, raise_server_exceptions=False)
# ════════════════════════════════════════════════════
# Write module — agent_superset.py
# ════════════════════════════════════════════════════
class TestAgentSqlLabExecute:
"""POST /api/agent/superset/sqllab/execute"""
def test_success(self):
client_sup = _make_mock_superset_client()
with patch("src.api.routes.agent_superset._get_superset_client", AsyncMock(return_value=client_sup)):
client = _make_client()
resp = client.post("/api/agent/superset/sqllab/execute?environment_id=env-1&database_id=1&sql=SELECT+1")
assert resp.status_code == 200
data = resp.json()
assert data["status"] == "success"
# Note: no 500-error test here because agent_superset routes use
# try/finally without except — RuntimeErrors propagate uncaught.
class TestAgentSqlLabFormat:
"""POST /api/agent/superset/sqllab/format"""
def test_success(self):
client_sup = _make_mock_superset_client()
with patch("src.api.routes.agent_superset._get_superset_client", AsyncMock(return_value=client_sup)):
client = _make_client()
resp = client.post("/api/agent/superset/sqllab/format?environment_id=env-1&sql=SELECT+*+FROM+foo")
assert resp.status_code == 200
assert resp.json()["result"] == "SELECT * FROM foo"
class TestAgentSqlLabEstimate:
"""POST /api/agent/superset/sqllab/estimate"""
def test_success(self):
client_sup = _make_mock_superset_client()
with patch("src.api.routes.agent_superset._get_superset_client", AsyncMock(return_value=client_sup)):
client = _make_client()
resp = client.post("/api/agent/superset/sqllab/estimate?environment_id=env-1&database_id=1&sql=SELECT+1")
assert resp.status_code == 200
assert resp.json()["cost"] == 10
class TestAgentDashboardCreate:
"""POST /api/agent/superset/dashboards"""
def test_success(self):
client_sup = _make_mock_superset_client()
with patch("src.api.routes.agent_superset._get_superset_client", AsyncMock(return_value=client_sup)):
client = _make_client()
resp = client.post("/api/agent/superset/dashboards?environment_id=env-1&dashboard_title=My+Dashboard")
assert resp.status_code == 200
assert resp.json()["id"] == 101
class TestAgentDashboardCopy:
"""POST /api/agent/superset/dashboards/{dashboard_id}/copy"""
def test_success(self):
client_sup = _make_mock_superset_client()
with patch("src.api.routes.agent_superset._get_superset_client", AsyncMock(return_value=client_sup)):
client = _make_client()
resp = client.post("/api/agent/superset/dashboards/42/copy?environment_id=env-1&dashboard_title=Copy")
assert resp.status_code == 200
assert resp.json()["id"] == 102
class TestAgentDashboardUpdate:
"""PUT /api/agent/superset/dashboards/{dashboard_id}"""
def test_success(self):
client_sup = _make_mock_superset_client()
with patch("src.api.routes.agent_superset._get_superset_client", AsyncMock(return_value=client_sup)):
client = _make_client()
resp = client.put("/api/agent/superset/dashboards/42?environment_id=env-1&dashboard_title=Updated")
assert resp.status_code == 200
assert resp.json()["slug"] == "updated-dash"
class TestAgentDatasetCreate:
"""POST /api/agent/superset/datasets"""
def test_success(self):
client_sup = _make_mock_superset_client()
with patch("src.api.routes.agent_superset._get_superset_client", AsyncMock(return_value=client_sup)):
client = _make_client()
resp = client.post("/api/agent/superset/datasets?environment_id=env-1&table_name=my_table&database=1")
assert resp.status_code == 200
assert resp.json()["table_name"] == "new_table"
class TestAgentDatasetDelete:
"""DELETE /api/agent/superset/datasets/{dataset_id}"""
def test_success(self):
client_sup = _make_mock_superset_client()
with patch("src.api.routes.agent_superset._get_superset_client", AsyncMock(return_value=client_sup)):
client = _make_client()
resp = client.delete("/api/agent/superset/datasets/42?environment_id=env-1")
assert resp.status_code == 200
assert resp.json()["status"] == "deleted"
class TestAgentDatasetDuplicate:
"""POST /api/agent/superset/datasets/{dataset_id}/duplicate"""
def test_success(self):
client_sup = _make_mock_superset_client()
with patch("src.api.routes.agent_superset._get_superset_client", AsyncMock(return_value=client_sup)):
client = _make_client()
resp = client.post("/api/agent/superset/datasets/42/duplicate?environment_id=env-1&table_name=dup_table")
assert resp.status_code == 200
assert resp.json()["table_name"] == "dup_table"
class TestAgentDatasetRefresh:
"""POST /api/agent/superset/datasets/{dataset_id}/refresh"""
def test_success(self):
client_sup = _make_mock_superset_client()
with patch("src.api.routes.agent_superset._get_superset_client", AsyncMock(return_value=client_sup)):
client = _make_client()
resp = client.post("/api/agent/superset/datasets/42/refresh?environment_id=env-1")
assert resp.status_code == 200
assert resp.json()["status"] == "refreshed"
# ════════════════════════════════════════════════════
# Explore module — agent_superset_explore.py
# ════════════════════════════════════════════════════
class TestAgentListDatabases:
"""GET /api/agent/superset/databases"""
def test_success(self):
client_sup = _make_mock_superset_client()
with patch("src.api.routes.agent_superset_explore._get_superset_client", AsyncMock(return_value=client_sup)):
client = _make_client()
resp = client.get("/api/agent/superset/databases?environment_id=env-1")
assert resp.status_code == 200
data = resp.json()
assert len(data) == 1
assert data[0]["name"] == "main"
# Note: no 500-error test here — explore routes use try/finally without except.
class TestAgentDatabaseSchemas:
"""GET /api/agent/superset/databases/{database_id}/schemas"""
def test_success(self):
client_sup = _make_mock_superset_client()
with patch("src.api.routes.agent_superset_explore._get_superset_client", AsyncMock(return_value=client_sup)):
client = _make_client()
resp = client.get("/api/agent/superset/databases/1/schemas?environment_id=env-1")
assert resp.status_code == 200
assert "public" in resp.json()
# Note: no 500-error test here — explore routes use try/finally without except.
class TestAgentDatabaseTables:
"""GET /api/agent/superset/databases/{database_id}/tables"""
def test_success(self):
client_sup = _make_mock_superset_client()
with patch("src.api.routes.agent_superset_explore._get_superset_client", AsyncMock(return_value=client_sup)):
client = _make_client()
resp = client.get("/api/agent/superset/databases/1/tables?environment_id=env-1")
assert resp.status_code == 200
assert resp.json()[0]["name"] == "users"
class TestAgentDatabaseTableMetadata:
"""GET /api/agent/superset/databases/{database_id}/table_metadata"""
def test_success(self):
client_sup = _make_mock_superset_client()
with patch("src.api.routes.agent_superset_explore._get_superset_client", AsyncMock(return_value=client_sup)):
client = _make_client()
resp = client.get("/api/agent/superset/databases/1/table_metadata?environment_id=env-1&table_name=users")
assert resp.status_code == 200
assert resp.json()["columns"][0]["name"] == "id"
class TestAgentDatabaseSelectStar:
"""GET /api/agent/superset/databases/{database_id}/select_star"""
def test_success(self):
client_sup = _make_mock_superset_client()
with patch("src.api.routes.agent_superset_explore._get_superset_client", AsyncMock(return_value=client_sup)):
client = _make_client()
resp = client.get("/api/agent/superset/databases/1/select_star?environment_id=env-1&table_name=users")
assert resp.status_code == 200
assert resp.json()["sql"] == "SELECT * FROM users"
class TestAgentDatabaseValidateSql:
"""POST /api/agent/superset/databases/{database_id}/validate_sql"""
def test_success(self):
client_sup = _make_mock_superset_client()
with patch("src.api.routes.agent_superset_explore._get_superset_client", AsyncMock(return_value=client_sup)):
client = _make_client()
resp = client.post("/api/agent/superset/databases/1/validate_sql?environment_id=env-1&sql=SELECT+1")
assert resp.status_code == 200
assert resp.json()["valid"] is True
class TestAgentDatabaseTestConnection:
"""POST /api/agent/superset/databases/test_connection"""
def test_success(self):
client_sup = _make_mock_superset_client()
with patch("src.api.routes.agent_superset_explore._get_superset_client", AsyncMock(return_value=client_sup)):
client = _make_client()
resp = client.post(
"/api/agent/superset/databases/test_connection?environment_id=env-1&database_name=test&sqlalchemy_uri=postgresql:///test"
)
assert resp.status_code == 200
assert resp.json()["status"] == "ok"
def test_with_extra(self):
client_sup = _make_mock_superset_client()
with patch("src.api.routes.agent_superset_explore._get_superset_client", AsyncMock(return_value=client_sup)):
client = _make_client()
resp = client.post(
"/api/agent/superset/databases/test_connection?environment_id=env-1&database_name=test&sqlalchemy_uri=postgresql:///test&extra={}"
)
assert resp.status_code == 200
class TestAgentAuditPermissions:
"""GET /api/agent/superset/audit/permissions"""
def test_success(self):
client_sup = _make_mock_superset_client()
with patch("src.api.routes.agent_superset_explore._get_superset_client", AsyncMock(return_value=client_sup)):
client = _make_client()
resp = client.get("/api/agent/superset/audit/permissions?environment_id=env-1")
assert resp.status_code == 200
assert resp.json()["total"] == 0
def test_with_filters(self):
client_sup = _make_mock_superset_client()
with patch("src.api.routes.agent_superset_explore._get_superset_client", AsyncMock(return_value=client_sup)):
client = _make_client()
resp = client.get("/api/agent/superset/audit/permissions?environment_id=env-1&page=1&page_size=50&username_filter=admin&include_admin=true")
assert resp.status_code == 200
class TestAgentSavedQueryList:
"""GET /api/agent/superset/saved_queries"""
def test_success(self):
client_sup = _make_mock_superset_client()
with patch("src.api.routes.agent_superset_explore._get_superset_client", AsyncMock(return_value=client_sup)):
client = _make_client()
resp = client.get("/api/agent/superset/saved_queries?environment_id=env-1")
assert resp.status_code == 200
assert resp.json()["count"] == 0
class TestAgentSavedQueryGet:
"""GET /api/agent/superset/saved_queries/{query_id}"""
def test_success(self):
client_sup = _make_mock_superset_client()
with patch("src.api.routes.agent_superset_explore._get_superset_client", AsyncMock(return_value=client_sup)):
client = _make_client()
resp = client.get("/api/agent/superset/saved_queries/1?environment_id=env-1")
assert resp.status_code == 200
assert resp.json()["id"] == 1
# ════════════════════════════════════════════════════
# Error — Superset client aclose always called
# ════════════════════════════════════════════════════
class TestSupersetClientCleanup:
"""Verify that aclose() is called after each request."""
def test_aclose_called_on_success(self):
client_sup = _make_mock_superset_client()
with patch("src.api.routes.agent_superset._get_superset_client", AsyncMock(return_value=client_sup)):
client = _make_client()
resp = client.post("/api/agent/superset/sqllab/execute?environment_id=env-1&database_id=1&sql=SELECT+1")
assert resp.status_code == 200
client_sup.aclose.assert_called_once()
def test_aclose_called_on_error(self):
client_sup = _make_mock_superset_client()
client_sup.execute_sql.side_effect = RuntimeError("boom")
with patch("src.api.routes.agent_superset._get_superset_client", AsyncMock(return_value=client_sup)):
client = _make_client()
resp = client.post("/api/agent/superset/sqllab/execute?environment_id=env-1&database_id=1&sql=SELECT+1")
assert resp.status_code == 500
client_sup.aclose.assert_called_once()
# #endregion Test.Api.AgentSupersetRoutes

731
backend/tests/api/test_git_release_routes.py Normal file
View File

@@ -0,0 +1,731 @@
# #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")
import sys
from datetime import datetime, timezone
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from fastapi import FastAPI, HTTPException
from fastapi.testclient import TestClient
from pydantic import BaseModel
_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 src.schemas.auth import RoleSchema, User as UserSchema
from datetime import datetime
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()
from datetime import timezone
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_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