import sys import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[1] RUNTIME_DIR = ROOT / "deployments" / "offline" / "runtime" SALESFORCE_MIGRATION_DIR = ROOT / "scripts" / "salesforce_migration" sys.path.insert(0, str(RUNTIME_DIR)) sys.path.insert(0, str(SALESFORCE_MIGRATION_DIR)) import deal_document_salesforce_recovery as recovery # noqa: E402 import file_compatibility_url_reconcile as url_reconcile # noqa: E402 import salesforce_file_projection_audit as projection_audit # noqa: E402 import salesforce_file_projection_raw_repair as raw_repair # noqa: E402 from scripts.salesforce_migration import ( # noqa: E402 load_salesforce_snapshot_to_tidb as snapshot_loader, ) class SalesforceFileRecoveryStorageContractTests(unittest.TestCase): envv = { "FILE_STORAGE_PROVIDER": "s3", "FILE_STORAGE_BUCKET": "customer-files", "FILE_STORAGE_REGION": "us-west-2", "FILE_STORAGE_PREFIX": "prod/customer-files/v1", } physical_destination = { "provider": "s3", "bucket": "customer-files", "region": "us-west-2", "key_prefix": "prod/customer-files/v1/salesforce/content/blobs", } def test_projection_audit_resolves_contract_number_as_a_parent_key(self): query = projection_audit.build_target_query( target_table="contract", target_columns=["Id", "Name", "ContractNumber"], target_key="HK-PNA-25010301", ) self.assertIn("`ContractNumber`", query) self.assertIn("'HK-PNA-25010301'", query) self.assertIn("LIMIT 2", query) def test_compatibility_url_reconcile_uses_one_current_file_and_relative_path(self): sql = url_reconcile.build_reconcile_update_sql( target_table="contract", compatibility_url_field="StandardUploadUrl__c", system_file_table="_System_File", parent_object_api_name="Contract", attachment_kind="archived_contract", target_parent_ids=["800-contract-id"], ) self.assertNotIn("ROW_NUMBER() OVER", sql) self.assertIn("ParentObjectApiName = 'Contract'", sql) self.assertIn("AttachmentKind = 'archived_contract'", sql) self.assertIn("SourceType = 'salesforce'", sql) self.assertIn("SourceContentVersionId IS NOT NULL", sql) self.assertIn("TRIM(CAST(SourceContentVersionId AS CHAR)) <> ''", sql) self.assertIn("StorageProvider = 's3'", sql) self.assertIn( "StorageBucket = 'nexuscrm-sf-export-604175033442-us-west-2'", sql, ) self.assertIn("StorageKey = CAST(SourceContentVersionId AS CHAR)", sql) self.assertIn("'salesforce/content/blobs/'", sql) self.assertIn("StoragePath = CONCAT('s3://'", sql) self.assertIn( "CAST(selected_file.Id AS CHAR) = SUBSTRING_INDEX(", sql, ) self.assertIn( "CAST(guard_file.Id AS CHAR) = SUBSTRING_INDEX(", sql, ) self.assertIn("'/files/'", sql) self.assertIn("'/download'", sql) self.assertIn( "CONCAT('/services/data/v63.0/files/', selected_file.Id, '/download')", sql, ) self.assertIn("REGEXP '^https?://'", sql) self.assertIn("CAST(guard_target.`Id` AS CHAR)", sql) self.assertIn("CAST(target_record.`Id` AS CHAR)", sql) self.assertIn("'800-contract-id'", sql) self.assertIn( "reconciliation_guard.existing_target_count = 1", sql, ) self.assertIn( "scoped_absolute_count = reconciliation_guard.resolvable_count", sql, ) def test_order_and_contract_projection_fixtures_share_relative_runtime_url(self): for parent_object, target_table, attachment_kind in ( ("Order", "order", "archived_order"), ("Contract", "contract", "archived_contract"), ): with self.subTest(parent_object=parent_object): sql = url_reconcile.build_reconcile_update_sql( target_table=target_table, compatibility_url_field="StandardUploadUrl__c", system_file_table="_System_File", parent_object_api_name=parent_object, attachment_kind=attachment_kind, target_parent_ids=["parent-1"], ) self.assertIn(f"ParentObjectApiName = '{parent_object}'", sql) self.assertIn(f"AttachmentKind = '{attachment_kind}'", sql) self.assertIn("SourceType = 'salesforce'", sql) self.assertIn( "CONCAT('/services/data/v63.0/files/', " "selected_file.Id, '/download')", sql, ) self.assertNotIn("CONCAT('http", sql) def test_compatibility_url_reconcile_guard_requires_the_complete_requested_batch(self): sql = url_reconcile.build_reconcile_update_sql( target_table="contract", compatibility_url_field="StandardUploadUrl__c", system_file_table="_System_File", parent_object_api_name="Contract", attachment_kind="archived_contract", target_parent_ids=["existing-contract", "stale-contract", "existing-contract"], ) self.assertIn( "COUNT(DISTINCT CAST(guard_target.Id AS CHAR)) AS existing_target_count", sql, ) self.assertIn("reconciliation_guard.existing_target_count = 2", sql) self.assertIn("'existing-contract', 'stale-contract'", sql) def test_compatibility_url_batch_state_owns_existence_and_url_counts(self): sql = url_reconcile.build_batch_state_sql( target_table="contract", compatibility_url_field="StandardUploadUrl__c", target_parent_ids=["existing-contract", "stale-contract"], ) self.assertIn("COUNT(*) AS existing_target_count", sql) self.assertIn("AS absolute_url_count", sql) self.assertIn("'existing-contract', 'stale-contract'", sql) def test_compatibility_url_reconcile_requires_an_explicit_bounded_target(self): with self.assertRaisesRegex( projection_audit.DeployError, "at least one --target-parent-id is required", ): url_reconcile.build_reconcile_update_sql( target_table="contract", compatibility_url_field="StandardUploadUrl__c", system_file_table="_System_File", parent_object_api_name="Contract", attachment_kind="archived_contract", target_parent_ids=[], ) def test_runtime_destination_derives_a_logical_storage_key_prefix(self): bound = recovery.bind_runtime_recovery_storage( self.envv, self.physical_destination, context="test destination", ) self.assertEqual( "salesforce/content/blobs", bound["logical_key_prefix"], ) self.assertEqual( "prod/customer-files/v1/salesforce/content/blobs", bound["key_prefix"], ) def test_runtime_root_cannot_be_used_as_an_ambiguous_recovery_prefix(self): destination = { **self.physical_destination, "key_prefix": "prod/customer-files/v1", } with self.assertRaisesRegex( recovery.DeployError, "must select a logical child", ): recovery.bind_runtime_recovery_storage( self.envv, destination, context="test destination", ) def test_unsafe_logical_destination_prefixes_are_rejected(self): unsafe_prefixes = ( "prod/customer-files/v1/salesforce/../other", "prod/customer-files/v1/salesforce\\other", "prod/customer-files/v1/salesforce//other", ) for key_prefix in unsafe_prefixes: with self.subTest(key_prefix=key_prefix): destination = { **self.physical_destination, "key_prefix": key_prefix, } with self.assertRaisesRegex( recovery.DeployError, "unsafe logical key prefix", ): recovery.bind_runtime_recovery_storage( self.envv, destination, context="test destination", ) def test_snapshot_projection_fixtures_map_order_and_contract(self): link = { "Id": "06A-link", "ContentDocumentId": "069-document", "LinkedEntityId": "legacy-parent-id", "__ResolvedParentId": "runtime-parent-id", } version = { "Id": "068-version", "Title": "historical-file", "PathOnClient": "historical-file.pdf", "ContentSize": "7", "FileExtension": "pdf", } for parent_object, attachment_kind in ( ("Order", "archived_order"), ("Contract", "archived_contract"), ): with self.subTest(parent_object=parent_object): row = snapshot_loader.build_system_file_snapshot_row( parent={ "parent_object_api_name": parent_object, "attachment_kind": attachment_kind, }, link=link, version=version, source_type="salesforce", storage_provider="s3", storage_bucket="salesforce-archive", storage_key_prefix="", storage_path_key_prefix="salesforce/content/blobs", ) self.assertEqual("sfcdl-06A-link", row[0]) self.assertEqual("runtime-parent-id", row[3]) self.assertEqual(parent_object, row[4]) self.assertEqual(attachment_kind, row[5]) self.assertEqual("salesforce", row[9]) self.assertEqual("069-document", row[10]) self.assertEqual("068-version", row[11]) self.assertEqual("Active", row[12]) def test_projection_persists_logical_key_and_physical_storage_path(self): row = snapshot_loader.build_system_file_snapshot_row( parent={ "parent_object_api_name": "Order", "attachment_kind": "archived_order", }, link={ "Id": "06A-link", "ContentDocumentId": "069-document", "__ResolvedParentId": "order-1", }, version={ "Id": "068-version", "Title": "example", "PathOnClient": "example.txt", "ContentSize": "4", }, source_type="salesforce", storage_provider="s3", storage_bucket="customer-files", storage_key_prefix="salesforce/content/blobs", storage_path_key_prefix=( "prod/customer-files/v1/salesforce/content/blobs" ), ) self.assertEqual( "salesforce/content/blobs/068-version", row[15], ) self.assertEqual( "s3://customer-files/" "prod/customer-files/v1/salesforce/content/blobs/068-version", row[16], ) def test_normal_salesforce_projection_uses_version_id_as_logical_key(self): row = snapshot_loader.build_system_file_snapshot_row( parent={ "parent_object_api_name": "Order", "attachment_kind": "archived_order", }, link={ "Id": "06A-link", "ContentDocumentId": "069-document", "__ResolvedParentId": "order-1", }, version={ "Id": "068-version", "Title": "example", "PathOnClient": "example.txt", "ContentSize": "4", }, source_type="salesforce", storage_provider="s3", storage_bucket="salesforce-archive", storage_key_prefix="", storage_path_key_prefix="salesforce/content/blobs", ) self.assertEqual("068-version", row[15]) self.assertEqual( "s3://salesforce-archive/salesforce/content/blobs/068-version", row[16], ) def test_normal_sql_projection_separates_logical_and_physical_keys(self): parent = { "parent_object_api_name": "Order", "target_table": "order", "attachment_kind": "archived_order", } cdl_columns = [ "Id", "ContentDocumentId", "LinkedEntityId", "CreatedDate", ] cv_columns = [ "Id", "Title", "PathOnClient", "ContentSize", "CreatedDate", "LastModifiedDate", "FileExtension", ] kwargs = { "parent": parent, "content_document_link_table": "sf_content_document_link_raw", "content_version_table": "sf_content_version_raw", "system_file_table": "_System_File", "cdl_columns": cdl_columns, "cv_columns": cv_columns, "source_type": "salesforce", "storage_provider": "s3", "storage_bucket": "salesforce-archive", "storage_key_prefix": "salesforce/content/blobs", } snapshot_sql = snapshot_loader.build_salesforce_file_projection_sql( **kwargs, ) raw_sql = raw_repair.build_raw_projection_insert_sql( **kwargs, system_file_columns=list(raw_repair.SYSTEM_FILE_BASE_INSERT_COLUMNS), ) for sql in (snapshot_sql, raw_sql): with self.subTest(sql_builder=sql[:40]): self.assertIn("CONCAT('', cv.`Id`) AS `StorageKey`", sql) self.assertIn( "'salesforce/content/blobs/', cv.`Id`) AS `StoragePath`", sql, ) def test_projection_builders_reject_unsafe_logical_prefixes(self): kwargs = { "parent": { "parent_object_api_name": "Order", "target_table": "order", "attachment_kind": "archived_order", }, "content_document_link_table": "sf_content_document_link_raw", "content_version_table": "sf_content_version_raw", "system_file_table": "_System_File", "cdl_columns": [ "Id", "ContentDocumentId", "LinkedEntityId", ], "cv_columns": ["Id", "Title", "ContentSize"], "source_type": "salesforce", "storage_provider": "s3", "storage_bucket": "salesforce-archive", "storage_key_prefix": "salesforce/content/blobs", } unsafe_prefixes = ( "../escape", "foo//bar", "foo\\bar", "/foo", "foo/", ) for logical_prefix in unsafe_prefixes: with self.subTest(logical_prefix=logical_prefix): with self.assertRaisesRegex( RuntimeError, "logical storage key prefix is not canonical", ): snapshot_loader.build_salesforce_file_projection_sql( **kwargs, logical_storage_key_prefix=logical_prefix, ) with self.assertRaisesRegex( recovery.DeployError, "logical storage key prefix is not canonical", ): raw_repair.build_raw_projection_insert_sql( **kwargs, system_file_columns=list( raw_repair.SYSTEM_FILE_BASE_INSERT_COLUMNS ), logical_storage_key_prefix=logical_prefix, ) def test_projection_audit_requires_logical_key_and_physical_path(self): row = { "SourceContentVersionId": "068-version", "StorageProvider": "s3", "StorageBucket": "salesforce-archive", "StorageKey": "068-version", "StoragePath": ( "s3://salesforce-archive/" "salesforce/content/blobs/068-version" ), } self.assertEqual( "covered", projection_audit.check_file_metadata( row, storage_provider="s3", storage_bucket="salesforce-archive", storage_key_prefix="salesforce/content/blobs", logical_storage_key_prefix="", ), ) self.assertEqual( "unexpected-storage-key", projection_audit.check_file_metadata( { **row, "StorageKey": "salesforce/content/blobs/068-version", }, storage_provider="s3", storage_bucket="salesforce-archive", storage_key_prefix="salesforce/content/blobs", logical_storage_key_prefix="", ), ) with self.assertRaisesRegex( recovery.DeployError, "logical storage key prefix is not canonical", ): projection_audit.check_file_metadata( row, storage_provider="s3", storage_bucket="salesforce-archive", storage_key_prefix="salesforce/content/blobs", logical_storage_key_prefix="../escape", ) if __name__ == "__main__": unittest.main()