"""Historical compatibility evidence is complete and clean-install only.""" import json import contextlib from concurrent.futures import ThreadPoolExecutor import io from pathlib import Path import sys import tempfile import threading import types import unittest from unittest.mock import Mock, patch sys.path.insert(0, str(Path(__file__).parent / 'containers')) import version_compatibility as versions class VersionCompatibility(unittest.TestCase): def rows(self): return [{'platform': platform, 'release': release, 'status': 'passed', 'scenarios': [{'name': name, 'status': 'passed'} for name in versions.SCENARIOS], 'image_id': 'sha256:immutable', 'architecture': 'arm64', 'python': '3.12', 'source_manifest_sha256': 'source', 'journal': '/p4/1/journals/journal', 'binary_builds': {name: 'Rev. binary/LINUX/' + release + '/1234' for name in ('p4', 'p4d', 'p4broker', 'p4p')}, 'checksums': {name: 'a' * 64 for name in ('p4', 'p4d', 'p4broker', 'p4p')}} for platform in versions.PLATFORMS for release in versions.RELEASES] def test_complete_fourteen_cell_matrix_passes(self): self.assertEqual(versions.matrix_errors(self.rows()), []) def test_missing_duplicate_blocked_or_shortened_cells_fail(self): rows = self.rows() self.assertTrue(versions.matrix_errors(rows[:-1])) self.assertTrue(versions.matrix_errors(rows + [rows[0]])) extra = dict(rows[0], platform='unexpected-platform') self.assertTrue(versions.matrix_errors(rows + [extra])) rows[0]['scenarios'][0]['status'] = 'blocked' self.assertTrue(versions.matrix_errors(rows)) rows[0]['scenarios'].pop() self.assertTrue(versions.matrix_errors(rows)) def test_wrong_release_or_missing_build_provenance_fails(self): for field in ('image_id', 'architecture', 'python', 'source_manifest_sha256', 'checksums'): rows = self.rows() rows[0].pop(field) self.assertTrue(versions.matrix_errors(rows), field) rows = self.rows() rows[0]['binary_builds']['p4d'] = '2026.1' self.assertTrue(versions.matrix_errors(rows)) rows = self.rows() rows[0]['journal'] = '/p4/1/logs/journal' self.assertTrue(versions.matrix_errors(rows)) self.assertTrue(versions.matrix_errors(self.rows(), source_hash='different-export')) def test_existing_populated_container_is_never_reused_or_removed(self): with tempfile.TemporaryDirectory() as tmp, patch.object(versions, 'verify_manifest', return_value='source'): args = types.SimpleNamespace(source=Path(tmp), results=Path(tmp), name_prefix='sdp-version-existing', docker='fake') matrix = versions.Matrix(args) calls = [] def command(arguments, **kwargs): calls.append(arguments) return types.SimpleNamespace(returncode=0, stdout='existing', stderr='') matrix.command = command row = matrix.cell('rhel8', '2023.1') self.assertEqual(row['status'], 'failed') self.assertIn('existing container', row['error']) self.assertEqual(calls, [['inspect', 'sdp-version-existing-rhel8-2023-1']]) self.assertEqual({r['name'] for r in row['scenarios']}, set(versions.SCENARIOS)) def test_bootstrap_has_fresh_install_guard_and_no_source_compiler(self): script = (Path(versions.__file__).parent / 'version_bootstrap.sh').read_text() self.assertIn('[[ ! -e /p4 && ! -L /p4 ]]', script) self.assertIn('--initialize --confirm-disposable', script) self.assertIn('/inputs/sdp/test/reset.sh', script) self.assertIn('download_binaries.sh', script) self.assertIn('--only-binary=:all:', script) self.assertNotIn('install_sdp_python.sh', script) self.assertNotIn('set -e', script) def test_old_server_diskspace_safety_gap_is_reported_without_hiding_count(self): warning = "Warning: Skipping safety check; p4d version is too old to support needed 'p4 diskspace' features.\n" self.assertEqual(versions.capability_gaps('configuration complete'), []) self.assertEqual(versions.capability_gaps(warning * 5), [ '5 installer disk-space safety checks are unavailable in this server release; see provision.log.']) config_gap = 'COMPATIBILITY: Skipping dm.user.hideinvalid; requires P4D 2024.1 or newer; running 2023.1.' self.assertEqual(versions.capability_gaps(config_gap + '\n' + config_gap), [config_gap]) def test_installer_success_cannot_hide_configuration_errors(self): self.assertEqual(versions.configuration_errors('Warning: older safety check skipped\nExitCode: 0\n'), []) self.assertEqual(versions.configuration_errors('Error: unsupported setting\nExitCode: 5\nProvisioning complete\n'), ['Error: unsupported setting', 'ExitCode: 5']) def test_clone_cleanup_waits_until_shutdown_finishes(self): with tempfile.TemporaryDirectory() as temp: clone = Path(temp) / 'clone' clone.mkdir() (clone / 'pid').write_text('12345\n') stop = Mock() def alive(pid): stop.assert_called_once_with() self.assertTrue(clone.is_dir()) self.assertEqual(pid, 12345) return alive.calls.pop(0) alive.calls = [True, False] with patch.object(versions, 'clone_process_alive', side_effect=alive), patch.object(versions.time, 'sleep'): versions.stop_clone(clone, stop) self.assertFalse(clone.exists()) def test_clone_shutdown_timeout_retains_recovery_data(self): with tempfile.TemporaryDirectory() as temp: clone = Path(temp) (clone / 'pid').write_text('12345\n') (clone / 'db.fixture').write_bytes(b'recovery data') with patch.object(versions, 'clone_process_alive', return_value=True), \ patch.object(versions.time, 'monotonic', side_effect=[0, 2]): with self.assertRaisesRegex(RuntimeError, 'scratch retained at'): versions.stop_clone(clone, Mock(), timeout=1) self.assertEqual((clone / 'db.fixture').read_bytes(), b'recovery data') def test_clone_stop_failure_retains_recovery_data(self): with tempfile.TemporaryDirectory() as temp: clone = Path(temp) (clone / 'pid').write_text('12345\n') stop = Mock(side_effect=RuntimeError('stop failed')) with self.assertRaisesRegex(RuntimeError, 'scratch retained at'): versions.stop_clone(clone, stop) self.assertTrue((clone / 'pid').is_file()) def test_two_jobs_retain_out_of_order_success_and_failure_deterministically(self): with tempfile.TemporaryDirectory() as temp, patch.object(versions, 'verify_manifest', return_value='source'): args = types.SimpleNamespace(source=Path(temp), results=Path(temp), name_prefix='sdp-version-parallel', docker='fake', jobs=2, release=versions.RELEASES[:2], platform=versions.PLATFORMS) matrix = versions.Matrix(args) expected = [(platform, release) for release in args.release for platform in args.platform] rows = {(row['platform'], row['release']): row for row in self.rows()} gate, started = threading.Event(), threading.Barrier(2) lock, counters, reports = threading.Lock(), {'active': 0, 'maximum': 0}, [] def cell(platform, release): key = (platform, release) index = expected.index(key) with lock: counters['active'] += 1 counters['maximum'] = max(counters['maximum'], counters['active']) try: if index < 2: started.wait(timeout=5) if index == 0 and not gate.wait(timeout=5): raise RuntimeError('Incremental report did not retain the second cell') if index == 2: if not gate.wait(timeout=5): raise RuntimeError('First incremental report was not written') raise RuntimeError('simulated collection failure') return rows[key] finally: with lock: counters['active'] -= 1 original_write = matrix.write def write(completed, releases, platforms): reports.append(list(completed)) result = original_write(completed, releases, platforms) gate.set() return result matrix.cell, matrix.write = cell, write self.assertEqual(matrix.run(), 1) self.assertEqual(counters, {'active': 0, 'maximum': 2}) self.assertEqual((reports[0][0]['platform'], reports[0][0]['release']), expected[1]) final = json.loads((Path(temp) / 'version-compatibility.json').read_text()) self.assertEqual([(row['platform'], row['release']) for row in final['cases']], expected) self.assertEqual([row['status'] for row in final['cases']], ['passed', 'passed', 'failed', 'passed']) self.assertIn('simulated collection failure', final['cases'][2]['error']) self.assertEqual(final['jobs'], 2) def test_jobs_default_and_explicit_limit_are_validated_before_execution(self): base = ['version_compatibility.py', '--source', '/source', '--results', '/results'] for extra, expected in [([], 1), (['--jobs', '2'], 2)]: with patch.object(sys, 'argv', base + extra), patch.object(versions, 'Matrix') as factory: factory.return_value.run.return_value = 0 self.assertEqual(versions.main(), 0) self.assertEqual(factory.call_args.args[0].jobs, expected) for extra in (['--jobs', '0'], ['--jobs', '3'], ['--release', '2023.1', '--release', '2023.1'], ['--platform', 'rhel8', '--platform', 'rhel8']): with patch.object(sys, 'argv', base + extra), patch.object(versions, 'Matrix') as factory, \ contextlib.redirect_stderr(io.StringIO()): with self.assertRaises(SystemExit) as error: versions.main() self.assertEqual(error.exception.code, 2) factory.assert_not_called() def test_container_stop_waits_for_other_startup_readiness(self): with tempfile.TemporaryDirectory() as temp, patch.object(versions, 'verify_manifest', return_value='source'): matrix = versions.Matrix(types.SimpleNamespace(source=Path(temp), results=Path(temp))) ready_entered, allow_ready, stopped = threading.Event(), threading.Event(), threading.Event() calls, created = [], Mock() def command(arguments, *args, **kwargs): if arguments[0] == 'exec': calls.append('readiness entered') ready_entered.set() if not allow_ready.wait(5): raise RuntimeError('Readiness test was not released') calls.append('ready') else: calls.append(arguments[0]) if arguments[0] == 'stop': stopped.set() return types.SimpleNamespace(returncode=0, stdout='running', stderr='') matrix.command = command with ThreadPoolExecutor(max_workers=2) as pool: start = pool.submit(matrix.start_container, 'fresh', {'Id': 'immutable', 'Architecture': 'arm64'}, created) self.assertTrue(ready_entered.wait(5)) stop = pool.submit(matrix.stop_container, 'completed', Path(temp) / 'stop.log') try: self.assertFalse(stopped.wait(.05)) finally: allow_ready.set() start.result(timeout=5) stop.result(timeout=5) created.assert_called_once_with() self.assertEqual(calls, ['run', 'readiness entered', 'ready', 'stop']) def test_failed_readiness_keeps_created_container_cleanup_ownership(self): with tempfile.TemporaryDirectory() as temp, patch.object(versions, 'verify_manifest', return_value='source'): args = types.SimpleNamespace(source=Path(temp), results=Path(temp), name_prefix='sdp-version-startup', docker='fake') matrix = versions.Matrix(args) calls = [] def command(arguments, *args, **kwargs): calls.append(arguments) output, code = '', 0 if arguments[0] == 'inspect': code = 1 elif arguments[:2] == ['image', 'inspect']: output = json.dumps([{'Id': 'immutable', 'Architecture': 'arm64'}]) elif arguments[0] == 'info': output = 'arm64' return types.SimpleNamespace(returncode=code, stdout=output, stderr='') matrix.command = command matrix.stop_container = Mock(return_value=types.SimpleNamespace(returncode=0)) with patch.object(versions.time, 'sleep'), patch.object(versions.time, 'monotonic', side_effect=[0, 0, 0, 181]): row = matrix.cell('rhel8', '2023.1') self.assertEqual(row['status'], 'failed') self.assertEqual(row['error'], 'Systemd startup timed out') self.assertIn(['exec', 'sdp-version-startup-rhel8-2023-1', 'systemctl', 'list-jobs', '--no-pager'], calls) matrix.stop_container.assert_called_once_with('sdp-version-startup-rhel8-2023-1', Path(temp).resolve() / 'rhel8-2023.1/stop.log') def test_readiness_allows_normal_systemd_timeout_without_accepting_initializing(self): with tempfile.TemporaryDirectory() as temp, patch.object(versions, 'verify_manifest', return_value='source'): matrix = versions.Matrix(types.SimpleNamespace(source=Path(temp), results=Path(temp))) states, calls = iter(['initializing', 'running']), [] def command(arguments, *args, **kwargs): calls.append(arguments) return types.SimpleNamespace(stdout=next(states) if arguments[0] == 'exec' else '', returncode=0) matrix.command = command with patch.object(versions.time, 'sleep'), \ patch.object(versions.time, 'monotonic', side_effect=[0, 0, 0, 91, 91]): matrix.start_container('fresh', {'Id': 'immutable', 'Architecture': 'arm64'}, Mock()) self.assertEqual(sum(arguments[0] == 'exec' for arguments in calls), 2) self.assertEqual(versions.SYSTEMD_START_TIMEOUT, 180) if __name__ == '__main__': unittest.main()