"""跨租户隔离 —— 本仓库**最重要的测试**（AGENTS.md 2.5）。

三层验证，任何一层失效都算事故：
1. 接口层：A 租户的令牌打不到 B 租户的资源。
2. 数据层：应用角色连库、绑定 A 的上下文，物理上看不到 B 的行（RLS 生效）。
3. 配置层：RLS 确实 ENABLE，且应用角色不是 superuser / BYPASSRLS（否则 RLS 被跳过）。
"""

from __future__ import annotations

import uuid

import pytest
from httpx import AsyncClient
from sqlalchemy import text
from sqlalchemy.exc import ProgrammingError

from app.db import SessionFactory, bind_tenant, current_app_role
from app.models import TENANT_TABLES

from .conftest import create_workspace, login, unwrap


async def _two_tenants(client: AsyncClient):
    alice = await login(client, "13900000001")
    bob = await login(client, "13900000002")
    ws_alice = await create_workspace(client, alice, "Alice 的团队")
    ws_bob = await create_workspace(client, bob, "Bob 的团队")
    return alice, bob, ws_alice, ws_bob


async def test_cannot_read_other_workspace(client: AsyncClient) -> None:
    alice, _bob, _ws_a, ws_b = await _two_tenants(client)
    resp = await client.get(f"/api/v1/workspaces/{ws_b['id']}", headers=alice["headers"])
    assert resp.status_code == 403
    assert resp.json()["code"] == 3003


async def test_cannot_list_other_workspace_members(client: AsyncClient) -> None:
    alice, _bob, _ws_a, ws_b = await _two_tenants(client)
    resp = await client.get(f"/api/v1/workspaces/{ws_b['id']}/members", headers=alice["headers"])
    assert resp.status_code == 403


async def test_cannot_invite_into_other_workspace(client: AsyncClient) -> None:
    alice, _bob, _ws_a, ws_b = await _two_tenants(client)
    resp = await client.post(
        f"/api/v1/workspaces/{ws_b['id']}/invitations",
        json={"role": "member"},
        headers=alice["headers"],
    )
    assert resp.status_code == 403


async def test_my_workspace_list_is_scoped(client: AsyncClient) -> None:
    alice, _bob, ws_a, ws_b = await _two_tenants(client)
    mine = unwrap(await client.get("/api/v1/workspaces", headers=alice["headers"]))
    ids = [w["id"] for w in mine["items"]]
    assert ids == [ws_a["id"]]
    assert ws_b["id"] not in ids


async def test_rls_hides_rows_at_database_level(client: AsyncClient) -> None:
    _alice, _bob, ws_a, ws_b = await _two_tenants(client)
    ws_a_id = uuid.UUID(ws_a["id"])
    ws_b_id = uuid.UUID(ws_b["id"])

    async with SessionFactory() as session, session.begin():
        await bind_tenant(session, ws_a_id, None)
        visible = (await session.execute(text("SELECT id FROM workspace"))).scalars().all()
        member_ids = (
            (await session.execute(text("SELECT workspace_id FROM member"))).scalars().all()
        )

    assert [str(v) for v in visible] == [ws_a["id"]]
    assert set(member_ids) == {ws_a_id}
    assert ws_b_id not in set(member_ids)


async def test_rls_without_context_returns_nothing(client: AsyncClient) -> None:
    await _two_tenants(client)
    async with SessionFactory() as session, session.begin():
        count = (await session.execute(text("SELECT count(*) FROM workspace"))).scalar_one()
    assert count == 0, "未绑定租户上下文时不应看到任何行"


async def test_rls_enabled_on_every_tenant_table(client: AsyncClient) -> None:
    async with SessionFactory() as session:
        rows = (
            await session.execute(
                text(
                    "SELECT relname, relrowsecurity FROM pg_class "
                    "WHERE relname = ANY(:names) AND relkind = 'r'"
                ),
                {"names": list(TENANT_TABLES)},
            )
        ).all()
    status = dict(rows)
    assert set(status) == set(TENANT_TABLES), "有租户表没建出来"
    assert all(status.values()), f"未启用 RLS 的表：{[k for k, v in status.items() if not v]}"


async def test_app_role_cannot_bypass_rls(client: AsyncClient) -> None:
    async with SessionFactory() as session, session.begin():
        role = await current_app_role(session)
    assert role["is_super"] is False, "应用角色绝不能是 superuser，否则 RLS 被跳过"
    assert role["bypass_rls"] is False, "应用角色不能带 BYPASSRLS"
    assert role["read_all"] is False


async def test_app_role_has_no_ddl(client: AsyncClient) -> None:
    """应用角色不能建表——否则一次 SQL 注入就能改结构。"""
    async with SessionFactory() as session, session.begin():
        with pytest.raises(ProgrammingError):  # 权限不足，DDL 必须失败
            await session.execute(text("CREATE TABLE should_not_exist (id int)"))