atproto pds in zig pds.zat.dev
pds atproto
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354#!/usr/bin/env python3"""Check that concurrent smoke failures remain CI failures, without a PDS."""import osfrom pathlib import Pathimport shutilimport subprocessimport tempfileimport unittest
class SmokeRunnerTest(unittest.TestCase): def run_suites(self, public=0, permissioned=0, build=0): with tempfile.TemporaryDirectory() as directory: root = Path(directory) (root / 'tools').mkdir() shutil.copyfile(Path(__file__).with_name('smoke-all.sh'), root / 'tools/smoke-all.sh') zig = root / 'zig' zig.write_text(f'#!/bin/sh\nexit {build}\n') zig.chmod(0o755) for name, other, code in [('smoke', 'smoke-permissioned', public), ('smoke-permissioned', 'smoke', permissioned)]: script = root / 'tools' / f'{name}.sh' # The rendezvous fails if the runner accidentally becomes serial. script.write_text(f'''#!/bin/shset -eutest "$ZDS_SMOKE_PREBUILT" = 1touch {name}.startedi=0while [ ! -f {other}.started ]; do i=$((i + 1)) test "$i" -lt 100 sleep 0.01doneexit {code}''') script.chmod(0o755) env = dict(os.environ, PATH=str(root) + os.pathsep + os.environ['PATH']) result = subprocess.run(['sh', 'tools/smoke-all.sh'], cwd=root, env=env, capture_output=True, text=True, timeout=30) if build: self.assertFalse(list(root.glob('*.started'))) return result.returncode
def test_failure_matrix(self): cases = [(0, 0, 0), (1, 0, 0), (0, 1, 0), (1, 1, 0), (0, 0, 1)] results = [self.run_suites(*case) for case in cases] for case, result in zip(cases, results): with self.subTest(public=case[0], permissioned=case[1], build=case[2]): self.assertEqual(result == 0, case == (0, 0, 0))
if __name__ == '__main__': unittest.main()