Files
ss-tools/backend/tests/api/test_admin.py
busya ce0369ae5c test(backend): add 55+ test files to push coverage to 98%
Subagents delivered tests across all uncovered backend modules:

Schemas (100%): agent, auth, health, profile, settings, validation
Services (98-100%): auth, profile, health, llm, mapping, resource,
  security, git, superset_lookup, sql_table_extractor, rbac
API routes (new): auth, admin, health, environments, plugins,
  dashboards (helpers, projection, actions, listing),
  git (config, deps, env, helpers)
Clean Release (100%): DTO, facade, policy_engine, stages,
  repos, preparation, source_isolation, compliance
Git services: base, remote_providers
Agent module: app, run, middleware, langgraph_setup
Core: trace, cleanup, ws_log_handler, timezone, auth (config/oauth/security), matching
Reports: normalizer, report_service, type_profiles
Notifications: service, providers

Also:
- .gitignore: add .coverage, *.cover, coverage-* dirs
- src/schemas/auth.py: fix AD group DN regex (comma in CN=...)
- Remove co-located src/services/__tests__/ (caused pytest module collision)
2026-06-15 13:55:57 +03:00

456 lines
17 KiB
Python

# #region Test.Api.Admin [C:3] [TYPE Module] [SEMANTICS test,admin,api]
# @BRIEF Unit tests for admin API routes — users, roles, permissions, AD mappings.
# @RELATION BINDS_TO -> [AdminApi]
# @TEST_EDGE: user_not_found -> 404
# @TEST_EDGE: username_exists -> 400
# @TEST_EDGE: role_not_found -> 404
# @TEST_EDGE: role_already_exists -> 400
import os
# Set env BEFORE any source imports
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")
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from fastapi import FastAPI, HTTPException
from fastapi.testclient import TestClient
_src = str(Path(__file__).resolve().parent.parent.parent / "src")
if _src not in sys.path:
sys.path.insert(0, _src)
def _make_client(overrides: dict | None = None) -> TestClient:
"""Build TestClient with admin router and all deps overridden."""
from src.api.routes.admin import router
from src.core.database import get_auth_db
from src.dependencies import get_current_user, has_permission
from src.schemas.auth import User, RoleSchema
app = FastAPI()
app.include_router(router)
mock_db = MagicMock()
mock_user = User(
id="admin-1",
username="admin",
email="admin@example.com",
auth_source="LOCAL",
is_active=True,
created_at=__import__("datetime").datetime.now(),
roles=[RoleSchema(id="role-1", name="Admin", description="Admin", permissions=[])],
)
defaults = {
get_auth_db: lambda: mock_db,
get_current_user: lambda: mock_user,
has_permission: lambda *args, **kwargs: lambda: None,
}
if overrides:
defaults.update(overrides)
for dep, mock_fn in defaults.items():
app.dependency_overrides[dep] = mock_fn
return TestClient(app)
# ── Users ──
class TestListUsers:
"""GET /api/admin/users"""
def test_list_users_success(self):
"""Happy path: list all users returns 200."""
from src.core.database import get_auth_db
mock_user = MagicMock()
mock_user.id = "user-1"
mock_user.username = "alice"
mock_user.email = "alice@example.com"
mock_user.is_active = True
mock_user.auth_source = "LOCAL"
mock_user.created_at = __import__("datetime").datetime.now()
mock_user.last_login = None
mock_user.roles = []
mock_session = MagicMock()
mock_session.query.return_value.all.return_value = [mock_user]
client = _make_client({get_auth_db: lambda: mock_session})
resp = client.get("/api/admin/users")
assert resp.status_code == 200
data = resp.json()
assert isinstance(data, list)
assert data[0]["username"] == "alice"
class TestCreateUser:
"""POST /api/admin/users"""
def test_create_user_success(self):
"""Happy path: user created with 201."""
from src.core.database import get_auth_db
mock_session = MagicMock()
mock_repo = MagicMock()
mock_repo.get_user_by_username.return_value = None
mock_repo.get_role_by_name.side_effect = lambda name: (
MagicMock(id=f"role-{name}", name=name, permissions=[]) if name == "Admin" else None
)
with patch("src.api.routes.admin.AuthRepository", return_value=mock_repo):
client = _make_client({get_auth_db: lambda: mock_session})
resp = client.post("/api/admin/users", json={
"username": "newuser",
"email": "new@example.com",
"password": "StrongPass1",
"is_active": True,
"roles": ["Admin"],
})
assert resp.status_code == 201
mock_session.add.assert_called_once()
mock_session.commit.assert_called_once()
def test_create_user_duplicate_username(self):
"""Username already exists returns 400."""
from src.core.database import get_auth_db
mock_session = MagicMock()
mock_repo = MagicMock()
mock_repo.get_user_by_username.return_value = MagicMock()
with patch("src.api.routes.admin.AuthRepository", return_value=mock_repo):
client = _make_client({get_auth_db: lambda: mock_session})
resp = client.post("/api/admin/users", json={
"username": "exists",
"email": "dup@example.com",
"password": "StrongPass1",
})
assert resp.status_code == 400
assert "Username already exists" in resp.text
def test_create_user_weak_password(self):
"""Weak password returns 422."""
client = _make_client()
resp = client.post("/api/admin/users", json={
"username": "weakuser",
"email": "weak@example.com",
"password": "short",
})
assert resp.status_code == 422
def test_create_user_no_permission(self):
"""User without write permission gets 403."""
from src.core.database import get_auth_db
from src.dependencies import get_current_user, has_permission
from src.schemas.auth import User, RoleSchema
app = FastAPI()
from src.api.routes.admin import router
app.include_router(router)
app.dependency_overrides[get_auth_db] = lambda: MagicMock()
app.dependency_overrides[get_current_user] = lambda: User(
id="user-1", username="regular", email="u@x.com", auth_source="LOCAL",
created_at=__import__("datetime").datetime.now(), roles=[]
)
# Keep has_permission as real — it will use the mock get_current_user
client = TestClient(app)
resp = client.post("/api/admin/users", json={
"username": "test",
"email": "test@example.com",
"password": "StrongPass1",
})
assert resp.status_code == 403
class TestUpdateUser:
"""PUT /api/admin/users/{user_id}"""
def test_update_user_success(self):
"""Happy path: user updated and returned."""
from src.core.database import get_auth_db
mock_session = MagicMock()
mock_repo = MagicMock()
existing_user = MagicMock()
existing_user.id = "user-1"
existing_user.username = "oldname"
existing_user.email = "old@example.com"
existing_user.is_active = True
existing_user.roles = []
mock_repo.get_user_by_id.return_value = existing_user
with patch("src.api.routes.admin.AuthRepository", return_value=mock_repo):
client = _make_client({get_auth_db: lambda: mock_session})
resp = client.put("/api/admin/users/user-1", json={"email": "new@example.com"})
assert resp.status_code == 200
assert existing_user.email == "new@example.com"
mock_session.commit.assert_called_once()
def test_update_user_not_found(self):
"""Non-existent user returns 404."""
from src.core.database import get_auth_db
mock_session = MagicMock()
mock_repo = MagicMock()
mock_repo.get_user_by_id.return_value = None
with patch("src.api.routes.admin.AuthRepository", return_value=mock_repo):
client = _make_client({get_auth_db: lambda: mock_session})
resp = client.put("/api/admin/users/user-999", json={"email": "x@y.com"})
assert resp.status_code == 404
class TestDeleteUser:
"""DELETE /api/admin/users/{user_id}"""
def test_delete_user_success(self):
"""Happy path: user deleted returns 204."""
from src.core.database import get_auth_db
mock_session = MagicMock()
mock_repo = MagicMock()
mock_repo.get_user_by_id.return_value = MagicMock(username="testuser")
with patch("src.api.routes.admin.AuthRepository", return_value=mock_repo):
client = _make_client({get_auth_db: lambda: mock_session})
resp = client.delete("/api/admin/users/user-1")
assert resp.status_code == 204
mock_session.delete.assert_called_once()
mock_session.commit.assert_called_once()
def test_delete_user_not_found(self):
"""Non-existent user returns 404."""
from src.core.database import get_auth_db
mock_session = MagicMock()
mock_repo = MagicMock()
mock_repo.get_user_by_id.return_value = None
with patch("src.api.routes.admin.AuthRepository", return_value=mock_repo):
client = _make_client({get_auth_db: lambda: mock_session})
resp = client.delete("/api/admin/users/user-999")
assert resp.status_code == 404
# ── Roles ──
class TestListRoles:
"""GET /api/admin/roles"""
def test_list_roles_success(self):
"""Happy path: list roles returns 200."""
mock_role = MagicMock()
mock_role.id = "role-1"
mock_role.name = "Admin"
mock_role.description = "Administrator"
mock_role.permissions = []
mock_session = MagicMock()
mock_session.query.return_value.all.return_value = [mock_role]
from src.core.database import get_auth_db
client = _make_client({get_auth_db: lambda: mock_session})
resp = client.get("/api/admin/roles")
assert resp.status_code == 200
assert isinstance(resp.json(), list)
class TestCreateRole:
"""POST /api/admin/roles"""
def test_create_role_success(self):
"""Happy path: role created with 201."""
mock_session = MagicMock()
mock_session.query.return_value.filter.return_value.first.return_value = None
mock_repo = MagicMock()
mock_repo.get_permission_by_id.return_value = MagicMock(id="perm-1")
with patch("src.api.routes.admin.AuthRepository", return_value=mock_repo):
from src.core.database import get_auth_db
client = _make_client({get_auth_db: lambda: mock_session})
resp = client.post("/api/admin/roles", json={
"name": "Editor",
"description": "Can edit",
"permissions": ["perm-1"],
})
assert resp.status_code == 201
mock_session.add.assert_called_once()
mock_session.commit.assert_called_once()
def test_create_role_duplicate(self):
"""Duplicate role name returns 400."""
mock_session = MagicMock()
mock_session.query.return_value.filter.return_value.first.return_value = MagicMock()
from src.core.database import get_auth_db
client = _make_client({get_auth_db: lambda: mock_session})
resp = client.post("/api/admin/roles", json={
"name": "Admin",
"description": "Already exists",
"permissions": [],
})
assert resp.status_code == 400
assert "Role already exists" in resp.text
class TestUpdateRole:
"""PUT /api/admin/roles/{role_id}"""
def test_update_role_success(self):
"""Happy path: role updated."""
mock_session = MagicMock()
mock_repo = MagicMock()
existing_role = MagicMock()
existing_role.id = "role-1"
existing_role.name = "OldName"
existing_role.description = "Old desc"
existing_role.permissions = []
mock_repo.get_role_by_id.return_value = existing_role
with patch("src.api.routes.admin.AuthRepository", return_value=mock_repo):
from src.core.database import get_auth_db
client = _make_client({get_auth_db: lambda: mock_session})
resp = client.put("/api/admin/roles/role-1", json={"name": "NewName", "description": "New desc"})
assert resp.status_code == 200
assert existing_role.name == "NewName"
mock_session.commit.assert_called_once()
def test_update_role_not_found(self):
"""Non-existent role returns 404."""
mock_session = MagicMock()
mock_repo = MagicMock()
mock_repo.get_role_by_id.return_value = None
with patch("src.api.routes.admin.AuthRepository", return_value=mock_repo):
from src.core.database import get_auth_db
client = _make_client({get_auth_db: lambda: mock_session})
resp = client.put("/api/admin/roles/role-999", json={"name": "Ghost"})
assert resp.status_code == 404
class TestDeleteRole:
"""DELETE /api/admin/roles/{role_id}"""
def test_delete_role_success(self):
"""Happy path: role deleted returns 204."""
mock_session = MagicMock()
mock_repo = MagicMock()
mock_repo.get_role_by_id.return_value = MagicMock()
with patch("src.api.routes.admin.AuthRepository", return_value=mock_repo):
from src.core.database import get_auth_db
client = _make_client({get_auth_db: lambda: mock_session})
resp = client.delete("/api/admin/roles/role-1")
assert resp.status_code == 204
mock_session.delete.assert_called_once()
mock_session.commit.assert_called_once()
def test_delete_role_not_found(self):
"""Non-existent role returns 404."""
mock_session = MagicMock()
mock_repo = MagicMock()
mock_repo.get_role_by_id.return_value = None
with patch("src.api.routes.admin.AuthRepository", return_value=mock_repo):
from src.core.database import get_auth_db
client = _make_client({get_auth_db: lambda: mock_session})
resp = client.delete("/api/admin/roles/role-999")
assert resp.status_code == 404
# ── Permissions ──
class TestListPermissions:
"""GET /api/admin/permissions"""
def test_list_permissions_success(self):
"""Happy path: returns permissions list."""
mock_session = MagicMock()
mock_repo = MagicMock()
mock_perm = MagicMock()
mock_perm.id = "perm-1"
mock_perm.resource = "users"
mock_perm.action = "read"
mock_repo.list_permissions.return_value = [mock_perm]
with patch("src.api.routes.admin.AuthRepository", return_value=mock_repo), \
patch("src.api.routes.admin.discover_declared_permissions", return_value=[]), \
patch("src.api.routes.admin.sync_permission_catalog", return_value=0):
from src.core.database import get_auth_db
from src.dependencies import get_plugin_loader
client = _make_client({
get_auth_db: lambda: mock_session,
get_plugin_loader: lambda: MagicMock(),
})
resp = client.get("/api/admin/permissions")
assert resp.status_code == 200
assert isinstance(resp.json(), list)
def test_list_permissions_with_sync(self):
"""When new permissions discovered, sync is called."""
mock_session = MagicMock()
mock_repo = MagicMock()
mock_repo.list_permissions.return_value = []
with patch("src.api.routes.admin.AuthRepository", return_value=mock_repo), \
patch("src.api.routes.admin.discover_declared_permissions", return_value=["new:perm"]), \
patch("src.api.routes.admin.sync_permission_catalog", return_value=2):
from src.core.database import get_auth_db
from src.dependencies import get_plugin_loader
client = _make_client({
get_auth_db: lambda: mock_session,
get_plugin_loader: lambda: MagicMock(),
})
resp = client.get("/api/admin/permissions")
assert resp.status_code == 200
# ── AD Mappings ──
class TestListAdMappings:
"""GET /api/admin/ad-mappings"""
def test_list_mappings_success(self):
"""Happy path: returns AD mappings."""
mock_mapping = MagicMock()
mock_mapping.id = "map-1"
mock_mapping.ad_group = "DOMAIN\\group1"
mock_mapping.role_id = "role-1"
mock_session = MagicMock()
mock_session.query.return_value.all.return_value = [mock_mapping]
from src.core.database import get_auth_db
client = _make_client({get_auth_db: lambda: mock_session})
resp = client.get("/api/admin/ad-mappings")
assert resp.status_code == 200
assert isinstance(resp.json(), list)
class TestCreateAdMapping:
"""POST /api/admin/ad-mappings"""
def test_create_mapping_success(self):
"""Happy path: AD mapping created."""
mock_session = MagicMock()
from src.core.database import get_auth_db
client = _make_client({get_auth_db: lambda: mock_session})
resp = client.post("/api/admin/ad-mappings", json={
"ad_group": "DOMAIN\\newgroup",
"role_id": "role-1",
})
assert resp.status_code == 200
mock_session.add.assert_called_once()
mock_session.commit.assert_called_once()
def test_create_mapping_invalid_ad_group(self):
"""Invalid AD group name returns 422."""
client = _make_client()
resp = client.post("/api/admin/ad-mappings", json={
"ad_group": "invalid group with spaces!!!",
"role_id": "role-1",
})
assert resp.status_code == 422
# #endregion Test.Api.Admin