T01: core infrastructure + auth + multi-tenant + RLS
- 10 models: tenants, users, user_tenants, roles, sessions, audit_log, deletion_log, notifications, password_reset_tokens, api_tokens - Session-based auth (Redis + PostgreSQL audit trail) - Multi-tenant with ORM-level filtering + PostgreSQL RLS (set_config) - RBAC with roles/permissions + field-level permissions - CSRF protection via Origin header validation - Auth rate limiting (Redis counters with TTL) - CORS with explicit origins (no wildcard) - Health endpoint (no auth required) - Notification service + audit log middleware - 29 tests, 26 ACs, all passing - Coverage: 62% (infrastructure modules pending coverage in later tasks)
This commit is contained in:
leocrm-bot
@@ -1 +0,0 @@
|
||||
"""Test package for the CRM system."""
|
||||
|
||||
+218
-270
@@ -1,19 +1,22 @@
|
||||
"""Test fixtures: in-memory SQLite DB, async test client, auth helpers.
|
||||
"""Test fixtures: PostgreSQL test DB, Redis, async test client, auth helpers.
|
||||
|
||||
Design:
|
||||
- One AsyncEngine + session_factory per test (function-scoped) so each test
|
||||
has a fresh, isolated DB.
|
||||
- The FastAPI app's get_db dependency is overridden to use the same engine.
|
||||
- register/headers/login helpers talk to the HTTP API (integration tests).
|
||||
Each test gets a fresh database schema (created from metadata) and a clean Redis.
|
||||
Auth helpers talk to the HTTP API (integration tests).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import uuid
|
||||
from collections.abc import AsyncGenerator
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
import redis.asyncio as aioredis
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.ext.asyncio import (
|
||||
AsyncEngine,
|
||||
AsyncSession,
|
||||
@@ -21,301 +24,246 @@ from sqlalchemy.ext.asyncio import (
|
||||
create_async_engine,
|
||||
)
|
||||
|
||||
from app.core.db import Base, get_db
|
||||
from app.config import get_settings
|
||||
from app.core.db import Base, reset_engine_for_testing, close_engine
|
||||
from app.core.auth import hash_password
|
||||
from app.models.tenant import Tenant
|
||||
from app.models.user import User, UserTenant
|
||||
from app.models.role import Role
|
||||
from app.models.company import Company
|
||||
from app.main import create_app
|
||||
|
||||
|
||||
TEST_DB_URL = "postgresql+asyncpg://leocrm:leocrm@localhost:5432/leocrm_test"
|
||||
|
||||
|
||||
def _get_sync_engine():
|
||||
"""Create a sync engine for DDL operations (drop/create schema).
|
||||
Uses postgres superuser because leocrm user doesn't own the public schema.
|
||||
"""
|
||||
from sqlalchemy import create_engine
|
||||
return create_engine(
|
||||
"postgresql+psycopg2://postgres@localhost:5432/leocrm_test",
|
||||
echo=False,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(scope="session", autouse=True)
|
||||
def db_setup():
|
||||
"""Drop and recreate all tables once per test session."""
|
||||
sync_eng = _get_sync_engine()
|
||||
with sync_eng.connect() as conn:
|
||||
# Drop all tables and types
|
||||
conn.execute(text("DROP SCHEMA public CASCADE;"))
|
||||
conn.execute(text("CREATE SCHEMA public;"))
|
||||
conn.execute(text("GRANT ALL ON SCHEMA public TO leocrm;"))
|
||||
conn.commit()
|
||||
sync_eng.dispose()
|
||||
|
||||
# Create tables using async engine
|
||||
async def _create():
|
||||
eng = create_async_engine(TEST_DB_URL, echo=False)
|
||||
async with eng.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
await eng.dispose()
|
||||
|
||||
asyncio.get_event_loop().run_until_complete(_create())
|
||||
yield
|
||||
|
||||
# Cleanup after session
|
||||
sync_eng = _get_sync_engine()
|
||||
with sync_eng.connect() as conn:
|
||||
conn.execute(text("DROP SCHEMA public CASCADE;"))
|
||||
conn.execute(text("CREATE SCHEMA public;"))
|
||||
conn.execute(text("GRANT ALL ON SCHEMA public TO leocrm;"))
|
||||
conn.commit()
|
||||
sync_eng.dispose()
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def clean_tables(db_setup):
|
||||
"""Clean all table data before each test (preserve schema)."""
|
||||
sync_eng = _get_sync_engine()
|
||||
with sync_eng.connect() as conn:
|
||||
# TRUNCATE all tables with CASCADE — fast and reliable isolation
|
||||
conn.execute(text("TRUNCATE TABLE api_tokens, password_reset_tokens, notifications, deletion_log, audit_log, sessions, roles, companies, user_tenants, users, tenants CASCADE;"))
|
||||
conn.commit()
|
||||
sync_eng.dispose()
|
||||
yield
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def redis_client() -> AsyncGenerator[aioredis.Redis, None]:
|
||||
"""Redis client for tests — flushes DB before and after."""
|
||||
r = aioredis.from_url("redis://localhost:6379/0", decode_responses=True)
|
||||
await r.flushdb()
|
||||
yield r
|
||||
await r.flushdb()
|
||||
await r.aclose()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def engine() -> AsyncGenerator[AsyncEngine, None]:
|
||||
"""Per-test in-memory SQLite engine with schema created from metadata."""
|
||||
eng = create_async_engine(
|
||||
"sqlite+aiosqlite:///:memory:",
|
||||
echo=False,
|
||||
future=True,
|
||||
)
|
||||
async with eng.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
"""Async engine for the test database."""
|
||||
eng = create_async_engine(TEST_DB_URL, echo=False)
|
||||
yield eng
|
||||
await eng.dispose()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def session_factory(
|
||||
engine: AsyncEngine,
|
||||
) -> async_sessionmaker[AsyncSession]:
|
||||
async def session_factory(engine: AsyncEngine) -> async_sessionmaker[AsyncSession]:
|
||||
"""Session factory bound to the test engine."""
|
||||
return async_sessionmaker(bind=engine, expire_on_commit=False, class_=AsyncSession)
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def client(
|
||||
session_factory: async_sessionmaker[AsyncSession],
|
||||
) -> AsyncGenerator[AsyncClient, None]:
|
||||
"""httpx.AsyncClient wired to a fresh FastAPI app with the test DB injected."""
|
||||
async def db_session(session_factory: async_sessionmaker[AsyncSession]) -> AsyncGenerator[AsyncSession, None]:
|
||||
"""Database session for direct DB operations in tests."""
|
||||
async with session_factory() as session:
|
||||
yield session
|
||||
await session.rollback()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def app(engine: AsyncEngine, redis_client: aioredis.Redis):
|
||||
"""FastAPI app with test engine injected."""
|
||||
reset_engine_for_testing(engine)
|
||||
app = create_app()
|
||||
yield app
|
||||
await close_engine()
|
||||
|
||||
async def _override_get_db() -> AsyncGenerator[AsyncSession, None]:
|
||||
async with session_factory() as session:
|
||||
try:
|
||||
yield session
|
||||
except Exception:
|
||||
await session.rollback()
|
||||
raise
|
||||
|
||||
app.dependency_overrides[get_db] = _override_get_db
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def client(app) -> AsyncGenerator[AsyncClient, None]:
|
||||
"""HTTP async test client."""
|
||||
transport = ASGITransport(app=app)
|
||||
async with AsyncClient(transport=transport, base_url="http://test") as c:
|
||||
yield c
|
||||
|
||||
|
||||
# === Convenience: register via API (covers bootstrap, user, token, headers) ===
|
||||
# ─── Seed Data Helpers ───
|
||||
|
||||
ORIGIN_HEADER = {"Origin": "http://localhost:5173"}
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def registered_user(
|
||||
client: AsyncClient,
|
||||
) -> dict[str, Any]:
|
||||
"""Register a bootstrap user via the API.
|
||||
|
||||
Returns a dict with user, token, headers, email, password, name.
|
||||
Use this fixture (or the dependent `auth_headers`) for all auth-required tests.
|
||||
async def seed_tenant_and_users(db: AsyncSession) -> dict[str, Any]:
|
||||
"""Seed two tenants with admin, editor, viewer users.
|
||||
Returns dict with all created entity IDs.
|
||||
"""
|
||||
payload = {
|
||||
"email": "admin@test.com",
|
||||
"password": "TestPass123!",
|
||||
"name": "Test Admin",
|
||||
}
|
||||
resp = await client.post("/api/v1/auth/register", json=payload)
|
||||
assert resp.status_code == 201, f"register failed: {resp.status_code} {resp.text}"
|
||||
data = resp.json()
|
||||
token = data["access_token"]
|
||||
tenant_a = Tenant(name="Tenant A", slug="tenant-a")
|
||||
tenant_b = Tenant(name="Tenant B", slug="tenant-b")
|
||||
db.add_all([tenant_a, tenant_b])
|
||||
await db.flush()
|
||||
|
||||
# Admin in tenant A
|
||||
admin_a = User(
|
||||
tenant_id=tenant_a.id,
|
||||
email="admin@tenanta.com",
|
||||
name="Admin A",
|
||||
password_hash=hash_password("TestPass123!"),
|
||||
role="admin",
|
||||
is_active=True,
|
||||
preferences={},
|
||||
)
|
||||
# Viewer in tenant A
|
||||
viewer_a = User(
|
||||
tenant_id=tenant_a.id,
|
||||
email="viewer@tenanta.com",
|
||||
name="Viewer A",
|
||||
password_hash=hash_password("TestPass123!"),
|
||||
role="viewer",
|
||||
is_active=True,
|
||||
preferences={},
|
||||
)
|
||||
# Editor in tenant A
|
||||
editor_a = User(
|
||||
tenant_id=tenant_a.id,
|
||||
email="editor@tenanta.com",
|
||||
name="Editor A",
|
||||
password_hash=hash_password("TestPass123!"),
|
||||
role="editor",
|
||||
is_active=True,
|
||||
preferences={},
|
||||
)
|
||||
# Admin in tenant B
|
||||
admin_b = User(
|
||||
tenant_id=tenant_b.id,
|
||||
email="admin@tenantb.com",
|
||||
name="Admin B",
|
||||
password_hash=hash_password("TestPass123!"),
|
||||
role="admin",
|
||||
is_active=True,
|
||||
preferences={},
|
||||
)
|
||||
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)
|
||||
ut2 = UserTenant(user_id=viewer_a.id, tenant_id=tenant_a.id, is_default=True)
|
||||
ut3 = UserTenant(user_id=editor_a.id, tenant_id=tenant_a.id, is_default=True)
|
||||
ut4 = UserTenant(user_id=admin_b.id, tenant_id=tenant_b.id, is_default=True)
|
||||
# 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)
|
||||
db.add_all([ut1, ut2, ut3, ut4, ut5])
|
||||
await db.flush()
|
||||
|
||||
# Create a custom role with field permissions in tenant A
|
||||
custom_role = Role(
|
||||
tenant_id=tenant_a.id,
|
||||
name="sales_rep",
|
||||
permissions={"companies": {"read": True, "create": True, "update": True, "delete": False}},
|
||||
field_permissions={"annual_revenue": "hidden"},
|
||||
)
|
||||
db.add(custom_role)
|
||||
await db.flush()
|
||||
|
||||
# Create a company in tenant A
|
||||
company_a = Company(
|
||||
tenant_id=tenant_a.id,
|
||||
name="Company Alpha",
|
||||
industry="IT",
|
||||
created_by=admin_a.id,
|
||||
updated_by=admin_a.id,
|
||||
)
|
||||
# Create a company in tenant B
|
||||
company_b = Company(
|
||||
tenant_id=tenant_b.id,
|
||||
name="Company Beta",
|
||||
industry="Finance",
|
||||
created_by=admin_b.id,
|
||||
updated_by=admin_b.id,
|
||||
)
|
||||
db.add_all([company_a, company_b])
|
||||
await db.flush()
|
||||
|
||||
await db.commit()
|
||||
|
||||
return {
|
||||
"user": data["user"],
|
||||
"token": token,
|
||||
"headers": {"Authorization": f"Bearer {token}"},
|
||||
"email": payload["email"],
|
||||
"password": payload["password"],
|
||||
"name": payload["name"],
|
||||
"tenant_a": tenant_a,
|
||||
"tenant_b": tenant_b,
|
||||
"admin_a": admin_a,
|
||||
"viewer_a": viewer_a,
|
||||
"editor_a": editor_a,
|
||||
"admin_b": admin_b,
|
||||
"company_a": company_a,
|
||||
"company_b": company_b,
|
||||
"custom_role": custom_role,
|
||||
}
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def auth_headers(registered_user: dict[str, Any]) -> dict[str, str]:
|
||||
"""Authorization headers for the bootstrap user."""
|
||||
return registered_user["headers"]
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def second_user(
|
||||
client: AsyncClient,
|
||||
registered_user: dict[str, Any],
|
||||
) -> dict[str, Any]:
|
||||
"""Register a second user (admin-only flow) and return its auth info."""
|
||||
# Use admin headers to create another user
|
||||
admin_headers = registered_user["headers"]
|
||||
payload = {
|
||||
"email": "rep@test.com",
|
||||
"password": "RepPass123!",
|
||||
"name": "Test Rep",
|
||||
"role": "sales_rep",
|
||||
}
|
||||
resp = await client.post("/api/v1/users/", json=payload, headers=admin_headers)
|
||||
assert resp.status_code == 201, f"create user failed: {resp.status_code} {resp.text}"
|
||||
new_user = resp.json()
|
||||
|
||||
# Log in as the new user to get a token
|
||||
login_resp = await client.post(
|
||||
async def login_client(client: AsyncClient, email: str, password: str = "TestPass123!") -> dict[str, str]:
|
||||
"""Login via HTTP API and return cookies dict."""
|
||||
resp = await client.post(
|
||||
"/api/v1/auth/login",
|
||||
data={"username": payload["email"], "password": payload["password"]},
|
||||
json={"email": email, "password": password},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert login_resp.status_code == 200, (
|
||||
f"login failed: {login_resp.status_code} {login_resp.text}"
|
||||
)
|
||||
token = login_resp.json()["access_token"]
|
||||
|
||||
return {
|
||||
"user": new_user,
|
||||
"token": token,
|
||||
"headers": {"Authorization": f"Bearer {token}"},
|
||||
"email": payload["email"],
|
||||
"password": payload["password"],
|
||||
"name": payload["name"],
|
||||
}
|
||||
assert resp.status_code == 200, f"Login failed: {resp.status_code} {resp.text}"
|
||||
return dict(resp.cookies)
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def seed_data(
|
||||
client: AsyncClient,
|
||||
registered_user: dict[str, Any],
|
||||
session_factory: async_sessionmaker[AsyncSession],
|
||||
) -> dict[str, Any]:
|
||||
"""Seed a representative data set for business-logic tests.
|
||||
|
||||
Creates (via the API to exercise FK constraints end-to-end):
|
||||
- 2 accounts owned by the bootstrap admin user
|
||||
- 3 contacts (2 attached to accounts, 1 standalone)
|
||||
- 5 deals across various stages
|
||||
- 10 activities (some overdue, some completed)
|
||||
- 3 tags (VIP, Strategic, Enterprise)
|
||||
- 4 notes (mixed parents)
|
||||
|
||||
Returns a dict with all IDs + a ref to the auth headers for convenience.
|
||||
"""
|
||||
headers = registered_user["headers"]
|
||||
|
||||
# 2 accounts
|
||||
acc1 = await client.post(
|
||||
"/api/v1/accounts/",
|
||||
json={
|
||||
"name": "Acme Corp",
|
||||
"industry": "sme",
|
||||
"size": "sme",
|
||||
"website": "https://acme.test",
|
||||
},
|
||||
headers=headers,
|
||||
)
|
||||
assert acc1.status_code == 201, f"seed acc1: {acc1.status_code} {acc1.text}"
|
||||
acc1_id = acc1.json()["id"]
|
||||
|
||||
acc2 = await client.post(
|
||||
"/api/v1/accounts/",
|
||||
json={"name": "Globex GmbH", "industry": "enterprise", "size": "enterprise"},
|
||||
headers=headers,
|
||||
)
|
||||
assert acc2.status_code == 201, f"seed acc2: {acc2.status_code} {acc2.text}"
|
||||
acc2_id = acc2.json()["id"]
|
||||
|
||||
# 3 contacts (2 with account, 1 standalone)
|
||||
c1 = await client.post(
|
||||
"/api/v1/contacts/",
|
||||
json={
|
||||
"first_name": "Anna",
|
||||
"last_name": "Schmidt",
|
||||
"email": "anna@acme.example",
|
||||
"account_id": acc1_id,
|
||||
},
|
||||
headers=headers,
|
||||
)
|
||||
assert c1.status_code == 201, c1.text
|
||||
c1_id = c1.json()["id"]
|
||||
|
||||
c2 = await client.post(
|
||||
"/api/v1/contacts/",
|
||||
json={
|
||||
"first_name": "Bob",
|
||||
"last_name": "Mueller",
|
||||
"email": "bob@globex.example",
|
||||
"account_id": acc2_id,
|
||||
},
|
||||
headers=headers,
|
||||
)
|
||||
assert c2.status_code == 201, c2.text
|
||||
c2_id = c2.json()["id"]
|
||||
|
||||
c3 = await client.post(
|
||||
"/api/v1/contacts/",
|
||||
json={"first_name": "Clara", "last_name": "Weber", "email": "clara@x.example"},
|
||||
headers=headers,
|
||||
)
|
||||
assert c3.status_code == 201, c3.text
|
||||
c3_id = c3.json()["id"]
|
||||
|
||||
# 5 deals (different stages)
|
||||
deal_ids: list[int] = []
|
||||
for i, (title, stage) in enumerate(
|
||||
[
|
||||
("Deal A", "lead"),
|
||||
("Deal B", "qualified"),
|
||||
("Deal C", "proposal"),
|
||||
("Deal D", "negotiation"),
|
||||
("Deal E", "won"),
|
||||
]
|
||||
):
|
||||
d = await client.post(
|
||||
"/api/v1/deals/",
|
||||
json={
|
||||
"title": title,
|
||||
"value": str(1000 * (i + 1)),
|
||||
"stage": stage,
|
||||
"account_id": acc1_id if i % 2 == 0 else acc2_id,
|
||||
},
|
||||
headers=headers,
|
||||
)
|
||||
assert d.status_code == 201, d.text
|
||||
deal_ids.append(d.json()["id"])
|
||||
|
||||
# 10 activities (4 overdue, 4 future, 2 completed)
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
past = (datetime.now(UTC) - timedelta(days=2)).isoformat()
|
||||
future = (datetime.now(UTC) + timedelta(days=2)).isoformat()
|
||||
future2 = (datetime.now(UTC) + timedelta(days=10)).isoformat()
|
||||
activity_ids: list[int] = []
|
||||
for i in range(10):
|
||||
if i < 4:
|
||||
due = past
|
||||
body_data: dict[str, object] = {
|
||||
"type": "task",
|
||||
"subject": f"Overdue task {i}",
|
||||
"due_date": due,
|
||||
"deal_id": deal_ids[i % 5],
|
||||
}
|
||||
elif i < 8:
|
||||
due = future if i % 2 == 0 else future2
|
||||
body_data = {
|
||||
"type": "call",
|
||||
"subject": f"Upcoming call {i}",
|
||||
"due_date": due,
|
||||
"account_id": acc1_id,
|
||||
}
|
||||
else:
|
||||
body_data = {"type": "meeting", "subject": f"Done meeting {i}", "account_id": acc1_id}
|
||||
a = await client.post("/api/v1/activities/", json=body_data, headers=headers)
|
||||
assert a.status_code == 201, a.text
|
||||
activity_ids.append(a.json()["id"])
|
||||
|
||||
# 3 tags
|
||||
tag_ids: list[int] = []
|
||||
for name in ["VIP", "Strategic", "Enterprise"]:
|
||||
t = await client.post(
|
||||
"/api/v1/tags/", json={"name": name, "color": "#FF0080"}, headers=headers
|
||||
)
|
||||
assert t.status_code == 201, t.text
|
||||
tag_ids.append(t.json()["id"])
|
||||
|
||||
# 4 notes (mixed parents: 2 account, 1 contact, 1 deal)
|
||||
n1 = await client.post(
|
||||
"/api/v1/notes/",
|
||||
json={"body": "Note on Acme", "parent_type": "account", "parent_id": acc1_id},
|
||||
headers=headers,
|
||||
)
|
||||
assert n1.status_code == 201, n1.text
|
||||
n2 = await client.post(
|
||||
"/api/v1/notes/",
|
||||
json={"body": "Note on Globex", "parent_type": "account", "parent_id": acc2_id},
|
||||
headers=headers,
|
||||
)
|
||||
assert n2.status_code == 201, n2.text
|
||||
n3 = await client.post(
|
||||
"/api/v1/notes/",
|
||||
json={"body": "Note on Anna", "parent_type": "contact", "parent_id": c1_id},
|
||||
headers=headers,
|
||||
)
|
||||
assert n3.status_code == 201, n3.text
|
||||
n4 = await client.post(
|
||||
"/api/v1/notes/",
|
||||
json={"body": "Note on Deal A", "parent_type": "deal", "parent_id": deal_ids[0]},
|
||||
headers=headers,
|
||||
)
|
||||
assert n4.status_code == 201, n4.text
|
||||
|
||||
return {
|
||||
"account_ids": [acc1_id, acc2_id],
|
||||
"contact_ids": [c1_id, c2_id, c3_id],
|
||||
"deal_ids": deal_ids,
|
||||
"activity_ids": activity_ids,
|
||||
"tag_ids": tag_ids,
|
||||
"note_ids": [n1.json()["id"], n2.json()["id"], n3.json()["id"], n4.json()["id"]],
|
||||
"headers": headers,
|
||||
}
|
||||
async def get_auth_client(client: AsyncClient, email: str, password: str = "TestPass123!") -> AsyncClient:
|
||||
"""Return a client that's logged in."""
|
||||
await login_client(client, email, password)
|
||||
return client
|
||||
|
||||
@@ -1,113 +0,0 @@
|
||||
"""Tests for FR-2 (Account entity). All 8 acceptance criteria covered."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_account(client: AsyncClient, auth_headers: dict[str, str]) -> None:
|
||||
resp = await client.post(
|
||||
"/api/v1/accounts/",
|
||||
json={"name": "Acme Corp", "industry": "sme", "size": "sme"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
data = resp.json()
|
||||
assert data["name"] == "Acme Corp"
|
||||
assert data["industry"] == "sme"
|
||||
assert "id" in data and data["id"] > 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_account_missing_required(
|
||||
client: AsyncClient, auth_headers: dict[str, str]
|
||||
) -> None:
|
||||
resp = await client.post(
|
||||
"/api/v1/accounts/",
|
||||
json={"industry": "sme"}, # name missing
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_accounts_paginated(client: AsyncClient, seed_data: dict) -> None:
|
||||
# seed_data creates 2 accounts
|
||||
resp = await client.get("/api/v1/accounts/?limit=20", headers=seed_data["headers"])
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert isinstance(body, list)
|
||||
assert len(body) >= 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_accounts_filter_industry(client: AsyncClient, seed_data: dict) -> None:
|
||||
resp = await client.get("/api/v1/accounts/?industry=enterprise", headers=seed_data["headers"])
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert all(a["industry"] == "enterprise" for a in body)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_account_with_relations(client: AsyncClient, seed_data: dict) -> None:
|
||||
acc_id = seed_data["account_ids"][0]
|
||||
resp = await client.get(f"/api/v1/accounts/{acc_id}", headers=seed_data["headers"])
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert body["id"] == acc_id
|
||||
# All base fields are present (relations are reachable via dedicated endpoints)
|
||||
for key in ("id", "name", "industry", "size", "owner_id", "created_at", "updated_at"):
|
||||
assert key in body
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_account(client: AsyncClient, seed_data: dict) -> None:
|
||||
acc_id = seed_data["account_ids"][0]
|
||||
resp = await client.patch(
|
||||
f"/api/v1/accounts/{acc_id}",
|
||||
json={"name": "Acme Corporation (Updated)"},
|
||||
headers=seed_data["headers"],
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["name"] == "Acme Corporation (Updated)"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_soft_delete_account(client: AsyncClient, seed_data: dict) -> None:
|
||||
acc_id = seed_data["account_ids"][0]
|
||||
resp = await client.delete(f"/api/v1/accounts/{acc_id}", headers=seed_data["headers"])
|
||||
assert resp.status_code == 204
|
||||
# Subsequent GET should not find it (or returns 404 because deleted_at is set)
|
||||
get_resp = await client.get(f"/api/v1/accounts/{acc_id}", headers=seed_data["headers"])
|
||||
assert get_resp.status_code in (404, 200)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_db_write_account_no_password_field(
|
||||
client: AsyncClient, auth_headers: dict[str, str], session_factory
|
||||
) -> None:
|
||||
"""Accounts have no password-related field; ensure schema is plaintext business fields."""
|
||||
from sqlalchemy import text
|
||||
|
||||
resp = await client.post(
|
||||
"/api/v1/accounts/",
|
||||
json={"name": "Schema Test Co", "industry": "startup"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
|
||||
async with session_factory() as session:
|
||||
result = await session.execute(
|
||||
text("SELECT name, industry FROM accounts WHERE name='Schema Test Co'")
|
||||
)
|
||||
row = result.fetchone()
|
||||
assert row is not None
|
||||
assert row[0] == "Schema Test Co"
|
||||
assert row[1] == "startup"
|
||||
# No password / hash columns exist on accounts
|
||||
cols = await session.execute(text("PRAGMA table_info(accounts)"))
|
||||
names = {c[1] for c in cols.fetchall()}
|
||||
assert "password_hash" not in names
|
||||
assert "hashed_password" not in names
|
||||
@@ -1,63 +0,0 @@
|
||||
"""Tests for FR-5 (Activity entity, polymorphic parent)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_activity_with_polymorphic_fk(
|
||||
client: AsyncClient, auth_headers: dict[str, str], seed_data: dict
|
||||
) -> None:
|
||||
deal_id = seed_data["deal_ids"][0]
|
||||
resp = await client.post(
|
||||
"/api/v1/activities/",
|
||||
json={"type": "task", "subject": "Follow up", "deal_id": deal_id},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
body = resp.json()
|
||||
assert body["deal_id"] == deal_id
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_activities_filter_overdue(client: AsyncClient, seed_data: dict) -> None:
|
||||
# seed_data inserts 4 overdue activities
|
||||
resp = await client.get("/api/v1/activities/?overdue=true", headers=seed_data["headers"])
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
now = datetime.now() # naive: SQLite returns naive datetimes
|
||||
for a in body:
|
||||
if a.get("due_date") and a.get("completed_at") is None:
|
||||
due = a["due_date"]
|
||||
if due.endswith("Z"):
|
||||
due = due.replace("Z", "+00:00")
|
||||
assert datetime.fromisoformat(due) < now
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_complete_activity(client: AsyncClient, seed_data: dict) -> None:
|
||||
a_id = seed_data["activity_ids"][0]
|
||||
resp = await client.patch(
|
||||
f"/api/v1/activities/{a_id}/complete",
|
||||
json={"outcome": "Resolved via email"},
|
||||
headers=seed_data["headers"],
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
body = resp.json()
|
||||
assert body["completed_at"] is not None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_activity_requires_at_least_one_parent(
|
||||
client: AsyncClient, auth_headers: dict[str, str]
|
||||
) -> None:
|
||||
resp = await client.post(
|
||||
"/api/v1/activities/",
|
||||
json={"type": "task", "subject": "Orphan task"}, # no account/contact/deal
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
+162
-215
@@ -1,232 +1,179 @@
|
||||
"""Auth tests: covers all 9 FR-1 acceptance criteria."""
|
||||
"""Auth tests — ACs 1-9: login, logout, me, switch-tenant, password reset, CSRF."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
from jose import jwt
|
||||
|
||||
from app.core.config import get_settings
|
||||
|
||||
settings = get_settings()
|
||||
from tests.conftest import ORIGIN_HEADER, seed_tenant_and_users, login_client
|
||||
|
||||
|
||||
# === FR-1.1 / FR-1.2 / Akzeptanzkriterium 1: register success ===
|
||||
@pytest.mark.asyncio
|
||||
class TestAuthLogin:
|
||||
"""ACs 1-3: Login valid, invalid, me without session."""
|
||||
|
||||
|
||||
async def test_register_success(client: AsyncClient) -> None:
|
||||
"""AC #1: POST /api/v1/auth/register mit gültigem Payload → 201 + User-Objekt + JWT."""
|
||||
payload = {
|
||||
"email": "alice@example.com",
|
||||
"password": "SecurePass123!",
|
||||
"name": "Alice",
|
||||
}
|
||||
resp = await client.post("/api/v1/auth/register", json=payload)
|
||||
assert resp.status_code == 201, resp.text
|
||||
data = resp.json()
|
||||
assert "user" in data
|
||||
assert "access_token" in data
|
||||
assert data["token_type"] == "bearer"
|
||||
assert data["expires_in"] > 0
|
||||
# User object has the expected fields (no password_hash leaked)
|
||||
user = data["user"]
|
||||
assert user["email"] == "alice@example.com"
|
||||
assert user["name"] == "Alice"
|
||||
assert "password_hash" not in user
|
||||
assert "id" in user
|
||||
assert "org_id" in user
|
||||
|
||||
|
||||
# === FR-1.1 / Akzeptanzkriterium 2: register duplicate email ===
|
||||
|
||||
|
||||
async def test_register_duplicate_email(client: AsyncClient, registered_user: dict) -> None:
|
||||
"""AC #2: POST /api/v1/auth/register mit existierender Email → 409."""
|
||||
# registered_user already exists; trying again with same email (and 2nd user
|
||||
# would also be blocked by bootstrap). 409 is correct because of the email conflict.
|
||||
# But bootstrap is also blocked → 403 is also acceptable. We accept either.
|
||||
payload = {
|
||||
"email": registered_user["email"],
|
||||
"password": "AnotherPass123!",
|
||||
"name": "Dup User",
|
||||
}
|
||||
resp = await client.post("/api/v1/auth/register", json=payload)
|
||||
assert resp.status_code in (403, 409), (
|
||||
f"Expected 403 (bootstrap) or 409 (email), got {resp.status_code}: {resp.text}"
|
||||
)
|
||||
|
||||
|
||||
async def test_register_bootstrap_blocked_after_first(
|
||||
client: AsyncClient, registered_user: dict
|
||||
) -> None:
|
||||
"""AC: Zweiter POST /api/v1/auth/register nach erfolgreichem ersten → 403."""
|
||||
payload = {
|
||||
"email": "other@example.com",
|
||||
"password": "AnotherPass123!",
|
||||
"name": "Other User",
|
||||
}
|
||||
resp = await client.post("/api/v1/auth/register", json=payload)
|
||||
assert resp.status_code == 403, (
|
||||
f"Bootstrap should be blocked after first user, got {resp.status_code}: {resp.text}"
|
||||
)
|
||||
|
||||
|
||||
# === FR-1.2 / Akzeptanzkriterium 3: register weak password ===
|
||||
|
||||
|
||||
async def test_register_weak_password(client: AsyncClient) -> None:
|
||||
"""AC #3: Schwaches Passwort (< 8 Zeichen) → 422."""
|
||||
payload = {
|
||||
"email": "weak@example.com",
|
||||
"password": "short", # < 8 chars
|
||||
"name": "Weak",
|
||||
}
|
||||
resp = await client.post("/api/v1/auth/register", json=payload)
|
||||
assert resp.status_code == 422, resp.text
|
||||
|
||||
|
||||
# === FR-1.2 / Akzeptanzkriterium 4: login success ===
|
||||
|
||||
|
||||
async def test_login_success(client: AsyncClient, registered_user: dict) -> None:
|
||||
"""AC #4: POST /api/v1/auth/login mit korrekten Credentials → 200 + JWT."""
|
||||
resp = await client.post(
|
||||
"/api/v1/auth/login",
|
||||
data={
|
||||
"username": registered_user["email"],
|
||||
"password": registered_user["password"],
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
data = resp.json()
|
||||
assert "access_token" in data
|
||||
assert data["token_type"] == "bearer"
|
||||
assert data["expires_in"] > 0
|
||||
|
||||
|
||||
# === FR-1.2 / Akzeptanzkriterium 5: login wrong password ===
|
||||
|
||||
|
||||
async def test_login_wrong_password(client: AsyncClient, registered_user: dict) -> None:
|
||||
"""AC #5: POST /api/v1/auth/login mit falschem Passwort → 401."""
|
||||
resp = await client.post(
|
||||
"/api/v1/auth/login",
|
||||
data={
|
||||
"username": registered_user["email"],
|
||||
"password": "WrongPassword123!",
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 401, resp.text
|
||||
|
||||
|
||||
# === FR-1.2 / Akzeptanzkriterium 6: login nonexistent user ===
|
||||
|
||||
|
||||
async def test_login_nonexistent_user(client: AsyncClient) -> None:
|
||||
"""AC #6: POST /api/v1/auth/login mit nicht existierendem User → 401."""
|
||||
resp = await client.post(
|
||||
"/api/v1/auth/login",
|
||||
data={"username": "nobody@example.com", "password": "AnyPass123!"},
|
||||
)
|
||||
assert resp.status_code == 401, resp.text
|
||||
|
||||
|
||||
# === FR-1.6 / Akzeptanzkriterium 7: get /me with valid JWT ===
|
||||
|
||||
|
||||
async def test_get_me_with_valid_jwt(
|
||||
client: AsyncClient, auth_headers: dict[str, str], registered_user: dict
|
||||
) -> None:
|
||||
"""AC #7: GET /api/v1/users/me mit gültigem JWT → 200 + User-Daten (ohne password_hash)."""
|
||||
resp = await client.get("/api/v1/users/me", headers=auth_headers)
|
||||
assert resp.status_code == 200, resp.text
|
||||
data = resp.json()
|
||||
assert data["email"] == registered_user["email"]
|
||||
assert data["name"] == registered_user["name"]
|
||||
assert "password_hash" not in data
|
||||
|
||||
|
||||
# === FR-1.6 / Akzeptanzkriterium 8: get /me without JWT ===
|
||||
|
||||
|
||||
async def test_get_me_without_jwt(client: AsyncClient) -> None:
|
||||
"""AC #8: GET /api/v1/users/me ohne JWT → 401."""
|
||||
resp = await client.get("/api/v1/users/me")
|
||||
assert resp.status_code == 401, resp.text
|
||||
|
||||
|
||||
# === FR-1.6 / Akzeptanzkriterium 9: get /me with expired JWT ===
|
||||
|
||||
|
||||
async def test_get_me_with_expired_jwt(client: AsyncClient) -> None:
|
||||
"""AC #9: GET /api/v1/users/me mit expired JWT → 401 + Hinweis 'token_expired'."""
|
||||
# Forge an expired token using the same secret/algorithm
|
||||
expired_payload = {
|
||||
"sub": "1",
|
||||
"org_id": 1,
|
||||
"role": "admin",
|
||||
"exp": int(time.time()) - 3600, # 1h in the past
|
||||
"iat": int(time.time()) - 7200,
|
||||
}
|
||||
expired_token = jwt.encode(
|
||||
expired_payload, settings.AUTH_SECRET, algorithm=settings.JWT_ALGORITHM
|
||||
)
|
||||
resp = await client.get(
|
||||
"/api/v1/users/me",
|
||||
headers={"Authorization": f"Bearer {expired_token}"},
|
||||
)
|
||||
assert resp.status_code == 401, resp.text
|
||||
# The 401 body should signal token is invalid (we use 'token_expired_or_invalid')
|
||||
body = resp.json()
|
||||
assert "detail" in body
|
||||
assert (
|
||||
"token" in body["detail"].lower()
|
||||
or "expired" in body["detail"].lower()
|
||||
or "credential" in body["detail"].lower()
|
||||
), f"Expected token-related 401 detail, got: {body}"
|
||||
|
||||
|
||||
# === Bonus: password is stored hashed, not plaintext ===
|
||||
|
||||
|
||||
async def test_db_user_has_hashed_password(
|
||||
client: AsyncClient, registered_user: dict, session_factory
|
||||
) -> None:
|
||||
"""AC: DB-User wird mit gehashtem password_hash angelegt (kein Klartext)."""
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.models.user import User
|
||||
|
||||
async with session_factory() as session:
|
||||
result = await session.execute(select(User).where(User.email == registered_user["email"]))
|
||||
user = result.scalar_one()
|
||||
# bcrypt hashes start with $2b$ (or $2a$ for passlib), never plain text
|
||||
assert user.password_hash.startswith("$"), (
|
||||
f"Password hash should be a bcrypt string, got: {user.password_hash!r}"
|
||||
async def test_login_valid_returns_200_and_cookie(self, client: AsyncClient, db_session):
|
||||
"""AC 1: POST /api/v1/auth/login valid -> 200 + Set-Cookie leocrm_session."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
resp = await client.post(
|
||||
"/api/v1/auth/login",
|
||||
json={"email": "admin@tenanta.com", "password": "TestPass123!"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert user.password_hash != registered_user["password"]
|
||||
assert len(user.password_hash) > 50, (
|
||||
f"Bcrypt hash should be ~60 chars long, got {len(user.password_hash)}"
|
||||
assert resp.status_code == 200
|
||||
assert "leocrm_session" in resp.headers.get("set-cookie", "").lower() or \
|
||||
"leocrm_session" in str(resp.cookies)
|
||||
data = resp.json()
|
||||
assert data["email"] == "admin@tenanta.com"
|
||||
assert data["role"] == "admin"
|
||||
|
||||
async def test_login_invalid_returns_401(self, client: AsyncClient, db_session):
|
||||
"""AC 2: POST /api/v1/auth/login invalid -> 401."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
resp = await client.post(
|
||||
"/api/v1/auth/login",
|
||||
json={"email": "admin@tenanta.com", "password": "WrongPassword!"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 401
|
||||
|
||||
async def test_me_without_session_returns_401(self, client: AsyncClient):
|
||||
"""AC 3: GET /api/v1/auth/me without session -> 401."""
|
||||
resp = await client.get("/api/v1/auth/me")
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
# === Bonus: no default admin bootstrap on startup ===
|
||||
@pytest.mark.asyncio
|
||||
class TestAuthMe:
|
||||
"""AC 4: me with valid session."""
|
||||
|
||||
async def test_me_with_valid_session_returns_200(self, client: AsyncClient, db_session):
|
||||
"""AC 4: GET /api/v1/auth/me with valid session -> 200 + user+tenant JSON."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
resp = await client.get("/api/v1/auth/me")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["email"] == "admin@tenanta.com"
|
||||
assert data["role"] == "admin"
|
||||
assert "tenant_id" in data
|
||||
assert "tenant_name" in data
|
||||
|
||||
|
||||
async def test_no_default_admin_on_startup(client: AsyncClient, session_factory) -> None:
|
||||
"""AC: KEIN admin/admin Bootstrap-User beim App-Start (Frisch-DB = leer)."""
|
||||
from sqlalchemy import func, select
|
||||
@pytest.mark.asyncio
|
||||
class TestAuthLogout:
|
||||
"""AC 5: logout."""
|
||||
|
||||
from app.models.user import User
|
||||
async def test_logout_returns_200_and_invalidates_session(self, client: AsyncClient, db_session):
|
||||
"""AC 5: POST /api/v1/auth/logout -> 200, session invalidated."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
# Verify we're logged in
|
||||
resp = await client.get("/api/v1/auth/me")
|
||||
assert resp.status_code == 200
|
||||
# Logout
|
||||
resp = await client.post("/api/v1/auth/logout", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 200
|
||||
# Verify session is invalidated
|
||||
resp = await client.get("/api/v1/auth/me")
|
||||
assert resp.status_code == 401
|
||||
|
||||
# Fresh DB → no users
|
||||
async with session_factory() as session:
|
||||
result = await session.execute(select(func.count()).select_from(User))
|
||||
count = result.scalar_one()
|
||||
assert count == 0, f"Fresh DB should have 0 users, found {count}"
|
||||
|
||||
# Also check: no user with role=admin and well-known email
|
||||
result = await session.execute(select(User).where(User.role == "admin"))
|
||||
admins = result.scalars().all()
|
||||
assert len(admins) == 0, f"Fresh DB should have no admin users, found {len(admins)}"
|
||||
@pytest.mark.asyncio
|
||||
class TestPasswordReset:
|
||||
"""ACs 6-8: password reset request, confirm valid, confirm expired."""
|
||||
|
||||
async def test_password_reset_request_always_200(self, client: AsyncClient, db_session):
|
||||
"""AC 6: POST /api/v1/auth/password-reset/request -> always 200 (no user enumeration)."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
# Existing email
|
||||
resp = await client.post(
|
||||
"/api/v1/auth/password-reset/request",
|
||||
json={"email": "admin@tenanta.com"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
# Non-existing email — still 200
|
||||
resp = await client.post(
|
||||
"/api/v1/auth/password-reset/request",
|
||||
json={"email": "nonexistent@example.com"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
|
||||
async def test_password_reset_confirm_valid_token(self, client: AsyncClient, db_session):
|
||||
"""AC 7: POST /api/v1/auth/password-reset/confirm valid token -> 200."""
|
||||
from app.services.auth_service import auth_service
|
||||
await seed_tenant_and_users(db_session)
|
||||
raw_token = await auth_service.get_password_reset_token_raw(db_session, "admin@tenanta.com")
|
||||
await db_session.commit()
|
||||
assert raw_token is not None
|
||||
|
||||
resp = await client.post(
|
||||
"/api/v1/auth/password-reset/confirm",
|
||||
json={"token": raw_token, "new_password": "NewPass456!"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
|
||||
async def test_password_reset_confirm_expired_token(self, client: AsyncClient, db_session):
|
||||
"""AC 8: POST /api/v1/auth/password-reset/confirm expired token -> 400."""
|
||||
from app.services.auth_service import auth_service
|
||||
await seed_tenant_and_users(db_session)
|
||||
raw_token = await auth_service.create_expired_reset_token(db_session, "admin@tenanta.com")
|
||||
await db_session.commit()
|
||||
assert raw_token is not None
|
||||
|
||||
resp = await client.post(
|
||||
"/api/v1/auth/password-reset/confirm",
|
||||
json={"token": raw_token, "new_password": "NewPass456!"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestSwitchTenant:
|
||||
"""AC 9: switch-tenant."""
|
||||
|
||||
async def test_switch_tenant_returns_200(self, client: AsyncClient, db_session):
|
||||
"""AC 9: POST /api/v1/auth/switch-tenant -> 200, session tenant_id updated."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
|
||||
# Get initial tenant
|
||||
resp = await client.get("/api/v1/auth/me")
|
||||
assert resp.status_code == 200
|
||||
initial_tenant = resp.json()["tenant_id"]
|
||||
|
||||
# Get tenant B ID
|
||||
from app.models.tenant import Tenant
|
||||
from sqlalchemy import select
|
||||
q = select(Tenant).where(Tenant.slug == "tenant-b")
|
||||
result = await db_session.execute(q)
|
||||
tenant_b = result.scalar_one()
|
||||
|
||||
# Switch to tenant B
|
||||
resp = await client.post(
|
||||
"/api/v1/auth/switch-tenant",
|
||||
json={"tenant_id": str(tenant_b.id)},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["tenant_id"] == str(tenant_b.id)
|
||||
assert resp.json()["tenant_id"] != initial_tenant
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestCSRF:
|
||||
"""AC 23: CSRF — POST without Origin header -> 403."""
|
||||
|
||||
async def test_post_without_origin_returns_403(self, client: AsyncClient, db_session):
|
||||
"""AC 23: POST without Origin header -> 403."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
resp = await client.post(
|
||||
"/api/v1/auth/login",
|
||||
json={"email": "admin@tenanta.com", "password": "TestPass123!"},
|
||||
# No Origin header
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
@@ -1,56 +0,0 @@
|
||||
"""Tests for FR-3 (Contact entity)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_contact_with_account(
|
||||
client: AsyncClient, auth_headers: dict[str, str], seed_data: dict
|
||||
) -> None:
|
||||
acc_id = seed_data["account_ids"][0]
|
||||
resp = await client.post(
|
||||
"/api/v1/contacts/",
|
||||
json={
|
||||
"first_name": "Diana",
|
||||
"last_name": "Prince",
|
||||
"email": "diana@x.example",
|
||||
"account_id": acc_id,
|
||||
},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
body = resp.json()
|
||||
assert body["account_id"] == acc_id
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_contact_invalid_account(
|
||||
client: AsyncClient, auth_headers: dict[str, str]
|
||||
) -> None:
|
||||
resp = await client.post(
|
||||
"/api/v1/contacts/",
|
||||
json={"first_name": "Eve", "last_name": "Adams", "account_id": 99999},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_contacts_filter_account(client: AsyncClient, seed_data: dict) -> None:
|
||||
acc_id = seed_data["account_ids"][0]
|
||||
resp = await client.get(f"/api/v1/contacts/?account_id={acc_id}", headers=seed_data["headers"])
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert all(c["account_id"] == acc_id for c in body)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_contacts_by_email(client: AsyncClient, seed_data: dict) -> None:
|
||||
# seed_data creates contact with email 'anna@acme.example' on account 0
|
||||
resp = await client.get("/api/v1/contacts/?q=anna@acme.example", headers=seed_data["headers"])
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert any("anna@acme.example" in (c.get("email") or "") for c in body)
|
||||
@@ -1,38 +0,0 @@
|
||||
"""Tests for FR-7 (Dashboard: KPIs + activity feed)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_kpis(client: AsyncClient, seed_data: dict) -> None:
|
||||
resp = await client.get("/api/v1/dashboard/kpis", headers=seed_data["headers"])
|
||||
assert resp.status_code == 200, resp.text
|
||||
body = resp.json()
|
||||
assert {"open_deals_count", "pipeline_value", "won_this_month", "conversion_rate"} <= set(
|
||||
body.keys()
|
||||
)
|
||||
assert isinstance(body["open_deals_count"], int)
|
||||
assert isinstance(body["pipeline_value"], (int, float))
|
||||
assert isinstance(body["won_this_month"], int)
|
||||
assert isinstance(body["conversion_rate"], (int, float))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_activity_feed(client: AsyncClient, seed_data: dict) -> None:
|
||||
resp = await client.get("/api/v1/dashboard/feed?limit=20", headers=seed_data["headers"])
|
||||
assert resp.status_code == 200, resp.text
|
||||
body = resp.json()
|
||||
assert isinstance(body, list)
|
||||
assert len(body) >= 1
|
||||
# Items are sorted by created_at desc (most recent first)
|
||||
if len(body) >= 2:
|
||||
assert body[0]["created_at"] >= body[-1]["created_at"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_dashboard_requires_auth(client: AsyncClient) -> None:
|
||||
resp = await client.get("/api/v1/dashboard/kpis")
|
||||
assert resp.status_code == 401
|
||||
@@ -1,90 +0,0 @@
|
||||
"""Tests for FR-4 (Deal entity) and stage-history audit."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_deal(
|
||||
client: AsyncClient, auth_headers: dict[str, str], seed_data: dict
|
||||
) -> None:
|
||||
acc_id = seed_data["account_ids"][0]
|
||||
resp = await client.post(
|
||||
"/api/v1/deals/",
|
||||
json={"title": "Big Deal", "value": "5000.00", "stage": "qualified", "account_id": acc_id},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
body = resp.json()
|
||||
assert body["title"] == "Big Deal"
|
||||
assert body["stage"] == "qualified"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_deal_stage_creates_history(
|
||||
client: AsyncClient, seed_data: dict, session_factory
|
||||
) -> None:
|
||||
deal_id = seed_data["deal_ids"][0]
|
||||
resp = await client.patch(
|
||||
f"/api/v1/deals/{deal_id}/stage",
|
||||
json={"stage": "qualified"},
|
||||
headers=seed_data["headers"],
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json()["stage"] == "qualified"
|
||||
|
||||
# Verify DealStageHistory row exists in DB
|
||||
from sqlalchemy import text
|
||||
|
||||
async with session_factory() as session:
|
||||
result = await session.execute(
|
||||
text("SELECT to_stage FROM deal_stage_history WHERE deal_id = :did ORDER BY id"),
|
||||
{"did": deal_id},
|
||||
)
|
||||
stages = [r[0] for r in result.fetchall()]
|
||||
assert "qualified" in stages
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_pipeline_grouped_by_stage(client: AsyncClient, seed_data: dict) -> None:
|
||||
resp = await client.get("/api/v1/deals/pipeline", headers=seed_data["headers"])
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
# Each entry has stage, count, total_value
|
||||
assert all({"stage", "count", "total_value"} <= set(item.keys()) for item in body)
|
||||
# Seeded deals cover at least 5 distinct stages
|
||||
stages = {item["stage"] for item in body}
|
||||
assert len(stages) >= 3
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_filter_deals_by_owner(client: AsyncClient, seed_data: dict) -> None:
|
||||
owner_id = seed_data["headers"].get("X-Test-User-Id") # may be None; fallback: just check 200
|
||||
# we filter by an owner-id that does not exist; expect empty list
|
||||
resp = await client.get("/api/v1/deals/?owner_id=99999", headers=seed_data["headers"])
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert body == [] or all(d.get("owner_id") == 99999 for d in body)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_db_write_deal(
|
||||
client: AsyncClient, auth_headers: dict[str, str], seed_data: dict, session_factory
|
||||
) -> None:
|
||||
"""DB roundtrip: read back a seeded deal directly via SQL."""
|
||||
from sqlalchemy import text
|
||||
|
||||
deal_id = seed_data["deal_ids"][2]
|
||||
async with session_factory() as session:
|
||||
result = await session.execute(
|
||||
text("SELECT title, value, stage, account_id FROM deals WHERE id = :did"),
|
||||
{"did": deal_id},
|
||||
)
|
||||
row = result.fetchone()
|
||||
assert row is not None
|
||||
assert row[0] is not None # title
|
||||
assert row[1] is not None # value
|
||||
assert row[2] is not None # stage
|
||||
assert row[3] is not None # account_id
|
||||
@@ -1,184 +0,0 @@
|
||||
"""Frontend-Asset Tests: verify JS modules and CSS are syntactically valid
|
||||
and export / register the expected symbols.
|
||||
|
||||
These tests perform static text-level checks on the JS / CSS files — no
|
||||
runtime execution, no nodejs dependency. They are sufficient to catch:
|
||||
- Missing exports (e.g. `export { api, ApiError }` removed from api.js)
|
||||
- Missing global Alpine.data() registrations for the components
|
||||
- CSS variables removed from the custom-overrides file
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
WEBUI = Path(__file__).resolve().parent.parent / "app" / "webui"
|
||||
|
||||
|
||||
def _read(rel: str) -> str:
|
||||
p = WEBUI / rel
|
||||
assert p.exists(), f"Missing asset: {p}"
|
||||
return p.read_text(encoding="utf-8")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# api.js
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_api_js_exports_api_and_ApiError():
|
||||
content = _read("js/api.js")
|
||||
assert "export class ApiError" in content
|
||||
assert "export const api" in content or "export default api" in content
|
||||
|
||||
|
||||
def test_api_js_uses_localStorage_for_jwt():
|
||||
"""R-1 / architecture: JWT must live in localStorage."""
|
||||
content = _read("js/api.js")
|
||||
assert "localStorage" in content
|
||||
assert "jwt" in content
|
||||
|
||||
|
||||
def test_api_js_handles_401_redirect():
|
||||
"""401 must clear the token and redirect to /index.html."""
|
||||
content = _read("js/api.js")
|
||||
assert "401" in content
|
||||
assert "/index.html" in content
|
||||
|
||||
|
||||
def test_api_js_handles_204_no_content():
|
||||
content = _read("js/api.js")
|
||||
assert "204" in content
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# store.js
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_store_js_defines_auth_and_notifications():
|
||||
content = _read("js/store.js")
|
||||
assert "Alpine.store('auth'" in content
|
||||
assert "Alpine.store('notifications'" in content
|
||||
|
||||
|
||||
def test_store_js_auth_has_login_and_logout():
|
||||
content = _read("js/store.js")
|
||||
assert "login" in content
|
||||
assert "logout" in content
|
||||
assert "fetchUser" in content
|
||||
|
||||
|
||||
def test_store_js_notifications_has_push_and_dismiss():
|
||||
content = _read("js/store.js")
|
||||
assert "push" in content
|
||||
assert "dismiss" in content
|
||||
assert "success" in content
|
||||
assert "error" in content
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# app.css
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_app_css_contains_tailwind_overrides():
|
||||
content = _read("css/app.css")
|
||||
# The file defines custom CSS variables (--crm-primary, etc.)
|
||||
assert "--crm-primary" in content
|
||||
assert "--crm-success" in content
|
||||
assert "--crm-danger" in content
|
||||
|
||||
|
||||
def test_app_css_has_loading_spinner():
|
||||
content = _read("css/app.css")
|
||||
assert "crm-spinner" in content
|
||||
assert "@keyframes crm-spin" in content
|
||||
|
||||
|
||||
def test_app_css_has_toast_styles():
|
||||
content = _read("css/app.css")
|
||||
assert "crm-toast" in content
|
||||
assert "toast-success" in content
|
||||
assert "toast-error" in content
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Alpine components
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
COMPONENTS = [
|
||||
("account-list.js", "accountList"),
|
||||
("deal-kanban.js", "dealKanban"),
|
||||
("activity-list.js", "activityList"),
|
||||
("dashboard-kpis.js", "dashboardKpis"),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("filename,factory", COMPONENTS)
|
||||
def test_alpine_component_defined(filename, factory):
|
||||
"""Each component file must define a named factory and register it
|
||||
globally with Alpine.data(name, factory)."""
|
||||
content = _read(f"components/{filename}")
|
||||
# factory function definition
|
||||
assert f"export function {factory}" in content, (
|
||||
f"{filename} missing `export function {factory}()`"
|
||||
)
|
||||
# Alpine.data() registration
|
||||
assert f"Alpine.data('{factory}'" in content, (
|
||||
f"{filename} missing `Alpine.data('{factory}', …)` registration"
|
||||
)
|
||||
|
||||
|
||||
def test_deal_kanban_handles_drag_and_drop():
|
||||
content = _read("components/deal-kanban.js")
|
||||
assert "draggable" in content or "dataTransfer" in content
|
||||
assert "onDrop" in content
|
||||
assert "/deals/" in content and "/stage" in content
|
||||
|
||||
|
||||
def test_activity_list_supports_overdue_filter():
|
||||
content = _read("components/activity-list.js")
|
||||
assert "overdue" in content
|
||||
assert "/activities/" in content
|
||||
assert "/complete" in content
|
||||
|
||||
|
||||
def test_dashboard_kpis_loads_both_kpis_and_feed():
|
||||
content = _read("components/dashboard-kpis.js")
|
||||
assert "/dashboard/kpis" in content
|
||||
assert "/dashboard/feed" in content
|
||||
|
||||
|
||||
def test_account_list_uses_pagination():
|
||||
content = _read("components/account-list.js")
|
||||
assert "skip" in content
|
||||
assert "limit" in content
|
||||
assert "q" in content
|
||||
assert "/accounts/" in content
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# notifications.js (toast component)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_notifications_js_registers_toast_container():
|
||||
content = _read("js/components/notifications.js")
|
||||
assert "toastContainer" in content
|
||||
assert "Alpine.data" in content
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# auth.js
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_auth_js_exports_login_logout_register():
|
||||
content = _read("js/auth.js")
|
||||
assert "export async function login" in content
|
||||
assert "export async function logout" in content
|
||||
assert "export async function register" in content
|
||||
assert "export async function refreshToken" in content
|
||||
@@ -1,111 +0,0 @@
|
||||
"""Frontend-Smoke-Tests: verify all 13 HTML pages exist and contain the
|
||||
expected keywords/structure.
|
||||
|
||||
These tests read the HTML files directly from `app/webui/` (rather than
|
||||
hitting the FastAPI server) because the static-file mount is added in a
|
||||
later phase by the orchestrator. Once that mount is in place, the same
|
||||
checks could be re-implemented as `httpx.AsyncClient.get("/…")`.
|
||||
|
||||
Each test loads exactly one page and asserts:
|
||||
- file exists and is non-empty
|
||||
- contains the page's title text (so a typo / removal is caught)
|
||||
- has Tailwind-CDN, Alpine-CDN, and the app.css link
|
||||
- has no x-html attribute (R-5)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
WEBUI_DIR = Path(__file__).resolve().parent.parent / "app" / "webui"
|
||||
|
||||
|
||||
def _read(name: str) -> str:
|
||||
"""Read a file from the webui directory."""
|
||||
p = WEBUI_DIR / name
|
||||
assert p.exists(), f"Expected frontend file {p} to exist"
|
||||
return p.read_text(encoding="utf-8")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Each page: (filename, [required substrings])
|
||||
# ---------------------------------------------------------------------------
|
||||
PAGES = [
|
||||
("index.html", ["Login", "Registrieren", "Anmelden", "Konto erstellen"]),
|
||||
("app.html", ["Dashboard", "Logout", "Pipeline"]),
|
||||
("accounts.html", ["Accounts", "Suche (Name)", "Branche", "+ New Account"]),
|
||||
("accounts-detail.html", ["Account-Detail", "Bearbeiten", "Löschen", "Info"]),
|
||||
("contacts.html", ["Contacts", "Vorname", "Nachname", "+ New Contact"]),
|
||||
("contacts-detail.html", ["Contact-Detail", "Bearbeiten", "Löschen"]),
|
||||
("pipeline.html", ["Pipeline", "Owner-ID (optional)", "+ New Deal", "Ziehe Karten"]),
|
||||
("activities.html", ["Activities", "+ New Activity", "Fällig"]),
|
||||
("settings-profile.html", ["Mein Profil", "Avatar-URL", "E-Mail-Benachrichtigungen"]),
|
||||
("settings-users.html", ["User-Verwaltung", "+ New User", "Rolle"]),
|
||||
("settings-org.html", ["Org-Einstellungen", "Aktuelle Organisation", "v1.1"]),
|
||||
("404.html", ["404", "Seite nicht gefunden", "Zum Dashboard"]),
|
||||
# Dashboard: ships inside app.html. We test the marker text directly there.
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("filename,required", PAGES)
|
||||
def test_frontend_page_loads(filename, required):
|
||||
"""Each HTML page must exist and contain its identifying substrings."""
|
||||
content = _read(filename)
|
||||
assert len(content) > 200, f"{filename} seems empty or too small"
|
||||
for needle in required:
|
||||
assert needle in content, f"{filename} missing required string {needle!r}"
|
||||
|
||||
|
||||
def test_index_html_contains_login_and_register():
|
||||
"""Index page (auth screen) must show both Login and Register tabs."""
|
||||
content = _read("index.html")
|
||||
assert "Login" in content
|
||||
assert "Registrieren" in content
|
||||
# Alpine-CDN required
|
||||
assert "alpinejs" in content
|
||||
|
||||
|
||||
def test_app_html_contains_dashboard_and_logout():
|
||||
"""app.html is the post-login shell — must show Dashboard + Logout."""
|
||||
content = _read("app.html")
|
||||
assert "Dashboard" in content
|
||||
assert "Logout" in content
|
||||
|
||||
|
||||
def test_13_pages_exist():
|
||||
"""Exactly 13 HTML pages must be present (index + app + 10 + 404)."""
|
||||
pages = sorted(p.name for p in WEBUI_DIR.glob("*.html"))
|
||||
assert len(pages) == 13, f"Expected 13 pages, found {len(pages)}: {pages}"
|
||||
|
||||
|
||||
def test_dashboard_section_inside_app_html():
|
||||
"""The dashboard content is rendered inside app.html (default landing)."""
|
||||
content = _read("app.html")
|
||||
# The 4 KPI labels from the dashboard section
|
||||
assert "Open Deals" in content
|
||||
assert "Pipeline Value" in content
|
||||
assert "Won this Month" in content
|
||||
assert "Conversion Rate" in content
|
||||
|
||||
|
||||
def test_each_page_links_app_css():
|
||||
"""Every page must reference /css/app.css."""
|
||||
for filename, _ in PAGES:
|
||||
content = _read(filename)
|
||||
assert "/css/app.css" in content, f"{filename} missing /css/app.css link"
|
||||
|
||||
|
||||
def test_each_page_includes_alpine_cdn():
|
||||
"""Every page must include the Alpine.js CDN script."""
|
||||
for filename, _ in PAGES:
|
||||
content = _read(filename)
|
||||
assert "alpinejs" in content, f"{filename} missing Alpine.js CDN"
|
||||
|
||||
|
||||
def test_each_page_includes_tailwind_cdn():
|
||||
"""Every page must include the Tailwind-CDN script (dev-mode)."""
|
||||
for filename, _ in PAGES:
|
||||
content = _read(filename)
|
||||
assert "cdn.tailwindcss.com" in content, f"{filename} missing Tailwind CDN"
|
||||
@@ -1,183 +0,0 @@
|
||||
"""Frontend-Security Tests: XSS-mitigation, JWT-handling, CSP-middleware.
|
||||
|
||||
These tests enforce the rules from architecture Section 13 + task graph R-5:
|
||||
|
||||
- R-5: No `x-html` attribute anywhere in any HTML template (only x-text).
|
||||
- R-1: JWT must be read from / written to localStorage (not cookies, not sessionStorage).
|
||||
- 13.3: CSP-Header is configured in app/main.py (security_headers_middleware),
|
||||
not in a reverse-proxy config file.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
WEBUI = ROOT / "app" / "webui"
|
||||
MAIN_PY = ROOT / "app" / "main.py"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# R-5: no x-html in any HTML template
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# Match `x-html` as a whole word (attribute, not part of `x-htmlSomething`).
|
||||
# We deliberately allow `x-html="..."` to fail loudly, and we also catch
|
||||
# any camel-cased variant like `xHtml` for paranoia.
|
||||
XHTML_PATTERN = re.compile(r"\bx[-_]?html\b", re.IGNORECASE)
|
||||
|
||||
|
||||
def _all_html_files() -> list[Path]:
|
||||
return sorted(WEBUI.glob("*.html"))
|
||||
|
||||
|
||||
def test_no_x_html_in_alpine_templates():
|
||||
"""R-5: no `x-html` attribute anywhere in any HTML file."""
|
||||
offenders: list[tuple[str, int, str]] = []
|
||||
for path in _all_html_files():
|
||||
for lineno, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
|
||||
# Strip HTML comments to avoid false positives in documentation
|
||||
stripped = re.sub(r"<!--.*?-->", "", line)
|
||||
if XHTML_PATTERN.search(stripped):
|
||||
offenders.append((path.name, lineno, line.strip()[:120]))
|
||||
assert not offenders, "x-html usage is forbidden (R-5); offenders:\n" + "\n".join(
|
||||
f" {n}:{ln}: {snippet}" for n, ln, snippet in offenders
|
||||
)
|
||||
|
||||
|
||||
def test_no_innerHTML_in_alpine_pages():
|
||||
"""Defence in depth: also flag direct `innerHTML` assignments."""
|
||||
for path in _all_html_files():
|
||||
content = path.read_text(encoding="utf-8")
|
||||
# allow within <script> blocks only if commented out / disabled
|
||||
assert "innerHTML" not in content, (
|
||||
f"{path.name} contains innerHTML which is forbidden (R-5)"
|
||||
)
|
||||
|
||||
|
||||
def test_x_text_is_used_instead():
|
||||
"""Sanity check: the auth/dashboard pages use x-text for user-derived strings."""
|
||||
index = (WEBUI / "index.html").read_text(encoding="utf-8")
|
||||
# The error message div should use x-text (so user input never gets HTML-parsed)
|
||||
assert 'x-text="error"' in index or 'x-text="error"' in index
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# R-1: JWT in localStorage
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_jwt_uses_localStorage():
|
||||
"""api.js must read/write the JWT in localStorage, not cookies/sessionStorage."""
|
||||
api = (WEBUI / "js" / "api.js").read_text(encoding="utf-8")
|
||||
# Must read the token from localStorage
|
||||
assert "localStorage.getItem('jwt')" in api
|
||||
# Must persist the token back to localStorage on login
|
||||
assert "localStorage.setItem('jwt'" in api
|
||||
# Must clear on logout
|
||||
assert "localStorage.removeItem('jwt')" in api
|
||||
|
||||
|
||||
def test_jwt_not_in_sessionStorage():
|
||||
"""Defence in depth: do NOT use sessionStorage for the JWT."""
|
||||
api = (WEBUI / "js" / "api.js").read_text(encoding="utf-8")
|
||||
assert "sessionStorage" not in api, "api.js must not use sessionStorage for the JWT"
|
||||
|
||||
|
||||
def test_jwt_not_in_cookies():
|
||||
"""Defence in depth: do NOT put the JWT in document.cookie."""
|
||||
for js in WEBUI.rglob("*.js"):
|
||||
content = js.read_text(encoding="utf-8")
|
||||
assert "document.cookie" not in content, (
|
||||
f"{js.relative_to(ROOT)} must not use document.cookie for the JWT"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 13.3: CSP-Header in main.py
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_csp_header_in_main_py():
|
||||
"""app/main.py must wire up a middleware (or response-header decorator)
|
||||
that sets Content-Security-Policy. We don't run the server here — we
|
||||
just verify the configuration is present in code."""
|
||||
assert MAIN_PY.exists(), f"Missing {MAIN_PY}"
|
||||
content = MAIN_PY.read_text(encoding="utf-8")
|
||||
# The middleware / decorator must mention the CSP header value
|
||||
assert "Content-Security-Policy" in content, "app/main.py does not set Content-Security-Policy"
|
||||
# The header must allow the Tailwind CDN at minimum
|
||||
assert "cdn.tailwindcss.com" in content, (
|
||||
"CSP-Header must allow https://cdn.tailwindcss.com in script-src"
|
||||
)
|
||||
# And block scripts from arbitrary origins
|
||||
assert "default-src 'self'" in content or 'default-src "self"' in content, (
|
||||
"CSP-Header must set default-src 'self'"
|
||||
)
|
||||
|
||||
|
||||
def test_csp_blocks_object_embedding():
|
||||
"""object-src 'none' is part of the agreed lockdown."""
|
||||
content = MAIN_PY.read_text(encoding="utf-8")
|
||||
assert "object-src 'none'" in content, "CSP-Header must set object-src 'none'"
|
||||
|
||||
|
||||
def test_csp_uses_starlette_middleware():
|
||||
"""Per architecture: CSP is set in a FastAPI/Starlette middleware,
|
||||
not in nginx / a response handler."""
|
||||
content = MAIN_PY.read_text(encoding="utf-8")
|
||||
assert '@app.middleware("http")' in content or '@app.middleware("http")' in content, (
|
||||
'CSP-Header must be set in an @app.middleware("http") decorator'
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Auth-gate on protected pages
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
PROTECTED_PAGES = [
|
||||
"accounts.html",
|
||||
"accounts-detail.html",
|
||||
"contacts.html",
|
||||
"contacts-detail.html",
|
||||
"pipeline.html",
|
||||
"activities.html",
|
||||
"settings-profile.html",
|
||||
"settings-users.html",
|
||||
"settings-org.html",
|
||||
"app.html",
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("filename", PROTECTED_PAGES)
|
||||
def test_protected_page_has_auth_gate(filename):
|
||||
"""Each non-public page must call Alpine.store('auth').init() in its
|
||||
outermost x-data, which acts as the auth-gate."""
|
||||
content = (WEBUI / filename).read_text(encoding="utf-8")
|
||||
# The auth-gate is `x-data="{ init: () => Alpine.store('auth').init() }"`
|
||||
# or `x-init="init()"` on a component that checks the JWT.
|
||||
assert "Alpine.store('auth').init()" in content or "$store.auth.init()" in content, (
|
||||
f"{filename} has no Alpine.store('auth').init() auth-gate"
|
||||
)
|
||||
|
||||
|
||||
def test_index_page_has_no_auth_gate():
|
||||
"""index.html is the LOGIN page itself — it must NOT redirect to itself."""
|
||||
content = (WEBUI / "index.html").read_text(encoding="utf-8")
|
||||
# The auth store's init() redirects to /index.html when no JWT exists,
|
||||
# so it must NOT be called from the login page.
|
||||
assert "Alpine.store('auth').init()" not in content, (
|
||||
"index.html (login page) must not call auth.init() (would cause loop)"
|
||||
)
|
||||
|
||||
|
||||
def test_404_page_has_no_auth_gate():
|
||||
"""404 page is reachable without auth (so users can find their way back)."""
|
||||
content = (WEBUI / "404.html").read_text(encoding="utf-8")
|
||||
# The 404 must not call auth.init() — it is intentionally public.
|
||||
assert "Alpine.store('auth').init()" not in content, (
|
||||
"404.html must not call auth.init() (public page)"
|
||||
)
|
||||
+12
-38
@@ -1,45 +1,19 @@
|
||||
"""Health endpoint tests."""
|
||||
"""Health endpoint test — AC 18."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
|
||||
async def test_health_ok(client: AsyncClient) -> None:
|
||||
"""GET /health returns 200 with status, db, version fields."""
|
||||
resp = await client.get("/health")
|
||||
assert resp.status_code == 200, resp.text
|
||||
data = resp.json()
|
||||
assert data["status"] == "ok"
|
||||
assert data["db"] == "ok"
|
||||
assert "version" in data
|
||||
assert isinstance(data["version"], str)
|
||||
@pytest.mark.asyncio
|
||||
class TestHealth:
|
||||
"""AC 18: GET /api/v1/health -> 200 without auth."""
|
||||
|
||||
|
||||
async def test_health_no_auth_required(client: AsyncClient) -> None:
|
||||
"""GET /health works without an Authorization header (for Coolify healthcheck)."""
|
||||
# No Authorization header at all
|
||||
resp = await client.get("/health")
|
||||
assert resp.status_code == 200, resp.text
|
||||
# Even with a bogus header, it should still be 200 (public endpoint)
|
||||
resp = await client.get("/health", headers={"Authorization": "Bearer not-a-real-token"})
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
|
||||
async def test_api_v1_health_ok(client: AsyncClient) -> None:
|
||||
"""GET /api/v1/health returns 200 (versioned healthcheck)."""
|
||||
resp = await client.get("/api/v1/health")
|
||||
assert resp.status_code == 200, resp.text
|
||||
data = resp.json()
|
||||
assert data["status"] == "ok"
|
||||
assert data["db"] == "ok"
|
||||
assert "version" in data
|
||||
|
||||
|
||||
async def test_security_headers_on_health(client: AsyncClient) -> None:
|
||||
"""Security headers (CSP, X-Frame-Options, X-Content-Type-Options) are set on responses."""
|
||||
resp = await client.get("/health")
|
||||
assert resp.status_code == 200
|
||||
assert "Content-Security-Policy" in resp.headers
|
||||
assert resp.headers["X-Frame-Options"] == "DENY"
|
||||
assert resp.headers["X-Content-Type-Options"] == "nosniff"
|
||||
async def test_health_returns_200_without_auth(self, client: AsyncClient):
|
||||
"""AC 18: GET /api/v1/health -> 200 without auth."""
|
||||
resp = await client.get("/api/v1/health")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["status"] == "ok"
|
||||
assert "version" in data
|
||||
|
||||
@@ -1,69 +0,0 @@
|
||||
"""Tests for FR-6 + R-2 (Note entity, polymorphic parent validation)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_note_with_valid_parent(
|
||||
client: AsyncClient, auth_headers: dict[str, str], seed_data: dict
|
||||
) -> None:
|
||||
acc_id = seed_data["account_ids"][0]
|
||||
resp = await client.post(
|
||||
"/api/v1/notes/",
|
||||
json={"body": "Great call", "parent_type": "account", "parent_id": acc_id},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
body = resp.json()
|
||||
assert body["parent_type"] == "account"
|
||||
assert body["parent_id"] == acc_id
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_note_with_invalid_parent(
|
||||
client: AsyncClient, auth_headers: dict[str, str]
|
||||
) -> None:
|
||||
"""R-2: parent_type+parent_id must point to an existing entity; else 404."""
|
||||
resp = await client.post(
|
||||
"/api/v1/notes/",
|
||||
json={"body": "Note into the void", "parent_type": "account", "parent_id": 99999},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_notes_for_parent(client: AsyncClient, seed_data: dict) -> None:
|
||||
acc_id = seed_data["account_ids"][0]
|
||||
resp = await client.get(
|
||||
f"/api/v1/notes/?parent_type=account&parent_id={acc_id}",
|
||||
headers=seed_data["headers"],
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert all(n["parent_type"] == "account" and n["parent_id"] == acc_id for n in body)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_polymorphic_note_for_account_vs_contact(
|
||||
client: AsyncClient, auth_headers: dict[str, str], seed_data: dict
|
||||
) -> None:
|
||||
acc_id = seed_data["account_ids"][0]
|
||||
contact_id = seed_data["contact_ids"][0]
|
||||
n_acc = await client.post(
|
||||
"/api/v1/notes/",
|
||||
json={"body": "Account note", "parent_type": "account", "parent_id": acc_id},
|
||||
headers=auth_headers,
|
||||
)
|
||||
n_contact = await client.post(
|
||||
"/api/v1/notes/",
|
||||
json={"body": "Contact note", "parent_type": "contact", "parent_id": contact_id},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert n_acc.status_code == 201, n_acc.text
|
||||
assert n_contact.status_code == 201, n_contact.text
|
||||
assert n_acc.json()["parent_type"] == "account"
|
||||
assert n_contact.json()["parent_type"] == "contact"
|
||||
@@ -0,0 +1,99 @@
|
||||
"""Notification tests — ACs 24-26: list, mark read, unread count."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from tests.conftest import ORIGIN_HEADER, seed_tenant_and_users, login_client
|
||||
from app.core.notifications import create_notification
|
||||
from app.models.notification import Notification
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestNotifications:
|
||||
"""ACs 24-26: Notification list, mark read, unread count."""
|
||||
|
||||
async def test_list_notifications_returns_200(self, client: AsyncClient, db_session):
|
||||
"""AC 24: GET /api/v1/notifications -> 200 + unread first."""
|
||||
seed = await seed_tenant_and_users(db_session)
|
||||
# Create some notifications for admin_a
|
||||
await create_notification(
|
||||
db_session, seed["tenant_a"].id, seed["admin_a"].id,
|
||||
"info", "Read Notif", "Already read",
|
||||
)
|
||||
await db_session.flush()
|
||||
# Mark the first one as read
|
||||
from sqlalchemy import select, update
|
||||
q = select(Notification).where(Notification.title == "Read Notif")
|
||||
result = await db_session.execute(q)
|
||||
first_notif = result.scalar_one()
|
||||
first_notif.read_at = datetime.now(timezone.utc)
|
||||
await db_session.flush()
|
||||
|
||||
# Create an unread one
|
||||
await create_notification(
|
||||
db_session, seed["tenant_a"].id, seed["admin_a"].id,
|
||||
"info", "Unread Notif", "Not read yet",
|
||||
)
|
||||
await db_session.commit()
|
||||
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
resp = await client.get("/api/v1/notifications")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "items" in data
|
||||
assert "total" in data
|
||||
# Unread should come first
|
||||
items = data["items"]
|
||||
if len(items) >= 2:
|
||||
# Find the unread and read ones
|
||||
unread = [i for i in items if i["read_at"] is None]
|
||||
read = [i for i in items if i["read_at"] is not None]
|
||||
if unread and read:
|
||||
# Unread should appear before read in the list
|
||||
unread_idx = items.index(unread[0])
|
||||
read_idx = items.index(read[0])
|
||||
assert unread_idx < read_idx
|
||||
|
||||
async def test_mark_notification_read_returns_200(self, client: AsyncClient, db_session):
|
||||
"""AC 25: PATCH /api/v1/notifications/{id}/read -> 200."""
|
||||
seed = await seed_tenant_and_users(db_session)
|
||||
notif = await create_notification(
|
||||
db_session, seed["tenant_a"].id, seed["admin_a"].id,
|
||||
"info", "Test Notif", "Test body",
|
||||
)
|
||||
await db_session.commit()
|
||||
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
resp = await client.patch(
|
||||
f"/api/v1/notifications/{notif.id}/read",
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["read_at"] is not None
|
||||
|
||||
async def test_unread_count_returns_200_with_integer(self, client: AsyncClient, db_session):
|
||||
"""AC 26: GET /api/v1/notifications/unread-count -> 200 + integer count."""
|
||||
seed = await seed_tenant_and_users(db_session)
|
||||
await create_notification(
|
||||
db_session, seed["tenant_a"].id, seed["admin_a"].id,
|
||||
"info", "Unread 1", "Body 1",
|
||||
)
|
||||
await create_notification(
|
||||
db_session, seed["tenant_a"].id, seed["admin_a"].id,
|
||||
"info", "Unread 2", "Body 2",
|
||||
)
|
||||
await db_session.commit()
|
||||
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
resp = await client.get("/api/v1/notifications/unread-count")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "count" in data
|
||||
assert isinstance(data["count"], int)
|
||||
assert data["count"] >= 2
|
||||
@@ -1,29 +0,0 @@
|
||||
"""Tests for soft-delete isolation (R-1)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_soft_deleted_records_excluded_from_queries(
|
||||
client: AsyncClient, auth_headers: dict[str, str], seed_data: dict
|
||||
) -> None:
|
||||
# Create a throwaway account, soft-delete it, then verify it does NOT appear in list
|
||||
create = await client.post(
|
||||
"/api/v1/accounts/",
|
||||
json={"name": "Throwaway Co", "industry": "other"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert create.status_code == 201
|
||||
acc_id = create.json()["id"]
|
||||
|
||||
delete = await client.delete(f"/api/v1/accounts/{acc_id}", headers=auth_headers)
|
||||
assert delete.status_code == 204
|
||||
|
||||
# List accounts; the soft-deleted one must NOT appear in results
|
||||
listed = await client.get("/api/v1/accounts/?q=Throwaway", headers=auth_headers)
|
||||
assert listed.status_code == 200
|
||||
ids = [a["id"] for a in listed.json()]
|
||||
assert acc_id not in ids
|
||||
@@ -1,61 +0,0 @@
|
||||
"""Tests for FR-6 + R-2 (Tag + TagLink, polymorphic parent validation, unique constraint)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_tag(client: AsyncClient, auth_headers: dict[str, str]) -> None:
|
||||
resp = await client.post(
|
||||
"/api/v1/tags/",
|
||||
json={"name": "Important", "color": "#FF0000"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
body = resp.json()
|
||||
assert body["name"] == "Important"
|
||||
assert body["color"] == "#FF0000"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_link_tag_to_valid_parent(
|
||||
client: AsyncClient, auth_headers: dict[str, str], seed_data: dict
|
||||
) -> None:
|
||||
tag_id = seed_data["tag_ids"][0]
|
||||
acc_id = seed_data["account_ids"][0]
|
||||
resp = await client.post(
|
||||
"/api/v1/tags/link",
|
||||
json={"tag_id": tag_id, "parent_type": "account", "parent_id": acc_id},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_link_tag_to_invalid_parent(
|
||||
client: AsyncClient, auth_headers: dict[str, str], seed_data: dict
|
||||
) -> None:
|
||||
"""R-2: invalid parent → 404."""
|
||||
tag_id = seed_data["tag_ids"][0]
|
||||
resp = await client.post(
|
||||
"/api/v1/tags/link",
|
||||
json={"tag_id": tag_id, "parent_type": "account", "parent_id": 99999},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unique_constraint_on_link(
|
||||
client: AsyncClient, auth_headers: dict[str, str], seed_data: dict
|
||||
) -> None:
|
||||
"""Second link with same (tag, parent_type, parent_id) → 409."""
|
||||
tag_id = seed_data["tag_ids"][0]
|
||||
acc_id = seed_data["account_ids"][0]
|
||||
payload = {"tag_id": tag_id, "parent_type": "account", "parent_id": acc_id}
|
||||
r1 = await client.post("/api/v1/tags/link", json=payload, headers=auth_headers)
|
||||
assert r1.status_code == 201, r1.text
|
||||
r2 = await client.post("/api/v1/tags/link", json=payload, headers=auth_headers)
|
||||
assert r2.status_code == 409
|
||||
@@ -0,0 +1,281 @@
|
||||
"""Tenant + RBAC + User/Role management tests — ACs 10-17, 21-22."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from tests.conftest import ORIGIN_HEADER, seed_tenant_and_users, login_client
|
||||
from app.models.audit import AuditLog
|
||||
from app.models.notification import Notification
|
||||
from app.models.company import Company
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestUserManagement:
|
||||
"""ACs 10-14: User CRUD as admin/viewer."""
|
||||
|
||||
async def test_list_users_as_admin_returns_200(self, client: AsyncClient, db_session):
|
||||
"""AC 10: GET /api/v1/users as admin -> 200 + paginated list."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
resp = await client.get("/api/v1/users")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "items" in data
|
||||
assert "total" in data
|
||||
assert "page" in data
|
||||
assert "page_size" in data
|
||||
assert data["total"] >= 3 # admin, viewer, editor
|
||||
|
||||
async def test_list_users_as_viewer_returns_403(self, client: AsyncClient, db_session):
|
||||
"""AC 11: GET /api/v1/users as viewer -> 403."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "viewer@tenanta.com")
|
||||
resp = await client.get("/api/v1/users")
|
||||
assert resp.status_code == 403
|
||||
|
||||
async def test_create_user_valid_data_returns_201(self, client: AsyncClient, db_session):
|
||||
"""AC 12: POST /api/v1/users valid data -> 201."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
resp = await client.post(
|
||||
"/api/v1/users",
|
||||
json={
|
||||
"email": "newuser@example.com",
|
||||
"name": "New User",
|
||||
"password": "NewPass123!",
|
||||
"role": "viewer",
|
||||
},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
assert resp.json()["email"] == "newuser@example.com"
|
||||
|
||||
async def test_update_user_returns_200(self, client: AsyncClient, db_session):
|
||||
"""AC 13: PATCH /api/v1/users/{id} -> 200."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
# Get viewer user ID
|
||||
from app.models.user import User
|
||||
q = select(User).where(User.email == "viewer@tenanta.com")
|
||||
result = await db_session.execute(q)
|
||||
viewer = result.scalar_one()
|
||||
resp = await client.patch(
|
||||
f"/api/v1/users/{viewer.id}",
|
||||
json={"name": "Updated Viewer"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["name"] == "Updated Viewer"
|
||||
|
||||
async def test_delete_user_returns_204(self, client: AsyncClient, db_session):
|
||||
"""AC 14: DELETE /api/v1/users/{id} -> 204."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
from app.models.user import User
|
||||
q = select(User).where(User.email == "viewer@tenanta.com")
|
||||
result = await db_session.execute(q)
|
||||
viewer = result.scalar_one()
|
||||
resp = await client.delete(
|
||||
f"/api/v1/users/{viewer.id}",
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 204
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestRoleManagement:
|
||||
"""ACs 15-16: Role list and create."""
|
||||
|
||||
async def test_list_roles_returns_200(self, client: AsyncClient, db_session):
|
||||
"""AC 15: GET /api/v1/roles -> 200 + list with permissions."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
resp = await client.get("/api/v1/roles")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "items" in data
|
||||
assert len(data["items"]) >= 1
|
||||
assert "permissions" in data["items"][0]
|
||||
assert "field_permissions" in data["items"][0]
|
||||
|
||||
async def test_create_role_with_custom_permissions_returns_201(self, client: AsyncClient, db_session):
|
||||
"""AC 16: POST /api/v1/roles with custom permissions -> 201."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
resp = await client.post(
|
||||
"/api/v1/roles",
|
||||
json={
|
||||
"name": "manager",
|
||||
"permissions": {"companies": {"read": True, "create": True, "update": True, "delete": True}},
|
||||
"field_permissions": {"annual_revenue": "read"},
|
||||
},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
assert resp.json()["name"] == "manager"
|
||||
assert "companies" in resp.json()["permissions"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestCrossTenant:
|
||||
"""AC 17: Cross-tenant access on company -> 404 (not 403)."""
|
||||
|
||||
async def test_cross_tenant_company_access_returns_404(self, client: AsyncClient, db_session):
|
||||
"""AC 17: Cross-tenant access on company -> 404 (not 403)."""
|
||||
seed = await seed_tenant_and_users(db_session)
|
||||
# Login as admin A (tenant A)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
# Try to access company B (tenant B)
|
||||
company_b_id = str(seed["company_b"].id)
|
||||
resp = await client.get(f"/api/v1/companies/{company_b_id}")
|
||||
assert resp.status_code == 404
|
||||
assert resp.status_code != 403
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestRBAC:
|
||||
"""AC 21: RBAC — viewer can read company but not create (POST -> 403)."""
|
||||
|
||||
async def test_viewer_can_read_company(self, client: AsyncClient, db_session):
|
||||
"""AC 21a: Viewer can GET /api/v1/companies."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "viewer@tenanta.com")
|
||||
resp = await client.get("/api/v1/companies")
|
||||
assert resp.status_code == 200
|
||||
|
||||
async def test_viewer_cannot_create_company(self, client: AsyncClient, db_session):
|
||||
"""AC 21b: Viewer cannot POST /api/v1/companies -> 403."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "viewer@tenanta.com")
|
||||
resp = await client.post(
|
||||
"/api/v1/companies",
|
||||
json={"name": "Test Company"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestFieldPermissions:
|
||||
"""AC 22: Field-level permissions — hidden field not in response."""
|
||||
|
||||
async def test_hidden_field_not_in_response(self, client: AsyncClient, db_session):
|
||||
"""AC 22: Field-level permissions — hidden field not in response.
|
||||
The sales_rep role has annual_revenue as hidden.
|
||||
We need a user with that role.
|
||||
"""
|
||||
seed = await seed_tenant_and_users(db_session)
|
||||
# Create a user with sales_rep role
|
||||
from app.core.auth import hash_password
|
||||
from app.models.user import User, UserTenant
|
||||
sales_user = User(
|
||||
tenant_id=seed["tenant_a"].id,
|
||||
email="sales@tenanta.com",
|
||||
name="Sales Rep",
|
||||
password_hash=hash_password("TestPass123!"),
|
||||
role="sales_rep",
|
||||
is_active=True,
|
||||
preferences={},
|
||||
)
|
||||
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)
|
||||
db_session.add(ut)
|
||||
await db_session.commit()
|
||||
|
||||
await login_client(client, "sales@tenanta.com")
|
||||
# Get company A
|
||||
resp = await client.get(f"/api/v1/companies/{seed['company_a'].id}")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
# annual_revenue should be filtered out for sales_rep role
|
||||
assert "annual_revenue" not in data
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestAuditLog:
|
||||
"""AC 19: Audit log entry created on company.create/update/delete."""
|
||||
|
||||
async def test_audit_log_on_company_create(self, client: AsyncClient, db_session):
|
||||
"""AC 19: Audit log entry created on company.create."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
resp = await client.post(
|
||||
"/api/v1/companies",
|
||||
json={"name": "Audit Test Co"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
|
||||
# Check audit_log table
|
||||
from sqlalchemy import select as sel
|
||||
q = sel(AuditLog).where(AuditLog.entity_type == "company", AuditLog.action == "create")
|
||||
result = await db_session.execute(q)
|
||||
entries = result.scalars().all()
|
||||
assert len(entries) >= 1
|
||||
|
||||
async def test_audit_log_on_company_update(self, client: AsyncClient, db_session):
|
||||
"""AC 19: Audit log entry created on company.update."""
|
||||
seed = await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
resp = await client.patch(
|
||||
f"/api/v1/companies/{seed['company_a'].id}",
|
||||
json={"name": "Updated Company Alpha"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
|
||||
from sqlalchemy import select as sel
|
||||
q = sel(AuditLog).where(AuditLog.entity_type == "company", AuditLog.action == "update")
|
||||
result = await db_session.execute(q)
|
||||
entries = result.scalars().all()
|
||||
assert len(entries) >= 1
|
||||
|
||||
async def test_audit_log_on_company_delete(self, client: AsyncClient, db_session):
|
||||
"""AC 19: Audit log entry created on company.delete."""
|
||||
seed = await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
resp = await client.delete(
|
||||
f"/api/v1/companies/{seed['company_a'].id}",
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 204
|
||||
|
||||
from sqlalchemy import select as sel
|
||||
q = sel(AuditLog).where(AuditLog.entity_type == "company", AuditLog.action == "delete")
|
||||
result = await db_session.execute(q)
|
||||
entries = result.scalars().all()
|
||||
assert len(entries) >= 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestNotificationOnAssign:
|
||||
"""AC 20: Notification created on user assign."""
|
||||
|
||||
async def test_notification_created_on_user_create(self, client: AsyncClient, db_session):
|
||||
"""AC 20: Notification created on user assign (via user creation)."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
resp = await client.post(
|
||||
"/api/v1/users",
|
||||
json={
|
||||
"email": "assigned@example.com",
|
||||
"name": "Assigned User",
|
||||
"password": "NewPass123!",
|
||||
"role": "viewer",
|
||||
},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
new_user_id = resp.json()["id"]
|
||||
|
||||
# Check notification was created for the new user
|
||||
from sqlalchemy import select as sel
|
||||
q = sel(Notification).where(Notification.user_id == new_user_id)
|
||||
result = await db_session.execute(q)
|
||||
notifs = result.scalars().all()
|
||||
assert len(notifs) >= 1
|
||||
@@ -1,191 +0,0 @@
|
||||
"""User /me endpoint tests + RBAC verification."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from httpx import AsyncClient
|
||||
|
||||
# === /users/me ===
|
||||
|
||||
|
||||
async def test_get_me_returns_user_data(
|
||||
client: AsyncClient, auth_headers: dict[str, str], registered_user: dict
|
||||
) -> None:
|
||||
"""GET /api/v1/users/me with a valid JWT returns 200 + email, name, role, org_id."""
|
||||
resp = await client.get("/api/v1/users/me", headers=auth_headers)
|
||||
assert resp.status_code == 200, resp.text
|
||||
data = resp.json()
|
||||
assert data["email"] == registered_user["email"]
|
||||
assert data["name"] == registered_user["name"]
|
||||
assert "role" in data
|
||||
assert "org_id" in data
|
||||
assert "id" in data
|
||||
assert "password_hash" not in data
|
||||
assert "avatar_url" in data
|
||||
assert "email_notifications" in data
|
||||
|
||||
|
||||
async def test_get_me_invalid_token_format(client: AsyncClient) -> None:
|
||||
"""GET /api/v1/users/me with a malformed token returns 401."""
|
||||
resp = await client.get(
|
||||
"/api/v1/users/me",
|
||||
headers={"Authorization": "Bearer this-is-not-a-jwt"},
|
||||
)
|
||||
assert resp.status_code == 401, resp.text
|
||||
|
||||
|
||||
async def test_get_me_bogus_token(client: AsyncClient) -> None:
|
||||
"""GET /api/v1/users/me with a token signed with the wrong key returns 401."""
|
||||
import time
|
||||
|
||||
from jose import jwt
|
||||
|
||||
bogus = jwt.encode(
|
||||
{
|
||||
"sub": "1",
|
||||
"org_id": 1,
|
||||
"role": "admin",
|
||||
"exp": int(time.time()) + 3600,
|
||||
"iat": int(time.time()),
|
||||
},
|
||||
"completely-different-secret-key-32chars-abc",
|
||||
algorithm="HS256",
|
||||
)
|
||||
resp = await client.get(
|
||||
"/api/v1/users/me",
|
||||
headers={"Authorization": f"Bearer {bogus}"},
|
||||
)
|
||||
assert resp.status_code == 401, resp.text
|
||||
|
||||
|
||||
# === PATCH /users/{id} ===
|
||||
|
||||
|
||||
async def test_update_profile_self(
|
||||
client: AsyncClient, auth_headers: dict[str, str], registered_user: dict
|
||||
) -> None:
|
||||
"""PATCH /api/v1/users/{self_id} updates the current user's name."""
|
||||
user_id = registered_user["user"]["id"]
|
||||
resp = await client.patch(
|
||||
f"/api/v1/users/{user_id}",
|
||||
json={"name": "Updated Name", "email_notifications": False},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
data = resp.json()
|
||||
assert data["name"] == "Updated Name"
|
||||
assert data["email_notifications"] is False
|
||||
# email and id should be unchanged
|
||||
assert data["email"] == registered_user["email"]
|
||||
assert data["id"] == user_id
|
||||
|
||||
|
||||
async def test_update_profile_cannot_change_other(
|
||||
client: AsyncClient, auth_headers: dict[str, str], registered_user: dict
|
||||
) -> None:
|
||||
"""Non-admin cannot update another user's profile (403)."""
|
||||
# Bootstrap user IS admin (first registered user). To test the 403 case,
|
||||
# create a sales_rep and try to update someone else.
|
||||
# First, create a 2nd user (admin can do this)
|
||||
create_resp = await client.post(
|
||||
"/api/v1/users/",
|
||||
json={
|
||||
"email": "rep@test.com",
|
||||
"password": "RepPass123!",
|
||||
"name": "Test Rep",
|
||||
"role": "sales_rep",
|
||||
},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert create_resp.status_code == 201, create_resp.text
|
||||
other_id = create_resp.json()["id"]
|
||||
|
||||
# Now log in as the rep and try to update the admin
|
||||
login_resp = await client.post(
|
||||
"/api/v1/auth/login",
|
||||
data={"username": "rep@test.com", "password": "RepPass123!"},
|
||||
)
|
||||
assert login_resp.status_code == 200, login_resp.text
|
||||
rep_token = login_resp.json()["access_token"]
|
||||
rep_headers = {"Authorization": f"Bearer {rep_token}"}
|
||||
|
||||
# Rep tries to update the admin → 403
|
||||
admin_id = registered_user["user"]["id"]
|
||||
resp = await client.patch(
|
||||
f"/api/v1/users/{admin_id}",
|
||||
json={"name": "Hacked"},
|
||||
headers=rep_headers,
|
||||
)
|
||||
assert resp.status_code == 403, resp.text
|
||||
|
||||
|
||||
# === GET /users/ (admin list) ===
|
||||
|
||||
|
||||
async def test_list_users_as_admin(
|
||||
client: AsyncClient, auth_headers: dict[str, str], registered_user: dict
|
||||
) -> None:
|
||||
"""GET /api/v1/users/ as admin returns 200 + paginated list."""
|
||||
resp = await client.get("/api/v1/users/", headers=auth_headers)
|
||||
assert resp.status_code == 200, resp.text
|
||||
data = resp.json()
|
||||
assert "items" in data
|
||||
assert "total" in data
|
||||
assert "page" in data
|
||||
assert "page_size" in data
|
||||
assert data["total"] >= 1
|
||||
assert len(data["items"]) >= 1
|
||||
# The bootstrap user is in the list
|
||||
emails = [u["email"] for u in data["items"]]
|
||||
assert registered_user["email"] in emails
|
||||
|
||||
|
||||
async def test_list_users_as_sales_rep_forbidden(
|
||||
client: AsyncClient, auth_headers: dict[str, str]
|
||||
) -> None:
|
||||
"""GET /api/v1/users/ as sales_rep → 403."""
|
||||
# Create a sales_rep via the admin
|
||||
create_resp = await client.post(
|
||||
"/api/v1/users/",
|
||||
json={
|
||||
"email": "rep@test.com",
|
||||
"password": "RepPass123!",
|
||||
"name": "Test Rep",
|
||||
"role": "sales_rep",
|
||||
},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert create_resp.status_code == 201, create_resp.text
|
||||
|
||||
# Log in as the rep
|
||||
login_resp = await client.post(
|
||||
"/api/v1/auth/login",
|
||||
data={"username": "rep@test.com", "password": "RepPass123!"},
|
||||
)
|
||||
assert login_resp.status_code == 200, login_resp.text
|
||||
rep_token = login_resp.json()["access_token"]
|
||||
rep_headers = {"Authorization": f"Bearer {rep_token}"}
|
||||
|
||||
# Rep tries to list all users → 403
|
||||
resp = await client.get("/api/v1/users/", headers=rep_headers)
|
||||
assert resp.status_code == 403, resp.text
|
||||
|
||||
|
||||
# === Refresh & Logout ===
|
||||
|
||||
|
||||
async def test_refresh_token(client: AsyncClient, auth_headers: dict[str, str]) -> None:
|
||||
"""POST /api/v1/auth/refresh issues a new token for the current user."""
|
||||
resp = await client.post("/api/v1/auth/refresh", headers=auth_headers)
|
||||
assert resp.status_code == 200, resp.text
|
||||
data = resp.json()
|
||||
assert "access_token" in data
|
||||
assert data["token_type"] == "bearer"
|
||||
assert data["expires_in"] > 0
|
||||
|
||||
|
||||
async def test_logout(client: AsyncClient, auth_headers: dict[str, str]) -> None:
|
||||
"""POST /api/v1/auth/logout returns 200 with confirmation message."""
|
||||
resp = await client.post("/api/v1/auth/logout", headers=auth_headers)
|
||||
assert resp.status_code == 200, resp.text
|
||||
data = resp.json()
|
||||
assert data["message"] == "logged out"
|
||||
Reference in New Issue
Block a user