from __future__ import annotations import hashlib import os from pathlib import Path import sys import unittest import uuid ROOT = Path(__file__).resolve().parents[1] sys.path[:0] = [str(ROOT / "scripts"), str(ROOT / "deployments/offline/runtime")] from common import db_exec, db_query_scalar from serviceification_d3_engine import ( CATALOG_AUTHORITY_SQL, DeployError, canonical_bytes, read_optional_catalog_authority, ) from standalone_apprelease_catalog import ( ACTIVE_AUTHORITY_DIAGNOSTIC_SQL, INSTALLATION_STATES_SQL, PENDING_UPGRADE_SQL, idempotency_key, operation_facts, read_catalog_start, ) TARGET = "sha256:" + "2" * 64 TREE = "sha256:" + "3" * 64 PRIOR = "sha256:" + "1" * 64 PUBLICATION = "sha256:" + "4" * 64 KEY = idempotency_key("test2", operation_facts("upgrade", TARGET, TREE, 14)) class CatalogStartTest(unittest.TestCase): def test_identities_are_byte_identical_to_both_existing_installers(self): for environment in ("test1", "test2"): original = {"operation": "upgrade", "appReleaseDigest": TARGET, "l2TreeDigest": TREE, "expectedCatalogRevision": 14} self.assertEqual(original, operation_facts("upgrade", TARGET, TREE, 14)) self.assertEqual(environment + "-apprelease-" + hashlib.sha256(canonical_bytes(original)).hexdigest()[:32], idempotency_key(environment, original)) def test_missing_schema_is_the_only_early_empty_result(self): def no_query(*args, **kwargs): self.fail("missing Catalog schema must not query installation tables") self.assertIsNone(read_catalog_start( {"TIDB_DATABASE": "test"}, environment="test2", operation="install", release_digest=TARGET, l2_tree_digest=TREE, query=no_query, tables_present=lambda _: False, )) def test_active_installation_may_lag_current_org_writer_lineage(self): row = "\t".join(( "696E7374616C6C6174696F6E2D31", "6F72672D31", PRIOR, "14", "6E6578757363726D2D72756E74696D652D617069", "15", "15", PUBLICATION, "9", "10", )) def query(env, sql, **kwargs): self.assertEqual(CATALOG_AUTHORITY_SQL, sql) return row authority = read_catalog_start( {"TIDB_DATABASE": "test"}, environment="test2", operation="upgrade", release_digest=TARGET, l2_tree_digest=TREE, query=query, tables_present=lambda _: True, ) self.assertEqual("installation-1", authority.app_installation_id) self.assertEqual(PRIOR, authority.active_release_digest) self.assertEqual(15, authority.catalog_revision) self.assertEqual(15, authority.security_epoch) def test_existing_inconsistent_or_non_upgrade_state_is_not_absence(self): for state in ("FAILED", "PREPARING", "REMOVING", "UPGRADING\nUPGRADING"): def query(env, sql, **kwargs): if sql == CATALOG_AUTHORITY_SQL: return "" self.assertEqual(INSTALLATION_STATES_SQL, sql) return state with self.subTest(state=state), self.assertRaisesRegex(DeployError, "not a fresh installation"): read_catalog_start( {"TIDB_DATABASE": "test"}, environment="test2", operation="upgrade", release_digest=TARGET, l2_tree_digest=TREE, query=query, tables_present=lambda _: True, ) def test_active_incoherence_reports_safe_catalog_facts(self): details = ( "installation_release=present installation_operation=idle " "installation_revision=22 org_state=present org_operation=busy " "org_revision=23 org_security_epoch=23 active_pointer=present " "publication=present publication_revision=23 publication_security_epoch=23" ) def query(env, sql, **kwargs): if sql == CATALOG_AUTHORITY_SQL: return "" if sql == INSTALLATION_STATES_SQL: return "ACTIVE" self.assertEqual(ACTIVE_AUTHORITY_DIAGNOSTIC_SQL, sql) return details with self.assertRaisesRegex(DeployError, "org_operation=busy"): read_catalog_start( {"TIDB_DATABASE": "test"}, environment="test2", operation="upgrade", release_digest=TARGET, l2_tree_digest=TREE, query=query, tables_present=lambda _: True, ) def test_both_installers_recheck_retry_authority_but_require_idle_final_state(self): for environment in ("test1", "test2"): source = (ROOT / f"scripts/manage_{environment}_apprelease.py").read_text() self.assertIn(f'environment="{environment}"', source) self.assertIn("before = read_start()", source) self.assertIn("if read_start() != before:", source) self.assertIn("read_optional_catalog_authority(envv), output", source) self.assertIn(f'idempotency_key("{environment}", operation_facts)', source) @unittest.skipUnless(os.environ.get("STANDALONE_CATALOG_TIDB") == "1", "requires disposable local TiDB") class CatalogStartTiDBTest(unittest.TestCase): """Execute the production SQL/reader against an isolated TiDB database. The five tables are a projection fixture for this read boundary, not a claim to reproduce the entire Catalog writer. No existing database is modified. """ @classmethod def setUpClass(cls): cls.database = "test_catalog_retry_" + uuid.uuid4().hex cls.env = {"TIDB_HOST": "127.0.0.1", "TIDB_PORT": os.environ["STANDALONE_CATALOG_TEST_PORT"], "TIDB_USER": "root", "TIDB_PASSWORD": "", "TIDB_TLS": "false", "TIDB_DATABASE": cls.database} db_exec(cls.env, f"CREATE DATABASE `{cls.database}`") cls.addClassCleanup(db_exec, cls.env, f"DROP DATABASE `{cls.database}`") db_exec(cls.env, """ CREATE TABLE _System_CatalogAppInstallation ( AppInstallationId VARCHAR(36), OrgId VARCHAR(255), AppId VARCHAR(128), State VARCHAR(32), ActiveReleaseDigest VARCHAR(71), CandidateReleaseDigest VARCHAR(71), CatalogRevision BIGINT, ActiveOperationId VARCHAR(36)); CREATE TABLE _System_CatalogOrgState ( OrgId VARCHAR(255), RuntimeAudience VARCHAR(255), CatalogRevision BIGINT, LastSecurityEpoch BIGINT, ActiveOperationId VARCHAR(36)); CREATE TABLE _System_CatalogPublication ( PublicationDigest VARCHAR(71), OrgId VARCHAR(255), CatalogRevision BIGINT, SecurityEpoch BIGINT, SchemaGeneration BIGINT, EngineGeneration BIGINT); CREATE TABLE _System_CatalogPublicationActive ( OrgId VARCHAR(255), ActivePublicationPresent INT, ActivePublicationDigest VARCHAR(71)); CREATE TABLE _System_CatalogAppOperation ( OperationId VARCHAR(36), AppInstallationId VARCHAR(36), OrgId VARCHAR(255), AppId VARCHAR(128), Kind VARCHAR(16), State VARCHAR(32), Step VARCHAR(64), IdempotencyKey VARCHAR(255), PriorReleaseDigest VARCHAR(71), TargetReleaseDigest VARCHAR(71)); """, database=cls.database) def execute(self, sql): db_exec(self.env, sql, database=self.database) def setUp(self): for table in ("AppInstallation", "OrgState", "Publication", "PublicationActive", "AppOperation"): self.execute(f"DELETE FROM _System_Catalog{table}") self.execute(f""" INSERT INTO _System_CatalogAppInstallation VALUES ('installation-1','org-1','io.nexuscrm.sfcrm','ACTIVE','{PRIOR}',NULL,14,NULL); INSERT INTO _System_CatalogOrgState VALUES ('org-1','nexuscrm-runtime-api',14,14,NULL); INSERT INTO _System_CatalogPublication VALUES ('{PUBLICATION}','org-1',14,14,9,10); INSERT INTO _System_CatalogPublicationActive VALUES ('org-1',1,'{PUBLICATION}'); """) self.original = read_optional_catalog_authority(self.env) self.assertIsNotNone(self.original) def pending(self): self.execute(f""" INSERT INTO _System_CatalogAppOperation VALUES ('operation-1','installation-1','org-1','io.nexuscrm.sfcrm', 'UPGRADE','UPGRADING','CANDIDATE_COMMITTED','{KEY}','{PRIOR}','{TARGET}'); UPDATE _System_CatalogAppInstallation SET State='UPGRADING', ActiveOperationId='operation-1', CandidateReleaseDigest='{TARGET}'; UPDATE _System_CatalogOrgState SET ActiveOperationId='operation-1'; """) def read(self, **changes): return read_catalog_start(self.env, **{ "environment": "test2", "operation": "upgrade", "release_digest": TARGET, "l2_tree_digest": TREE, **changes, }) def test_reproduces_old_false_absence_then_recovers_same_start(self): self.pending() self.assertIsNone(read_optional_catalog_authority(self.env), "old reader must reproduce Test2 failure") self.assertEqual(self.original, self.read()) # Recovery itself is read-only; Core still owns continuation. self.assertEqual("UPGRADING", db_query_scalar(self.env, "SELECT State FROM _System_CatalogAppInstallation", database=self.database)) def test_idle_installation_and_fresh_database_are_unchanged(self): self.assertEqual(self.original, self.read()) self.execute("DELETE FROM _System_CatalogAppInstallation") self.assertIsNone(self.read(operation="install")) def test_idle_installation_uses_newer_active_org_publication(self): newer_publication = "sha256:" + "7" * 64 self.execute(f""" INSERT INTO _System_CatalogPublication VALUES ('{newer_publication}','org-1',15,15,11,12); UPDATE _System_CatalogPublicationActive SET ActivePublicationDigest='{newer_publication}'; UPDATE _System_CatalogOrgState SET CatalogRevision=15, LastSecurityEpoch=15; """) authority = self.read() self.assertEqual(PRIOR, authority.active_release_digest) self.assertEqual(15, authority.catalog_revision) self.assertEqual(15, authority.security_epoch) self.assertEqual(newer_publication, authority.active_publication_digest) self.assertEqual(11, authority.schema_generation) self.assertEqual(12, authority.engine_generation) def test_explicit_install_cannot_misinterpret_a_pending_upgrade(self): self.pending() with self.assertRaisesRegex(DeployError, "not a fresh installation"): self.read(operation="install") def test_wrong_release_tree_or_environment_cannot_take_over_pending_work(self): self.pending() for changed in ({"release_digest": PRIOR}, {"l2_tree_digest": "sha256:" + "5" * 64}, {"environment": "test1"}): with self.subTest(changed=changed), self.assertRaisesRegex(DeployError, "different release inputs"): self.read(**changed) def test_catalog_owns_later_recovery_phases_and_previous_publication_is_retained(self): self.pending() for step in ("CANDIDATE_COMMITTED", "SERVING_ARTIFACTS_COMMITTED", "PROJECTION_ACKNOWLEDGED"): self.execute(f"UPDATE _System_CatalogAppOperation SET Step='{step}'") self.assertEqual(self.original, self.read()) self.execute(f""" INSERT INTO _System_CatalogPublication VALUES ('{'sha256:' + '6' * 64}','org-1',15,15,9,10); UPDATE _System_CatalogPublicationActive SET ActivePublicationDigest='{'sha256:' + '6' * 64}'; """) self.assertEqual(self.original, self.read()) def test_mismatched_operation_owner_is_not_recovered(self): self.pending() for field in ("OrgId", "AppInstallationId", "AppId"): original = {"OrgId": "org-1", "AppInstallationId": "installation-1", "AppId": "io.nexuscrm.sfcrm"}[field] self.execute(f"UPDATE _System_CatalogAppOperation SET {field}='other'") with self.subTest(field=field), self.assertRaisesRegex(DeployError, "starting authority"): self.read() self.execute(f"UPDATE _System_CatalogAppOperation SET {field}='{original}'") def test_missing_or_ambiguous_starting_publication_stops(self): self.pending() self.execute(f"INSERT INTO _System_CatalogPublication VALUES ('{'sha256:' + '7' * 64}','org-1',14,14,9,10)") with self.assertRaisesRegex(DeployError, "starting authority"): self.read() self.execute("DELETE FROM _System_CatalogPublication") with self.assertRaisesRegex(DeployError, "starting authority"): self.read() if __name__ == "__main__": unittest.main()