"""Tests for new Module Switches catalog + Migration export/import endpoints.""" import os import io import json import zipfile import requests BASE_URL = os.environ.get("REACT_APP_BACKEND_URL", "https://offline-billing-pro-2.preview.emergentagent.com").rstrip("/") class TestModuleSwitches: def test_catalog_has_all_categories(self, admin_session): r = admin_session.get(f"{BASE_URL}/api/admin/features/catalog", timeout=15) assert r.status_code == 200, r.text data = r.json() assert isinstance(data, list) assert len(data) >= 40, f"Expected ≥40 modules, got {len(data)}" categories = {m["category"] for m in data} # Must cover the major buckets from user's spec for must in ("Transactions", "Masters", "HR & Ops", "Sales Channels", "Integrations", "Platform"): assert must in categories, f"Missing category: {must}" def test_catalog_module_shape(self, admin_session): r = admin_session.get(f"{BASE_URL}/api/admin/features/catalog", timeout=15).json() sample = r[0] for k in ("key", "label", "category", "status", "enabled", "hide_when_disabled"): assert k in sample, f"Missing field: {k}" # Status must be one of the allowed values statuses = {m["status"] for m in r} assert statuses.issubset({"active", "trial", "premium", "coming_soon", "expired"}) def test_toggle_disable_then_enable(self, admin_session): r = admin_session.put(f"{BASE_URL}/api/admin/features/barcode", json={"enabled": False}, timeout=15) assert r.status_code == 200, r.text assert r.json()["barcode"] is False r = admin_session.put(f"{BASE_URL}/api/admin/features/barcode", json={"enabled": True}, timeout=15) assert r.status_code == 200 assert r.json()["barcode"] is True def test_coming_soon_cannot_be_enabled(self, admin_session): """Coming-soon modules are placeholders — UI should reject enable attempts.""" r = admin_session.put(f"{BASE_URL}/api/admin/features/payroll", json={"enabled": True}, timeout=15) assert r.status_code == 400, r.text assert "future release" in r.json()["detail"].lower() or "implement" in r.json()["detail"].lower() def test_coming_soon_can_be_explicitly_disabled(self, admin_session): """Disabling a coming_soon module is a no-op but must not 400.""" r = admin_session.put(f"{BASE_URL}/api/admin/features/payroll", json={"enabled": False}, timeout=15) assert r.status_code == 200, r.text def test_unknown_feature_returns_404(self, admin_session): r = admin_session.put(f"{BASE_URL}/api/admin/features/nonexistent_module", json={"enabled": True}, timeout=15) assert r.status_code == 404, r.text def test_bulk_toggle(self, admin_session): r = admin_session.post( f"{BASE_URL}/api/admin/features/bulk-toggle", json={"keys": ["barcode", "qrcode", "pos"], "enabled": True}, timeout=15, ) assert r.status_code == 200, r.text data = r.json() assert data["updated"] == 3 assert data["features"]["barcode"] is True def test_bulk_toggle_skips_coming_soon(self, admin_session): r = admin_session.post( f"{BASE_URL}/api/admin/features/bulk-toggle", json={"keys": ["barcode", "payroll", "attendance"], "enabled": True}, timeout=15, ) assert r.status_code == 200, r.text data = r.json() # barcode = updated; payroll + attendance = skipped (coming_soon) assert data["updated"] == 1 skipped_keys = {s["key"] for s in data["skipped"]} assert "payroll" in skipped_keys and "attendance" in skipped_keys def test_features_requires_admin(self): r = requests.get(f"{BASE_URL}/api/admin/features/catalog", timeout=15) assert r.status_code == 401 class TestMigration: def test_info_returns_metrics(self, admin_session): r = admin_session.get(f"{BASE_URL}/api/admin/migrate/info", timeout=20) assert r.status_code == 200, r.text d = r.json() for k in ("format_version", "db_collections", "db_documents", "db_counts", "gridfs_files"): assert k in d assert d["format_version"] == 1 assert d["db_collections"] > 0 def test_info_requires_admin(self): r = requests.get(f"{BASE_URL}/api/admin/migrate/info", timeout=15) assert r.status_code == 401 def test_export_returns_valid_zip(self, admin_session): r = admin_session.post( f"{BASE_URL}/api/admin/migrate/export", json={"label": "pytest", "encrypt": True, "include_uploads": True}, timeout=120, ) assert r.status_code == 200, r.text body = r.content assert len(body) > 1000, "Suspiciously small export" zf = zipfile.ZipFile(io.BytesIO(body)) names = set(zf.namelist()) assert "manifest.json" in names assert "db.snapshot" in names assert "README.txt" in names mf = json.loads(zf.read("manifest.json")) assert mf["kind"] == "rbs-regal-migration" assert mf["format_version"] == 1 assert mf["db_encrypted"] is True assert mf.get("jwt_secret_fingerprint") # non-empty assert mf.get("db_counts"), "manifest must include per-collection counts" assert mf.get("source_host") def test_export_unencrypted(self, admin_session): r = admin_session.post( f"{BASE_URL}/api/admin/migrate/export", json={"label": "plain", "encrypt": False, "include_uploads": False}, timeout=120, ) assert r.status_code == 200, r.text zf = zipfile.ZipFile(io.BytesIO(r.content)) mf = json.loads(zf.read("manifest.json")) assert mf["db_encrypted"] is False def test_import_round_trip_merge(self, admin_session): # Export r = admin_session.post( f"{BASE_URL}/api/admin/migrate/export", json={"label": "roundtrip", "encrypt": True, "include_uploads": False}, timeout=120, ) assert r.status_code == 200 pkg = r.content # Import in merge mode sess = requests.Session() sess.cookies.update(admin_session.cookies) files = {"file": ("test.zip", pkg, "application/zip")} r2 = sess.post(f"{BASE_URL}/api/admin/migrate/import?mode=merge", files=files, timeout=240) assert r2.status_code == 200, r2.text d = r2.json() assert d["ok"] is True assert d["restored_collections"] > 0 assert d["restored_documents"] > 0 assert d["label"] == "roundtrip" def test_import_rejects_non_zip(self, admin_session): sess = requests.Session() sess.cookies.update(admin_session.cookies) r = sess.post( f"{BASE_URL}/api/admin/migrate/import?mode=merge", files={"file": ("notazip.zip", b"this is not a zip file", "application/zip")}, timeout=30, ) assert r.status_code == 400, r.text def test_import_rejects_zip_missing_manifest(self, admin_session): # Build a bogus zip with random content buf = io.BytesIO() with zipfile.ZipFile(buf, "w") as z: z.writestr("random.txt", b"hello") sess = requests.Session() sess.cookies.update(admin_session.cookies) r = sess.post( f"{BASE_URL}/api/admin/migrate/import?mode=merge", files={"file": ("bad.zip", buf.getvalue(), "application/zip")}, timeout=30, ) assert r.status_code == 400, r.text assert "manifest" in r.json()["detail"].lower() or "db.snapshot" in r.json()["detail"].lower() def test_import_invalid_mode(self, admin_session): # Create a real export first exp = admin_session.post( f"{BASE_URL}/api/admin/migrate/export", json={"label": "mode-test", "encrypt": True, "include_uploads": False}, timeout=120, ) sess = requests.Session() sess.cookies.update(admin_session.cookies) r = sess.post( f"{BASE_URL}/api/admin/migrate/import?mode=nuke", files={"file": ("p.zip", exp.content, "application/zip")}, timeout=30, ) assert r.status_code == 400, r.text def test_export_requires_admin(self): r = requests.post(f"{BASE_URL}/api/admin/migrate/export", json={}, timeout=15) assert r.status_code == 401