leocrm/tests/test_no_legacy_tenant_var.py

"""CI test: verify no RLS policy uses legacy app.tenant_id variable.

After alembic upgrade head, all RLS policies must use app.current_tenant_id
exclusively. This test fails if any policy in the database still references
the old app.tenant_id variable.
"""

from __future__ import annotations

import os

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

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!!")


_ADMIN_DB_URL = os.environ.get(
    "RLS_TEST_ADMIN_DB_URL",
    "postgresql+asyncpg://postgres@localhost:5432/leocrm_test",
)


def _skip_if_no_db():
    try:
        import asyncio
        eng = create_async_engine(_ADMIN_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


@pytest_asyncio.fixture
async def admin_session():
    eng = create_async_engine(_ADMIN_DB_URL, echo=False)
    async with eng.connect() as conn:
        session = AsyncSession(bind=conn, expire_on_commit=False)
        yield session
        await session.rollback()
        await conn.rollback()
    await eng.dispose()


@pytest.mark.asyncio
@pytest.mark.skipif(_skip_if_no_db(), reason="Test database not available")
async def test_no_policy_uses_legacy_app_tenant_id(admin_session: AsyncSession):
    """No RLS policy should reference the legacy app.tenant_id variable.

    All policies must use app.current_tenant_id exclusively.
    This test runs after alembic upgrade head to verify the final state.
    """
    result = await admin_session.execute(text("""
        SELECT tablename, policyname, qual, with_check
        FROM pg_policies
        WHERE schemaname = 'public'
        AND (
            qual ILIKE '%app.tenant_id%'
            OR with_check ILIKE '%app.tenant_id%'
        )
    """))
    legacy_policies = result.fetchall()
    assert len(legacy_policies) == 0, \
        f"RLS policies still using legacy app.tenant_id: {legacy_policies}"


@pytest.mark.asyncio
@pytest.mark.skipif(_skip_if_no_db(), reason="Test database not available")
async def test_all_tenant_policies_use_current_tenant_id(admin_session: AsyncSession):
    """All tenant isolation policies must use app.current_tenant_id."""
    result = await admin_session.execute(text("""
        SELECT tablename, policyname
        FROM pg_policies
        WHERE schemaname = 'public'
        AND policyname LIKE '%tenant_isolation%'
        AND (
            qual NOT ILIKE '%app.current_tenant_id%'
            AND with_check NOT ILIKE '%app.current_tenant_id%'
        )
    """))
    wrong_policies = result.fetchall()
    assert len(wrong_policies) == 0, \
        f"Tenant policies not using app.current_tenant_id: {wrong_policies}"