"""Explicit CI-only coexistence proof on its disposable Linux Docker host.""" import json import os from pathlib import Path import subprocess import tempfile import unittest import uuid from network_inventory import collect, network_record from network_prepare import prepare from network_runtime import run ENABLED = os.environ.get('GOWALK_NETWORK_DOCKER_TEST') == '1' IMAGE = 'busybox:1.37.0' def docker(*args): result = subprocess.run(['docker', *args], stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=60) if result.returncode: raise AssertionError('Isolated Docker command failed: ' + args[0]) return result.stdout.decode().strip() def project_file(root, project): source = {'name': project, 'services': { 'database': {'image': IMAGE, 'command': ['sh', '-c', 'mkdir -p /www; printf ' + project + ' > /www/index.html; httpd -f -p 8080 -h /www']}, 'api': {'image': IMAGE, 'command': ['sleep', '180']}}, 'networks': {'default': {'name': project + '_default'}}} path = root / (project + '.json') path.write_text(json.dumps(source)) rendered = json.loads(docker('compose', '-f', str(path), '--project-name', project, 'config', '--format', 'json')) return path, rendered def container_id(path, project, service): return docker('compose', '-f', str(path), '--project-name', project, 'ps', '-q', service) def fixture_ipam_evidence(observed): """Project validated address fields from only this disposable CI fixture before cleanup.""" network = network_record(observed)[0] configurations = [] for pool in network['IPAM']['Config']: row = {key: pool[key] for key in ('Subnet', 'Gateway', 'IPRange') if key in pool} row['auxiliary_address_count'] = len(pool.get('AuxiliaryAddresses') or {}) configurations.append(row) return {'schema': 'gowalk-cicd/backend-network-fixture-ipam.v1', 'name': network['Name'], 'driver': network['IPAM']['Driver'], 'option_count': len(network['IPAM']['Options']), 'config': configurations} def cleanup(projects): for path, app in reversed(projects): raw = docker('network', 'ls', '--filter', 'label=com.docker.compose.project=' + app, '-q') for identity in raw.splitlines(): observed = json.loads(docker('network', 'inspect', identity))[0] if (observed['Labels'].get('com.docker.compose.project') != app or observed['Labels'].get('com.docker.compose.network') != 'default' or observed['Name'] != app + '_default'): raise AssertionError('Owned test cleanup scope changed') print(json.dumps(fixture_ipam_evidence(observed)), flush=True) docker('compose', '-f', str(path), '--project-name', app, 'down', '--remove-orphans') remaining = docker('network', 'ls', '--filter', 'label=com.docker.compose.project=' + app, '-q') if remaining: raise AssertionError('Owned test network remains') @unittest.skipUnless(ENABLED, 'requires explicit isolated CI Docker admission') class DockerProof(unittest.TestCase): def test_six_networks_coexist_and_compose_replays_without_replacement(self): self.assertEqual(os.geteuid(), 0, 'The namespace proof runs on the isolated host as root') baseline = collect(run) self.assertTrue({'host', 'null'} <= {row['Driver'] for row in baseline['networks']}) print(json.dumps({'schema': 'gowalk-cicd/backend-network-fixture-host.v1', **{ key: baseline[key] for key in ('docker_version', 'compose_version', 'python_version', 'pool_source')}}), flush=True) docker('pull', IMAGE) projects, networks, containers = [], [], [] prefix = 'cicd-net-proof-' + uuid.uuid4().hex[:12] with tempfile.TemporaryDirectory(prefix='gowalk-network-proof-') as temporary: try: for index in range(6): app = prefix + '-' + str(index) path, rendered = project_file(Path(temporary), app) projects.append((path, app)) receipt = prepare(rendered, app) networks.append(receipt['networks'][0]) docker('compose', '-f', str(path), '--project-name', app, 'up', '-d', '--pull', 'never') containers.append(container_id(path, app, 'api')) self.assertEqual(len({row['subnets'][0] for row in networks}), 6) self.assertEqual(len({row['id'] for row in networks}), 6) self.verify_projects(projects, networks, containers) after = collect(run) old = {row['Id']: row for row in after['networks']} for network in baseline['networks']: self.assertIn(network['Id'], old) self.assertEqual(old[network['Id']]['IPAM'], network['IPAM']) print(json.dumps({'schema': 'gowalk-cicd/backend-network-coexistence.v1', 'ok': True, 'docker_version': after['docker_version'], 'compose_version': after['compose_version'], 'python_version': after['python_version'], 'networks': networks, 'own_service_http_probes': 6, 'cross_project_http_refusals': 6, 'unchanged_replay_ids': 6, 'preserved_networks': len(baseline['networks'])})) finally: cleanup(projects) def verify_projects(self, projects, networks, containers): for index, (path, app) in enumerate(projects): api = containers[index] self.assertEqual(docker('exec', api, 'wget', '-qO-', '-T', '3', 'http://database:8080'), app) foreign_path, foreign_app = projects[(index + 1) % len(projects)] foreign = container_id(foreign_path, foreign_app, 'database') destination = docker('inspect', '--format', '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}', foreign) denied = subprocess.run(['docker', 'exec', api, 'wget', '-qO-', '-T', '2', 'http://' + destination + ':8080'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=5) self.assertNotEqual(denied.returncode, 0, 'Another project network became reachable') rendered = json.loads(docker('compose', '-f', str(path), '--project-name', app, 'config', '--format', 'json')) receipt = prepare(rendered, app) self.assertEqual(receipt['networks'][0]['id'], networks[index]['id']) docker('compose', '-f', str(path), '--project-name', app, 'up', '-d', '--pull', 'never') self.assertEqual(container_id(path, app, 'api'), api) if __name__ == '__main__': unittest.main()