from concurrent.futures import ThreadPoolExecutor from contextlib import contextmanager from datetime import datetime, timezone from types import SimpleNamespace from uuid import UUID, uuid4 import pytest from fastapi.testclient import TestClient from psycopg.conninfo import conninfo_to_dict from app.auth.project_dependencies import get_metadata_repository from app.infra.db.project_routing import ActiveProjectRouting from samples.peripheral import auth from samples.peripheral.api import create_app from samples.peripheral.config import SampleSettings from samples.peripheral.errors import SampleFailure from samples.peripheral.store import PostgresStore @pytest.fixture def project_api(monkeypatch): projects = {str(uuid4()): "tjwater_v2", str(uuid4()): "fengyang_v2"} connections = [] permissions = ["simulation.view"] missing = set() route_overrides = {} class Repository: async def get_project_db_routing(self, project_id, role): code = projects[str(project_id)] if (code, role) in missing: return None database = code + ("_history" if role == "iot_data" else "_business") return SimpleNamespace(**({ "dsn": f"postgresql://test@localhost/{database}", "db_type": "timescaledb" if role == "iot_data" else "postgresql", } | route_overrides.get((code, role), {}))) class Connection: def __init__(self, database): self.database = database self.info = SimpleNamespace(dbname=database) self.rows = [] self.readonly = False self.statements = [] def execute(self, statement, parameters=None): self.statements.append((statement, parameters)) self.rows = [] if statement == "SET TRANSACTION READ ONLY": self.readonly = True if "current_database()" in statement: self.rows = [{"name": self.database}] elif "peripheral_sample.scenarios" in statement: self.rows = [] elif "to_regclass('asset.scada_devices')" in statement: self.rows = [{"table_name": "asset.scada_devices"}] elif "FROM asset.scada_devices" in statement: self.rows = [{"device_id": self.database, "device_type": "pressure", "measurement_unit": "MPa", "api_query_id": "point-1"}] elif "FROM scada.measurements" in statement: self.rows = [{"time": datetime(2026, 10, 21, tzinfo=timezone.utc), "monitored_value": 0.3}] elif "to_regclass('workbench.domain_event_log')" in statement: self.rows = [{"event_log": "workbench.domain_event_log"}] elif "RETURNING event_seq" in statement: self.rows = [{"event_seq": 42}] return self def fetchone(self): return self.rows[0] if self.rows else None def fetchall(self): return self.rows @contextmanager def connect(*, conninfo, **kwargs): connection = Connection(conninfo_to_dict(conninfo)["dbname"]) connections.append(connection) yield connection monkeypatch.setattr("samples.peripheral.store.psycopg.connect", connect) monkeypatch.setattr(auth, "next_get", lambda request, path: { "project_id": request.headers["x-project-id"], "user_id": str(uuid4()), "network": projects[request.headers["x-project-id"]], "permissions": permissions, }) app = create_app(SampleSettings()) app.dependency_overrides[get_metadata_repository] = lambda: Repository() with TestClient(app) as client: yield SimpleNamespace(client=client, projects=projects, connections=connections, permissions=permissions, missing=missing, route_overrides=route_overrides) def headers(project): return {"Authorization": "Bearer TEST_ONLY", "X-Project-Id": project} def test_concurrent_project_reads_use_separate_metadata_databases(project_api): api = project_api def read(project): response = api.client.get("/api/sample/v1/monitoring/points", headers=headers(project)) assert response.status_code == 200, response.text return project, response.json() with ThreadPoolExecutor(max_workers=2) as pool: results = list(pool.map(read, list(api.projects) * 3)) for project, payload in results: database = api.projects[project] + "_business" assert payload["source_id"] == database + ":asset.scada_devices" assert payload["items"][0]["point_id"] == database assert all(connection.readonly for connection in api.connections) def test_history_uses_separate_project_timescale_route(project_api): api = project_api project, code = list(api.projects.items())[1] response = api.client.get("/api/sample/v1/monitoring/observations", headers=headers(project), params={ "point_id": "point-1", "start": "2026-10-21T00:00:00Z", "end": "2026-10-22T00:00:00Z", }) assert response.status_code == 200, response.text assert response.json()["items"][0]["value"] == 0.3 assert response.json()["source_id"] == code + "_history:scada.measurements" assert [connection.database for connection in api.connections] == [code + "_business", code + "_history"] assert all(connection.readonly for connection in api.connections) def test_missing_routing_never_falls_back_to_default_database(project_api): api = project_api project, code = list(api.projects.items())[1] api.missing.add((code, "biz_data")) response = api.client.get("/api/sample/v1/monitoring/points", headers=headers(project)) assert response.status_code == 503 assert response.json()["code"] == "PROJECT_STORAGE_NOT_CONFIGURED" assert not api.connections def test_work_order_writes_and_events_use_the_selected_project(project_api): api = project_api api.permissions.append("simulation.run") project, code = list(api.projects.items())[1] response = api.client.post("/api/sample/v1/work-orders", headers=headers(project), json={ "request_id": str(uuid4()), "title": "routing test", "summary": "test only", }) assert response.status_code == 201, response.text assert response.json()["project_id"] == project assert len(api.connections) == 1 connection = api.connections[0] assert connection.database == code + "_business" assert not connection.readonly event = next(params for sql, params in connection.statements if "INSERT INTO workbench.domain_event_log" in sql) assert event[0] == project assert event[2] == response.json()["work_order_id"] @pytest.mark.parametrize("override", [ {"dsn": "postgresql://test@localhost"}, {"dsn": "not-a-valid-dsn"}, {"db_type": "timescaledb"}, ]) def test_invalid_project_routes_fail_without_opening_storage(project_api, override): api = project_api project, code = list(api.projects.items())[1] api.route_overrides[(code, "biz_data")] = override response = api.client.get("/api/sample/v1/monitoring/points", headers=headers(project)) assert response.status_code == 503 assert response.json()["code"] == "PROJECT_STORAGE_NOT_CONFIGURED" assert not api.connections def test_crud_reads_do_not_require_timescale_configuration(project_api): api = project_api project, code = list(api.projects.items())[1] api.missing.add((code, "iot_data")) response = api.client.get("/api/sample/v1/monitoring/points", headers=headers(project)) assert response.status_code == 200 unavailable = api.client.get("/api/sample/v1/monitoring/observations", headers=headers(project), params={ "point_id": "point-1", "start": "2026-10-21T00:00:00Z", "end": "2026-10-22T00:00:00Z", }) assert unavailable.status_code == 503 @pytest.mark.parametrize("method,permissions", [("GET", []), ("POST", ["simulation.view"])]) def test_project_routing_preserves_read_and_write_permissions(project_api, method, permissions): api = project_api api.permissions[:] = permissions response = api.client.request(method, "/api/sample/v1/scenario", headers=headers(next(iter(api.projects)))) assert response.status_code == 403 assert response.json()["code"] == "PERMISSION_DENIED" assert not api.connections def test_mismatched_identity_is_rejected_before_storage(project_api, monkeypatch): monkeypatch.setattr(auth, "next_get", lambda request, path: { "project_id": str(uuid4()), "user_id": str(uuid4()), "network": "fengyang_v2", "permissions": ["simulation.view", "simulation.run"], }) response = project_api.client.get("/api/sample/v1/context", headers=headers(next(iter(project_api.projects)))) assert response.status_code == 403 assert response.json()["code"] == "PROJECT_MISMATCH" assert not project_api.connections def test_routed_connection_does_not_use_legacy_settings_or_missing_timescale(project_api): project = next(iter(project_api.projects)) routing = ActiveProjectRouting("logical-project", "dbname=physical-business", project_id=UUID(project)) store = PostgresStore(SampleSettings(db_name="unrelated"), routing=routing) assert store.settings.db_name == "physical-business" with store.connection(readonly=True): pass with pytest.raises(SampleFailure) as failure: with store.connection(readonly=True, timeseries=True): pass assert failure.value.code == "PROJECT_STORAGE_NOT_CONFIGURED" assert [connection.database for connection in project_api.connections] == ["physical-business"]