Agent Zero
abbe7a18fc
- P0: hooks.py 3-tuple fix, trigger_dispatcher Contract, contacts/plugin unregister_actions_by_owner - P0: 5 test files — check_permission mocks removed, hardcoded DB credential → env var - P1: attachment_service DmsFile via Contract helper, restore_registry/history_hooks dedup - P1: mail/plugin restore unregister, mcp_client datetime.now(UTC), saved_views/filters patterns - P1: ProtectedRoute fail-closed, 13 test assertion fixes (bcrypt, DB-URLs, SECRET_KEYs) - P2: deprecated notifications → post_system_message (3 files), forgejo Base, report_generator lazy import - P2: webhooks permissions, deps.py/roles.py plugin perms removed, import_export default - P2: address/tags/entity_links patterns removed, worker.py Contract-Umgehungen fixed - P2: 28 frontend TODOs (hardcoded constants, deprecated notification API) - P3: dead code, duplicates, deprecated imports, private attr, __import__ inline - P3: 8 frontend TODOs (LucideIcons, inline styles, XSS, i18n) - ruff: 838 → 0 (612 auto-fix + 246 manual + 27 F821 regression fix) - F821: 30 → 0 (AutomationDefinition, DmsFile, user_id, Path, Any, String) - Contract-Umgehungen: 2 neue gefunden (worker.py:169, worker.py:280) und gefixt
94 lines
3.1 KiB
Python
94 lines
3.1 KiB
Python
"""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}"
|