"""Tests for Iteration 5 — Advanced Role/Permission System + Full Duplex Sync. Covers: • /api/permissions/catalog, /roles, /me CRUD + 409 guards • /api/permissions/users/{uid}/role assignment + self-demote guard • /api/users POST with new role values (cashier, manager, etc.) • /api/sync/push (empty, valid op, idempotency, invalid url) • /api/sync/pull, /api/sync/status, /api/sync/log """ import os import uuid import requests BASE_URL = os.environ.get("REACT_APP_BACKEND_URL", "https://offline-billing-pro-2.preview.emergentagent.com").rstrip("/") # ---------- Permissions Catalog & Preset Roles ---------- class TestPermissionsCatalog: def test_catalog_returns_17_modules(self, admin_session): r = admin_session.get(f"{BASE_URL}/api/permissions/catalog", timeout=30) assert r.status_code == 200, r.text data = r.json() assert "modules" in data mods = data["modules"] assert len(mods) == 17, f"Expected 17 modules, got {len(mods)}" # Each module has key/label/actions/permissions for m in mods: assert "key" in m and "label" in m and "actions" in m and "permissions" in m assert isinstance(m["actions"], list) and len(m["actions"]) >= 1 # permissions are "mod.action" for p in m["permissions"]: assert p.startswith(f"{m['key']}.") assert isinstance(data.get("all_permissions"), list) def test_roles_returns_5_presets(self, admin_session): r = admin_session.get(f"{BASE_URL}/api/permissions/roles", timeout=30) assert r.status_code == 200, r.text roles = r.json()["roles"] names = {r["name"]: r for r in roles if r.get("preset")} for expected in ["admin", "manager", "accountant", "cashier", "viewer"]: assert expected in names, f"Missing preset {expected}" assert names[expected]["preset"] is True # Admin has wildcard assert names["admin"]["permissions"] == ["*"] # Approximate counts (relax exact equality to avoid brittle tests) assert 50 <= len(names["manager"]["permissions"]) <= 80, len(names["manager"]["permissions"]) assert 30 <= len(names["accountant"]["permissions"]) <= 55 assert 15 <= len(names["cashier"]["permissions"]) <= 35 assert 10 <= len(names["viewer"]["permissions"]) <= 25 def test_me_for_admin(self, admin_session): r = admin_session.get(f"{BASE_URL}/api/permissions/me", timeout=30) assert r.status_code == 200, r.text data = r.json() assert data["role"] == "admin" assert data["is_super"] is True assert data["permissions"] == ["*"] def test_catalog_requires_auth(self): r = requests.get(f"{BASE_URL}/api/permissions/catalog", timeout=30) assert r.status_code in (401, 403) # ---------- Custom Role CRUD ---------- class TestCustomRoles: ROLE_NAME = f"test_lite_{uuid.uuid4().hex[:6]}" def test_create_custom_role(self, admin_session): payload = { "name": self.ROLE_NAME, "label": "Test Lite", "permissions": ["invoicing.view", "invoicing.create"], } r = admin_session.post(f"{BASE_URL}/api/permissions/roles", json=payload, timeout=30) assert r.status_code in (200, 201), r.text data = r.json() assert data["name"] == self.ROLE_NAME assert data["preset"] is False assert set(data["permissions"]) == {"invoicing.view", "invoicing.create"} # Verify appears in listing as preset=false r2 = admin_session.get(f"{BASE_URL}/api/permissions/roles", timeout=30) roles = r2.json()["roles"] found = [x for x in roles if x["name"] == self.ROLE_NAME] assert found and found[0]["preset"] is False def test_update_custom_role(self, admin_session): payload = {"name": self.ROLE_NAME, "permissions": ["invoicing.view", "items.view"]} r = admin_session.put(f"{BASE_URL}/api/permissions/roles/{self.ROLE_NAME}", json=payload, timeout=30) assert r.status_code == 200, r.text # Verify r2 = admin_session.get(f"{BASE_URL}/api/permissions/roles", timeout=30) roles = r2.json()["roles"] found = [x for x in roles if x["name"] == self.ROLE_NAME][0] assert set(found["permissions"]) == {"invoicing.view", "items.view"} def test_update_preset_admin_409(self, admin_session): r = admin_session.put( f"{BASE_URL}/api/permissions/roles/admin", json={"name": "admin", "permissions": ["invoicing.view"]}, timeout=30, ) assert r.status_code == 409, r.text def test_create_preset_name_409(self, admin_session): r = admin_session.post( f"{BASE_URL}/api/permissions/roles", json={"name": "admin", "permissions": []}, timeout=30, ) assert r.status_code == 409, r.text def test_delete_preset_admin_409(self, admin_session): r = admin_session.delete(f"{BASE_URL}/api/permissions/roles/admin", timeout=30) assert r.status_code == 409, r.text def test_delete_custom_role(self, admin_session): r = admin_session.delete(f"{BASE_URL}/api/permissions/roles/{self.ROLE_NAME}", timeout=30) assert r.status_code == 200, r.text # Verify removed r2 = admin_session.get(f"{BASE_URL}/api/permissions/roles", timeout=30) names = [x["name"] for x in r2.json()["roles"]] assert self.ROLE_NAME not in names # ---------- Role Assignment / User creation ---------- class TestRoleAssignment: def test_create_user_with_cashier_role(self, admin_session): email = f"TEST_user_{uuid.uuid4().hex[:8]}@rmregal.com" r = admin_session.post( f"{BASE_URL}/api/users", json={"email": email, "password": "Pass@1234", "name": "Cashier User", "role": "cashier"}, timeout=30, ) assert r.status_code == 200, r.text data = r.json() assert data["role"] == "cashier" assert data["email"] == email.lower() uid = data["id"] # Assign manager role r2 = admin_session.put( f"{BASE_URL}/api/permissions/users/{uid}/role", json={"role": "manager"}, timeout=30, ) assert r2.status_code == 200, r2.text assert r2.json()["role"] == "manager" # Verify via effective endpoint r3 = admin_session.get(f"{BASE_URL}/api/permissions/users/{uid}/effective", timeout=30) assert r3.status_code == 200 assert r3.json()["role"] == "manager" assert len(r3.json()["effective_permissions"]) > 30 # Cleanup admin_session.delete(f"{BASE_URL}/api/users/{uid}", timeout=30) def test_self_demote_blocked(self, admin_session): me = admin_session.get(f"{BASE_URL}/api/auth/me", timeout=30).json() admin_uid = me.get("id") or me.get("_id") assert admin_uid, f"auth/me missing id: {me}" r = admin_session.put( f"{BASE_URL}/api/permissions/users/{admin_uid}/role", json={"role": "cashier"}, timeout=30, ) assert r.status_code == 409, r.text # ---------- Sync Push / Pull / Status / Log ---------- class TestSync: def test_push_empty_ops(self, admin_session): r = admin_session.post(f"{BASE_URL}/api/sync/push", json={"ops": []}, timeout=30) assert r.status_code == 200, r.text data = r.json() assert data["processed"] == 0 assert data["ok"] == 0 assert data["skipped"] == 0 assert data["failed"] == 0 def test_push_valid_op_then_idempotent_replay(self, admin_session, default_company_id): op_id = f"test-op-{uuid.uuid4()}" party_name = f"TEST_Sync_Party_{uuid.uuid4().hex[:6]}" op = { "op_id": op_id, "kind": "party", "method": "POST", "url": f"/api/parties?company_id={default_company_id}", "body": {"name": party_name, "type": "customer", "phone": "9999999999"}, "summary": f"Create party {party_name}", } # First push: ok=1 r = admin_session.post(f"{BASE_URL}/api/sync/push", json={"ops": [op]}, timeout=60) assert r.status_code == 200, r.text data = r.json() assert data["processed"] == 1, data assert data["ok"] == 1, data assert data["failed"] == 0, data # Replay: should be skipped (idempotent) r2 = admin_session.post(f"{BASE_URL}/api/sync/push", json={"ops": [op]}, timeout=60) assert r2.status_code == 200, r2.text d2 = r2.json() assert d2["skipped"] == 1, d2 assert d2["ok"] == 0, d2 # Cleanup the created party parties = admin_session.get( f"{BASE_URL}/api/parties?company_id={default_company_id}", timeout=30 ).json() for p in parties: if p.get("name") == party_name: admin_session.delete(f"{BASE_URL}/api/parties/{p['id']}", timeout=30) break def test_push_invalid_url_marked_error(self, admin_session): op = { "op_id": f"test-bad-{uuid.uuid4()}", "kind": "item", "method": "POST", "url": "/items", # missing /api/ prefix "body": {"name": "x"}, } r = admin_session.post(f"{BASE_URL}/api/sync/push", json={"ops": [op]}, timeout=30) assert r.status_code == 200, r.text data = r.json() assert data["failed"] == 1, data assert data["ok"] == 0, data assert data["results"][0]["status"] == "error" def test_pull_returns_arrays(self, admin_session, default_company_id): r = admin_session.get( f"{BASE_URL}/api/sync/pull?company_id={default_company_id}", timeout=60 ) assert r.status_code == 200, r.text data = r.json() assert "server_ts" in data assert isinstance(data.get("parties"), list) assert isinstance(data.get("items"), list) assert isinstance(data.get("invoices"), list) assert "counts" in data server_ts = data["server_ts"] # Second pull with since=server_ts should be near-zero r2 = admin_session.get( f"{BASE_URL}/api/sync/pull?company_id={default_company_id}&since={server_ts}", timeout=60, ) assert r2.status_code == 200 d2 = r2.json() # Allow up to small number of newly-created docs from concurrent tests assert d2["counts"]["parties"] <= 5 assert d2["counts"]["invoices"] <= 5 def test_status(self, admin_session): r = admin_session.get(f"{BASE_URL}/api/sync/status", timeout=30) assert r.status_code == 200, r.text data = r.json() assert "total_synced" in data assert "failed" in data assert "last_sync_ts" in data assert isinstance(data["total_synced"], int) def test_log(self, admin_session): r = admin_session.get(f"{BASE_URL}/api/sync/log", timeout=30) assert r.status_code == 200, r.text logs = r.json() assert isinstance(logs, list) if logs: assert "status" in logs[0] assert "op_id" in logs[0]