#!/usr/bin/env python3 """Temporary CSVs only; no business configuration, database, or real matrix writes.""" import contextlib import csv import importlib.util import io import json from pathlib import Path import tempfile import unittest HERE = Path(__file__).resolve().parent spec = importlib.util.spec_from_file_location('integrate', HERE / 'integrate_20261010.py') tool = importlib.util.module_from_spec(spec) spec.loader.exec_module(tool) original = {k: getattr(tool, k) for k in ['ROOT', 'COVERAGE', 'CONTRACT']} class IntegrationTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() tool.ROOT = Path(self.tmp.name) tool.COVERAGE = tool.ROOT / 'coverage.csv' tool.CONTRACT = tool.ROOT / 'contract.csv' (tool.ROOT / 'REPORT.md').write_text('fixture evidence\n') self.target = dict(repo='store', method='GET', path='/api/example/list', module='fixture', terminal='PC', role='fixture', risk='read', scope='fixture-read', evidence='REPORT.md', from_repo='unreviewed', from_source='apifox-2026-05-08') self.reset_inputs() def tearDown(self): for key, value in original.items(): setattr(tool, key, value) self.tmp.cleanup() def write(self, path, fields, rows): with path.open('w', newline='') as output: writer = csv.DictWriter(output, fieldnames=fields) writer.writeheader(); writer.writerows(rows) def reset_inputs(self, coverage_owner='unreviewed', source='apifox-2026-05-08', contract_path='/api/example/list'): for real, temp, column, path, owner in [ (original['COVERAGE'], tool.COVERAGE, 'route_or_command', '/api/example/list', coverage_owner), (original['CONTRACT'], tool.CONTRACT, 'path', contract_path, 'unreviewed')]: with real.open(newline='') as file: fields = next(csv.reader(file)) row = dict.fromkeys(fields, '') row.update(repo=owner, method='GET', source=source) row[column] = path if 'kind' in row: row['kind'] = 'http' self.write(temp, fields, [row]) def run_tool(self): with contextlib.redirect_stdout(io.StringIO()): tool.main([self.target], tool.ROOT / 'audit.json', lambda: None) def test_exact_owner_resolution_preserves_open_gate_and_is_idempotent(self): self.run_tool() audit = json.loads((tool.ROOT / 'audit.json').read_text()) self.assertEqual(len(audit['ownership_resolution']), 2) with tool.CONTRACT.open(newline='') as file: row = next(csv.DictReader(file)) self.assertEqual(row['repo'], 'store'); self.assertEqual(row['coverage_gate'], 'open') before = [p.read_bytes() for p in [tool.COVERAGE, tool.CONTRACT]] self.run_tool() self.assertEqual(before, [p.read_bytes() for p in [tool.COVERAGE, tool.CONTRACT]]) self.assertEqual(len(json.loads((tool.ROOT / 'audit.json').read_text())['ownership_resolution']), 2) def test_missing_contract_target_does_not_partially_write_coverage(self): self.reset_inputs(contract_path='/api/other/list') before = [p.read_bytes() for p in [tool.COVERAGE, tool.CONTRACT]] with self.assertRaisesRegex(RuntimeError, 'Missing registered target'): self.run_tool() self.assertEqual(before, [p.read_bytes() for p in [tool.COVERAGE, tool.CONTRACT]]) def test_verified_retained_test_data_is_not_falsely_labeled_deleted(self): self.target['cleanup'] = 'retained-owned-marked-ledger-verified-credentials-and-workers-released' self.run_tool() with tool.CONTRACT.open(newline='') as file: row = next(csv.DictReader(file)) self.assertEqual(row['cleanup'], self.target['cleanup']) self.assertEqual(row['coverage_gate'], 'open') def test_existing_confirmed_delivery_scope_is_not_downgraded(self): with tool.CONTRACT.open(newline='') as file: reader = csv.DictReader(file); fields = reader.fieldnames; rows = list(reader) rows[0].update(repo='store', delivery_scope='confirmed-in-scope-device') self.write(tool.CONTRACT, fields, rows) self.run_tool() with tool.CONTRACT.open(newline='') as file: row = next(csv.DictReader(file)) self.assertEqual(row['delivery_scope'], 'confirmed-in-scope-device') def test_shared_original_failure_is_not_labeled_positive(self): self.target.update(mysql_state='tested-fail-shared-original-schema', dm_state='tested-fail-shared-original-schema', success_case='tested-fail-shared-original-schema-not-positive') self.run_tool() with tool.COVERAGE.open(newline='') as file: row = next(csv.DictReader(file)) self.assertEqual(row['mysql_http_or_command'], self.target['mysql_state']) self.assertEqual(row['dm_http_or_command'], self.target['dm_state']) with tool.CONTRACT.open(newline='') as file: row = next(csv.DictReader(file)) self.assertEqual(row['success_case'], self.target['success_case']) self.assertEqual(row['coverage_gate'], 'open') def test_unknown_original_source_is_rejected_without_write(self): self.reset_inputs(source='unknown-source') before = [p.read_bytes() for p in [tool.COVERAGE, tool.CONTRACT]] with self.assertRaisesRegex(RuntimeError, 'Unexpected original owner/source'): self.run_tool() self.assertEqual(before, [p.read_bytes() for p in [tool.COVERAGE, tool.CONTRACT]]) def add_case_alias_rows(self, third=False): for path,column in [(tool.COVERAGE,'route_or_command'),(tool.CONTRACT,'path')]: with path.open(newline='') as file: reader=csv.DictReader(file);fields=reader.fieldnames;rows=list(reader) alias=dict(rows[0]);alias[column]='/API/example/list';rows.append(alias) if third: alias=dict(rows[0]);alias[column]='/Api/example/list';rows.append(alias) self.write(path,fields,rows) def test_verified_case_only_source_aliases_keep_both_raw_rows_one_target(self): self.add_case_alias_rows() self.target['verified_source_case_aliases']=['/api/example/list','/API/example/list'] self.run_tool() for path in [tool.COVERAGE,tool.CONTRACT]: with path.open(newline='') as file:rows=list(csv.DictReader(file)) self.assertEqual(len(rows),2) self.assertTrue(all(row['repo']=='store' for row in rows)) audit=json.loads((tool.ROOT/'audit.json').read_text()) self.assertEqual(audit['target_count'],1) self.assertEqual(len(audit['case_alias_resolution']),2) self.assertEqual(len(audit['ownership_resolution']),4) def test_unlisted_source_case_alias_rejected_before_either_file_written(self): self.add_case_alias_rows(third=True) self.target['verified_source_case_aliases']=['/api/example/list','/API/example/list'] before=[p.read_bytes() for p in [tool.COVERAGE,tool.CONTRACT]] with self.assertRaisesRegex(RuntimeError,'Case alias evidence does not cover original rows'):self.run_tool() self.assertEqual(before,[p.read_bytes() for p in [tool.COVERAGE,tool.CONTRACT]]) def test_real_any_route_is_not_converted_to_one_method(self): with tool.COVERAGE.open(newline='') as file: reader = csv.DictReader(file); fields = reader.fieldnames; rows = list(reader) rows[0].update(repo='store', source='explicit-ANY-route', method='*') self.write(tool.COVERAGE, fields, rows) before = tool.COVERAGE.read_bytes() with self.assertRaisesRegex(RuntimeError, 'genuine ANY'): self.run_tool() self.assertEqual(before, tool.COVERAGE.read_bytes()) def prepare_new_runtime_target(self): for path,column in [(tool.COVERAGE,'route_or_command'),(tool.CONTRACT,'path')]: with path.open(newline='') as file: reader=csv.DictReader(file);fields=reader.fieldnames;rows=list(reader) rows[0][column]='/api/existing/list' self.write(path,fields,rows) self.target.update(register_new_runtime_key=True,runtime_observed_method='GET', runtime_client_source='fixture-client.js',runtime_controller_source='fixture-controller.php') def test_new_actual_runtime_key_registered_in_both_inputs_with_open_gate(self): self.prepare_new_runtime_target();self.run_tool() for path,column in [(tool.COVERAGE,'route_or_command'),(tool.CONTRACT,'path')]: with path.open(newline='') as file:rows=list(csv.DictReader(file)) added=[row for row in rows if row[column]=='/api/example/list'] self.assertEqual(len(rows),2);self.assertEqual(len(added),1) self.assertEqual(added[0]['source'],'current-client-controller-runtime-observed') audit=json.loads((tool.ROOT/'audit.json').read_text()) self.assertEqual(len(audit['runtime_key_registration']),2) with tool.CONTRACT.open(newline='') as file:rows=list(csv.DictReader(file)) self.assertEqual(rows[1]['coverage_gate'],'open') def test_new_runtime_key_missing_client_proof_rejected_without_write(self): self.prepare_new_runtime_target();del self.target['runtime_client_source'] before=[p.read_bytes() for p in [tool.COVERAGE,tool.CONTRACT]] with self.assertRaisesRegex(RuntimeError,'New runtime key requires actual method and client/controller proof'):self.run_tool() self.assertEqual(before,[p.read_bytes() for p in [tool.COVERAGE,tool.CONTRACT]]) def test_direct_controller_http_is_recorded_without_fabricating_product_client(self): self.prepare_new_runtime_target();del self.target['runtime_client_source'] self.target.update(runtime_inventory_scope='controller-direct-http-observed', runtime_direct_http_proof='REPORT.md') self.run_tool() for path in [tool.COVERAGE,tool.CONTRACT]: with path.open(newline='') as file:rows=list(csv.DictReader(file)) self.assertEqual(rows[1]['source'],'controller-direct-http-runtime-observed') audit=json.loads((tool.ROOT/'audit.json').read_text()) entry=audit['runtime_key_registration'][0] self.assertEqual(entry['client_source'],'') self.assertEqual(entry['direct_http_proof'],'REPORT.md') self.assertIn('Product-client usage is unverified',entry['meaning']) with tool.CONTRACT.open(newline='') as file:rows=list(csv.DictReader(file)) self.assertEqual(rows[1]['coverage_gate'],'open') def test_direct_controller_claim_without_existing_wire_report_cannot_register(self): self.prepare_new_runtime_target();del self.target['runtime_client_source'] self.target.update(runtime_inventory_scope='controller-direct-http-observed', runtime_direct_http_proof='missing-report.md') before=[p.read_bytes() for p in [tool.COVERAGE,tool.CONTRACT]] with self.assertRaisesRegex(RuntimeError,'New runtime key requires actual method'):self.run_tool() self.assertEqual(before,[p.read_bytes() for p in [tool.COVERAGE,tool.CONTRACT]]) if __name__ == '__main__': unittest.main()