"""Exercise release projection -> serving config -> supervised inventory/ingress. Schema fixtures are not signature verification or live user-permission evidence. """ import copy import json import tempfile import unittest from pathlib import Path from types import SimpleNamespace from scripts import validate_serviceification_bom as bom from scripts import render_serviceification_test_stack as render from scripts import serviceification_d3_engine as engine from scripts import run_serviceification_d3_staging as staging from scripts import run_serviceification_d3_prod as prod ROOT = Path(__file__).resolve().parents[1] MCP_IMAGE = 'ghcr.io/pingcap-inc/nexuscrm-mcp-worker' def sealed_fixture(with_mcp=True): value = json.loads((ROOT / 'tests/fixtures/serviceification-independent-sources-v10.json').read_text()) if with_mcp: value['services']['mcp_worker'] = { 'oci_repository': MCP_IMAGE, 'source_repository': 'pingcap-inc/nexuscrm-app-packs', 'source_commit': 'a' * 40, 'digest': 'sha256:' + 'b' * 64, 'provenance_digest': 'sha256:' + 'c' * 64, 'sbom_digest': 'sha256:' + 'd' * 64, } value['execution']['topology'] += ';mcp_worker>runtime_api' value['desired_bom_sha256'] = bom.canonical_json_sha256(bom.desired_v10_projection_from_resolved(value)) return value class MCPReleaseTests(unittest.TestCase): def test_local_cli_default_selects_six_roles_and_explicit_bom_is_not_filtered(self): from scripts import run_serviceification_d3_test as runner argv = ['prepare-local'] for name in ('env-file', 'output-dir', 'core-root', 'app-packs-root', 'l2-root', 'deployment-tools-root', 'approval-publication'): argv.extend(['--' + name, '/tmp/local-input']) for role in ('runtime-api', 'runtime-worker', 'integration-service', 'sfcrm-app-service', 'catalog-reconcile', 'frontend'): argv.extend(['--' + role + '-image', 'sha256:' + 'a' * 64]) args = runner.parse_local_preparation_args(argv) self.assertIsNone(args.desired_bom) source = ROOT / 'release/serviceification-d3-bom.json' before = source.read_bytes() local, excluded = runner.local_preparation_bom(args.desired_bom) self.assertEqual(['mcp_worker'], excluded) self.assertNotIn('mcp_worker', local['services']) self.assertNotIn('mcp_worker', local['execution']['topology']) self.assertEqual([], bom.validate_bom(local, require_desired=True)) self.assertEqual(before, source.read_bytes()) explicit = runner.parse_local_preparation_args(argv + ['--desired-bom', str(source)]) with self.assertRaisesRegex(engine.DeployError, 'explicit desired BOM selects MCP'): runner.local_preparation_bom(explicit.desired_bom) with tempfile.TemporaryDirectory() as directory: path = Path(directory) / 'local.json' path.write_text(json.dumps(local)) self.assertEqual((local, []), runner.local_preparation_bom(path)) def test_independent_source_survives_sealed_and_execution_projection(self): sealed = sealed_fixture() self.assertEqual([], bom.validate_bom(sealed, require_deployable=True)) view = bom.deployment_bom_view(sealed) self.assertEqual([], bom.validate_deployment_bom_view(view)) self.assertEqual(sealed['services']['mcp_worker'], view['services']['mcp_worker']) self.assertNotEqual(view['services']['sfcrm_app_service']['source_commit'], 'a' * 40) self.assertIn('mcp_worker', bom.publication_receipt_roles(sealed)) self.assertEqual(MCP_IMAGE + '@sha256:' + 'b' * 64, engine.exact_images(view)['mcp_worker']) for mutation in ('missing_digest', 'wrong_owner', 'competing_identity', 'missing_api', 'altered_service'): candidate = copy.deepcopy(sealed) if mutation == 'missing_digest': del candidate['services']['mcp_worker']['digest'] elif mutation == 'wrong_owner': candidate['services']['mcp_worker']['source_repository'] = 'wrong/owner' elif mutation == 'competing_identity': candidate['catalog_consumers']['authorized_workload_ids'].append('mcp-worker') elif mutation == 'missing_api': candidate['app_installation']['capabilities'] = [c for c in candidate['app_installation']['capabilities'] if c['runtime_capability_id'] != 'runtime.records.mutate.v1'] else: candidate['services']['mcp_worker']['source_commit'] = 'e' * 40 self.assertTrue(bom.validate_bom(candidate, require_deployable=True), mutation) def test_selected_digest_cannot_be_replaced_by_another_valid_receipt(self): from scripts.resolve_serviceification_bom import _load_publication, ValidationFailure as ResolutionFailure sealed = sealed_fixture() request = bom.desired_v10_projection_from_resolved(sealed)['services']['mcp_worker'] receipt = {**sealed['services']['mcp_worker'], 'verification_receipt_digest': 'sha256:'+'f'*64} with tempfile.TemporaryDirectory() as directory: path = Path(directory)/'receipt.json' path.write_text(json.dumps(receipt)) self.assertEqual(receipt['digest'], _load_publication(path, request, 'services.mcp_worker')['digest']) receipt['digest'] = 'sha256:'+'e'*64 path.write_text(json.dumps(receipt)) with self.assertRaises(ResolutionFailure): _load_publication(path, request, 'services.mcp_worker') def test_legacy_bom_does_not_gain_a_worker_or_authority(self): value = sealed_fixture(False) self.assertEqual([], bom.validate_bom(value, require_deployable=True)) self.assertNotIn('mcp_worker', bom.publication_receipt_roles(value)) self.assertNotIn('mcp_worker', engine.exact_images(bom.deployment_bom_view(value))) def test_runtime_admits_client_without_reassigning_handlers(self): view = bom.deployment_bom_view(sealed_fixture()) frontend = {**render.frontend_receipt_identity(view), 'requiredCompatibility': {'hostSdk': {}, 'contributionSchema': {}}} catalog = { 'schema': view['execution']['catalog_reconcile_output_schema'], 'orgId': 'org', 'appInstallationId': 'installation', 'appReleaseDigest': view['app_installation']['app_release_digest'], 'metadataSourceDigest': view['l2_metadata']['tree_sha256'], 'workloadId': 'sfcrm-app-service', 'serviceReleaseDigest': 'sha256:' + 'f' * 64, 'operationId': 'operation', 'catalogRevision': 1, 'schemaGeneration': 1, 'engineGeneration': 1, 'securityEpoch': 1, 'catalogSnapshotDigest': 'sha256:' + 'f' * 64, 'frontendReleaseDigest': view['frontend']['publication']['digest'], 'frontendCompatibilityReceiptDigest': render.sha256_digest(frontend), 'capabilities': [{ 'appCapabilityId': c['capability_id'], 'runtimeCapabilityId': c['runtime_capability_id'], 'handlerBindingId': str(i), 'ownershipEpoch': 1, 'runtimeMetadataViewDigest': 'sha256:' + 'f' * 64, 'compatibilityReceiptDigest': 'sha256:' + 'f' * 64, } for i, c in enumerate(view['app_installation']['capabilities'])], } result = render.render_serving_inputs(view, catalog, frontend, replica_id='replica', org_id='org', workload_id='sfcrm-app-service', catalog_store_root='/catalog', application_owner_id='crm') self.assertEqual(['mcp-worker'], result['user_api_callers']) self.assertIn('mcp-worker', result['authorized_workload_ids']) self.assertTrue(all(c['authorized_workload_id'] != 'mcp-worker' for c in result['capabilities'])) self.assertTrue(all(c['authorized_workload_id'] != 'mcp-worker' for c in result['workload_capabilities'])) with tempfile.TemporaryDirectory() as directory: tokens = engine.provision_workload_identities(Path(directory), mcp=True) self.assertEqual(3, len(set(tokens.values()))) self.assertEqual(0o400, (Path(directory) / 'mcp.token').stat().st_mode & 0o777) identities = json.loads((Path(directory) / 'runtime-workload-identities.json').read_text()) self.assertEqual(set(tokens), {i['workloadId'] for i in identities['identities']}) def test_mcp_is_in_start_health_and_reverse_stop_inventory(self): stems = engine.LEGACY_WORKER_STEMS ports = engine.ports_with_mcp(engine.D3Ports(3101, 3203, 3501, 3502, {s: 3510+i for i, s in enumerate(stems)}), {'mcp_worker': 'image'}) state = {'workerStems': list(stems), 'containers': {'runtime_api': 'runtime', 'integration_service': 'integration', 'sfcrm_app_service': 'sfcrm', 'mcp_worker': 'mcp', 'workers': {s: s for s in stems}}} context = SimpleNamespace(images={'mcp_worker': 'image'}, ports=ports) order = engine.expected_serving_container_order(state) self.assertEqual('mcp', order[-1]) targets = dict(engine.serving_health_targets(context, state)) self.assertEqual('http://127.0.0.1:3503/readyz', targets['mcp']) self.assertEqual('http://127.0.0.1:3101/readyz', targets['sfcrm']) with self.assertRaises(engine.DeployError): engine.ports_with_mcp(engine.D3Ports(3503, 3203, 3501, 3502, {}), {'mcp_worker': 'image'}) def test_restart_role_never_changes_when_mcp_is_appended(self): state = {'containers': {'runtime_api':'api-42', 'integration_service':'integration-73', 'sfcrm_app_service':'sfcrm-98', 'workers':{s:'worker-'+s for s in engine.LEGACY_WORKER_STEMS}}} for with_mcp in (False, True): if with_mcp: state['containers']['mcp_worker'] = 'mcp-101' for role, expected in (('runtime-api','api-42'), ('integration-service','integration-73'), ('sfcrm-app-service','sfcrm-98')): self.assertEqual(expected, engine.restart_container_for_role(state, role)) if with_mcp: self.assertEqual('mcp-101', engine.restart_container_for_role(state, 'mcp-worker')) else: with self.assertRaises(engine.DeployError): engine.restart_container_for_role(state, 'mcp-worker') def test_interrupted_service_replacement_keeps_its_closed_three_role_map(self): view = bom.deployment_bom_view(sealed_fixture()) images = engine.exact_images(view) context = SimpleNamespace(images=images) previous = {k: images[k] for k in ('runtime_api', 'integration_service', 'sfcrm_app_service')} candidate = {**previous, 'sfcrm_app_service': previous['sfcrm_app_service'].split('@')[0] + '@sha256:' + 'e' * 64} state = {'phase': 'service-image-replace-in-progress', 'serviceImageTransition': { 'schema': engine.L0_SERVICE_IMAGE_TRANSITION_SCHEMA, 'previousImages': previous, 'candidateImages': candidate, 'requestedRoles': ['sfcrm_app_service'], }} self.assertEqual(candidate, engine.service_image_transition_candidate_images(context, state)) state['serviceImageTransition']['candidateImages']['mcp_worker'] = images['mcp_worker'] with self.assertRaises(engine.DeployError): engine.service_image_transition_candidate_images(context, state) def test_test_slot_mcp_routes_reenter_and_remove_without_touching_siblings(self): from scripts import run_serviceification_d3_test as test blocks = [] for slot in ('test1', 'test3'): c = test.SLOT_CONTRACT[slot] blocks.append('server {\n server_name '+c.server_name+';\n'+''.join( ' location '+path+' {\n proxy_pass http://127.0.0.1:'+str(c.backend_port)+';\n }\n' for path in ('/mcp', '/.well-known/oauth-protected-resource'))+'}\n') text = ''.join(blocks) contract = test.SLOT_CONTRACT['test3'] ports = engine.D3Ports(contract.backend_port, contract.integration_port, contract.runtime_grpc, contract.runtime_admin, {}) context = SimpleNamespace(envv={'ENV_NAME':'test-test3'}, images={'mcp_worker':'image'}, ports=engine.ports_with_mcp(ports, {'mcp_worker':'image'})) mapped = test.map_slot_mcp_proxy(text, context, enabled=True) self.assertTrue(mapped.startswith(blocks[0])) self.assertEqual(mapped, test.map_slot_mcp_proxy(mapped, context, enabled=True)) self.assertEqual(text, test.map_slot_mcp_proxy(mapped, context, enabled=False)) self.assertIn(':'+str(context.ports.mcp_http)+';', mapped) def test_public_routes_select_worker_while_crm_remains_separate(self): context = SimpleNamespace(run_id='d3-staging-blue-1-1', images={'mcp_worker': 'image'}, ports=engine.D3Ports(3101, 3203, 3501, 3502, {}, 3503), envv={'ENV_NAME': 'staging-blue', 'SERVER_NAME': 'nexuscrm-new-staging.pingcap.com', 'PUBLIC_BASE_URL': 'https://nexuscrm-new-staging.pingcap.com'}) text = staging.render_d3_nginx_server(context, Path('/tmp/frontend')) for path in ('/mcp', '/.well-known/oauth-protected-resource'): block = text.split('location '+path+' {', 1)[1].split('}', 1)[0] self.assertIn('proxy_pass http://127.0.0.1:3503;', block) self.assertIn('proxy_pass http://127.0.0.1:3101/readyz;', text) with self.assertRaises(engine.DeployError): engine.route_mcp_ingress('server {}', context) if __name__ == '__main__': unittest.main()