219 lines
7.2 KiB
Python
219 lines
7.2 KiB
Python
"""Shared pytest fixtures: sqlite-backed DB + FastAPI TestClient."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
# Ensure test settings before any app import resolves them.
|
|
os.environ.setdefault("SESSION_SECRET", "test-session-secret-value-123456")
|
|
os.environ.setdefault("SECRET_ENCRYPTION_KEY", "test-encryption-key-value-1234")
|
|
os.environ.setdefault("ADMIN_INITIAL_PASSWORD", "test-admin-password-123")
|
|
os.environ.setdefault("MOSQUITTO_DYNSEC_PASSWORD", "test-dynsec-password")
|
|
os.environ.setdefault("DATABASE_URL", "sqlite+pysqlite:///:memory:")
|
|
|
|
from sqlalchemy import create_engine # noqa: E402
|
|
from sqlalchemy.orm import Session, sessionmaker # noqa: E402
|
|
from sqlalchemy.pool import StaticPool # noqa: E402
|
|
|
|
from app import db as db_module # noqa: E402
|
|
from app import main as main_module # noqa: E402
|
|
from app.models import ManagementRole, ManagementUser # noqa: E402
|
|
from app.security import hash_password # noqa: E402
|
|
|
|
|
|
@pytest.fixture()
|
|
def db_session():
|
|
"""Fresh in-memory database per test with schema + seeded users.
|
|
|
|
StaticPool + check_same_thread=False: the TestClient serves requests
|
|
on a different thread than the fixture, but all access must share ONE
|
|
sqlite connection (in-memory DBs are per-connection).
|
|
"""
|
|
engine = create_engine(
|
|
"sqlite+pysqlite:///:memory:",
|
|
future=True,
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
db_module.Base.metadata.create_all(engine)
|
|
factory = sessionmaker(bind=engine, future=True)
|
|
session = factory()
|
|
# Seed one super admin and one auditor for RBAC tests.
|
|
session.add_all(
|
|
[
|
|
ManagementUser(
|
|
username="admin",
|
|
password_hash=hash_password("admin-password-123"),
|
|
role=ManagementRole.SUPER_ADMIN,
|
|
must_change_password=False,
|
|
),
|
|
ManagementUser(
|
|
username="auditor",
|
|
password_hash=hash_password("auditor-password-123"),
|
|
role=ManagementRole.AUDITOR,
|
|
must_change_password=False,
|
|
),
|
|
]
|
|
)
|
|
session.commit()
|
|
yield session
|
|
session.close()
|
|
|
|
@pytest.fixture()
|
|
def client(db_session, monkeypatch):
|
|
"""TestClient wired to the fixture DB via dependency_overrides."""
|
|
|
|
def _override_get_db():
|
|
# Yield the SAME session the test asserts against; the request
|
|
# handler shares its transaction for simplicity in tests.
|
|
try:
|
|
yield db_session
|
|
finally:
|
|
pass # session lifetime managed by the db_session fixture
|
|
|
|
main_module.app.dependency_overrides[db_module.get_db] = _override_get_db
|
|
|
|
# Skip the lifespan bootstrap (create_all + admin seed) against a
|
|
# separate engine: tests manage the schema themselves.
|
|
from app import main as _m
|
|
|
|
async def _noop_lifespan(app):
|
|
yield
|
|
|
|
monkeypatch.setattr(_m, "_bootstrap_admin", lambda: None)
|
|
|
|
# Fake dynsec so no broker is needed for API tests.
|
|
class FakeDynsec:
|
|
"""Mirrors the corrected wire contract: _command returns the `data`
|
|
object; client objects use `disabled` (absent == enabled)."""
|
|
|
|
def __init__(self):
|
|
self.clients: dict[str, dict] = {}
|
|
self.roles: dict[str, dict] = {}
|
|
self.groups: dict[str, dict] = {}
|
|
|
|
def _command(self, command, **kwargs):
|
|
return {}
|
|
|
|
def create_client(self, username, password, **kwargs):
|
|
self.clients[username] = {"username": username, "disabled": False, **kwargs}
|
|
|
|
def list_clients(self):
|
|
return list(self.clients.values())
|
|
|
|
def delete_client(self, username):
|
|
self.clients.pop(username, None)
|
|
|
|
def disable_client(self, username):
|
|
self.clients[username]["disabled"] = True
|
|
|
|
def enable_client(self, username):
|
|
self.clients[username]["disabled"] = False
|
|
|
|
def set_client_password(self, username, password):
|
|
pass
|
|
|
|
def get_client(self, username):
|
|
return self.clients[username]
|
|
|
|
def create_group(self, groupname, roles=None):
|
|
self.groups[groupname] = {"groupname": groupname, "roles": roles or []}
|
|
|
|
def list_groups(self):
|
|
return list(self.groups.values())
|
|
|
|
def delete_group(self, groupname):
|
|
self.groups.pop(groupname, None)
|
|
|
|
def create_role(self, rolename, acls=None):
|
|
self.roles[rolename] = {"rolename": rolename, "acls": acls or []}
|
|
|
|
def list_roles(self):
|
|
return list(self.roles.values())
|
|
|
|
def delete_role(self, rolename):
|
|
self.roles.pop(rolename, None)
|
|
|
|
def set_role_acls(self, rolename, acls):
|
|
self.roles[rolename]["acls"] = acls
|
|
|
|
def get_role(self, rolename):
|
|
return self.roles[rolename]
|
|
|
|
def add_client_role(self, username, rolename, priority=-1):
|
|
client = self.clients.setdefault(username, {"username": username, "disabled": False})
|
|
client["roles"] = [
|
|
*client.get("roles", []),
|
|
{"rolename": rolename, "priority": priority},
|
|
]
|
|
|
|
def remove_client_role(self, username, rolename):
|
|
client = self.clients.get(username, {})
|
|
client["roles"] = [
|
|
r for r in client.get("roles", []) if r["rolename"] != rolename
|
|
]
|
|
|
|
def add_group_client(self, groupname, username, priority=1):
|
|
self.clients.setdefault(username, {}).setdefault("groups", []).append(
|
|
{"groupname": groupname, "priority": priority}
|
|
)
|
|
|
|
def remove_group_client(self, groupname, username):
|
|
client = self.clients.get(username, {})
|
|
client["groups"] = [
|
|
g for g in client.get("groups", []) if g["groupname"] != groupname
|
|
]
|
|
|
|
def add_group_role(self, groupname, rolename, priority=-1):
|
|
group = self.groups.setdefault(groupname, {"groupname": groupname, "roles": []})
|
|
group["roles"] = [
|
|
*group.get("roles", []),
|
|
{"rolename": rolename, "priority": priority},
|
|
]
|
|
|
|
def remove_group_role(self, groupname, rolename):
|
|
group = self.groups.get(groupname, {})
|
|
group["roles"] = [
|
|
r for r in group.get("roles", []) if r["rolename"] != rolename
|
|
]
|
|
|
|
def connect(self):
|
|
pass
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
fake = FakeDynsec()
|
|
monkeypatch.setattr(main_module, "dynsec_client", fake)
|
|
|
|
from app.routers import mqtt as mqtt_router
|
|
|
|
monkeypatch.setattr(mqtt_router, "dynsec_client", fake)
|
|
|
|
with TestClient(main_module.app) as test_client:
|
|
yield test_client
|
|
|
|
main_module.app.dependency_overrides.clear()
|
|
|
|
|
|
def login(client: TestClient, username: str, password: str):
|
|
response = client.post(
|
|
"/api/v1/auth/login", json={"username": username, "password": password}
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
return response
|
|
|
|
|
|
@pytest.fixture()
|
|
def admin_client(client):
|
|
login(client, "admin", "admin-password-123")
|
|
return client
|
|
|
|
|
|
@pytest.fixture()
|
|
def auditor_client(client):
|
|
login(client, "auditor", "auditor-password-123")
|
|
return client |