"""Tenant isolation in the database itself (db/schema.sql's Row-Level
Security section): the API's own role sees only the current tenant's rows
whatever a query forgets to filter, can't write another tenant's rows, and
fails outright with no tenant set. And the catalog checks that keep it so:
every table with a tenant_id has a policy that binds the API's role, and
that role isn't one RLS lets through."""

import uuid

import pytest
from sqlalchemy import text
from sqlalchemy.exc import DBAPIError

from app.core.database import engine
from tests.conftest import AsyncSessionLocal

# Tables with a tenant_id and no policy, on purpose: workers read jobs
# across tenants; a sign-on attempt is looked up before its tenant is known.
NO_POLICY = {"jobs", "sso_states"}
# Policy, not forced: forced RLS would stop Postgres using their JSONB
# indexes. Read only through app/crud/record_sql.py, which scopes them.
NOT_FORCED = {"master_records", "form_submissions"}


async def test_the_api_role_is_one_rls_binds():
    async with engine.connect() as conn:
        row = (await conn.execute(text("SELECT rolsuper, rolbypassrls FROM pg_roles WHERE rolname = current_user"))).one()
    assert not row.rolsuper and not row.rolbypassrls, (
        "The API connects as a superuser or a BYPASSRLS role — row-level security doesn't apply to it. "
        "Connect as an ordinary role (scripts/create_app_role.py)."
    )


async def test_every_tenant_table_has_a_policy_that_binds_the_api():
    async with engine.connect() as conn:
        rows = (
            await conn.execute(
                text(
                    "SELECT c.relname, c.relrowsecurity, c.relforcerowsecurity, "
                    "(SELECT count(*) FROM pg_policy p WHERE p.polrelid = c.oid) AS policies "
                    "FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace "
                    "WHERE n.nspname = 'public' AND c.relkind IN ('r', 'p') AND NOT c.relispartition AND EXISTS ("
                    "  SELECT 1 FROM pg_attribute a WHERE a.attrelid = c.oid AND a.attname = 'tenant_id' AND NOT a.attisdropped)"
                )
            )
        ).all()
    problems = []
    for name, enabled, forced, policies in rows:
        if name in NO_POLICY:
            continue
        if not enabled or not policies:
            problems.append(f"{name}: no row-level security policy")
        elif not forced and name not in NOT_FORCED:
            problems.append(f"{name}: policy not forced (FORCE ROW LEVEL SECURITY)")
    assert not problems, "Add the table to db/schema.sql's Row-Level Security section (and a migration): " + "; ".join(problems)


@pytest.fixture
async def other_tenant():
    tenant_id = uuid.uuid4()
    async with AsyncSessionLocal() as db:
        async with db.begin():
            await db.execute(text("INSERT INTO tenants (id, name, status) VALUES (:id, :n, 'active')"), {"id": tenant_id, "n": f"Pytest Tenant {tenant_id.hex[:8]}"})
            await db.execute(
                text("INSERT INTO connectors (tenant_id, name, type, configuration, status) VALUES (:t, 'Other tenant hook', 'webhook', '{}', 'enabled')"),
                {"t": tenant_id},
            )
    yield tenant_id
    async with AsyncSessionLocal() as db:
        async with db.begin():
            await db.execute(text("DELETE FROM connectors WHERE tenant_id = :t"), {"t": tenant_id})
            await db.execute(text("DELETE FROM tenants WHERE id = :t"), {"t": tenant_id})


async def _as_tenant(conn, tenant_id) -> None:
    await conn.execute(text("SELECT set_config('app.current_tenant_id', :v, true)"), {"v": str(tenant_id)})


async def test_a_query_without_a_tenant_filter_still_sees_only_its_tenant(test_tenant, other_tenant):
    async with engine.connect() as conn:
        async with conn.begin():
            await _as_tenant(conn, test_tenant["tenant_id"])
            # No WHERE tenant_id — the database adds it.
            tenants = {r[0] for r in await conn.execute(text("SELECT DISTINCT tenant_id FROM users"))}
            hooks = (await conn.execute(text("SELECT count(*) FROM connectors WHERE name = 'Other tenant hook'"))).scalar()
    assert tenants == {test_tenant["tenant_id"]}
    assert hooks == 0


async def test_another_tenants_rows_cant_be_written(test_tenant, other_tenant):
    async with engine.connect() as conn:
        async with conn.begin():
            await _as_tenant(conn, test_tenant["tenant_id"])
            # Invisible, so an update or delete touches nothing.
            changed = await conn.execute(text("UPDATE connectors SET name = 'hijacked' WHERE tenant_id = :t"), {"t": other_tenant})
            assert changed.rowcount == 0
        with pytest.raises(DBAPIError, match="row-level security"):
            async with conn.begin():
                await _as_tenant(conn, test_tenant["tenant_id"])
                await conn.execute(
                    text("INSERT INTO connectors (tenant_id, name, type, configuration, status) VALUES (:t, 'x', 'webhook', '{}', 'enabled')"), {"t": other_tenant}
                )
    async with AsyncSessionLocal() as db:
        assert (await db.execute(text("SELECT name FROM connectors WHERE tenant_id = :t"), {"t": other_tenant})).scalar() == "Other tenant hook"


async def test_no_tenant_set_is_an_error_not_every_tenant(test_tenant):
    async with engine.connect() as conn:
        with pytest.raises(DBAPIError):
            async with conn.begin():
                await conn.execute(text("SELECT count(*) FROM users"))