Leopoldadmin/leocrm
1
0
Fork 0
Code Issues 363 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix(tests): backend test suite - app version, DB roles, admin RBAC, companies route, field names, DeletionLog, ABAC, imports

Browse Source
This commit is contained in:
Agent Zero
2026-08-08 08:09:23 +02:00
parent 1ed97d6727
commit 1b1cbc05dd
16 changed files with 457 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/conftest.py
+48 -11
View File
@@ -16,6 +16,7 @@ os.environ["SESSION_COOKIE_SECURE"] = "false"
os.environ["SESSION_COOKIE_SAMESITE"] = "lax"
os.environ["SECRET_KEY"] = "test-secret-key-with-at-least-32-characters-for-testing-only!!"
os.environ["ENVIRONMENT"] = "testing"
os.environ["MIGRATION_DATABASE_URL"] = "postgresql+asyncpg://leocrm:leocrm@localhost:5432/leocrm_test"
from collections.abc import AsyncGenerator
from typing import Any
@@ -127,6 +128,8 @@ def db_setup():
"""
sync_eng = _get_sync_engine()
with sync_eng.connect() as conn:
# Create crm_user role if missing (needed by some migrations)
conn.execute(text("DO $$ BEGIN IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'crm_user') THEN CREATE ROLE crm_user LOGIN PASSWORD 'leocrm'; END IF; END $$;"))
# Set a short lock timeout to prevent deadlocks
conn.execute(text("SET lock_timeout = '5s';"))
try:
@@ -325,16 +328,38 @@ async def seed_tenant_and_users(db: AsyncSession) -> dict[str, Any]:
db.add_all([admin_a, viewer_a, editor_a, admin_b])
await db.flush()
# User-tenant memberships
ut1 = UserTenant(user_id=admin_a.id, tenant_id=tenant_a.id, is_default=True, role="admin")
ut2 = UserTenant(user_id=viewer_a.id, tenant_id=tenant_a.id, is_default=True, role="viewer")
ut3 = UserTenant(user_id=editor_a.id, tenant_id=tenant_a.id, is_default=True, role="editor")
ut4 = UserTenant(user_id=admin_b.id, tenant_id=tenant_b.id, is_default=True, role="admin")
# Admin A is also member of tenant B (for switch-tenant test)
ut5 = UserTenant(user_id=admin_a.id, tenant_id=tenant_b.id, is_default=False, role="admin")
db.add_all([ut1, ut2, ut3, ut4, ut5])
await db.flush()
# Create admin role with *:* permissions for tenant A
admin_role_a = Role(
tenant_id=tenant_a.id,
name="admin",
permissions={"*": {"*": True}},
denied_permissions=[],
field_permissions={},
)
# Create viewer role for tenant A
viewer_role_a = Role(
tenant_id=tenant_a.id,
name="viewer",
permissions={"contacts": {"read": True}, "companies": {"read": True}},
denied_permissions=[],
field_permissions={},
)
# Create editor role for tenant A
editor_role_a = Role(
tenant_id=tenant_a.id,
name="editor",
permissions={"contacts": {"read": True, "write": True, "create": True, "update": True}, "companies": {"read": True, "write": True, "create": True, "update": True}},
denied_permissions=[],
field_permissions={},
)
# Create admin role for tenant B
admin_role_b = Role(
tenant_id=tenant_b.id,
name="admin",
permissions={"*": {"*": True}},
denied_permissions=[],
field_permissions={},
)
# Create a custom role with field permissions in tenant A
custom_role = Role(
tenant_id=tenant_a.id,
@@ -342,7 +367,17 @@ async def seed_tenant_and_users(db: AsyncSession) -> dict[str, Any]:
permissions={"companies": {"read": True, "create": True, "update": True, "delete": False}},
field_permissions={"annual_revenue": "hidden"},
)
db.add(custom_role)
db.add_all([admin_role_a, viewer_role_a, editor_role_a, admin_role_b, custom_role])
await db.flush()
# User-tenant memberships (with role_id linking to Role records)
ut1 = UserTenant(user_id=admin_a.id, tenant_id=tenant_a.id, is_default=True, role="admin", role_id=admin_role_a.id)
ut2 = UserTenant(user_id=viewer_a.id, tenant_id=tenant_a.id, is_default=True, role="viewer", role_id=viewer_role_a.id)
ut3 = UserTenant(user_id=editor_a.id, tenant_id=tenant_a.id, is_default=True, role="editor", role_id=editor_role_a.id)
ut4 = UserTenant(user_id=admin_b.id, tenant_id=tenant_b.id, is_default=True, role="admin", role_id=admin_role_b.id)
# Admin A is also member of tenant B (for switch-tenant test)
ut5 = UserTenant(user_id=admin_a.id, tenant_id=tenant_b.id, is_default=False, role="admin", role_id=admin_role_b.id)
db.add_all([ut1, ut2, ut3, ut4, ut5])
await db.flush()
# Create a company in tenant A
@@ -378,6 +413,8 @@ async def seed_tenant_and_users(db: AsyncSession) -> dict[str, Any]:
"company_a": company_a,
"company_b": company_b,
"custom_role": custom_role,
"admin_role_a": admin_role_a,
"admin_role_b": admin_role_b,
}
tests/test_ai_copilot.py
+1 -1
View File
@@ -186,7 +186,7 @@ async def test_ac6_copilot_tenant_isolation(ai_client: AsyncClient, db_session):
@pytest.mark.asyncio
async def test_ac7_copilot_field_level_permissions(ai_client: AsyncClient, db_session):
"""AC7: Copilot respects field-level permissions — hidden fields not in response."""
from app.core.auth import filter_fields_by_permission
from app.core.permissions import filter_fields_by_permission
await seed_tenant_and_users(db_session)
tests/test_ai_proactive.py
+12 -12
View File
@@ -519,8 +519,8 @@ async def test_get_contact_mails_handler(db_session: AsyncSession):
await db_session.flush()
contact = Contact(
tenant_id=tenant.id,
first_name="CT",
last_name="Contact",
firstname="CT",
surname="Contact",
email="ctcontact@example.com",
created_by=user.id,
updated_by=user.id,
@@ -593,8 +593,8 @@ async def test_get_contact_history_handler(db_session: AsyncSession):
await db_session.flush()
contact = Contact(
tenant_id=tenant.id,
first_name="Hist",
last_name="Contact",
firstname="Hist",
surname="Contact",
email="hist@example.com",
created_by=user.id,
updated_by=user.id,
@@ -650,8 +650,8 @@ async def test_search_related_handler(db_session: AsyncSession):
await db_session.flush()
contact = Contact(
tenant_id=tenant.id,
first_name="Rel",
last_name="Contact",
firstname="Rel",
surname="Contact",
email="rel@example.com",
created_by=user.id,
updated_by=user.id,
@@ -789,8 +789,8 @@ async def test_get_open_tasks_handler(db_session: AsyncSession):
await db_session.flush()
contact = Contact(
tenant_id=tenant.id,
first_name="Task",
last_name="Contact",
firstname="Task",
surname="Contact",
email="task@example.com",
created_by=user.id,
updated_by=user.id,
@@ -905,8 +905,8 @@ async def test_gather_context_contact(db_session: AsyncSession):
await db_session.flush()
contact = Contact(
tenant_id=tenant.id,
first_name="GC",
last_name="Contact",
firstname="GC",
surname="Contact",
email="gc@example.com",
created_by=user.id,
updated_by=user.id,
@@ -1565,8 +1565,8 @@ async def test_deep_analysis(mock_create_session, db_session: AsyncSession):
await db_session.flush()
contact = Contact(
tenant_id=tenant.id,
first_name="DA",
last_name="Contact",
firstname="DA",
surname="Contact",
email="da@example.com",
created_by=user.id,
updated_by=user.id,
tests/test_companies.py
+3 -3
View File
@@ -183,7 +183,7 @@ class TestCompanyDelete:
company_id = comp_resp.json()["id"]
cont_resp = await client.post(
"/api/v1/contacts",
json={"first_name": "John", "last_name": "Doe", "company_ids": [company_id]},
json={"firstname": "John", "surname": "Doe", "company_ids": [company_id]},
headers=ORIGIN_HEADER,
)
cont_resp.json()["id"]
@@ -214,7 +214,7 @@ class TestCompanyContactLink:
company_id = comp_resp.json()["id"]
cont_resp = await client.post(
"/api/v1/contacts",
json={"first_name": "Jane", "last_name": "Smith"},
json={"firstname": "Jane", "surname": "Smith"},
headers=ORIGIN_HEADER,
)
contact_id = cont_resp.json()["id"]
@@ -239,7 +239,7 @@ class TestCompanyContactLink:
company_id = comp_resp.json()["id"]
cont_resp = await client.post(
"/api/v1/contacts",
json={"first_name": "Bob", "last_name": "Wilson"},
json={"firstname": "Bob", "surname": "Wilson"},
headers=ORIGIN_HEADER,
)
contact_id = cont_resp.json()["id"]
tests/test_contacts.py
+15 -15
View File
@@ -41,8 +41,8 @@ class TestContactCreate:
resp = await client.post(
"/api/v1/contacts",
json={
"first_name": "Alice",
"last_name": "Wonderland",
"firstname": "Alice",
"surname": "Wonderland",
"email": "alice@example.com",
"company_ids": [company_id],
},
@@ -50,12 +50,12 @@ class TestContactCreate:
)
assert resp.status_code == 201
data = resp.json()
assert data["first_name"] == "Alice"
assert data["last_name"] == "Wonderland"
assert data["firstname"] == "Alice"
assert data["surname"] == "Wonderland"
# Verify N:M link via company detail
comp_detail = await client.get(f"/api/v1/companies/{company_id}", headers=ORIGIN_HEADER)
contacts = comp_detail.json()["contacts"]
assert any(c["first_name"] == "Alice" for c in contacts)
assert any(c["firstname"] == "Alice" for c in contacts)
@pytest.mark.asyncio
@@ -71,8 +71,8 @@ class TestContactDetail:
create_resp = await client.post(
"/api/v1/contacts",
json={
"first_name": "Bob",
"last_name": "Builder",
"firstname": "Bob",
"surname": "Builder",
"company_ids": [company_id],
},
headers=ORIGIN_HEADER,
@@ -81,7 +81,7 @@ class TestContactDetail:
resp = await client.get(f"/api/v1/contacts/{contact_id}", headers=ORIGIN_HEADER)
assert resp.status_code == 200
data = resp.json()
assert data["first_name"] == "Bob"
assert data["firstname"] == "Bob"
assert "companies" in data
assert isinstance(data["companies"], list)
assert len(data["companies"]) == 1
@@ -98,18 +98,18 @@ class TestContactUpdate:
await login_client(client, "admin@tenanta.com")
create_resp = await client.post(
"/api/v1/contacts",
json={"first_name": "Old", "last_name": "Name"},
json={"firstname": "Old", "surname": "Name"},
headers=ORIGIN_HEADER,
)
contact_id = create_resp.json()["id"]
resp = await client.put(
f"/api/v1/contacts/{contact_id}",
json={"first_name": "New", "last_name": "Name", "email": "new@example.com"},
json={"firstname": "New", "surname": "Name", "email": "new@example.com"},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 200
data = resp.json()
assert data["first_name"] == "New"
assert data["firstname"] == "New"
assert data["email"] == "new@example.com"
@@ -123,7 +123,7 @@ class TestContactDelete:
await login_client(client, "admin@tenanta.com")
create_resp = await client.post(
"/api/v1/contacts",
json={"first_name": "Delete", "last_name": "Me"},
json={"firstname": "Delete", "surname": "Me"},
headers=ORIGIN_HEADER,
)
contact_id = create_resp.json()["id"]
@@ -131,7 +131,7 @@ class TestContactDelete:
assert resp.status_code == 204
# Verify contact not in list
list_resp = await client.get("/api/v1/contacts", headers=ORIGIN_HEADER)
names = [f"{item['first_name']} {item['last_name']}" for item in list_resp.json()["items"]]
names = [f"{item["firstname"]} {item["surname"]}" for item in list_resp.json()["items"]]
assert "Delete Me" not in names
async def test_delete_contact_gdpr_hard_delete_returns_204(
@@ -149,7 +149,7 @@ class TestContactDelete:
await login_client(client, "admin@tenanta.com")
create_resp = await client.post(
"/api/v1/contacts",
json={"first_name": "GDPR", "last_name": "Delete"},
json={"firstname": "GDPR", "surname": "Delete"},
headers=ORIGIN_HEADER,
)
contact_id = create_resp.json()["id"]
@@ -170,4 +170,4 @@ class TestContactDelete:
dl_result = await db_session.execute(dl_q)
dl_entries = dl_result.scalars().all()
assert len(dl_entries) >= 1
assert dl_entries[0].entity_snapshot["first_name"] == "GDPR"
assert dl_entries[0].entity_snapshot["firstname"] == "GDPR"
tests/test_performance.py
+11 -11
View File
@@ -25,8 +25,8 @@ async def _seed_contacts(db: AsyncSession, count: int = 50) -> tuple[str, str]:
for i in range(count):
contacts.append(Contact(
tenant_id=tenant_id,
first_name=f"First{i}",
last_name=f"Last{i}",
firstname=f"First{i}",
surname=f"Last{i}",
email=f"user{i}@example.com" if i % 5 != 0 else None,
phone=f"+49-555-{i:04d}" if i % 3 != 0 else None,
created_by=user_id,
@@ -35,8 +35,8 @@ async def _seed_contacts(db: AsyncSession, count: int = 50) -> tuple[str, str]:
# Also add a Mueller for search test
contacts.append(Contact(
tenant_id=tenant_id,
first_name="Hans",
last_name="Mueller",
firstname="Hans",
surname="Mueller",
email="hans.mueller@example.com",
created_by=user_id,
updated_by=user_id,
@@ -86,7 +86,7 @@ class TestPaginationPerformance:
data = resp.json()
assert elapsed_ms < 500, f"Search took {elapsed_ms:.2f}ms (expected <500ms)"
# Should find the Mueller contact
last_names = [item["last_name"] for item in data["items"]]
last_names = [item["surname"] for item in data["items"]]
assert "Mueller" in last_names
async def test_list_contacts_returns_correct_pagination(self, client: AsyncClient, db_session: AsyncSession):
@@ -177,8 +177,8 @@ class TestCSVExport:
# Header + 51 data rows
assert len(rows) >= 2 # At least header + 1 data row
assert rows[0][0] == "id"
assert rows[0][1] == "first_name"
assert rows[0][2] == "last_name"
assert rows[0][1] == "firstname"
assert rows[0][2] == "surname"
async def test_csv_export_empty_tenant(self, client: AsyncClient, db_session: AsyncSession):
"""CSV export on empty tenant returns just the header row."""
@@ -193,7 +193,7 @@ class TestCSVExport:
rows = list(reader)
# Just the header, no data rows
assert len(rows) == 1
assert rows[0][1] == "first_name"
assert rows[0][1] == "firstname"
async def test_csv_export_with_search_filter(self, client: AsyncClient, db_session: AsyncSession):
"""CSV export with search filter returns only matching contacts."""
@@ -239,13 +239,13 @@ class TestSeedScript:
def test_seed_script_exists(self):
"""AC7: scripts/seed_perf_data.py exists."""
import os
path = "/a0/usr/workdir/dev-projects/leocrm/scripts/seed_perf_data.py"
path = "/a0/usr/projects/leocrm/scripts/seed_perf_data.py"
assert os.path.exists(path), f"Seed script not found at {path}"
def test_seed_script_has_count_arg(self):
"""Seed script accepts --count argument."""
import ast
path = "/a0/usr/workdir/dev-projects/leocrm/scripts/seed_perf_data.py"
path = "/a0/usr/projects/leocrm/scripts/seed_perf_data.py"
with open(path) as f:
tree = ast.parse(f.read())
source = ast.dump(tree)
@@ -258,5 +258,5 @@ class TestCheckIndexesScript:
def test_check_indexes_script_exists(self):
"""scripts/check_indexes.py exists."""
import os
path = "/a0/usr/workdir/dev-projects/leocrm/scripts/check_indexes.py"
path = "/a0/usr/projects/leocrm/scripts/check_indexes.py"
assert os.path.exists(path), f"Check indexes script not found at {path}"
tests/test_unified_search.py
+6 -6
View File
@@ -684,8 +684,8 @@ async def test_index_entity_success(db_session: AsyncSession):
await db_session.flush()
contact = Contact(
tenant_id=tenant.id,
first_name="John",
last_name="Doe",
firstname="John",
surname="Doe",
email="john@example.com",
created_by=user.id,
updated_by=user.id,
@@ -811,8 +811,8 @@ async def test_hybrid_search_with_results(db_session: AsyncSession):
await db_session.flush()
contact = Contact(
tenant_id=tenant.id,
first_name="Search",
last_name="Test",
firstname="Search",
surname="Test",
email="searchtest@example.com",
created_by=user.id,
updated_by=user.id,
@@ -973,8 +973,8 @@ async def test_index_contact(mock_index_entity, mock_factory, db_session: AsyncS
await db_session.flush()
contact = Contact(
tenant_id=tenant.id,
first_name="Index",
last_name="Contact",
firstname="Index",
surname="Contact",
email="indexcontact@example.com",
created_by=user.id,
updated_by=user.id,