"""Tenant licensing (app/crud/licensing.py): seats for active Studio and Portal accounts, the dates a license is valid between, and the Super Admin's routes that set it.""" import uuid from datetime import date, timedelta import pytest from sqlalchemy import text from app.core.auth import create_platform_admin_token, hash_password from app.crud import licensing, oidc from app.crud.sign_in_policy import SecuritySettings, SsoSettings from app.models.identity import Role, Tenant from app.models.platform_admin import PlatformAdmin from tests.conftest import TEST_ADMIN_PASSWORD, AsyncSessionLocal TODAY = licensing.today() async def _set_license(tenant_id, **fields) -> None: async with AsyncSessionLocal() as db: async with db.begin(): tenant = await db.get(Tenant, tenant_id) for key, value in fields.items(): setattr(tenant, f"license_{key}", value) licensing.forget(tenant_id) def _new_user(email: str, status: str = "active") -> dict: return {"name": "Someone", "email": email, "password": "a-password-1", "status": status} @pytest.fixture async def super_admin(): async with AsyncSessionLocal() as db: async with db.begin(): admin = PlatformAdmin( name="Pytest Super Admin", email=f"pytest-super-{uuid.uuid4().hex[:8]}@example.com", password_hash=hash_password("pytest-password-123"), status="active", ) db.add(admin) await db.flush() admin_id = admin.id yield {"Authorization": f"Bearer {create_platform_admin_token(admin_id=admin_id)}"} async with AsyncSessionLocal() as db: async with db.begin(): await db.execute(text("DELETE FROM platform_admins WHERE id = :id"), {"id": admin_id}) # ---- dates ------------------------------------------------------------------------------------------------ def test_the_period_includes_both_end_dates(): start, end = date(2027, 1, 1), date(2027, 12, 31) assert licensing.period_state(None, None, start) == "unlimited" assert licensing.period_state(start, end, start) == "active" assert licensing.period_state(start, end, end) == "active" assert licensing.period_state(start, end, start - timedelta(days=1)) == "not_started" assert licensing.period_state(start, end, end + timedelta(days=1)) == "expired" assert licensing.period_state(None, end, date(2000, 1, 1)) == "active" assert licensing.period_message(start, end, end + timedelta(days=1)) == ( "This workspace's license ended on 31 December 2027. Contact your administrator to renew it." ) assert "starts on 1 January 2027" in licensing.period_message(start, None, date(2026, 6, 1)) assert licensing.period_message(start, end, start) is None async def test_outside_its_dates_nobody_signs_in_or_uses_a_session(client, test_tenant, auth_headers): login = {"tenant_id": str(test_tenant["tenant_id"]), "email": test_tenant["admin_email"], "password": TEST_ADMIN_PASSWORD} assert (await client.get("/users", headers=auth_headers)).status_code == 200 await _set_license(test_tenant["tenant_id"], ends_on=TODAY - timedelta(days=1)) resp = await client.post("/auth/login", json=login) assert resp.status_code == 403 assert resp.json()["detail"]["code"] == "license_inactive" assert "license ended on" in resp.json()["detail"]["message"] # A session from before is turned away too. resp = await client.get("/users", headers=auth_headers) assert resp.status_code == 403 and resp.json()["detail"]["code"] == "license_inactive" # A wrong password still just says so — the license isn't told to strangers. assert (await client.post("/auth/login", json={**login, "password": "wrong-password"})).status_code == 401 # Portal registration is closed. resp = await client.post( "/auth/register", json={"tenant_id": str(test_tenant["tenant_id"]), "organization_name": "Acme", "name": "Vee", "email": "vee@example.com", "password": "a-password-1"}, ) assert resp.status_code == 403 await _set_license(test_tenant["tenant_id"], starts_on=TODAY + timedelta(days=3), ends_on=None) resp = await client.post("/auth/login", json=login) assert resp.status_code == 403 and "starts on" in resp.json()["detail"]["message"] await _set_license(test_tenant["tenant_id"], starts_on=TODAY, ends_on=TODAY) assert (await client.post("/auth/login", json=login)).status_code == 200 assert (await client.get("/users", headers=auth_headers)).status_code == 200 # ---- seats ------------------------------------------------------------------------------------------------ async def test_active_studio_accounts_are_limited_to_the_seats(client, test_tenant, auth_headers): tid = test_tenant["tenant_id"] await _set_license(tid, max_users=2) # the admin + one more assert (await client.post("/users", headers=auth_headers, json=_new_user("one@example.com"))).status_code == 201 resp = await client.post("/users", headers=auth_headers, json=_new_user("two@example.com")) assert resp.status_code == 403 assert resp.json()["detail"]["code"] == "license_seats" assert "licensed for 2 active users" in resp.json()["detail"]["message"] # An inactive account takes no seat — but activating it needs one. inactive = await client.post("/users", headers=auth_headers, json=_new_user("three@example.com", "inactive")) assert inactive.status_code == 201 three = inactive.json()["id"] assert (await client.patch(f"/users/{three}", headers=auth_headers, json={"status": "active"})).status_code == 403 # Renaming an account that's already active needs no seat. users = (await client.get("/users", headers=auth_headers)).json() one = next(u["id"] for u in users if u["email"] == "one@example.com") assert (await client.patch(f"/users/{one}", headers=auth_headers, json={"name": "Renamed", "status": "active"})).status_code == 200 # Freeing a seat lets it in. assert (await client.patch(f"/users/{one}", headers=auth_headers, json={"status": "inactive"})).status_code == 200 assert (await client.patch(f"/users/{three}", headers=auth_headers, json={"status": "active"})).status_code == 200 status_ = (await client.get("/tenants/me/license", headers=auth_headers)).json() assert status_["usage"] == {"users": 2, "portal_users": 0} assert status_["license"]["max_users"] == 2 assert status_["state"] == "unlimited" and status_["days_left"] is None async def test_portal_accounts_have_their_own_seats(client, test_tenant, auth_headers): tid = test_tenant["tenant_id"] await _set_license(tid, max_users=1, max_portal_users=1) reg = {"tenant_id": str(tid), "organization_name": "Acme", "name": "Vee", "phone": "+15555550100", "password": "a-password-1"} # Registering creates an inactive account, so it's never refused for seats. assert (await client.post("/auth/register", json={**reg, "email": "v1@example.com"})).status_code == 201 assert (await client.post("/auth/register", json={**reg, "email": "v2@example.com"})).status_code == 201 users = {u["email"]: u["id"] for u in (await client.get("/users", headers=auth_headers)).json()} # Studio seats are all taken (the admin), Portal ones aren't. assert (await client.patch(f"/users/{users['v1@example.com']}", headers=auth_headers, json={"status": "active"})).status_code == 200 resp = await client.patch(f"/users/{users['v2@example.com']}", headers=auth_headers, json={"status": "active"}) assert resp.status_code == 403 and "1 active Portal user," in resp.json()["detail"]["message"] async def test_single_sign_on_creates_no_account_past_the_seats(test_tenant): tid = test_tenant["tenant_id"] await _set_license(tid, max_users=1) async with AsyncSessionLocal() as db: async with db.begin(): await db.execute(text("SELECT set_config('app.current_tenant_id', :v, true)"), {"v": str(tid)}) tenant = await db.get(Tenant, tid) rules = SecuritySettings(sso=SsoSettings(enabled=True, auto_create_users=True, default_role_id=str(test_tenant["admin_role_id"]))) # The admin role is a system role, scoped internal by default. role = await db.get(Role, test_tenant["admin_role_id"]) assert role.scope == "internal" with pytest.raises(oidc.OidcError, match="licensed for 1 active user,"): await oidc.resolve_user(db, tenant, rules, {"sub": "new-person", "email": "new@example.com", "email_verified": True}) # ---- the Super Admin -------------------------------------------------------------------------------------- async def test_the_super_admin_sets_the_license_and_sees_usage(client, test_tenant, super_admin, auth_headers): tid = test_tenant["tenant_id"] ends = TODAY + timedelta(days=9) resp = await client.put(f"/admin/tenants/{tid}/license", headers=super_admin, json={"max_users": 1, "starts_on": str(TODAY), "ends_on": str(ends)}) assert resp.status_code == 200, resp.text body = resp.json() assert body["state"] == "active" and body["days_left"] == 10 assert body["usage"] == {"users": 1, "portal_users": 0} assert body["license"] == {"max_users": 1, "max_portal_users": None, "starts_on": str(TODAY), "ends_on": str(ends)} # Over the limit, the Super Admin can't add an active account either. resp = await client.post(f"/admin/tenants/{tid}/users", headers=super_admin, json=_new_user("extra@example.com")) assert resp.status_code == 403 and resp.json()["detail"]["code"] == "license_seats" usage = (await client.get("/admin/license-usage", headers=super_admin)).json() assert usage[str(tid)] == {"users": 1, "portal_users": 0} listed = next(t for t in (await client.get("/admin/tenants", headers=super_admin)).json() if t["id"] == str(tid)) assert listed["license"]["ends_on"] == str(ends) # The change is in the tenant's own audit log. async with AsyncSessionLocal() as db: entry = ( await db.execute(text("SELECT details FROM audit_log_entries WHERE tenant_id = :t AND action = 'tenant.license_updated'"), {"t": tid}) ).scalar_one() assert entry["after"]["max_users"] == 1 and entry["before"]["max_users"] is None # Ending it takes effect at once for this process's sessions. yesterday = TODAY - timedelta(days=1) resp = await client.put(f"/admin/tenants/{tid}/license", headers=super_admin, json={"starts_on": str(yesterday - timedelta(days=5)), "ends_on": str(yesterday)}) assert resp.json()["state"] == "expired" and resp.json()["days_left"] == 0 assert (await client.get("/users", headers=auth_headers)).status_code == 403 # The Super Admin still reaches the tenant. assert (await client.get(f"/admin/tenants/{tid}/users", headers=super_admin)).status_code == 200 bad = await client.put(f"/admin/tenants/{tid}/license", headers=super_admin, json={"starts_on": str(TODAY), "ends_on": str(yesterday)}) assert bad.status_code == 422 assert (await client.put(f"/admin/tenants/{tid}/license", headers=super_admin, json={"max_users": 0})).status_code == 422 async def test_a_new_tenant_can_start_with_a_license(client, super_admin): resp = await client.post( "/admin/tenants", headers=super_admin, json={"name": f"Pytest Tenant {uuid.uuid4().hex[:6]}", "license": {"max_users": 5, "ends_on": "2030-01-31"}} ) assert resp.status_code == 201, resp.text tenant_id = resp.json()["id"] try: assert resp.json()["license"] == {"max_users": 5, "max_portal_users": None, "starts_on": None, "ends_on": "2030-01-31"} finally: async with AsyncSessionLocal() as db: async with db.begin(): await db.execute(text("DELETE FROM roles WHERE tenant_id = :tid"), {"tid": tenant_id}) await db.execute(text("DELETE FROM tenants WHERE id = :tid"), {"tid": tenant_id})