"""Cross-Tenant Security Tests v2 — RLS enforcement with unprivileged DB roles.

These tests verify that PostgreSQL Row Level Security (RLS) actually blocks
cross-tenant access when using the unprivileged ``crm_api`` role
(NOSUPERUSER, NOBYPASSRLS, not table owner).

Unlike v1 tests which run as ``crm_user`` (superuser, RLS bypassed),
these tests connect as ``crm_api`` to verify RLS enforcement at the DB level.

Test matrix:
- No tenant context  → SELECT returns 0 rows, INSERT fails
- Tenant A context   → only Tenant A rows visible, Tenant B insert blocked
- Tenant B context   → only Tenant B rows visible, Tenant A insert blocked
- Cross-tenant write → WITH CHECK blocks wrong tenant_id on INSERT/UPDATE

Requirements:
- PostgreSQL with RLS enabled
- ``crm_api`` role (NOSUPERUSER, NOBYPASSRLS)
- ``crm_api`` has SELECT/INSERT/UPDATE/DELETE on tenant tables
- ``crm_api`` is NOT the table owner
"""

from __future__ import annotations

import os
import uuid

import pytest
import pytest_asyncio
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

# Set test environment BEFORE any app imports
os.environ["SESSION_COOKIE_SECURE"] = "false"
os.environ["SESSION_COOKIE_SAMESITE"] = "lax"
os.environ["ENVIRONMENT"] = "testing"
os.environ.setdefault("SECRET_KEY", "test-secret-key-with-at-least-32-characters-for-testing-only!!")

from app.core.db import set_tenant_context
from app.models.contact import Contact
from app.models.tenant import Tenant
from app.models.user import User, UserTenant


# Use the unprivileged crm_api role for RLS testing
# Falls back to leocrm user if crm_api is not available (local dev)
_API_DB_URL = os.environ.get(
    "RLS_TEST_DB_URL",
    "postgresql+asyncpg://crm_api:crm_api_password@localhost:5432/leocrm_test",
)

# Superuser URL for setup (creating tenants, users, etc.)
_ADMIN_DB_URL = os.environ.get(
    "RLS_TEST_ADMIN_DB_URL",
    "postgresql+asyncpg://postgres@localhost:5432/leocrm_test",
)


def _skip_if_no_rls_role():
    """Skip tests if the unprivileged RLS role is not available."""
    try:
        import asyncio
        eng = create_async_engine(_API_DB_URL, echo=False)
        async def _check():
            async with eng.connect() as conn:
                await conn.execute(text("SELECT 1"))
        asyncio.get_event_loop().run_until_complete(_check())
        eng.dispose()
        return False
    except Exception:
        eng.dispose()
        return True


_skip_reason = (
    "Unprivileged crm_api role not available for RLS testing. "
    "Set RLS_TEST_DB_URL to a connection string using a NOSUPERUSER/NOBYPASSRLS role."
)


@pytest_asyncio.fixture
async def admin_engine():
    """Admin engine for setup (creates tenants, users, contacts)."""
    eng = create_async_engine(_ADMIN_DB_URL, echo=False)
    yield eng
    await eng.dispose()


@pytest_asyncio.fixture
async def api_engine():
    """Unprivileged engine using crm_api role — RLS enforced."""
    eng = create_async_engine(_API_DB_URL, echo=False)
    yield eng
    await eng.dispose()


@pytest_asyncio.fixture
async def admin_session(admin_engine):
    """Admin session for data setup."""
    async with admin_engine.connect() as conn:
        await conn.begin()
        session = AsyncSession(bind=conn, expire_on_commit=False)
        yield session
        await session.rollback()
        await conn.rollback()


@pytest_asyncio.fixture
async def api_session(api_engine):
    """Unprivileged session using crm_api — RLS enforced."""
    async with api_engine.connect() as conn:
        await conn.begin()
        session = AsyncSession(bind=conn, expire_on_commit=False)
        yield session
        await session.rollback()
        await conn.rollback()


@pytest_asyncio.fixture
async def seed_data(admin_session: AsyncSession):
    """Seed two tenants with contacts using admin (superuser) connection."""
    from app.core.auth import hash_password

    tenant_a = Tenant(id=uuid.uuid4(), name="RLS Tenant A", slug=f"rls-a-{uuid.uuid4().hex[:8]}")
    tenant_b = Tenant(id=uuid.uuid4(), name="RLS Tenant B", slug=f"rls-b-{uuid.uuid4().hex[:8]}")
    admin_session.add_all([tenant_a, tenant_b])
    await admin_session.flush()

    user_a = User(
        id=uuid.uuid4(),
        name="RLS User A",
        email=f"rls-a-{uuid.uuid4().hex[:8]}@test.local",
        password_hash=hash_password("TestPass123!"),
        is_active=True,
        is_system_admin=False,
    )
    user_b = User(
        id=uuid.uuid4(),
        name="RLS User B",
        email=f"rls-b-{uuid.uuid4().hex[:8]}@test.local",
        password_hash=hash_password("TestPass123!"),
        is_active=True,
        is_system_admin=False,
    )
    admin_session.add_all([user_a, user_b])
    await admin_session.flush()

    ut_a = UserTenant(user_id=user_a.id, tenant_id=tenant_a.id, role="admin", status="active", is_default=True)
    ut_b = UserTenant(user_id=user_b.id, tenant_id=tenant_b.id, role="admin", status="active", is_default=True)
    admin_session.add_all([ut_a, ut_b])
    await admin_session.flush()

    contact_a = Contact(
        id=uuid.uuid4(),
        tenant_id=tenant_a.id,
        firstname="RLS",
        surname="Alpha",
        email_1=f"rls-alpha-{uuid.uuid4().hex[:8]}@contact.local",
        owner_id=user_a.id,
        created_by=user_a.id,
        updated_by=user_a.id,
    )
    contact_b = Contact(
        id=uuid.uuid4(),
        tenant_id=tenant_b.id,
        firstname="RLS",
        surname="Beta",
        email_1=f"rls-beta-{uuid.uuid4().hex[:8]}@contact.local",
        owner_id=user_b.id,
        created_by=user_b.id,
        updated_by=user_b.id,
    )
    admin_session.add_all([contact_a, contact_b])
    await admin_session.flush()

    return {
        "tenant_a": tenant_a,
        "tenant_b": tenant_b,
        "user_a": user_a,
        "user_b": user_b,
        "contact_a": contact_a,
        "contact_b": contact_b,
    }


# ── RLS Enforcement Tests with Unprivileged Role ─────────────────────────────


@pytest.mark.asyncio
@pytest.mark.skipif(_skip_if_no_rls_role(), reason=_skip_reason)
async def test_rls_no_tenant_context_returns_zero_rows(api_session: AsyncSession, seed_data):
    """Without tenant context, SELECT on tenant table must return 0 rows."""
    result = await api_session.execute(
        text("SELECT count(*) FROM contacts WHERE deleted_at IS NULL")
    )
    count = result.scalar()
    assert count == 0, f"RLS fail-open: {count} rows visible without tenant context!"


@pytest.mark.asyncio
@pytest.mark.skipif(_skip_if_no_rls_role(), reason=_skip_reason)
async def test_rls_tenant_a_sees_only_own_rows(api_session: AsyncSession, seed_data):
    """With tenant A context, only tenant A contacts are visible."""
    tenant_a = seed_data["tenant_a"]
    tenant_b = seed_data["tenant_b"]

    await set_tenant_context(api_session, tenant_a.id)
    result = await api_session.execute(
        text("SELECT tenant_id FROM contacts WHERE deleted_at IS NULL")
    )
    rows = result.fetchall()
    for row in rows:
        assert row[0] == str(tenant_a.id), \
            f"RLS leak: tenant A context shows row from {row[0]}"
    # Tenant B's contact must not be visible
    tenant_b_ids = [r[0] for r in rows if r[0] == str(tenant_b.id)]
    assert len(tenant_b_ids) == 0, "RLS failed: Tenant B data visible in Tenant A context!"


@pytest.mark.asyncio
@pytest.mark.skipif(_skip_if_no_rls_role(), reason=_skip_reason)
async def test_rls_tenant_b_sees_only_own_rows(api_session: AsyncSession, seed_data):
    """With tenant B context, only tenant B contacts are visible."""
    tenant_a = seed_data["tenant_a"]
    tenant_b = seed_data["tenant_b"]

    await set_tenant_context(api_session, tenant_b.id)
    result = await api_session.execute(
        text("SELECT tenant_id FROM contacts WHERE deleted_at IS NULL")
    )
    rows = result.fetchall()
    for row in rows:
        assert row[0] == str(tenant_b.id), \
            f"RLS leak: tenant B context shows row from {row[0]}"
    tenant_a_ids = [r[0] for r in rows if r[0] == str(tenant_a.id)]
    assert len(tenant_a_ids) == 0, "RLS failed: Tenant A data visible in Tenant B context!"


@pytest.mark.asyncio
@pytest.mark.skipif(_skip_if_no_rls_role(), reason=_skip_reason)
async def test_rls_blocks_cross_tenant_insert(api_session: AsyncSession, seed_data):
    """RLS WITH CHECK must block INSERT with wrong tenant_id."""
    tenant_a = seed_data["tenant_a"]
    tenant_b = seed_data["tenant_b"]
    user_a = seed_data["user_a"]

    await set_tenant_context(api_session, tenant_a.id)

    # Try to insert a contact with tenant B's ID while in tenant A context
    new_id = uuid.uuid4()
    await api_session.execute(
        text(
            "INSERT INTO contacts (id, tenant_id, firstname, surname, email_1, "
            "owner_id, created_by, updated_by, type, displayname) "
            "VALUES (:id, :tenant_id, :firstname, :surname, :email, :owner, :creator, :updater, :ctype, :dname)"
        ),
        {
            "id": str(new_id),
            "tenant_id": str(tenant_b.id),  # Wrong tenant!
            "firstname": "Cross",
            "surname": "Tenant",
            "email": f"cross-{uuid.uuid4().hex[:8]}@test.local",
            "owner": str(user_a.id),
            "creator": str(user_a.id),
            "updater": str(user_a.id),
            "ctype": "person",
            "dname": "Cross Tenant",
        },
    )

    # The INSERT should fail due to RLS WITH CHECK
    with pytest.raises(Exception) as exc_info:
        await api_session.flush()
    assert "row level security" in str(exc_info.value).lower() or "rls" in str(exc_info.value).lower(), \
        f"Expected RLS error, got: {exc_info.value}"
    await api_session.rollback()


@pytest.mark.asyncio
@pytest.mark.skipif(_skip_if_no_rls_role(), reason=_skip_reason)
async def test_rls_blocks_cross_tenant_update(api_session: AsyncSession, seed_data):
    """RLS must block UPDATE of tenant B's row from tenant A context."""
    tenant_a = seed_data["tenant_a"]
    contact_b = seed_data["contact_b"]

    await set_tenant_context(api_session, tenant_a.id)

    # Try to update tenant B's contact from tenant A context
    result = await api_session.execute(
        text("UPDATE contacts SET surname = 'Hacked' WHERE id = :id"),
        {"id": str(contact_b.id)},
    )
    # Should affect 0 rows (RLS hides tenant B's row from tenant A context)
    assert result.rowcount == 0, \
        f"RLS failed: UPDATE affected {result.rowcount} rows in cross-tenant context!"


@pytest.mark.asyncio
@pytest.mark.skipif(_skip_if_no_rls_role(), reason=_skip_reason)
async def test_rls_blocks_cross_tenant_delete(api_session: AsyncSession, seed_data):
    """RLS must block DELETE of tenant B's row from tenant A context."""
    tenant_a = seed_data["tenant_a"]
    contact_b = seed_data["contact_b"]

    await set_tenant_context(api_session, tenant_a.id)

    result = await api_session.execute(
        text("DELETE FROM contacts WHERE id = :id"),
        {"id": str(contact_b.id)},
    )
    assert result.rowcount == 0, \
        f"RLS failed: DELETE affected {result.rowcount} rows in cross-tenant context!"


@pytest.mark.asyncio
@pytest.mark.skipif(_skip_if_no_rls_role(), reason=_skip_reason)
async def test_rls_tenant_a_insert_own_succeeds(api_session: AsyncSession, seed_data):
    """RLS allows INSERT with correct tenant_id in tenant A context."""
    tenant_a = seed_data["tenant_a"]
    user_a = seed_data["user_a"]

    await set_tenant_context(api_session, tenant_a.id)

    new_id = uuid.uuid4()
    await api_session.execute(
        text(
            "INSERT INTO contacts (id, tenant_id, firstname, surname, email_1, "
            "owner_id, created_by, updated_by, type, displayname) "
            "VALUES (:id, :tenant_id, :firstname, :surname, :email, :owner, :creator, :updater, :ctype, :dname)"
        ),
        {
            "id": str(new_id),
            "tenant_id": str(tenant_a.id),  # Correct tenant!
            "firstname": "Own",
            "surname": "Tenant",
            "email": f"own-{uuid.uuid4().hex[:8]}@test.local",
            "owner": str(user_a.id),
            "creator": str(user_a.id),
            "updater": str(user_a.id),
            "ctype": "person",
            "dname": "Own Tenant",
        },
    )
    await api_session.flush()
    # Verify the row is visible
    result = await api_session.execute(
        text("SELECT id FROM contacts WHERE id = :id"),
        {"id": str(new_id)},
    )
    assert result.fetchone() is not None, "RLS blocked valid same-tenant INSERT!"
    await api_session.rollback()


@pytest.mark.asyncio
@pytest.mark.skipif(_skip_if_no_rls_role(), reason=_skip_reason)
async def test_rls_role_is_not_superuser(api_session: AsyncSession):
    """Verify the test role is not superuser and cannot bypass RLS."""
    result = await api_session.execute(
        text(
            "SELECT rolsuper, rolbypassrls FROM pg_roles WHERE rolname = current_user"
        )
    )
    row = result.fetchone()
    assert row is not None, "Could not query role properties"
    assert row[0] is False, f"Test role {row} is SUPERUSER — RLS tests are meaningless!"
    assert row[1] is False, f"Test role {row} has BYPASSRLS — RLS tests are meaningless!"


@pytest.mark.asyncio
@pytest.mark.skipif(_skip_if_no_rls_role(), reason=_skip_reason)
async def test_rls_role_is_not_table_owner(api_session: AsyncSession):
    """Verify the test role is not the owner of tenant tables."""
    result = await api_session.execute(
        text(
            "SELECT tableowner FROM pg_tables WHERE schemaname='public' AND tablename='contacts'"
        )
    )
    owner = result.scalar()
    current_user_result = await api_session.execute(text("SELECT current_user"))
    current_user = current_user_result.scalar()
    assert owner != current_user, \
        f"Test role '{current_user}' owns contacts table — RLS is bypassed for owners!"


@pytest.mark.asyncio
@pytest.mark.skipif(_skip_if_no_rls_role(), reason=_skip_reason)
async def test_rls_no_bootstrap_fallback_policy(api_session: AsyncSession):
    """Verify no fail-open/bootstrap RLS policy exists on tenant tables.

    A fail-open policy would allow access when tenant context is missing.
    This test checks that no policy uses IS NULL, = '', or COALESCE patterns
    that would grant access without a tenant context.
    """
    result = await api_session.execute(
        text("""
            SELECT tablename, policyname, qual, with_check
            FROM pg_policies
            WHERE schemaname = 'public'
            AND tablename IN ('contacts', 'tasks', 'workspaces', 'files')
            AND (
                qual ILIKE '%IS NULL%'
                OR qual ILIKE '%= ''%'
                OR qual ILIKE '%COALESCE%'
                OR with_check ILIKE '%IS NULL%'
                OR with_check ILIKE '%= ''%'
                OR with_check ILIKE '%COALESCE%'
            )
        """)
    )
    bad_policies = result.fetchall()
    assert len(bad_policies) == 0, \
        f"Fail-open RLS policies found: {bad_policies}"