patrickandClaude Opus 4.8
898a81f499
Mandantenfähigkeit ausgebaut: - Neue Rolle RESELLER (company_id NULL); companies.reseller_id + is_active - RLS-Erweiterung (Migration 0034): companies/users zusätzlich auf app.reseller_id gefenced → Reseller sieht/verwaltet DB-seitig nur eigene Firmen, keine personenbezogenen Zeit-/Abwesenheitsdaten (DSGVO: nur Verwaltung) - get_current_user setzt app.reseller_id + Bypass aus für RESELLER - tenant_service: Firma + Erst-Admin (Einladung), Übersicht mit Kennzahlen - Router /reseller/* (Self-Service) und /admin/* (SUPER_ADMIN: Mandanten + Reseller) - Login-Sperre bei deaktiviertem Mandanten - Frontend: TenantsPage (/admin/tenants), eigene ResellerCompaniesPage (/reseller), rollenbasierte Login-Weiterleitung, Nav "Mandanten" für SUPER_ADMIN - 4 neue Tests inkl. Cross-Reseller-RLS-Isolation; 172/172 grün Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
110 lines
4.4 KiB
Python
110 lines
4.4 KiB
Python
"""Tests für Reseller-Rolle + Mandantenverwaltung + RLS-Isolation."""
|
|
import uuid
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy import text
|
|
|
|
from app.core.security import hash_password
|
|
from app.models.user import User, UserRole
|
|
|
|
|
|
async def _make_user(db, *, email, role, company_id=None, password="Secret123"):
|
|
await db.execute(text("SET LOCAL app.bypass_rls = 'on'"))
|
|
user = User(
|
|
id=uuid.uuid4(), company_id=company_id, email=email,
|
|
password_hash=hash_password(password),
|
|
first_name=role.title(), last_name="User",
|
|
role=UserRole(role), is_active=True,
|
|
)
|
|
db.add(user)
|
|
await db.commit()
|
|
return user.id
|
|
|
|
|
|
async def _login(client, email, password="Secret123"):
|
|
r = await client.post("/api/v1/auth/login", json={"email": email, "password": password})
|
|
assert r.status_code == 200, r.text
|
|
return {"Authorization": f"Bearer {r.json()['access_token']}"}
|
|
|
|
|
|
@pytest.mark.asyncio(loop_scope="session")
|
|
async def test_superadmin_creates_reseller(client: AsyncClient, db_session):
|
|
await _make_user(db_session, email="sa@platform.de", role="SUPER_ADMIN")
|
|
h = await _login(client, "sa@platform.de")
|
|
|
|
r = await client.post("/api/v1/admin/resellers", json={
|
|
"email": "reseller-a@partner.de", "first_name": "Rita", "last_name": "Reseller",
|
|
}, headers=h)
|
|
assert r.status_code == 201, r.text
|
|
assert r.json()["company_count"] == 0
|
|
|
|
lst = await client.get("/api/v1/admin/resellers", headers=h)
|
|
assert any(x["email"] == "reseller-a@partner.de" for x in lst.json())
|
|
|
|
|
|
@pytest.mark.asyncio(loop_scope="session")
|
|
async def test_reseller_creates_and_sees_only_own_company(client: AsyncClient, db_session):
|
|
rid_a = await _make_user(db_session, email="res-a@p.de", role="RESELLER")
|
|
rid_b = await _make_user(db_session, email="res-b@p.de", role="RESELLER")
|
|
|
|
ha = await _login(client, "res-a@p.de")
|
|
hb = await _login(client, "res-b@p.de")
|
|
|
|
# Reseller A legt eine Firma an
|
|
r = await client.post("/api/v1/reseller/companies", json={
|
|
"name": "Alpha GmbH", "admin_email": "admin@alpha.de",
|
|
"admin_first_name": "Al", "admin_last_name": "Pha",
|
|
}, headers=ha)
|
|
assert r.status_code == 201, r.text
|
|
assert r.json()["name"] == "Alpha GmbH"
|
|
|
|
# Reseller B legt eine andere Firma an
|
|
r2 = await client.post("/api/v1/reseller/companies", json={
|
|
"name": "Beta GmbH", "admin_email": "admin@beta.de",
|
|
"admin_first_name": "Be", "admin_last_name": "Ta",
|
|
}, headers=hb)
|
|
assert r2.status_code == 201, r2.text
|
|
|
|
# A sieht nur Alpha, NICHT Beta (RLS-Isolation)
|
|
la = await client.get("/api/v1/reseller/companies", headers=ha)
|
|
names_a = {c["name"] for c in la.json()}
|
|
assert "Alpha GmbH" in names_a
|
|
assert "Beta GmbH" not in names_a
|
|
|
|
lb = await client.get("/api/v1/reseller/companies", headers=hb)
|
|
names_b = {c["name"] for c in lb.json()}
|
|
assert "Beta GmbH" in names_b
|
|
assert "Alpha GmbH" not in names_b
|
|
assert rid_a != rid_b
|
|
|
|
|
|
@pytest.mark.asyncio(loop_scope="session")
|
|
async def test_reseller_cannot_access_admin_endpoints(client: AsyncClient, db_session):
|
|
await _make_user(db_session, email="res-c@p.de", role="RESELLER")
|
|
h = await _login(client, "res-c@p.de")
|
|
r = await client.get("/api/v1/admin/tenants", headers=h)
|
|
assert r.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio(loop_scope="session")
|
|
async def test_deactivated_tenant_blocks_login(client: AsyncClient, db_session):
|
|
rid = await _make_user(db_session, email="res-d@p.de", role="RESELLER")
|
|
h = await _login(client, "res-d@p.de")
|
|
create = await client.post("/api/v1/reseller/companies", json={
|
|
"name": "Gamma GmbH", "admin_email": "admin@gamma.de",
|
|
"admin_first_name": "Ga", "admin_last_name": "Mma",
|
|
}, headers=h)
|
|
company_id = create.json()["id"]
|
|
|
|
# Direkt einen aktiven Mitarbeiter in Gamma anlegen (per DB) und Login testen
|
|
await _make_user(db_session, email="emp@gamma.de", role="EMPLOYEE", company_id=company_id)
|
|
ok = await client.post("/api/v1/auth/login", json={"email": "emp@gamma.de", "password": "Secret123"})
|
|
assert ok.status_code == 200
|
|
|
|
# Firma deaktivieren → Login gesperrt
|
|
await client.patch(f"/api/v1/reseller/companies/{company_id}", json={"is_active": False}, headers=h)
|
|
blocked = await client.post("/api/v1/auth/login", json={"email": "emp@gamma.de", "password": "Secret123"})
|
|
assert blocked.status_code == 403
|
|
assert rid is not None
|