from __future__ import annotations import unittest import argparse import json from pathlib import Path import tempfile from scripts import apply_runtime_app_shell as shell def clone_navigation_table(args, source, table): shell.mysql_query(args, f"CREATE TABLE `{table}` LIKE `{source}`.`{table}`") if table == "_System_ApplicationNavigationReceipt": # CREATE TABLE LIKE omits foreign keys. Preserve the canonical L0 # receipt lifecycle required by the current Controlplane image. shell.mysql_query(args, "ALTER TABLE `_System_ApplicationNavigationReceipt` " "ADD FOREIGN KEY (ApplicationId) REFERENCES `_System_App` (Id) ON DELETE CASCADE") class NavigationInitializationTests(unittest.TestCase): def config(self): return {"id": "app", "label": "App", "frontendAppId": "crm", "appMenuItemId": "menu", "startUrl": "/app/app", "isDefault": True, "navigationItems": [{"id": "home", "label": "Home", "target": {"page": {"id": "Home"}}}], "tabPermissions": ["ExplicitGrant"]} def test_shell_never_overwrites_current_navigation_or_derives_tab_grants(self): sql = shell.build_apply_sql(self.config()) self.assertNotIn("NavigationItems =", sql) self.assertIn("CAST('null' AS JSON)", sql) self.assertIn("ExplicitGrant", sql) self.assertNotIn("'Home'", sql) def test_new_template_is_accepted_and_legacy_requires_sealed_translation(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "shell.json" config = self.config() path.write_text(json.dumps(config)) loaded, _ = shell.load_app_shell(path) self.assertEqual(config["navigationItems"], loaded["navigationItems"]) config["navigationItems"] = [{"id": "home", "label": "Home", "type": "page", "pageUrl": "/"}] path.write_text(json.dumps(config)) loaded, _ = shell.load_app_shell(path) with self.assertRaisesRegex(shell.AppShellError, "unique L1 identity"): shell.translate_navigation(loaded["navigationItems"], {}) def test_initialization_rejects_mutable_or_partial_image_references(self): for image in ("catalog:latest", "sha256:abcd", "catalog@sha256:abcd", "catalog:tag", "evil.example/catalog@sha256:" + "a" * 64, "sha256:" + "a" * 64): with self.subTest(image=image): args = argparse.Namespace(controlplane_bin=None, controlplane_image=image) with self.assertRaisesRegex(shell.AppShellError, "immutable digest"): shell.validate_controlplane(args) class AppShellFailureTests(unittest.TestCase): def test_untrusted_image_fails_before_database_access(self): import subprocess import sys with tempfile.TemporaryDirectory() as directory: config = Path(directory) / "config.json" config.write_text(json.dumps(NavigationInitializationTests().config())) result = subprocess.run( [sys.executable, shell.__file__, "--config", str(config), "--controlplane-image", "other.example/tool@sha256:" + "a" * 64, "--db-host", "invalid.invalid", "--db-user", "unused", "--db-database", "unused"], capture_output=True, text=True, timeout=10, ) self.assertNotEqual(0, result.returncode) self.assertIn("owner repository", result.stderr) self.assertNotIn("database command", result.stderr) def test_invalid_navigation_entrypoint_leaves_shared_database_unchanged(self): import os import subprocess import sys import time binary = os.environ.get("APP_SHELL_CONTROLPLANE_BIN") candidate = os.environ.get("APP_SHELL_LOCAL_CANDIDATE") if os.environ.get("NEXUSCRM_APP_SHELL_TIDB_TEST") != "1" or not (binary or candidate): self.skipTest("requires TiDB and a built Controlplane executable or local candidate") args = argparse.Namespace(db_host=os.environ.get("TIDB_HOST", "127.0.0.1"), db_port=int(os.environ.get("TIDB_PORT", "4407")), db_user="root", db_password=os.environ.get("TIDB_PASSWORD", ""), db_database="test", db_tls="false") database = "shell_preflight_" + str(time.time_ns()) source = os.environ.get("APP_SHELL_SCHEMA_SOURCE", "nexuscrm_d3_local") self.assertTrue(source.replace("_", "").isalnum()) shell.mysql_query(args, f"CREATE DATABASE `{database}`") args.db_database = database tables = ("_System_App", "AppMenuItem", "Profile", "ProfileAppVisibility", "ProfileTabSetting", "_System_SetupAuditTrail", "_System_ApplicationNavigationReceipt") try: for table in tables: clone_navigation_table(args, source, table) shell.mysql_query(args, "INSERT INTO Profile (Id,Name) VALUES ('profile','Profile')") with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "config.json" output = Path(directory) / "receipt.json" selection = ["--controlplane-bin", binary] if binary else [] if not binary: from scripts import serviceification_d3_engine as d3 image = d3.exact_images(d3.load_bom(Path(candidate)))["catalog_reconcile"] selection = ["--controlplane-image", image, "--local-candidate", candidate] command = [sys.executable, shell.__file__, "--config", str(path), "--output", str(output), *selection, "--db-host", args.db_host, "--db-port", str(args.db_port), "--db-user", args.db_user, "--db-password", args.db_password, "--db-database", database, "--db-tls", "false"] for existing in (False, True): if existing: shell.mysql_apply(args, shell.build_apply_sql(NavigationInitializationTests().config())) before = {table: shell.mysql_query(args, f"SELECT * FROM `{table}` ORDER BY Id") for table in tables} for kind in ("empty target", "unknown target field", "duplicate item"): with self.subTest(existing=existing, invalid=kind): config = NavigationInitializationTests().config() config.update(label="Must not commit", tabPermissions=["MustNotGrant"]) if kind == "empty target": config["navigationItems"][0]["target"]["page"]["id"] = "" elif kind == "unknown target field": config["navigationItems"][0]["target"]["page"]["unknown"] = True else: config["navigationItems"] *= 2 path.write_text(json.dumps(config)) result = subprocess.run(command, capture_output=True, text=True, timeout=30) self.assertNotEqual(0, result.returncode) expected = "duplicate navigation item id" if kind == "duplicate item" else "Controlplane navigation initialization failed" self.assertIn(expected, result.stderr) self.assertFalse(output.exists()) self.assertEqual(before, {table: shell.mysql_query(args, f"SELECT * FROM `{table}` ORDER BY Id") for table in tables}) path.write_text(json.dumps(NavigationInitializationTests().config())) result = subprocess.run(command, capture_output=True, text=True, timeout=90) self.assertEqual(0, result.returncode, result.stderr) self.assertTrue(output.exists()) initialized = shell.mysql_query(args, "SELECT NavigationRevision,NavigationItems FROM `_System_App`") row_counts = {table: shell.mysql_query(args, f"SELECT COUNT(*) FROM `{table}`") for table in tables} output.unlink() # A lost receipt retries the complete entrypoint. result = subprocess.run(command, capture_output=True, text=True, timeout=90) self.assertEqual(0, result.returncode, result.stderr) self.assertTrue(output.exists()) self.assertEqual(initialized, shell.mysql_query(args, "SELECT NavigationRevision,NavigationItems FROM `_System_App`")) self.assertEqual(row_counts, {table: shell.mysql_query(args, f"SELECT COUNT(*) FROM `{table}`") for table in tables}) finally: args.db_database = "test" shell.mysql_query(args, f"DROP DATABASE `{database}`") def test_app_shell_batch_rolls_back_and_retries_on_tidb(self): import os import time if os.environ.get("NEXUSCRM_APP_SHELL_TIDB_TEST") != "1": self.skipTest("requires a local initialized TiDB schema") args = argparse.Namespace( db_host=os.environ.get("TIDB_HOST", "127.0.0.1"), db_port=int(os.environ.get("TIDB_PORT", "4407")), db_user="root", db_password=os.environ.get("TIDB_PASSWORD", ""), db_database="test", db_tls="false", ) database = "shell_failure_" + str(time.time_ns()) source = os.environ.get("APP_SHELL_SCHEMA_SOURCE", "nexuscrm_d3_local") self.assertTrue(source.replace("_", "").isalnum()) shell.mysql_query(args, f"CREATE DATABASE `{database}`") args.db_database = database try: tables = ("_System_App", "AppMenuItem", "Profile", "ProfileAppVisibility", "ProfileTabSetting", "_System_SetupAuditTrail") for table in tables: clone_navigation_table(args, source, table) shell.mysql_query(args, "INSERT INTO Profile (Id,Name) VALUES ('profile','Profile')") config = NavigationInitializationTests().config() shell.mysql_apply(args, shell.build_apply_sql(config)) before = {table: shell.mysql_query(args, f"SELECT * FROM `{table}` ORDER BY Id") for table in tables} changed = dict(config, label="Changed", isDefault=False) # Fail after the permission DELETE, before its replacement INSERT. sql = shell.build_apply_sql(changed).replace( "INSERT INTO `ProfileTabSetting`", "INSERT INTO `MissingPermissionTable`" ) with self.assertRaises(shell.AppShellError): shell.mysql_apply(args, sql) self.assertEqual(before, {table: shell.mysql_query(args, f"SELECT * FROM `{table}` ORDER BY Id") for table in tables}) shell.mysql_apply(args, shell.build_apply_sql(changed)) shell.mysql_apply(args, shell.build_apply_sql(changed)) self.assertEqual("Changed", shell.mysql_query(args, "SELECT Label FROM `_System_App`").strip()) self.assertEqual("1", shell.mysql_query(args, "SELECT COUNT(*) FROM ProfileTabSetting").strip()) candidate_path = os.environ.get("APP_SHELL_LOCAL_CANDIDATE") if candidate_path: from scripts import serviceification_d3_engine as d3 candidate = Path(candidate_path) args.controlplane_image = d3.exact_images(d3.load_bom(candidate))["catalog_reconcile"] args.controlplane_bin = None args.local_candidate = candidate args.assert_only = False args.output = None clone_navigation_table(args, source, "_System_ApplicationNavigationReceipt") shell.validate_controlplane(args) invalid = dict(changed, navigationItems=[{"id": "home", "label": "Home", "target": {"page": {"id": ""}}}]) with self.assertRaises(shell.AppShellError): shell.initialize_navigation(args, invalid) self.assertEqual("0", shell.mysql_query(args, "SELECT NavigationRevision FROM `_System_App`").strip()) receipt = shell.initialize_navigation(args, changed) self.assertEqual("initialized", receipt["disposition"]) # Treat the previous receipt as lost: repeat the complete install. shell.mysql_apply(args, shell.build_apply_sql(changed)) replay = shell.initialize_navigation(args, changed) self.assertEqual("preserved", replay["disposition"]) self.assertEqual(receipt["document"], replay["document"]) finally: args.db_database = "test" shell.mysql_query(args, f"DROP DATABASE `{database}`") import argparse import json import os import subprocess import sys import tempfile from pathlib import Path import unittest import uuid from scripts.apply_runtime_app_shell import ( AppShellError, build_apply_sql, initialize_navigation, load_app_shell, mysql_apply, mysql_query, sql_string, translate_navigation, ) class NavigationTranslationTests(unittest.TestCase): def test_release_navigation_exposes_org_with_distinct_identity(self): root = Path(__file__).resolve().parents[1] config, _ = load_app_shell(root / "configs/pingcap-crm-app-shell.json") items = translate_navigation(config["navigationItems"], {"Home": "/"}) orgs = [item for item in items if item.get("target", {}).get("resource", {}).get("resource_type") == "Org__c"] self.assertEqual(1, len(orgs)) account = next(item for item in items if item.get("target", {}).get("resource", {}).get("resource_type") == "Account") self.assertNotEqual(account["id"], orgs[0]["id"]) self.assertNotEqual(account["icon"], orgs[0]["icon"]) def test_preserves_order_labels_and_explicit_empty(self): items = [{"id": "custom", "type": "object", "objectApiName": "Account", "label": "My clients"}, {"id": "landing", "type": "page", "pageUrl": "/", "label": "Welcome", "icon": "House"}] self.assertEqual(translate_navigation(items, {"Home": "/"}), [ {"id": "custom", "label": "My clients", "target": {"resource": {"resource_type": "Account"}}}, {"id": "landing", "label": "Welcome", "icon": "House", "target": {"page": {"id": "Home"}}}, ]) self.assertEqual(translate_navigation([], {"Home": "/"}), []) def test_unknown_or_ambiguous_page_cannot_replace_saved_configuration(self): item = {"id": "x", "type": "page", "pageUrl": "/custom", "label": "Custom"} for pages in ({"Home": "/"}, {"A": "/custom", "B": "/custom"}): with self.assertRaisesRegex(AppShellError, "unique L1 identity"): translate_navigation([item], pages) @unittest.skipUnless(os.environ.get("NEXUSCRM_NAVIGATION_TEST_DATABASE"), "requires an isolated TiDB database with canonical L0 schema") class PublishedNavigationInitializationTests(unittest.TestCase): def test_empty_templates_and_explicit_empty_permissions_install_and_retry(self): args = argparse.Namespace(db_host="127.0.0.1", db_port=int(os.environ.get("TIDB_PORT", "4001")), db_user="root", db_password=os.environ.get("TIDB_PASSWORD", ""), db_database=os.environ["NEXUSCRM_NAVIGATION_TEST_DATABASE"], db_tls="false") config = NavigationInitializationTests().config() app_id = "empty-" + uuid.uuid4().hex config.update(id=app_id, appMenuItemId="menu-" + app_id) config.pop("tabPermissions") config["navigationItems"] = [] identity = sql_string(app_id) before = mysql_query(args, "SELECT * FROM ProfileTabSetting ORDER BY Id") try: with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "shell.json" output = Path(directory) / "receipt.json" for items in ([], NavigationInitializationTests().config()["navigationItems"]): config["navigationItems"] = items if items: config["tabPermissions"] = [] path.write_text(json.dumps(config)) for _ in range(2): result = subprocess.run([sys.executable, shell.__file__, "--config", str(path), "--output", str(output), "--controlplane-image", os.environ["NEXUSCRM_NAVIGATION_CATALOG_IMAGE"], "--db-host", args.db_host, "--db-port", str(args.db_port), "--db-user", args.db_user, "--db-password", args.db_password, "--db-database", args.db_database, "--db-tls", "false"], text=True, capture_output=True, timeout=90) self.assertEqual(0, result.returncode, result.stderr) receipt = json.loads(output.read_text()) self.assertEqual([], receipt["navigation_item_ids"]) self.assertEqual(1, receipt["navigation_revision"]) self.assertEqual(before, mysql_query(args, "SELECT * FROM ProfileTabSetting ORDER BY Id")) finally: mysql_apply(args, f"DELETE FROM ProfileAppVisibility WHERE AppId={identity}; DELETE FROM AppMenuItem WHERE AppId={identity}; DELETE FROM _System_App WHERE Id={identity}") def test_deployment_migrates_legacy_then_preserves_edited_and_empty_documents(self): root = Path(__file__).resolve().parents[1] config, _ = load_app_shell(root / "configs/pingcap-crm-app-shell.json") app_id = "navigation-proof-" + uuid.uuid4().hex config.update(id=app_id, name=app_id, appMenuItemId="menu-" + app_id) args = argparse.Namespace( db_host=os.environ.get("TIDB_HOST", "127.0.0.1"), db_port=int(os.environ.get("TIDB_PORT", "4001")), db_user=os.environ.get("TIDB_USER", "root"), db_password=os.environ.get("TIDB_PASSWORD", ""), db_database=os.environ["NEXUSCRM_NAVIGATION_TEST_DATABASE"], db_tls=os.environ.get("TIDB_TLS", "false"), navigation_catalog_image=os.environ["NEXUSCRM_NAVIGATION_CATALOG_IMAGE"], navigation_pages=Path(os.environ["NEXUSCRM_NAVIGATION_PAGES"]), assert_only=False, ) identity = sql_string(app_id) try: mysql_apply(args, build_apply_sql(config)) # Existing legacy edits must win over the release template. legacy = [{"id": "custom", "label": "My start", "type": "page", "pageUrl": "/"}] mysql_apply(args, f"UPDATE _System_App SET NavigationItems={sql_string(json.dumps(legacy))} WHERE Id={identity}") first = initialize_navigation(args, config) self.assertEqual("migrated", first["disposition"]) self.assertEqual(1, first["revision"]) current = json.loads(mysql_query(args, f"SELECT NavigationItems FROM _System_App WHERE Id={identity}")) self.assertEqual([{"id": "custom", "label": "My start", "target": {"page": {"id": "Home"}}}], current) for items in (current, []): mysql_apply(args, f"UPDATE _System_App SET NavigationItems={sql_string(json.dumps(items))}, NavigationRevision=2 WHERE Id={identity}") mysql_apply(args, build_apply_sql(config)) repeated = initialize_navigation(args, config) self.assertEqual({"disposition": "preserved", "revision": 2}, {key: repeated[key] for key in ("disposition", "revision")}) self.assertEqual(items, json.loads(mysql_query(args, f"SELECT NavigationItems FROM _System_App WHERE Id={identity}"))) finally: mysql_apply(args, f"DELETE FROM ProfileAppVisibility WHERE AppId={identity}; DELETE FROM AppMenuItem WHERE AppId={identity}; DELETE FROM _System_App WHERE Id={identity}") def test_invalid_navigation_fails_before_any_shell_write(self): root = Path(__file__).resolve().parents[1] config, _ = load_app_shell(root / "configs/pingcap-crm-app-shell.json") app_id = "navigation-preflight-" + uuid.uuid4().hex config.update(id=app_id, name=app_id, appMenuItemId="menu-" + app_id) args = argparse.Namespace( db_host=os.environ.get("TIDB_HOST", "127.0.0.1"), db_port=int(os.environ.get("TIDB_PORT", "4001")), db_user=os.environ.get("TIDB_USER", "root"), db_password=os.environ.get("TIDB_PASSWORD", ""), db_database=os.environ["NEXUSCRM_NAVIGATION_TEST_DATABASE"], db_tls=os.environ.get("TIDB_TLS", "false"), ) identity = sql_string(app_id) def snapshot(): return mysql_query(args, ";".join( f"SELECT * FROM `{table}` ORDER BY Id" for table in ("_System_App", "AppMenuItem", "ProfileAppVisibility", "ProfileTabSetting", "_System_SetupAuditTrail") )) with tempfile.TemporaryDirectory(prefix="navigation-preflight-") as raw: temporary = Path(raw) config_path = temporary / "shell.json" pages_path = temporary / "pages.json" command = [sys.executable, str(root / "scripts/apply_runtime_app_shell.py"), "--config", str(config_path), "--output", str(temporary / "result.json"), "--navigation-catalog-image", os.environ["NEXUSCRM_NAVIGATION_CATALOG_IMAGE"], "--navigation-pages", str(pages_path), "--db-host", args.db_host, "--db-port", str(args.db_port), "--db-user", args.db_user, "--db-database", args.db_database, "--db-tls", args.db_tls] env = {**os.environ, "TIDB_PASSWORD": args.db_password} try: config_path.write_text(json.dumps(config)) before = snapshot() # A new shell must not be created when its projection is absent. result = subprocess.run(command, env=env, capture_output=True, text=True, timeout=90) self.assertNotEqual(0, result.returncode) self.assertEqual(before, snapshot(), "missing projection mutated a new shell") pages_path.write_text(Path(os.environ["NEXUSCRM_NAVIGATION_PAGES"]).read_text()) result = subprocess.run(command, env=env, capture_output=True, text=True, timeout=90) self.assertEqual(0, result.returncode, result.stderr) evidence = json.loads((temporary / "result.json").read_text()) self.assertEqual("initialized", evidence["navigation_initialization"]["disposition"]) mysql_apply(args, f"UPDATE _System_App SET NavigationRevision=0, NavigationItems={sql_string(json.dumps(config['navigationItems']))} WHERE Id={identity}") config["label"] = "Must not be applied" config_path.write_text(json.dumps(config)) cases = [ ("malformed projection", "{", None), ("wrong projection shape", "[]", None), ("unknown legacy route", '{"Home":"/"}', "/legacy"), ("ambiguous legacy route", '{"Home":"/","A":"/legacy","B":"/legacy"}', "/legacy"), ] for name, pages, route in cases: with self.subTest(case=name): pages_path.write_text(pages) if route: current = [{"id": "old", "label": "Keep", "type": "page", "pageUrl": route}] mysql_apply(args, f"UPDATE _System_App SET NavigationItems={sql_string(json.dumps(current))} WHERE Id={identity}") before = snapshot() result = subprocess.run(command, env=env, capture_output=True, text=True, timeout=90) self.assertNotEqual(0, result.returncode) self.assertEqual(before, snapshot(), name + " mutated shell metadata") # L0 semantic validation also precedes writes, not just JSON parsing. pages_path.write_text('{"Home":"/"}') invalid = [{"id": "unsafe", "label": "Keep", "type": "web", "pageUrl": "javascript:alert(1)"}] mysql_apply(args, f"UPDATE _System_App SET NavigationItems={sql_string(json.dumps(invalid))} WHERE Id={identity}") before = snapshot() result = subprocess.run(command, env=env, capture_output=True, text=True, timeout=90) self.assertNotEqual(0, result.returncode) self.assertEqual(before, snapshot(), "invalid semantic target mutated shell metadata") finally: mysql_apply(args, f"DELETE FROM ProfileAppVisibility WHERE AppId={identity}; DELETE FROM AppMenuItem WHERE AppId={identity}; DELETE FROM _System_App WHERE Id={identity}")