198 lines
6.9 KiB
Python
198 lines
6.9 KiB
Python
# #region Test.Api.Mappings [C:3] [TYPE Module] [SEMANTICS test,mappings,routes]
|
|
# @BRIEF Tests for mappings API routes — get, create, suggest.
|
|
# @RELATION BINDS_TO -> [MappingsApi]
|
|
|
|
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
|
|
|
|
_src = str(Path(__file__).resolve().parent.parent.parent / "src")
|
|
if _src not in sys.path:
|
|
sys.path.insert(0, _src)
|
|
|
|
|
|
def _make_mapping(**kw):
|
|
m = MagicMock()
|
|
m.id = kw.get("id", "map-1")
|
|
m.source_env_id = kw.get("source_env_id", "env-1")
|
|
m.target_env_id = kw.get("target_env_id", "env-2")
|
|
m.source_db_uuid = kw.get("source_db_uuid", "src-uuid")
|
|
m.target_db_uuid = kw.get("target_db_uuid", "tgt-uuid")
|
|
m.source_db_name = kw.get("source_db_name", "Source DB")
|
|
m.target_db_name = kw.get("target_db_name", "Target DB")
|
|
m.engine = kw.get("engine", "postgresql")
|
|
return m
|
|
|
|
|
|
def _make_client(overrides: dict | None = None) -> TestClient:
|
|
from src.api.routes.mappings import router
|
|
from src.core.database import get_db
|
|
from src.dependencies import get_config_manager, get_current_user
|
|
from src.schemas.auth import User, RoleSchema
|
|
|
|
app = FastAPI()
|
|
app.include_router(router, prefix="/api/mappings")
|
|
|
|
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
|
|
app.dependency_overrides[get_config_manager] = lambda: MagicMock()
|
|
if overrides:
|
|
for dep, fn in overrides.items():
|
|
app.dependency_overrides[dep] = fn
|
|
return TestClient(app)
|
|
|
|
|
|
# ── get_mappings ──
|
|
|
|
class TestGetMappings:
|
|
"""GET /mappings"""
|
|
|
|
def test_success(self):
|
|
mock_db = MagicMock()
|
|
mock_db.query.return_value.all.return_value = [
|
|
_make_mapping(id="m1"),
|
|
_make_mapping(id="m2"),
|
|
]
|
|
|
|
from src.core.database import get_db
|
|
client = _make_client({get_db: lambda: mock_db})
|
|
resp = client.get("/api/mappings")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert len(data) == 2
|
|
assert data[0]["id"] == "m1"
|
|
|
|
def test_filter_by_source(self):
|
|
mock_db = MagicMock()
|
|
mock_filter = MagicMock()
|
|
mock_db.query.return_value.filter.return_value = mock_filter
|
|
mock_filter.filter.return_value = mock_filter
|
|
mock_filter.all.return_value = [_make_mapping(id="m1")]
|
|
|
|
from src.core.database import get_db
|
|
client = _make_client({get_db: lambda: mock_db})
|
|
resp = client.get("/api/mappings?source_env_id=env-1")
|
|
assert resp.status_code == 200
|
|
|
|
def test_filter_by_target(self):
|
|
mock_db = MagicMock()
|
|
mock_filter = MagicMock()
|
|
mock_db.query.return_value.filter.return_value = mock_filter
|
|
mock_filter.filter.return_value = mock_filter
|
|
mock_filter.all.return_value = []
|
|
|
|
from src.core.database import get_db
|
|
client = _make_client({get_db: lambda: mock_db})
|
|
resp = client.get("/api/mappings?target_env_id=env-2")
|
|
assert resp.status_code == 200
|
|
|
|
def test_empty(self):
|
|
mock_db = MagicMock()
|
|
mock_db.query.return_value.all.return_value = []
|
|
|
|
from src.core.database import get_db
|
|
client = _make_client({get_db: lambda: mock_db})
|
|
resp = client.get("/api/mappings")
|
|
assert resp.status_code == 200
|
|
assert resp.json() == []
|
|
|
|
|
|
# ── create_mapping ──
|
|
|
|
class TestCreateMapping:
|
|
"""POST /mappings"""
|
|
|
|
CREATE_PAYLOAD = {
|
|
"source_env_id": "env-1",
|
|
"target_env_id": "env-2",
|
|
"source_db_uuid": "uuid-1",
|
|
"target_db_uuid": "uuid-2",
|
|
"source_db_name": "Source DB",
|
|
"target_db_name": "Target DB",
|
|
"engine": "postgresql",
|
|
}
|
|
|
|
def test_create_new(self):
|
|
mock_db = MagicMock()
|
|
mock_db.query.return_value.filter.return_value.first.return_value = None
|
|
|
|
# Simulate db.refresh setting the id
|
|
def _refresh(obj):
|
|
obj.id = "map-new-1"
|
|
mock_db.refresh.side_effect = _refresh
|
|
|
|
from src.core.database import get_db
|
|
client = _make_client({get_db: lambda: mock_db})
|
|
resp = client.post("/api/mappings", json=self.CREATE_PAYLOAD)
|
|
assert resp.status_code == 200
|
|
mock_db.add.assert_called_once()
|
|
mock_db.commit.assert_called_once()
|
|
|
|
def test_update_existing(self):
|
|
existing = _make_mapping()
|
|
mock_db = MagicMock()
|
|
mock_db.query.return_value.filter.return_value.first.return_value = existing
|
|
|
|
from src.core.database import get_db
|
|
client = _make_client({get_db: lambda: mock_db})
|
|
resp = client.post("/api/mappings", json=self.CREATE_PAYLOAD)
|
|
assert resp.status_code == 200
|
|
mock_db.add.assert_not_called()
|
|
mock_db.commit.assert_called_once()
|
|
assert existing.target_db_uuid == "uuid-2"
|
|
assert existing.target_db_name == "Target DB"
|
|
|
|
|
|
# ── suggest_mappings_api ──
|
|
|
|
class TestSuggestMappingsApi:
|
|
"""POST /mappings/suggest"""
|
|
|
|
def test_success(self):
|
|
mock_config = MagicMock()
|
|
mock_suggestions = [{"source": "db1", "target": "db2", "score": 0.95}]
|
|
|
|
from src.dependencies import get_config_manager
|
|
with patch("src.services.mapping_service.MappingService") as MockMappingSvc:
|
|
instance = MagicMock()
|
|
instance.get_suggestions = AsyncMock(return_value=mock_suggestions)
|
|
MockMappingSvc.return_value = instance
|
|
client = _make_client({get_config_manager: lambda: mock_config})
|
|
resp = client.post("/api/mappings/suggest", json={
|
|
"source_env_id": "env-1", "target_env_id": "env-2",
|
|
})
|
|
assert resp.status_code == 200
|
|
assert resp.json() == mock_suggestions
|
|
|
|
def test_error(self):
|
|
mock_config = MagicMock()
|
|
from src.dependencies import get_config_manager
|
|
with patch("src.services.mapping_service.MappingService") as MockMappingSvc:
|
|
instance = MagicMock()
|
|
instance.get_suggestions = AsyncMock(side_effect=RuntimeError("suggestion failed"))
|
|
MockMappingSvc.return_value = instance
|
|
client = _make_client({get_config_manager: lambda: mock_config})
|
|
resp = client.post("/api/mappings/suggest", json={
|
|
"source_env_id": "env-1", "target_env_id": "env-2",
|
|
})
|
|
assert resp.status_code == 500
|
|
# #endregion Test.Api.Mappings
|