"""Covers the Workspace feature (app/routers/workspaces.py): dynamic Postgres database provisioning via app/core/db_provisioning.py, credential encryption at rest (app/core/crypto.py), and the per-workspace engine registry (app/core/workspace_engines.py) via the "Test connection" action. Provisioning tests run against the SAME local Postgres server the test suite's own DATABASE_URL points at (settings.provisioning_superuser is "postgres"/"postgres" by default, a real superuser on a local dev instance) — each test creates its own uniquely-named throwaway database/ role and tears it down via DELETE ?drop_database=true (exercising that endpoint) or, on failure paths where nothing was ever created, there is nothing to tear down.""" import json import uuid from datetime import datetime, timezone import asyncpg import pytest from sqlalchemy import text from sqlalchemy.engine import make_url from app.core.auth import create_access_token from app.core.config import settings from app.core.crypto import decrypt_secret, encrypt_secret from tests.conftest import AsyncSessionLocal from app.core.db_provisioning import ProvisioningError, validate_identifier from app.models.identity import Role as RoleModel, Tenant, User, UserRole # The server the suite's own database is on (CI's is a service named "db", not localhost). _LOCAL_HOST = make_url(settings.database_url).host or "localhost" _LOCAL_PORT = make_url(settings.database_url).port or 5432 def _unique(prefix: str) -> str: return f"{prefix}_{uuid.uuid4().hex[:8]}" async def _vendor_token(tenant_id: uuid.UUID) -> str: """A vendor user holding an is_system_role role — the strongest possible action_matrix a caller could have — to prove deny_vendor blocks on vendor_id alone, independent of permissions (same shape as test_vendor_portal.py's own helper).""" async with AsyncSessionLocal() as db: async with db.begin(): role = RoleModel(tenant_id=tenant_id, name="Vendor Super Role", is_system_role=True, action_matrix={}) db.add(role) await db.flush() user = User(tenant_id=tenant_id, name="Vendor", email=f"vendor-{uuid.uuid4()}@example.com", status="active") db.add(user) await db.flush() db.add(UserRole(user_id=user.id, role_id=role.id)) await db.flush() user_id = user.id return create_access_token(tenant_id=tenant_id, user_id=user_id, vendor_id=uuid.uuid4()) async def _no_permission_token(tenant_id: uuid.UUID) -> str: """A non-vendor user with a real (non-system) role that grants nothing — proves require_permission("workspaces", ...) itself is enforced, not just deny_vendor.""" async with AsyncSessionLocal() as db: async with db.begin(): role = RoleModel(tenant_id=tenant_id, name="No Access Role", action_matrix={}) db.add(role) await db.flush() user = User(tenant_id=tenant_id, name="Limited", email=f"limited-{uuid.uuid4()}@example.com", status="active") db.add(user) await db.flush() db.add(UserRole(user_id=user.id, role_id=role.id)) await db.flush() user_id = user.id return create_access_token(tenant_id=tenant_id, user_id=user_id, vendor_id=None) # ---- pure unit coverage ----------------------------------------------------------------------------------- def test_validate_identifier_accepts_valid_names(): validate_identifier("acme_staging", "db_name") # does not raise validate_identifier("a12", "db_name") @pytest.mark.parametrize( "bad_name", ["Bad-Name", "1starts_with_digit", "has space", "has'quote", "UPPERCASE", "a" * 64, "ab", "drop table x;--"], ) def test_validate_identifier_rejects_invalid_names(bad_name): with pytest.raises(ProvisioningError): validate_identifier(bad_name, "db_name") def test_encrypt_decrypt_round_trip(): ciphertext = encrypt_secret("s3cret-P@ss") assert ciphertext != "s3cret-P@ss" assert decrypt_secret(ciphertext) == "s3cret-P@ss" # ---- permission gating ------------------------------------------------------------------------------------- async def test_vendor_cannot_list_workspaces(client, test_tenant): token = await _vendor_token(test_tenant["tenant_id"]) resp = await client.get("/workspaces", headers={"Authorization": f"Bearer {token}"}) assert resp.status_code == 403 async def test_vendor_cannot_create_workspace(client, test_tenant): token = await _vendor_token(test_tenant["tenant_id"]) resp = await client.post( "/workspaces", headers={"Authorization": f"Bearer {token}"}, json={"name": "x", "db_host": _LOCAL_HOST, "db_port": _LOCAL_PORT, "db_name": "x", "db_username": "x", "db_password": "x"}, ) assert resp.status_code == 403 async def test_role_without_workspaces_permission_cannot_create(client, test_tenant): token = await _no_permission_token(test_tenant["tenant_id"]) resp = await client.post( "/workspaces", headers={"Authorization": f"Bearer {token}"}, json={"name": "x", "db_host": _LOCAL_HOST, "db_port": _LOCAL_PORT, "db_name": "x", "db_username": "x", "db_password": "x"}, ) assert resp.status_code == 403 # ---- real provisioning end-to-end ------------------------------------------------------------------------- async def test_create_workspace_provisions_real_database(client, auth_headers, test_tenant): db_name = _unique("pytest_ws") db_username = _unique("pytest_ws_role") resp = await client.post( "/workspaces", headers=auth_headers, json={ "name": "Pytest Workspace", "db_host": _LOCAL_HOST, "db_port": _LOCAL_PORT, "db_name": db_name, "db_username": db_username, "db_password": "Pytest-Pass123", }, ) assert resp.status_code == 201 body = resp.json() assert body["status"] == "active" assert "db_password" not in body and "db_password_encrypted" not in body workspace_id = body["id"] # The provisioned database is real — confirm independently of the app's # own registry, via a direct connection as the new role itself. conn = await asyncpg.connect(host=_LOCAL_HOST, port=_LOCAL_PORT, database=db_name, user=db_username, password="Pytest-Pass123") try: count = await conn.fetchval("SELECT count(*) FROM information_schema.tables WHERE table_schema = 'public'") assert count > 30 # the full db/schema.sql table set, not an empty database finally: await conn.close() test_resp = await client.post(f"/workspaces/{workspace_id}/test-connection", headers=auth_headers) assert test_resp.status_code == 200 assert test_resp.json()["success"] is True delete_resp = await client.delete(f"/workspaces/{workspace_id}?drop_database=true", headers=auth_headers) assert delete_resp.status_code == 204 # Actually gone — a fresh connection attempt as the dropped role fails. with pytest.raises(Exception): await asyncpg.connect(host=_LOCAL_HOST, port=_LOCAL_PORT, database=db_name, user=db_username, password="Pytest-Pass123") async def test_create_workspace_rejects_invalid_db_name(client, auth_headers, test_tenant): resp = await client.post( "/workspaces", headers=auth_headers, json={ "name": "Bad Name Workspace", "db_host": _LOCAL_HOST, "db_port": _LOCAL_PORT, "db_name": "Not-A-Valid-Identifier!", "db_username": _unique("pytest_ws_role"), "db_password": "whatever123", }, ) # A tracking row is still created (201) — its status reports the failure # rather than the whole request 4xx/5xx-ing and losing the record. assert resp.status_code == 201 body = resp.json() assert body["status"] == "failed" assert "identifier" in body["status_detail"] async def test_create_workspace_duplicate_db_name_fails_cleanly(client, auth_headers, test_tenant): db_name = _unique("pytest_ws_dup") first_username = _unique("pytest_ws_role") second_username = _unique("pytest_ws_role") first = await client.post( "/workspaces", headers=auth_headers, json={ "name": "First", "db_host": _LOCAL_HOST, "db_port": _LOCAL_PORT, "db_name": db_name, "db_username": first_username, "db_password": "Pytest-Pass123", }, ) assert first.json()["status"] == "active" second = await client.post( "/workspaces", headers=auth_headers, json={ "name": "Second", "db_host": _LOCAL_HOST, "db_port": _LOCAL_PORT, "db_name": db_name, "db_username": second_username, "db_password": "Pytest-Pass123", }, ) assert second.status_code == 201 assert second.json()["status"] == "failed" # Clean up the one database that did get created. cleanup = await client.delete(f"/workspaces/{first.json()['id']}?drop_database=true", headers=auth_headers) assert cleanup.status_code == 204 # ---- copy_tenant_data ---------------------------------------------------------------------------------------- async def _create_master_with_record(client, auth_headers, name: str) -> tuple[str, str]: resp = await client.post( "/masters", headers=auth_headers, json={"name": name, "field_schema": [{"name": "revenue", "type": "number"}]}, ) master_id = resp.json()["id"] record = await client.post( f"/masters/{master_id}/records", headers=auth_headers, json={"field_values": {"revenue": 42}} ) return master_id, record.json()["id"] async def _create_workspace(client, auth_headers, *, copy_tenant_data: bool) -> tuple[dict, str, str]: db_name = _unique("pytest_ws_copy") db_username = _unique("pytest_ws_role") resp = await client.post( "/workspaces", headers=auth_headers, json={ "name": "Copy Test Workspace", "db_host": _LOCAL_HOST, "db_port": _LOCAL_PORT, "db_name": db_name, "db_username": db_username, "db_password": "Pytest-Pass123", "copy_tenant_data": copy_tenant_data, }, ) return resp.json(), db_name, db_username async def _connect_as_superuser(db_name: str) -> asyncpg.Connection: """Verification queries below read RLS-protected application tables (masters, users, ...) — connecting as the granted app role would hit Postgres's "unrecognized configuration parameter" error, since RLS's `current_setting('app.current_tenant_id')` (no missing_ok fallback — see db/schema.sql) throws on a session that's never had it SET, and these tests never call SET on their throwaway connection. The superuser bypasses RLS entirely, which is exactly what a "did the copy land" check wants anyway.""" return await asyncpg.connect( host=_LOCAL_HOST, port=_LOCAL_PORT, database=db_name, user=settings.provisioning_superuser, password=settings.provisioning_superuser_password, ) async def test_create_workspace_with_copy_tenant_data_duplicates_rows(client, auth_headers, test_tenant): master_id, record_id = await _create_master_with_record(client, auth_headers, "Copy Source Master") body, db_name, db_username = await _create_workspace(client, auth_headers, copy_tenant_data=True) assert body["status"] == "active" conn = await _connect_as_superuser(db_name) try: tenant_row = await conn.fetchrow("SELECT id FROM tenants WHERE id = $1", test_tenant["tenant_id"]) assert tenant_row is not None master_row = await conn.fetchrow("SELECT id, name FROM masters WHERE id = $1", uuid.UUID(master_id)) assert master_row is not None assert master_row["name"] == "Copy Source Master" record_row = await conn.fetchrow("SELECT id, field_values FROM master_records WHERE id = $1", uuid.UUID(record_id)) assert record_row is not None # Raw asyncpg (unlike SQLAlchemy's own dialect, which is what the # actual copy code uses) doesn't auto-decode jsonb to a dict. assert json.loads(record_row["field_values"]) == {"revenue": 42} # The admin user's own role assignment — proves the no-tenant_id, # JOIN-filtered user_roles path copied correctly. role_count = await conn.fetchval( "SELECT count(*) FROM user_roles ur JOIN users u ON u.id = ur.user_id WHERE u.id = $1", test_tenant["admin_id"], ) assert role_count == 1 finally: await conn.close() await client.delete(f"/workspaces/{body['id']}?drop_database=true", headers=auth_headers) async def test_copy_tenant_data_does_not_leak_other_tenants(client, auth_headers, test_tenant): """A second, unrelated tenant's data must never show up in the first tenant's copy — proves the copy is scoped by tenant_id, not a blind dump of the whole shared database.""" async with AsyncSessionLocal() as db: async with db.begin(): other_tenant = Tenant(name=f"Other Tenant {uuid.uuid4().hex[:6]}", status="active", license_set_at=datetime.now(timezone.utc)) db.add(other_tenant) await db.flush() other_role = RoleModel(tenant_id=other_tenant.id, name="Admin", is_system_role=True, action_matrix={}) db.add(other_role) await db.flush() other_admin = User(tenant_id=other_tenant.id, name="Other Admin", email=f"other-{uuid.uuid4()}@example.com", status="active") db.add(other_admin) await db.flush() db.add(UserRole(user_id=other_admin.id, role_id=other_role.id)) await db.flush() other_tenant_id = other_tenant.id other_token = create_access_token(tenant_id=other_tenant_id, user_id=other_admin.id, vendor_id=None) other_headers = {"Authorization": f"Bearer {other_token}"} _other_master_id, _other_record_id = await _create_master_with_record(client, other_headers, "Other Tenant Master") await _create_master_with_record(client, auth_headers, "My Own Master") body, db_name, db_username = await _create_workspace(client, auth_headers, copy_tenant_data=True) assert body["status"] == "active" conn = await _connect_as_superuser(db_name) try: names = {r["name"] for r in await conn.fetch("SELECT name FROM masters")} assert names == {"My Own Master"} other_tenant_row = await conn.fetchrow("SELECT id FROM tenants WHERE id = $1", other_tenant_id) assert other_tenant_row is None finally: await conn.close() await client.delete(f"/workspaces/{body['id']}?drop_database=true", headers=auth_headers) async with AsyncSessionLocal() as db: async with db.begin(): await db.execute(text("DELETE FROM user_roles WHERE user_id = :uid"), {"uid": str(other_admin.id)}) await db.execute(text("DELETE FROM master_records WHERE tenant_id = :tid"), {"tid": str(other_tenant_id)}) await db.execute(text("DELETE FROM masters WHERE tenant_id = :tid"), {"tid": str(other_tenant_id)}) await db.execute(text("DELETE FROM audit_log_entries WHERE tenant_id = :tid"), {"tid": str(other_tenant_id)}) await db.execute(text("DELETE FROM users WHERE tenant_id = :tid"), {"tid": str(other_tenant_id)}) await db.execute(text("DELETE FROM roles WHERE tenant_id = :tid"), {"tid": str(other_tenant_id)}) await db.execute(text("DELETE FROM tenants WHERE id = :tid"), {"tid": str(other_tenant_id)}) async def test_create_workspace_without_copy_tenant_data_stays_empty(client, auth_headers, test_tenant): await _create_master_with_record(client, auth_headers, "Should Not Be Copied") body, db_name, db_username = await _create_workspace(client, auth_headers, copy_tenant_data=False) assert body["status"] == "active" conn = await _connect_as_superuser(db_name) try: assert await conn.fetchval("SELECT count(*) FROM tenants") == 0 assert await conn.fetchval("SELECT count(*) FROM masters") == 0 finally: await conn.close() await client.delete(f"/workspaces/{body['id']}?drop_database=true", headers=auth_headers) async def test_copy_tenant_data_failure_rolls_back_whole_workspace(client, auth_headers, test_tenant, monkeypatch): import app.core.db_provisioning as provisioning monkeypatch.setattr(provisioning, "_TENANT_DATA_TABLES", ["this_table_does_not_exist"]) body, db_name, db_username = await _create_workspace(client, auth_headers, copy_tenant_data=True) assert body["status"] == "failed" assert "copying tenant data failed" in body["status_detail"].lower() with pytest.raises(Exception): await asyncpg.connect(host=_LOCAL_HOST, port=_LOCAL_PORT, database=db_name, user=db_username, password="Pytest-Pass123") async def test_delete_without_drop_database_leaves_database_intact(client, auth_headers, test_tenant): db_name = _unique("pytest_ws_keep") db_username = _unique("pytest_ws_role") created = await client.post( "/workspaces", headers=auth_headers, json={ "name": "Keep The Database", "db_host": _LOCAL_HOST, "db_port": _LOCAL_PORT, "db_name": db_name, "db_username": db_username, "db_password": "Pytest-Pass123", }, ) assert created.json()["status"] == "active" workspace_id = created.json()["id"] delete_resp = await client.delete(f"/workspaces/{workspace_id}", headers=auth_headers) assert delete_resp.status_code == 204 # The tracking row is gone, but the real database/role are still there. conn = await asyncpg.connect(host=_LOCAL_HOST, port=_LOCAL_PORT, database=db_name, user=db_username, password="Pytest-Pass123") await conn.close() # Now actually clean it up (not via the API — the row no longer exists). admin = await asyncpg.connect(host=_LOCAL_HOST, port=_LOCAL_PORT, database="postgres", user=settings.provisioning_superuser, password=settings.provisioning_superuser_password) try: await admin.execute(f'DROP DATABASE IF EXISTS "{db_name}"') await admin.execute(f'DROP ROLE IF EXISTS "{db_username}"') finally: await admin.close()