"""Pruebas de aislamiento Row-Level Security por ``tenant_id`` y ``company_id``.

Las políticas se crean en la migración
``d1a2b3c4e5f6_enable_rls_tenant_company`` y dependen de las GUCs
``app.tenant_id`` / ``app.company_id`` fijadas por la aplicación.

Como ``postgres`` hace BYPASSRLS por defecto, los tests cambian de rol
dentro de la transacción a un usuario sin ese atributo (``anexo76_rls_test``)
antes de contar filas. Si la conexión de pruebas no tiene privilegios para
crear el rol, el set completo se salta con ``pytest.skip``.
"""

from __future__ import annotations

import uuid
from typing import Iterator

import pytest
from sqlalchemy import text
from sqlalchemy.exc import DBAPIError, ProgrammingError
from sqlalchemy.orm import Session

from tests.conftest import TestingSessionLocal, engine
from tests.fixtures.builders import ensure_tenant_company


TEST_ROLE = "anexo76_rls_test"
RLS_SCHEMAS = ("core", "a76", "a24", "public")


@pytest.fixture(scope="module")
def rls_role() -> str:
    """Crea (idempotente) un rol no-superusuario con los privilegios mínimos
    necesarios para ejercitar las políticas RLS en los tests."""
    try:
        with engine.begin() as conn:
            conn.execute(
                text(
                    f"""
                    DO $$
                    BEGIN
                        IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = '{TEST_ROLE}') THEN
                            CREATE ROLE {TEST_ROLE} NOLOGIN NOBYPASSRLS;
                        END IF;
                    END $$;
                    """
                )
            )
            for schema in RLS_SCHEMAS:
                conn.execute(text(f'GRANT USAGE ON SCHEMA "{schema}" TO {TEST_ROLE}'))
                conn.execute(
                    text(
                        f'GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA "{schema}" TO {TEST_ROLE}'
                    )
                )
                conn.execute(
                    text(
                        f'GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA "{schema}" TO {TEST_ROLE}'
                    )
                )
    except ProgrammingError as exc:
        pytest.skip(f"No hay privilegio para preparar rol de pruebas RLS: {exc}")
    return TEST_ROLE


@pytest.fixture
def rls_session() -> Iterator[Session]:
    """Sesión con transacción externa + rollback (no contamina BD real)."""
    connection = engine.connect()
    transaction = connection.begin()
    session = TestingSessionLocal(bind=connection)
    try:
        yield session
    finally:
        session.close()
        transaction.rollback()
        connection.close()


def _allocate_id() -> int:
    return 1_700_000_000 + (uuid.uuid4().int % 90_000_000)


def _bootstrap_two_tenants(session: Session) -> tuple[tuple[int, int], tuple[int, int]]:
    tid_a, tid_b = _allocate_id(), _allocate_id()
    cid_a, cid_b = _allocate_id(), _allocate_id()
    ensure_tenant_company(session, tenant_id=tid_a, company_id=cid_a)
    ensure_tenant_company(session, tenant_id=tid_b, company_id=cid_b)
    return (tid_a, cid_a), (tid_b, cid_b)


def _seed_clients(session: Session, tenant_id: int, company_id: int, *, count: int) -> None:
    """Inserta filas en ``a76.clients_and_providers`` con SQL crudo para
    aislarnos del ORM y medir filtrado puro a nivel de BD."""
    for i in range(count):
        session.execute(
            text(
                """
                INSERT INTO a76.clients_and_providers
                    (tenant_id, company_id, name, client_or_provider, is_active)
                VALUES
                    (:tid, :cid, :name, 'BOTH', true)
                """
            ),
            {"tid": tenant_id, "cid": company_id, "name": f"Seed {tenant_id}/{company_id}/{i}"},
        )
    session.flush()


def _count_under_role(
    session: Session,
    tenant_id: int | None,
    company_id: int | None,
    *,
    role: str,
    table: str = "a76.clients_and_providers",
) -> int:
    """Cuenta filas tras cambiar a un rol sin BYPASSRLS y fijar el contexto."""
    session.execute(text(f"SET LOCAL ROLE {role}"))
    session.execute(
        text("SELECT set_config('app.tenant_id', :t, true), set_config('app.company_id', :c, true)"),
        {
            "t": "" if tenant_id is None else str(tenant_id),
            "c": "" if company_id is None else str(company_id),
        },
    )
    try:
        return int(session.execute(text(f"SELECT count(*) FROM {table}")).scalar() or 0)
    finally:
        session.execute(text("RESET ROLE"))


def test_tenant_isolation_hides_rows_from_other_tenant(
    rls_session: Session, rls_role: str
) -> None:
    (tid_a, cid_a), (tid_b, cid_b) = _bootstrap_two_tenants(rls_session)
    _seed_clients(rls_session, tid_a, cid_a, count=3)
    _seed_clients(rls_session, tid_b, cid_b, count=2)

    visible_a = _count_under_role(rls_session, tid_a, None, role=rls_role)
    visible_b = _count_under_role(rls_session, tid_b, None, role=rls_role)

    assert visible_a == 3, "Tenant A debe ver solo sus 3 filas"
    assert visible_b == 2, "Tenant B debe ver solo sus 2 filas"


def test_company_scope_narrows_within_tenant(
    rls_session: Session, rls_role: str
) -> None:
    (tid_a, cid_a), _ = _bootstrap_two_tenants(rls_session)

    second_company_id = _allocate_id()
    ensure_tenant_company(rls_session, tenant_id=tid_a, company_id=second_company_id)

    _seed_clients(rls_session, tid_a, cid_a, count=4)
    _seed_clients(rls_session, tid_a, second_company_id, count=7)

    without_company = _count_under_role(rls_session, tid_a, None, role=rls_role)
    with_company_a = _count_under_role(rls_session, tid_a, cid_a, role=rls_role)
    with_second = _count_under_role(rls_session, tid_a, second_company_id, role=rls_role)

    assert without_company == 11, "Sin company context el tenant ve ambas compañías"
    assert with_company_a == 4
    assert with_second == 7


def test_missing_tenant_context_returns_zero_rows(
    rls_session: Session, rls_role: str
) -> None:
    (tid_a, cid_a), _ = _bootstrap_two_tenants(rls_session)
    _seed_clients(rls_session, tid_a, cid_a, count=5)

    visible = _count_under_role(rls_session, None, None, role=rls_role)
    assert visible == 0, "Sin app.tenant_id la política debe devolver 0 filas"


def test_insert_violation_respects_tenant_policy(
    rls_session: Session, rls_role: str
) -> None:
    """La cláusula ``WITH CHECK`` debe rechazar inserts fuera de contexto."""
    (tid_a, cid_a), (tid_b, _) = _bootstrap_two_tenants(rls_session)

    rls_session.execute(text(f"SET LOCAL ROLE {rls_role}"))
    rls_session.execute(
        text("SELECT set_config('app.tenant_id', :t, true), set_config('app.company_id', :c, true)"),
        {"t": str(tid_a), "c": str(cid_a)},
    )
    try:
        with pytest.raises(DBAPIError):
            rls_session.execute(
                text(
                    """
                    INSERT INTO a76.clients_and_providers
                        (tenant_id, company_id, name, client_or_provider, is_active)
                    VALUES
                        (:tid, :cid, 'Cross-tenant attempt', 'BOTH', true)
                    """
                ),
                {"tid": tid_b, "cid": cid_a},
            )
    finally:
        rls_session.rollback()
