setCollapsed(!collapsed)}
>
{collapsed ? (
diff --git a/frontend/src/hooks/useCommWebSocket.ts b/frontend/src/hooks/useCommWebSocket.ts
index 9c94769..904cbca 100644
--- a/frontend/src/hooks/useCommWebSocket.ts
+++ b/frontend/src/hooks/useCommWebSocket.ts
@@ -1,4 +1,3 @@
-// TODO: P3-F12 — Replace console.log with structured logger
import { useEffect, useRef } from 'react';
import { useCommStore } from '@/store/commStore';
import type { Conversation, Message } from '@/store/commStore';
@@ -29,7 +28,7 @@ export function useCommWebSocket() {
ws.onopen = () => {
reconnectAttempts = 0;
- console.log('Comm WebSocket connected');
+ console.debug('Comm WebSocket connected');
};
ws.onmessage = (event) => {
diff --git a/frontend/src/pages/Settings.tsx b/frontend/src/pages/Settings.tsx
index f18fd36..af77493 100644
--- a/frontend/src/pages/Settings.tsx
+++ b/frontend/src/pages/Settings.tsx
@@ -1,10 +1,21 @@
-// TODO: P3-F1/F2 — Replace import * as LucideIcons with explicit icon imports
import React, { useMemo } from 'react';
-// TODO: P2-F2 — Replace hardcoded settings nav items with dynamic config
import { NavLink, Outlet } from 'react-router-dom';
import { useTranslation } from 'react-i18next';
import { usePluginStore } from '@/store/pluginStore';
-import * as LucideIcons from 'lucide-react';
+import { Settings, Mail, Bell, Sparkles, Bot, Shield, Users, UsersRound, Package } from 'lucide-react';
+
+const ICON_MAP: Record> = {
+ Settings,
+ Mail,
+ Bell,
+ Sparkles,
+ Bot,
+ Shield,
+ Users,
+ UsersRound,
+};
+
+const FALLBACK_ICON = Package;
export function SettingsPage() {
const { t } = useTranslation();
@@ -37,8 +48,8 @@ export function SettingsPage() {
to: `/settings/${p.path}`,
label: t(p.label_key, p.label),
icon: (() => {
- const Icon = (LucideIcons as any)[p.icon];
- return Icon ? React.createElement(Icon, { className: 'w-4 h-4' }) : '\ud83d\udce6';
+ const Icon = ICON_MAP[p.icon] ?? FALLBACK_ICON;
+ return React.createElement(Icon, { className: 'w-4 h-4' });
})(),
}));
diff --git a/pyproject.toml b/pyproject.toml
index 6570e64..0508bc3 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -13,6 +13,7 @@ asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "session"
asyncio_default_test_loop_scope = "session"
testpaths = ["tests"]
+pythonpath = [".", "scripts"]
python_files = ["test_*.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
diff --git a/tests/conftest.py b/tests/conftest.py
index 397bcdd..4e6f924 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -434,6 +434,49 @@ async def seed_tenant_and_users(db: AsyncSession) -> dict[str, Any]:
}
+async def create_no_perm_user(db: AsyncSession, seed: dict[str, Any]) -> User:
+ """Create a user in tenant A with no permissions (for RBAC tests).
+
+ Returns the created user. The user has a role with empty permissions,
+ so any require_permission check will return 403.
+ """
+ from app.core.auth import hash_password
+ from app.models.role import Role
+ from app.models.user import User, UserTenant
+
+ no_perm_role = Role(
+ tenant_id=seed["tenant_a"].id,
+ name="no_perm",
+ permissions={},
+ denied_permissions=[],
+ field_permissions={},
+ )
+ db.add(no_perm_role)
+ await db.flush()
+
+ no_perm_user = User(
+ email="noperm@tenanta.com",
+ name="No Perm",
+ password_hash=hash_password("TestPass123!"),
+ is_active=True,
+ preferences={},
+ )
+ db.add(no_perm_user)
+ await db.flush()
+
+ ut = UserTenant(
+ user_id=no_perm_user.id,
+ tenant_id=seed["tenant_a"].id,
+ is_default=True,
+ role="no_perm",
+ role_id=no_perm_role.id,
+ )
+ db.add(ut)
+ await db.flush()
+ await db.commit()
+ return no_perm_user
+
+
async def login_client(
client: AsyncClient, email: str, password: str = "TestPass123!"
) -> dict[str, str]:
diff --git a/tests/test_agent_subtasks.py b/tests/test_agent_subtasks.py
index 70d1aea..21b388d 100644
--- a/tests/test_agent_subtasks.py
+++ b/tests/test_agent_subtasks.py
@@ -26,11 +26,6 @@ def db_setup():
yield
-@pytest.fixture(autouse=True)
-def clean_tables(db_setup):
- """No-op override of conftest clean_tables."""
- yield
-
@pytest.fixture(autouse=True)
def clean_tables(db_setup):
"""No-op override of conftest clean_tables."""
diff --git a/tests/test_ai_deploy.py b/tests/test_ai_deploy.py
index bfaa5ab..af56459 100644
--- a/tests/test_ai_deploy.py
+++ b/tests/test_ai_deploy.py
@@ -4,14 +4,10 @@ from __future__ import annotations
import json
import os
-import sys
from unittest.mock import patch
import pytest
-# Add scripts dir to path
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "scripts"))
-
def test_phase_result_to_dict():
"""PhaseResult.to_dict should serialize correctly."""
diff --git a/tests/test_ai_health_check.py b/tests/test_ai_health_check.py
index 77e9f96..01c2864 100644
--- a/tests/test_ai_health_check.py
+++ b/tests/test_ai_health_check.py
@@ -3,15 +3,10 @@
from __future__ import annotations
import json
-import os
-import sys
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
-# Add scripts dir to path
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "scripts"))
-
def test_enumerate_routes_returns_api_routes():
"""enumerate_routes should return only /api/ routes with methods."""
diff --git a/tests/test_api_documentation.py b/tests/test_api_documentation.py
index 149a130..2d296ab 100644
--- a/tests/test_api_documentation.py
+++ b/tests/test_api_documentation.py
@@ -3,12 +3,9 @@
from __future__ import annotations
import os
-import sys
import pytest
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-
def test_openapi_tags_configured():
"""FastAPI app should have openapi_tags configured with descriptions."""
diff --git a/tests/test_backup_restore.py b/tests/test_backup_restore.py
index d49de63..a2ff9cd 100644
--- a/tests/test_backup_restore.py
+++ b/tests/test_backup_restore.py
@@ -4,16 +4,12 @@ from __future__ import annotations
import json
import os
-import sys
import tempfile
from pathlib import Path
from unittest.mock import patch
import pytest
-# Add scripts dir to path
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "scripts"))
-
def test_get_db_connection_params_from_asyncpg_url():
"""get_db_connection_params should convert asyncpg URL to standard postgresql URL."""
diff --git a/tests/test_calendar.py b/tests/test_calendar.py
index 8841173..86add4d 100644
--- a/tests/test_calendar.py
+++ b/tests/test_calendar.py
@@ -1078,3 +1078,38 @@ async def test_ac30_private_entry_visibility(calendar_app, db_session):
# Direct access to private entry → 403
resp = await viewer_c.get(f"/api/v1/calendar/entries/{entry_id}", headers=ORIGIN_HEADER)
assert resp.status_code == 403
+
+
+# ─── Cross-tenant isolation test ───
+
+
+@pytest.mark.asyncio
+async def test_cross_tenant_calendar_isolation(calendar_app, db_session):
+ """Calendar created in tenant A is not accessible from tenant B."""
+ from httpx import ASGITransport, AsyncClient
+
+ from tests.conftest import seed_tenant_and_users
+
+ await seed_tenant_and_users(db_session)
+ transport = ASGITransport(app=calendar_app)
+
+ async with AsyncClient(transport=transport, base_url="http://test") as client_a:
+ await login_client(client_a, "admin@tenanta.com")
+ resp = await client_a.post(
+ "/api/v1/calendars",
+ json={"name": "Tenant A Calendar"},
+ headers=ORIGIN_HEADER,
+ )
+ assert resp.status_code == 201
+ cal_id = resp.json()["id"]
+
+ # Tenant B admin must not see tenant A's calendar
+ async with AsyncClient(transport=transport, base_url="http://test") as client_b:
+ await login_client(client_b, "admin@tenantb.com")
+ list_resp = await client_b.get("/api/v1/calendars", headers=ORIGIN_HEADER)
+ assert list_resp.status_code == 200
+ assert all(c["id"] != cal_id for c in list_resp.json())
+
+ # Direct access to tenant A's calendar → 404
+ get_resp = await client_b.get(f"/api/v1/calendars/{cal_id}", headers=ORIGIN_HEADER)
+ assert get_resp.status_code == 404
diff --git a/tests/test_companies.py b/tests/test_companies.py
index 53ba68a..b31b600 100644
--- a/tests/test_companies.py
+++ b/tests/test_companies.py
@@ -351,3 +351,45 @@ class TestCompanyAuditAndSoftDelete:
names = [item["name"] for item in list_resp.json()["items"]]
assert "SoftDelete Corp" not in names
assert "Company Alpha" in names # Seed company still present
+
+
+# ── Visibility filter test ──
+
+
+@pytest.mark.asyncio
+class TestCompanyVisibilityFilter:
+ """Row-level visibility filter hides non-owned, non-shared companies."""
+
+ async def test_visibility_filter_hides_owned_company_from_viewer(self, client: AsyncClient, db_session):
+ """A company owned by admin_a is not visible to viewer_a in the contacts list."""
+ await seed_tenant_and_users(db_session)
+ await login_client(client, "admin@tenanta.com")
+ # Create a company via /api/v1/contacts so owner_id is set to admin_a
+ create_resp = await client.post(
+ "/api/v1/contacts",
+ json={"type": "company", "name": "Owned Corp", "displayname": "Owned Corp"},
+ headers=ORIGIN_HEADER,
+ )
+ assert create_resp.status_code == 201
+ owned_id = create_resp.json()["id"]
+
+ # Admin (owner) sees it in the contacts list with type=company
+ list_resp = await client.get("/api/v1/contacts?type=company", headers=ORIGIN_HEADER)
+ assert list_resp.status_code == 200
+ assert any(c["id"] == owned_id for c in list_resp.json()["items"])
+
+ # Viewer (non-owner, not shared) must NOT see it
+ await login_client(client, "viewer@tenanta.com")
+ viewer_list = await client.get("/api/v1/contacts?type=company", headers=ORIGIN_HEADER)
+ assert viewer_list.status_code == 200
+ assert all(c["id"] != owned_id for c in viewer_list.json()["items"])
+
+ async def test_visibility_filter_shows_tenant_owned_company(self, client: AsyncClient, db_session):
+ """A tenant-owned company (owner_id NULL) is visible to all users with read permission."""
+ await seed_tenant_and_users(db_session)
+ await login_client(client, "viewer@tenanta.com")
+ # Seed company 'Company Alpha' has owner_id NULL → tenant-owned → visible to viewer
+ list_resp = await client.get("/api/v1/contacts?type=company", headers=ORIGIN_HEADER)
+ assert list_resp.status_code == 200
+ names = [c["name"] for c in list_resp.json()["items"]]
+ assert "Company Alpha" in names
diff --git a/tests/test_contacts.py b/tests/test_contacts.py
index e06599f..5f544ad 100644
--- a/tests/test_contacts.py
+++ b/tests/test_contacts.py
@@ -171,3 +171,45 @@ class TestContactDelete:
al_entries = al_result.scalars().all()
assert len(al_entries) >= 1
assert any(e.action == "hard_delete" for e in al_entries)
+
+
+# ── Visibility filter test ──
+
+
+@pytest.mark.asyncio
+class TestContactVisibilityFilter:
+ """Row-level visibility filter hides non-owned, non-shared contacts."""
+
+ async def test_visibility_filter_hides_owned_contact_from_viewer(self, client: AsyncClient, db_session):
+ """A contact owned by admin_a is not visible to viewer_a in the list."""
+ await seed_tenant_and_users(db_session)
+ await login_client(client, "admin@tenanta.com")
+ # Create a contact as admin (owner_id = admin_a)
+ create_resp = await client.post(
+ "/api/v1/contacts",
+ json={"firstname": "Owned", "surname": "Contact"},
+ headers=ORIGIN_HEADER,
+ )
+ assert create_resp.status_code == 201
+ owned_id = create_resp.json()["id"]
+
+ # Admin (owner) sees it in the list
+ list_resp = await client.get("/api/v1/contacts", headers=ORIGIN_HEADER)
+ assert list_resp.status_code == 200
+ assert any(c["id"] == owned_id for c in list_resp.json()["items"])
+
+ # Viewer (non-owner, not shared) must NOT see it
+ await login_client(client, "viewer@tenanta.com")
+ viewer_list = await client.get("/api/v1/contacts", headers=ORIGIN_HEADER)
+ assert viewer_list.status_code == 200
+ assert all(c["id"] != owned_id for c in viewer_list.json()["items"])
+
+ async def test_visibility_filter_shows_tenant_owned_contact(self, client: AsyncClient, db_session):
+ """A tenant-owned contact (owner_id NULL) is visible to all users with read permission."""
+ await seed_tenant_and_users(db_session)
+ await login_client(client, "viewer@tenanta.com")
+ # Seed company 'Company Alpha' has owner_id NULL → tenant-owned → visible to viewer
+ list_resp = await client.get("/api/v1/contacts?type=company", headers=ORIGIN_HEADER)
+ assert list_resp.status_code == 200
+ names = [c["name"] for c in list_resp.json()["items"]]
+ assert "Company Alpha" in names
diff --git a/tests/test_cross_tenant_security.py b/tests/test_cross_tenant_security.py
index 9e35f27..b7984c0 100644
--- a/tests/test_cross_tenant_security.py
+++ b/tests/test_cross_tenant_security.py
@@ -190,7 +190,6 @@ async def contact_b(db_session: AsyncSession, tenant_b: Tenant, user_b: User):
# ── Cross-Tenant RLS Tests ────────────────────────────────────────────────────
-@pytest.mark.asyncio
@pytest.mark.asyncio
async def test_rls_blocks_cross_tenant_insert(
db_session: AsyncSession,
diff --git a/tests/test_custom_fields.py b/tests/test_custom_fields.py
index eb89587..467471d 100644
--- a/tests/test_custom_fields.py
+++ b/tests/test_custom_fields.py
@@ -162,3 +162,50 @@ class TestCustomFieldDefinition:
)
assert len(manifest.custom_fields) == 1
assert manifest.custom_fields[0].name == "score"
+
+
+# ── Cross-tenant isolation test ──
+@pytest.mark.asyncio
+class TestCustomFieldsCrossTenant:
+ """Custom fields must not leak across tenants."""
+
+ async def test_cross_tenant_isolation(self, client: AsyncClient, db_session):
+ """Tenant B cannot access custom fields of tenant A's contact."""
+ from httpx import ASGITransport
+ from httpx import AsyncClient as AC
+
+ import app.main
+
+ await seed_tenant_and_users(db_session)
+ await login_client(client, "admin@tenanta.com")
+ # Create a contact in tenant A
+ create_resp = await client.post(
+ "/api/v1/contacts",
+ json={"type": "company", "name": "Tenant A Custom Corp", "displayname": "Tenant A Custom Corp"},
+ headers=ORIGIN_HEADER,
+ )
+ assert create_resp.status_code == 201
+ contact_id = create_resp.json()["id"]
+
+ # Tenant B admin must not access tenant A's contact custom fields
+ app_instance = app.main.app
+ async with AC(transport=ASGITransport(app=app_instance), base_url="http://test") as client_b:
+ await login_client(client_b, "admin@tenantb.com")
+ resp = await client_b.get(
+ f"/api/v1/contacts/{contact_id}/custom-fields",
+ headers=ORIGIN_HEADER,
+ )
+ assert resp.status_code == 404
+
+ async def test_rbac_no_permission(self, client: AsyncClient, db_session):
+ """User without contacts:read permission gets 403 on GET custom fields."""
+ from tests.conftest import create_no_perm_user
+
+ seed = await seed_tenant_and_users(db_session)
+ await create_no_perm_user(db_session, seed)
+ await login_client(client, "noperm@tenanta.com")
+ resp = await client.get(
+ "/api/v1/contacts/00000000-0000-0000-0000-000000000000/custom-fields",
+ headers=ORIGIN_HEADER,
+ )
+ assert resp.status_code == 403
diff --git a/tests/test_dashboard.py b/tests/test_dashboard.py
index d6393e8..f3c4907 100644
--- a/tests/test_dashboard.py
+++ b/tests/test_dashboard.py
@@ -44,3 +44,50 @@ class TestDashboardWidgets:
assert "id" in widget
assert "component" in widget
assert "label_key" in widget
+
+
+# ── Cross-tenant isolation test ──
+@pytest.mark.asyncio
+class TestDashboardCrossTenant:
+ """Dashboard counts must not leak across tenants."""
+
+ async def test_cross_tenant_isolation(self, client: AsyncClient, db_session):
+ """Tenant B admin does not see tenant A's contacts in dashboard counts."""
+ from httpx import ASGITransport
+ from httpx import AsyncClient as AC
+
+ import app.main
+
+ await seed_tenant_and_users(db_session)
+ await login_client(client, "admin@tenanta.com")
+ # Create a company in tenant A
+ create_resp = await client.post(
+ "/api/v1/companies",
+ json={"name": "Tenant A Dashboard Corp"},
+ headers=ORIGIN_HEADER,
+ )
+ assert create_resp.status_code == 201
+
+ # Tenant A admin sees it
+ counts_a = await client.get("/api/v1/dashboard/counts", headers=ORIGIN_HEADER)
+ assert counts_a.status_code == 200
+ assert counts_a.json()["companies"] >= 1
+
+ # Tenant B admin must not see it
+ app_instance = app.main.app
+ async with AC(transport=ASGITransport(app=app_instance), base_url="http://test") as client_b:
+ await login_client(client_b, "admin@tenantb.com")
+ counts_b = await client_b.get("/api/v1/dashboard/counts", headers=ORIGIN_HEADER)
+ assert counts_b.status_code == 200
+ # Tenant B only has its own seeded company (Company Beta)
+ assert counts_b.json()["companies"] == 1
+
+ async def test_rbac_no_permission(self, client: AsyncClient, db_session):
+ """User without dashboard:read permission gets 403."""
+ from tests.conftest import create_no_perm_user
+
+ seed = await seed_tenant_and_users(db_session)
+ await create_no_perm_user(db_session, seed)
+ await login_client(client, "noperm@tenanta.com")
+ resp = await client.get("/api/v1/dashboard/widgets", headers=ORIGIN_HEADER)
+ assert resp.status_code == 403
diff --git a/tests/test_dms.py b/tests/test_dms.py
index c286f02..41e8e59 100644
--- a/tests/test_dms.py
+++ b/tests/test_dms.py
@@ -736,3 +736,38 @@ async def test_ac19_bulk_delete(authed_client):
for fid in file_ids:
resp = await client.get(f"/api/v1/dms/files/{fid}", headers=ORIGIN_HEADER)
assert resp.status_code == 404
+
+
+# ─── Cross-tenant isolation test ───
+
+
+@pytest.mark.asyncio
+async def test_cross_tenant_dms_isolation(dms_app, db_session):
+ """File uploaded in tenant A is not visible to tenant B."""
+ from httpx import ASGITransport, AsyncClient
+
+ from tests.conftest import seed_tenant_and_users
+
+ await seed_tenant_and_users(db_session)
+ transport = ASGITransport(app=dms_app)
+
+ async with AsyncClient(transport=transport, base_url="http://test") as client_a:
+ await login_client(client_a, "admin@tenanta.com")
+ upload_resp = await client_a.post(
+ "/api/v1/dms/files/upload",
+ files={"file": ("tenant_a.pdf", PDF_CONTENT, "application/pdf")},
+ headers=ORIGIN_HEADER,
+ )
+ assert upload_resp.status_code == 201
+ file_id = upload_resp.json()["id"]
+
+ # Tenant B admin must not see tenant A's file in the list
+ async with AsyncClient(transport=transport, base_url="http://test") as client_b:
+ await login_client(client_b, "admin@tenantb.com")
+ list_resp = await client_b.get("/api/v1/dms/files", headers=ORIGIN_HEADER)
+ assert list_resp.status_code == 200
+ assert all(f["id"] != file_id for f in list_resp.json())
+
+ # Direct access to tenant A's file → 404
+ get_resp = await client_b.get(f"/api/v1/dms/files/{file_id}", headers=ORIGIN_HEADER)
+ assert get_resp.status_code == 404
diff --git a/tests/test_dms_coverage.py b/tests/test_dms_coverage.py
index b83402b..d9e3a5c 100644
--- a/tests/test_dms_coverage.py
+++ b/tests/test_dms_coverage.py
@@ -8,7 +8,6 @@ shared-with-me with data, bulk mixed IDs, tenant isolation.
from __future__ import annotations
-import os
import uuid
from unittest.mock import patch
diff --git a/tests/test_entity_links.py b/tests/test_entity_links.py
index fad8dad..390027e 100644
--- a/tests/test_entity_links.py
+++ b/tests/test_entity_links.py
@@ -485,3 +485,45 @@ async def test_list_contact_files_empty(authed_client: AsyncClient):
resp = await client.get(f"/api/v1/contacts/{contact_id}/files", headers=ORIGIN_HEADER)
assert resp.status_code == 200
assert resp.json() == []
+
+
+# ─── Cross-tenant isolation test ───
+
+
+@pytest.mark.asyncio
+async def test_cross_tenant_entity_link_isolation(plugin_app, db_session):
+ """Entity link created in tenant A is not visible to tenant B."""
+ from tests.conftest import seed_tenant_and_users
+
+ seed = await seed_tenant_and_users(db_session)
+ transport = ASGITransport(app=plugin_app)
+
+ async with AsyncClient(transport=transport, base_url="http://test") as client_a:
+ await login_client(client_a, "admin@tenanta.com")
+ # Upload a file in tenant A
+ upload_resp = await client_a.post(
+ "/api/v1/dms/files/upload",
+ files={"file": ("tenant_a_link.txt", b"tenant a link", "text/plain")},
+ headers=ORIGIN_HEADER,
+ )
+ assert upload_resp.status_code == 201
+ file_id = upload_resp.json()["id"]
+ company_id = str(seed["company_a"].id)
+
+ # Link file to company_a in tenant A
+ link_resp = await client_a.post(
+ f"/api/v1/entity-links/files/{file_id}/link",
+ json={"entity_type": "company", "entity_id": company_id},
+ headers=ORIGIN_HEADER,
+ )
+ assert link_resp.status_code == 200
+
+ # Tenant B admin must not see tenant A's link
+ async with AsyncClient(transport=transport, base_url="http://test") as client_b:
+ await login_client(client_b, "admin@tenantb.com")
+ links_resp = await client_b.get(
+ f"/api/v1/entity-links/files/{file_id}/links",
+ headers=ORIGIN_HEADER,
+ )
+ assert links_resp.status_code == 200
+ assert links_resp.json() == []
diff --git a/tests/test_notifications.py b/tests/test_notifications.py
index 8a5ebfd..1c65618 100644
--- a/tests/test_notifications.py
+++ b/tests/test_notifications.py
@@ -116,3 +116,52 @@ class TestNotifications:
assert "count" in data
assert isinstance(data["count"], int)
assert data["count"] >= 2
+
+
+# ── Cross-tenant isolation + RBAC tests ──
+
+
+@pytest.mark.asyncio
+class TestNotificationCrossTenant:
+ """Notifications must not leak across tenants."""
+
+ async def test_cross_tenant_isolation(self, client: AsyncClient, db_session):
+ """Notification for tenant A user is not visible to tenant B admin."""
+ from httpx import ASGITransport
+ from httpx import AsyncClient as AC
+
+ import app.main
+
+ seed = await seed_tenant_and_users(db_session)
+ await create_notification(
+ db_session,
+ seed["tenant_a"].id,
+ seed["admin_a"].id,
+ "info",
+ "Tenant A Notif",
+ "Body",
+ )
+ await db_session.commit()
+
+ await login_client(client, "admin@tenanta.com")
+ resp = await client.get("/api/v1/notifications")
+ assert resp.status_code == 200
+ assert any(i["title"] == "Tenant A Notif" for i in resp.json()["items"])
+
+ # Tenant B admin must not see tenant A's notification
+ app_instance = app.main.app
+ async with AC(transport=ASGITransport(app=app_instance), base_url="http://test") as client_b:
+ await login_client(client_b, "admin@tenantb.com")
+ resp_b = await client_b.get("/api/v1/notifications")
+ assert resp_b.status_code == 200
+ assert all(i["title"] != "Tenant A Notif" for i in resp_b.json()["items"])
+
+ async def test_rbac_no_permission(self, client: AsyncClient, db_session):
+ """User without notifications:read permission gets 403 on list."""
+ from tests.conftest import create_no_perm_user
+
+ seed = await seed_tenant_and_users(db_session)
+ await create_no_perm_user(db_session, seed)
+ await login_client(client, "noperm@tenanta.com")
+ resp = await client.get("/api/v1/notifications")
+ assert resp.status_code == 403
diff --git a/tests/test_saved_filters.py b/tests/test_saved_filters.py
index 4524d00..877a87d 100644
--- a/tests/test_saved_filters.py
+++ b/tests/test_saved_filters.py
@@ -126,3 +126,55 @@ class TestSavedFilterDelete:
await login_client(client, "admin@tenanta.com")
resp = await client.delete("/api/v1/saved-filters/not-a-uuid", headers=ORIGIN_HEADER)
assert resp.status_code == 400
+
+
+# ── Cross-tenant isolation test ──
+@pytest.mark.asyncio
+class TestSavedFilterCrossTenant:
+ """Saved filters must not leak across tenants."""
+
+ async def test_cross_tenant_isolation(self, client: AsyncClient, db_session):
+ """Filter created in tenant A is not visible to tenant B."""
+ from httpx import ASGITransport
+ from httpx import AsyncClient as AC
+
+ import app.main
+
+ await seed_tenant_and_users(db_session)
+ await login_client(client, "admin@tenanta.com")
+ # Create a filter in tenant A
+ create_resp = await client.post(
+ "/api/v1/saved-filters",
+ json={"name": "Tenant A Filter", "entity_type": "contacts", "filter_criteria": {}},
+ headers=ORIGIN_HEADER,
+ )
+ assert create_resp.status_code == 201
+ filter_id = create_resp.json()["id"]
+
+ # Login as tenant B admin
+ app_instance = app.main.app
+ async with AC(transport=ASGITransport(app=app_instance), base_url="http://test") as client_b:
+ await login_client(client_b, "admin@tenantb.com")
+ # List must not contain tenant A's filter
+ list_resp = await client_b.get("/api/v1/saved-filters", headers=ORIGIN_HEADER)
+ assert list_resp.status_code == 200
+ assert all(f["id"] != filter_id for f in list_resp.json())
+ # Delete must 404 (not found in tenant B)
+ del_resp = await client_b.delete(
+ f"/api/v1/saved-filters/{filter_id}", headers=ORIGIN_HEADER
+ )
+ assert del_resp.status_code == 404
+
+ async def test_rbac_no_permission(self, client: AsyncClient, db_session):
+ """User without contacts:read permission gets 403 on create."""
+ from tests.conftest import create_no_perm_user
+
+ seed = await seed_tenant_and_users(db_session)
+ await create_no_perm_user(db_session, seed)
+ await login_client(client, "noperm@tenanta.com")
+ resp = await client.post(
+ "/api/v1/saved-filters",
+ json={"name": "No Perm Filter", "entity_type": "contacts", "filter_criteria": {}},
+ headers=ORIGIN_HEADER,
+ )
+ assert resp.status_code == 403
diff --git a/tests/test_tags.py b/tests/test_tags.py
index b016765..35c582e 100644
--- a/tests/test_tags.py
+++ b/tests/test_tags.py
@@ -513,3 +513,51 @@ async def test_list_tag_entities_invalid_id(authed_client: AsyncClient):
"""GET /api/v1/tags/{invalid}/entities → 400."""
resp = await authed_client.get("/api/v1/tags/bad-uuid/entities", headers=ORIGIN_HEADER)
assert resp.status_code == 400
+
+
+# ── Cross-tenant isolation + RBAC tests ──
+
+
+@pytest.mark.asyncio
+async def test_cross_tenant_tag_isolation(plugin_app, db_session):
+ """Tag created in tenant A is not visible to tenant B."""
+ from tests.conftest import seed_tenant_and_users
+
+ await seed_tenant_and_users(db_session)
+ transport = ASGITransport(app=plugin_app)
+
+ async with AsyncClient(transport=transport, base_url="http://test") as client_a:
+ await login_client(client_a, "admin@tenanta.com")
+ create_resp = await client_a.post(
+ "/api/v1/tags",
+ json={"name": "Tenant A Tag", "color": "#FF0000"},
+ headers=ORIGIN_HEADER,
+ )
+ assert create_resp.status_code == 201
+ tag_id = create_resp.json()["id"]
+
+ # Tenant B admin must not see tenant A's tag
+ async with AsyncClient(transport=transport, base_url="http://test") as client_b:
+ await login_client(client_b, "admin@tenantb.com")
+ list_resp = await client_b.get("/api/v1/tags", headers=ORIGIN_HEADER)
+ assert list_resp.status_code == 200
+ assert all(t["id"] != tag_id for t in list_resp.json())
+
+
+@pytest.mark.asyncio
+async def test_rbac_tag_no_permission(plugin_app, db_session):
+ """User without tags:write permission gets 403 on create."""
+ from tests.conftest import create_no_perm_user, seed_tenant_and_users
+
+ seed = await seed_tenant_and_users(db_session)
+ await create_no_perm_user(db_session, seed)
+ transport = ASGITransport(app=plugin_app)
+
+ async with AsyncClient(transport=transport, base_url="http://test") as client:
+ await login_client(client, "noperm@tenanta.com")
+ resp = await client.post(
+ "/api/v1/tags",
+ json={"name": "No Perm Tag", "color": "#00FF00"},
+ headers=ORIGIN_HEADER,
+ )
+ assert resp.status_code == 403
diff --git a/tests/test_tasks.py b/tests/test_tasks.py
index 2d5b9fc..4cf019e 100644
--- a/tests/test_tasks.py
+++ b/tests/test_tasks.py
@@ -180,3 +180,49 @@ class TestTaskDelete:
# Verify it's gone from list
list_resp = await tasks_client.get("/api/v1/tasks", headers=ORIGIN_HEADER)
assert not any(t["id"] == task_id for t in list_resp.json()["items"])
+
+
+# ── Cross-tenant isolation test ──
+@pytest.mark.asyncio
+class TestTaskCrossTenant:
+ """Tasks must not leak across tenants."""
+
+ async def test_cross_tenant_isolation(self, tasks_app, db_session):
+ """Task created in tenant A is not visible to tenant B."""
+ from httpx import ASGITransport, AsyncClient
+
+ await seed_tenant_and_users(db_session)
+ transport = ASGITransport(app=tasks_app)
+ async with AsyncClient(transport=transport, base_url="http://test") as client_a:
+ await login_client(client_a, "admin@tenanta.com")
+ create_resp = await client_a.post(
+ "/api/v1/tasks",
+ json={"title": "Tenant A Task"},
+ headers=ORIGIN_HEADER,
+ )
+ assert create_resp.status_code == 201
+ task_id = create_resp.json()["id"]
+
+ async with AsyncClient(transport=transport, base_url="http://test") as client_b:
+ await login_client(client_b, "admin@tenantb.com")
+ # List must not contain tenant A's task
+ list_resp = await client_b.get("/api/v1/tasks", headers=ORIGIN_HEADER)
+ assert list_resp.status_code == 200
+ assert all(t["id"] != task_id for t in list_resp.json()["items"])
+ # Get must 404
+ get_resp = await client_b.get(f"/api/v1/tasks/{task_id}", headers=ORIGIN_HEADER)
+ assert get_resp.status_code == 404
+
+ async def test_rbac_no_permission(self, tasks_client: AsyncClient, db_session):
+ """User without tasks:write permission gets 403 on create."""
+ from tests.conftest import create_no_perm_user
+
+ seed = await seed_tenant_and_users(db_session)
+ await create_no_perm_user(db_session, seed)
+ await login_client(tasks_client, "noperm@tenanta.com")
+ resp = await tasks_client.post(
+ "/api/v1/tasks",
+ json={"title": "No Perm Task"},
+ headers=ORIGIN_HEADER,
+ )
+ assert resp.status_code == 403
diff --git a/tests/test_tenant.py b/tests/test_tenant.py
index adcd0b3..46f08af 100644
--- a/tests/test_tenant.py
+++ b/tests/test_tenant.py
@@ -186,7 +186,8 @@ class TestFieldPermissions:
)
db_session.add(sales_user)
await db_session.flush()
- ut = UserTenant(user_id=sales_user.id, tenant_id=seed["tenant_a"].id, is_default=True, role="sales_rep", role_id=seed["custom_role"].id)
+ custom_role = seed.get("custom_role")
+ ut = UserTenant(user_id=sales_user.id, tenant_id=seed["tenant_a"].id, is_default=True, role="sales_rep", role_id=custom_role.id if custom_role else None)
db_session.add(ut)
await db_session.commit()
diff --git a/tests/test_workflows.py b/tests/test_workflows.py
index e65aabd..6523205 100644
--- a/tests/test_workflows.py
+++ b/tests/test_workflows.py
@@ -1777,3 +1777,52 @@ async def test_service_get_instance_not_found(db_session):
result = await get_instance(db_session, tenant_id, str(uuid.uuid4()))
assert result is None
+
+
+# ── Cross-tenant isolation + RBAC tests ──
+
+
+@pytest.mark.asyncio
+async def test_cross_tenant_workflow_isolation(client: AsyncClient, db_session):
+ """Workflow created in tenant A is not visible to tenant B."""
+ from httpx import ASGITransport
+ from httpx import AsyncClient as AC
+
+ import app.main
+
+ await seed_tenant_and_users(db_session)
+ await login_client(client, "admin@tenanta.com")
+ create_resp = await client.post(
+ "/api/v1/workflows",
+ json={"name": "Tenant A WF", "steps": VALID_STEPS},
+ headers=ORIGIN_HEADER,
+ )
+ assert create_resp.status_code == 201
+ wf_id = create_resp.json()["id"]
+
+ # Tenant B admin must not see tenant A's workflow
+ app_instance = app.main.app
+ async with AC(transport=ASGITransport(app=app_instance), base_url="http://test") as client_b:
+ await login_client(client_b, "admin@tenantb.com")
+ list_resp = await client_b.get("/api/v1/workflows")
+ assert list_resp.status_code == 200
+ assert all(w["id"] != wf_id for w in list_resp.json()["items"])
+ # Direct access to tenant A's workflow → 404
+ get_resp = await client_b.get(f"/api/v1/workflows/{wf_id}")
+ assert get_resp.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_rbac_workflow_no_permission(client: AsyncClient, db_session):
+ """User without workflows:write permission gets 403 on create."""
+ from tests.conftest import create_no_perm_user
+
+ seed = await seed_tenant_and_users(db_session)
+ await create_no_perm_user(db_session, seed)
+ await login_client(client, "noperm@tenanta.com")
+ resp = await client.post(
+ "/api/v1/workflows",
+ json={"name": "No Perm WF", "steps": VALID_STEPS},
+ headers=ORIGIN_HEADER,
+ )
+ assert resp.status_code == 403