import base64 import csv import hashlib import json import os import shutil import subprocess import sys import tempfile import unittest from unittest import mock import zipfile APP_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'app')) if APP_DIR not in sys.path: sys.path.insert(0, APP_DIR) import worker_package import worker_package_builder import runtime_security from lifecycle_authority import ( GIT_MANIFEST_NAME, REMOTE_WORKER_CODE_AUTHORITY_FILES, TRUFFLEHOG_MANIFEST_NAME, ) from result_bundle import FORMAT_VERSION from scan_execution import PROTOCOL_VERSION def package_manifest(platform_tag=None): files = {} for name in REMOTE_WORKER_CODE_AUTHORITY_FILES: files[name] = {'path': f'app/{name}', 'sha256': '1' * 64} return { 'schema': worker_package.WORKER_PACKAGE_SCHEMA, 'protocol_version': PROTOCOL_VERSION, 'bundle_format_version': FORMAT_VERSION, 'platform_tag': platform_tag or worker_package.local_platform_tag(), 'capabilities': [ { 'source': 'gitlab', 'platform': 'gitlab', 'planning_kind': 'exact_git_v1', }, { 'source': 'github', 'platform': 'github', 'planning_kind': 'exact_git_v1', }, ], 'app_root': 'app', 'files': files, 'executables': { TRUFFLEHOG_MANIFEST_NAME: { 'path': 'bin/trufflehog.exe', 'sha256': '2' * 64, }, GIT_MANIFEST_NAME: { 'path': 'runtime/git/cmd/git.exe', 'sha256': '3' * 64, }, }, 'assets': { 'detector_policy': { 'path': 'app/detectors.yaml', 'sha256': '4' * 64, }, }, 'runtime_trees': { 'git': { 'path': 'runtime/git', 'sha256': '5' * 64, 'file_count': 7, }, **({ 'python': { 'path': 'runtime/python', 'sha256': '6' * 64, 'file_count': 3, }, } if (platform_tag or worker_package.local_platform_tag()).startswith('windows-') else {}), }, } class WorkerPackageTests(unittest.TestCase): def test_windows_launchers_expose_cli_and_keep_foreground_alias(self): with tempfile.TemporaryDirectory() as root: worker_package_builder._windows_support_files(root) cli = open(os.path.join(root, 'truf-worker.cmd'), encoding='ascii').read() alias = open(os.path.join(root, 'run-worker.cmd'), encoding='ascii').read() prepare = open(os.path.join(root, 'prepare-worker.ps1'), encoding='ascii').read() self.assertIn('remote_worker_bootstrap.py" -- %*', cli) self.assertIn('remote_worker_bootstrap.py" -- run %*', alias) self.assertIn('"*S-1-5-32-544`:(OI)(CI)F" | Out-Null', prepare) self.assertIn("(Join-Path $root '*') /inheritance:d /T /C", prepare) def test_linux_launchers_expose_cli_and_keep_foreground_alias(self): with tempfile.TemporaryDirectory() as root: worker_package_builder._linux_support_files(root) cli = open(os.path.join(root, 'truf-worker'), encoding='ascii').read() alias = open(os.path.join(root, 'run-worker'), encoding='ascii').read() prepare = open(os.path.join(root, 'prepare-worker.sh'), encoding='ascii').read() self.assertIn('remote_worker_bootstrap.py" -- "$@"', cli) self.assertIn('remote_worker_bootstrap.py" -- run "$@"', alias) self.assertTrue(cli.startswith('#!/bin/sh\nset -eu\n')) self.assertIn('prepare-worker.sh requires sudo', prepare) self.assertIn('chown -R 0:0 "$root/bin" "$root/runtime"', prepare) @unittest.skipUnless(os.name == 'nt', 'Windows ACL regression') def test_windows_preparation_resets_children_to_private_inherited_acls(self): with tempfile.TemporaryDirectory() as parent: root = os.path.join(parent, 'worker') child = os.path.join(root, 'runtime', 'python', 'python.exe') os.makedirs(os.path.dirname(child)) with open(child, 'wb') as handle: handle.write(b'python') worker_package_builder._windows_support_files(root) subprocess.run( [ 'powershell.exe', '-NoProfile', '-NonInteractive', '-ExecutionPolicy', 'Bypass', '-File', os.path.join(root, 'prepare-worker.ps1'), ], check=True, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) self.assertTrue(runtime_security.private_directory_ready(root)) self.assertTrue(runtime_security.private_file_ready(child)) def test_windows_dependency_normalization_removes_launcher_timestamp_variance(self): normalized = [] with tempfile.TemporaryDirectory() as root: for index, second in enumerate((0, 2)): dependencies = os.path.join(root, str(index)) bin_root = os.path.join(dependencies, 'bin') dist_info = os.path.join(dependencies, 'idna-1.0.dist-info') os.makedirs(bin_root) os.makedirs(dist_info) launcher = os.path.join(bin_root, 'idna.exe') with open(launcher, 'wb') as handle: handle.write(b'MZ' + (b'\x00' * 62)) with zipfile.ZipFile(launcher, 'a') as archive: item = zipfile.ZipInfo('__main__.py', (2026, 9, 22, 12, 0, second)) archive.writestr(item, b'print("idna")\n') digest = base64.urlsafe_b64encode( hashlib.sha256(open(launcher, 'rb').read()).digest() ).decode('ascii').rstrip('=') record = os.path.join(dist_info, 'RECORD') with open(record, 'w', encoding='utf-8', newline='') as handle: csv.writer(handle, lineterminator='\r\n').writerows(( ('../../bin/idna.exe', 'sha256=' + digest, str(os.path.getsize(launcher))), ('idna-1.0.dist-info/RECORD', '', ''), )) worker_package_builder._normalize_windows_dependency_artifacts(dependencies) with zipfile.ZipFile(launcher) as archive: self.assertEqual(archive.read('__main__.py'), b'print("idna")\n') normalized.append(( open(launcher, 'rb').read(), open(record, 'rb').read(), )) self.assertEqual(normalized[0], normalized[1]) def _assembled_manifest(self, root): app_root = os.path.join(root, 'app') os.makedirs(app_root, exist_ok=True) for index, name in enumerate(REMOTE_WORKER_CODE_AUTHORITY_FILES): path = os.path.join(app_root, *name.split('/')) os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, 'wb') as handle: handle.write(f'authority-{index}'.encode('ascii')) dependency = os.path.join(app_root, 'dependencies', 'fixture_dependency.py') os.makedirs(os.path.dirname(dependency), exist_ok=True) with open(dependency, 'wb') as handle: handle.write(b'FIXTURE = True\n') paths = { 'trufflehog': os.path.join('bin', 'trufflehog.exe'), 'git': os.path.join('runtime', 'git', 'cmd', 'git.exe'), 'policy': os.path.join('app', 'detectors.yaml'), } for label, relative in paths.items(): path = os.path.join(root, relative) os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, 'wb') as handle: handle.write(label.encode('ascii')) python_root = None if worker_package.local_platform_tag().startswith('windows-'): python_root = os.path.join('runtime', 'python') os.makedirs(os.path.join(root, python_root), exist_ok=True) with open(os.path.join(root, python_root, 'python.exe'), 'wb') as handle: handle.write(b'python') return worker_package.build_worker_package_manifest( root, trufflehog_path=paths['trufflehog'].replace(os.sep, '/'), git_path=paths['git'].replace(os.sep, '/'), detector_policy_path=paths['policy'].replace(os.sep, '/'), git_root='runtime/git', python_root=(python_root or '').replace(os.sep, '/') or None, capabilities=[ { 'source': 'gitlab', 'platform': 'gitlab', 'planning_kind': 'exact_git_v1', }, { 'source': 'github', 'platform': 'github', 'planning_kind': 'exact_git_v1', }, ], ) def test_manifest_is_path_neutral_canonical_and_build_identity_is_stable(self): manifest = package_manifest('linux-aarch64') normalized = worker_package.normalize_worker_package_manifest(manifest) self.assertEqual( normalized['capabilities'], [ { 'source': 'github', 'platform': 'github', 'planning_kind': 'exact_git_v1', }, { 'source': 'gitlab', 'platform': 'gitlab', 'planning_kind': 'exact_git_v1', }, ], ) self.assertEqual( worker_package.worker_package_manifest_sha256(manifest), worker_package.worker_package_manifest_sha256(normalized), ) build = worker_package.worker_package_build_compatibility(manifest) self.assertEqual(build['platform_tag'], 'linux-aarch64') self.assertEqual(build['detector_policy_sha256'], '4' * 64) def test_manifest_rejects_native_or_escaping_paths(self): manifest = package_manifest() manifest['assets']['detector_policy']['path'] = '../detectors.yaml' with self.assertRaises(worker_package.WorkerPackageError): worker_package.normalize_worker_package_manifest(manifest) manifest = package_manifest() manifest['executables'][GIT_MANIFEST_NAME]['path'] = r'runtime\git.exe' with self.assertRaises(worker_package.WorkerPackageError): worker_package.normalize_worker_package_manifest(manifest) def test_manifest_requires_every_worker_authority(self): self.assertIn('worker_contracts.py', REMOTE_WORKER_CODE_AUTHORITY_FILES) self.assertIn('worker_assignment_runner.py', REMOTE_WORKER_CODE_AUTHORITY_FILES) self.assertIn('worker_cli.py', REMOTE_WORKER_CODE_AUTHORITY_FILES) self.assertIn('worker_local_state.py', REMOTE_WORKER_CODE_AUTHORITY_FILES) self.assertIn('worker_supervisor.py', REMOTE_WORKER_CODE_AUTHORITY_FILES) manifest = package_manifest() manifest['files'].pop(REMOTE_WORKER_CODE_AUTHORITY_FILES[0]) with self.assertRaisesRegex(worker_package.WorkerPackageError, 'incomplete'): worker_package.normalize_worker_package_manifest(manifest) def test_manifest_forbids_server_and_detailed_keycheck_code(self): self.assertIn('keycheck_candidates.py', REMOTE_WORKER_CODE_AUTHORITY_FILES) self.assertNotIn('keycheck_runner.py', REMOTE_WORKER_CODE_AUTHORITY_FILES) self.assertFalse(any( name.startswith('keycheckers/') for name in REMOTE_WORKER_CODE_AUTHORITY_FILES )) for name in ( 'dashboard.py', 'keycheck_runner.py', 'keycheckers/openai/Keycheck.py', 'dependencies/keycheck_runner.py', ): with self.subTest(name=name): manifest = package_manifest() manifest['files'][name] = { 'path': f'app/{name}', 'sha256': '5' * 64, } with self.assertRaises(worker_package.WorkerPackageError): worker_package.normalize_worker_package_manifest(manifest) def test_manifest_rejects_unknown_or_inconsistent_capabilities(self): manifest = package_manifest() manifest['capabilities'][0]['source'] = 'keychecks' with self.assertRaisesRegex(worker_package.WorkerPackageError, 'capability'): worker_package.normalize_worker_package_manifest(manifest) manifest = package_manifest() manifest['capabilities'][0]['platform'] = 'docker' with self.assertRaisesRegex(worker_package.WorkerPackageError, 'capability'): worker_package.normalize_worker_package_manifest(manifest) manifest = package_manifest() manifest['capabilities'].append(dict(manifest['capabilities'][0])) with self.assertRaisesRegex(worker_package.WorkerPackageError, 'duplicated'): worker_package.normalize_worker_package_manifest(manifest) def test_manifest_rejects_legacy_shape_and_noninteger_versions(self): manifest = package_manifest() manifest['sources'] = ['github'] with self.assertRaisesRegex(worker_package.WorkerPackageError, 'shape'): worker_package.normalize_worker_package_manifest(manifest) for name, value in ( ('schema', '3'), ('protocol_version', 2.0), ('bundle_format_version', True), ): with self.subTest(name=name): manifest = package_manifest() manifest[name] = value with self.assertRaises(worker_package.WorkerPackageError): worker_package.normalize_worker_package_manifest(manifest) def test_verify_maps_only_manifested_paths_and_requires_private_authority(self): manifest = package_manifest() with tempfile.TemporaryDirectory() as root: manifest_path = os.path.join(root, 'worker-package.json') with open(manifest_path, 'w', encoding='utf-8') as handle: json.dump(manifest, handle) with mock.patch.object( worker_package, 'verify_code_manifest', side_effect=lambda value, **_kwargs: value, ) as verify, mock.patch.object( worker_package, '_verify_runtime_tree', return_value=os.path.join(root, 'runtime'), ): result = worker_package.verify_worker_package(manifest_path) verify.assert_called_once() self.assertTrue(verify.call_args.kwargs['require_private_acl']) self.assertEqual(result['build_compatibility']['platform_tag'], manifest['platform_tag']) self.assertTrue(result['trufflehog_path'].endswith(os.path.join('bin', 'trufflehog.exe'))) self.assertTrue(result['git_path'].endswith(os.path.join('runtime', 'git', 'cmd', 'git.exe'))) self.assertTrue(result['detector_policy_path'].endswith(os.path.join('app', 'detectors.yaml'))) def test_verify_rejects_foreign_platform_before_authority_check(self): local = worker_package.local_platform_tag() foreign = 'linux-aarch64' if local != 'linux-aarch64' else 'windows-x86_64' manifest = package_manifest(foreign) with tempfile.TemporaryDirectory() as root: manifest_path = os.path.join(root, 'worker-package.json') with open(manifest_path, 'w', encoding='utf-8') as handle: json.dump(manifest, handle) with mock.patch.object(worker_package, 'verify_code_manifest') as verify: with self.assertRaisesRegex(worker_package.WorkerPackageError, 'local platform'): worker_package.verify_worker_package(manifest_path) verify.assert_not_called() def test_builder_hashes_preassembled_package_and_writes_canonical_manifest(self): with tempfile.TemporaryDirectory() as root: manifest = self._assembled_manifest(root) self.assertEqual( [item['source'] for item in manifest['capabilities']], ['github', 'gitlab'], ) self.assertNotEqual( manifest['files'][REMOTE_WORKER_CODE_AUTHORITY_FILES[0]]['sha256'], '1' * 64, ) self.assertIn('remote_worker_client.py', manifest['files']) self.assertIn('remote_worker_bootstrap.py', manifest['files']) self.assertIn('docker_depth_experiment.py', manifest['files']) self.assertIn('query_policy.py', manifest['files']) self.assertIn('worker_contracts.py', manifest['files']) self.assertIn('dependencies/fixture_dependency.py', manifest['files']) manifest_path = worker_package.write_worker_package_manifest( os.path.join(root, 'worker-package.json'), manifest, ) loaded = worker_package.load_worker_package_manifest(manifest_path) self.assertEqual(loaded, manifest) self.assertEqual( worker_package.worker_package_manifest_sha256(loaded), worker_package.worker_package_manifest_sha256(manifest), ) def test_manifest_bytes_loader_is_bounded_and_uses_existing_normalization(self): manifest = package_manifest() payload = json.dumps(manifest).encode('utf-8') self.assertEqual( worker_package.load_worker_package_manifest_bytes(payload), worker_package.normalize_worker_package_manifest(manifest), ) for invalid in ( 'not-bytes', b'\xff', b'{', b'x' * (worker_package.MAX_WORKER_PACKAGE_MANIFEST_BYTES + 1), ): with self.subTest(invalid=type(invalid).__name__): with self.assertRaises(worker_package.WorkerPackageError): worker_package.load_worker_package_manifest_bytes(invalid) def test_manifest_loader_rejects_hardlinked_path(self): with tempfile.TemporaryDirectory() as root: manifest_path = os.path.join(root, 'worker-package.json') linked_path = os.path.join(root, 'linked-worker-package.json') with open(manifest_path, 'w', encoding='utf-8') as handle: json.dump(package_manifest(), handle) try: os.link(manifest_path, linked_path) except OSError as exc: self.skipTest(f'hardlinks unavailable: {exc}') with self.assertRaisesRegex(worker_package.WorkerPackageError, 'regular file'): worker_package.load_worker_package_manifest(linked_path) def test_builder_fails_closed_when_authority_file_is_missing(self): with tempfile.TemporaryDirectory() as root: app_root = os.path.join(root, 'app') os.makedirs(app_root) with self.assertRaises((FileNotFoundError, worker_package.WorkerPackageError)): worker_package.build_worker_package_manifest( root, trufflehog_path='bin/trufflehog', git_path='bin/git', detector_policy_path='app/detectors.yaml', git_root='runtime/git', capabilities=[{ 'source': 'github', 'platform': 'github', 'planning_kind': 'exact_git_v1', }], ) def test_package_assembler_copies_only_worker_authority_and_runtime_trees(self): with tempfile.TemporaryDirectory() as root: dependencies = os.path.join(root, 'dependencies') os.makedirs(dependencies) with open(os.path.join(dependencies, 'fixture_dependency.py'), 'wb') as handle: handle.write(b'FIXTURE = True\n') git_root = os.path.join(root, 'git-source') git_executable = 'cmd/git.exe' if os.name == 'nt' else 'bin/git' git_path = os.path.join(git_root, *git_executable.split('/')) os.makedirs(os.path.dirname(git_path)) shutil.copyfile(sys.executable, git_path) with open(os.path.join(git_root, 'helper.fixture'), 'wb') as handle: handle.write(b'complete-runtime-tree') python_root = None if os.name == 'nt': python_root = os.path.join(root, 'python-source') os.makedirs(python_root) shutil.copyfile(sys.executable, os.path.join(python_root, 'python.exe')) output = os.path.join(root, 'package') manifest = worker_package_builder.assemble_worker_package( output, source_app=APP_DIR, dependencies_root=dependencies, detector_policy_source=os.path.join(APP_DIR, 'trufflehog-custom-detectors.yaml'), trufflehog_source=sys.executable, git_source_root=git_root, git_executable=git_executable, python_source_root=python_root, operator_readme_source=os.path.join( os.path.dirname(APP_DIR), 'docs', 'remote-worker-quickstart-ru.md', ), operator_cheatsheet_sources=[ os.path.join( os.path.dirname(APP_DIR), 'docs', f'remote-worker-cheatsheet-{name}-ru.md', ) for name in ('windows', 'linux', 'docker') ], platform_tag=worker_package.local_platform_tag(), ) expected = set(REMOTE_WORKER_CODE_AUTHORITY_FILES) | { 'dependencies/fixture_dependency.py', 'trufflehog-custom-detectors.yaml', } self.assertEqual(set(manifest['files']), expected) self.assertEqual( manifest['capabilities'], [ { 'source': 'dockerhub', 'platform': 'docker', 'planning_kind': 'docker_direct_v1', }, { 'source': 'gitlab', 'platform': 'gitlab', 'planning_kind': 'exact_git_v1', }, { 'source': 'huggingface', 'platform': 'huggingface', 'planning_kind': 'huggingface_space_v1', }, ], ) self.assertNotIn('worker_package_builder.py', manifest['files']) self.assertFalse(os.path.exists(os.path.join(output, 'app', 'keycheck_runner.py'))) self.assertEqual( set(manifest['runtime_trees']), {'git', 'python'} if os.name == 'nt' else {'git'}, ) self.assertEqual(manifest['runtime_trees']['git']['file_count'], 2) self.assertTrue(os.path.isfile(os.path.join(output, 'worker-package.json'))) self.assertIn( 'установка и работа', open(os.path.join(output, 'README_RU.md'), encoding='utf-8').read(), ) for name in ('windows', 'linux', 'docker'): self.assertTrue(os.path.isfile(os.path.join( output, f'remote-worker-cheatsheet-{name}-ru.md', ))) if os.name == 'nt': self.assertTrue(os.path.isfile(os.path.join(output, 'truf-worker.cmd'))) self.assertTrue(os.path.isfile(os.path.join(output, 'run-worker.cmd'))) self.assertTrue(os.path.isfile(os.path.join(output, 'prepare-worker.ps1'))) self.assertIn( 'remote_worker_bootstrap.py" -- %*', open(os.path.join(output, 'truf-worker.cmd'), encoding='ascii').read(), ) self.assertIn( 'remote_worker_bootstrap.py" -- run %*', open(os.path.join(output, 'run-worker.cmd'), encoding='ascii').read(), ) else: launcher = os.path.join(output, 'truf-worker') run_launcher = os.path.join(output, 'run-worker') prepare = os.path.join(output, 'prepare-worker.sh') self.assertTrue(os.access(launcher, os.X_OK)) self.assertTrue(os.access(run_launcher, os.X_OK)) self.assertTrue(os.access(prepare, os.X_OK)) self.assertIn( 'remote_worker_bootstrap.py" -- "$@"', open(launcher, encoding='ascii').read(), ) self.assertIn( 'remote_worker_bootstrap.py" -- run "$@"', open(run_launcher, encoding='ascii').read(), ) subprocess.run(['sh', '-n', launcher], check=True) subprocess.run(['sh', '-n', run_launcher], check=True) subprocess.run(['sh', '-n', prepare], check=True) def test_real_worker_authority_verification_detects_tampering(self): with tempfile.TemporaryDirectory() as root: manifest = self._assembled_manifest(root) manifest_path = worker_package.write_worker_package_manifest( os.path.join(root, 'worker-package.json'), manifest, ) with mock.patch( 'lifecycle_authority.private_file_ready', return_value=True, ), mock.patch( 'lifecycle_authority.require_trusted_native_executable', return_value=None, ), mock.patch.object( worker_package, '_runtime_tree_permissions_ready', return_value=True, ): verified = worker_package.verify_worker_package(manifest_path) self.assertEqual( set(verified['code_manifest']['files']), set(manifest['files']), ) with open( os.path.join(root, 'app', 'remote_worker_client.py'), 'ab', ) as handle: handle.write(b'tampered') with self.assertRaisesRegex(Exception, 'drifted'): worker_package.verify_worker_package(manifest_path) def test_isolated_bootstrap_verifies_application_before_entrypoint(self): with tempfile.TemporaryDirectory() as root: app_root = os.path.join(root, 'app') os.makedirs(os.path.join(app_root, 'dependencies')) bootstrap = os.path.join(app_root, 'remote_worker_bootstrap.py') client = os.path.join(app_root, 'worker_cli.py') shutil.copyfile( os.path.join(APP_DIR, 'remote_worker_bootstrap.py'), bootstrap, ) with open(client, 'w', encoding='utf-8') as handle: handle.write("import sys\nprint('bootstrap-ok:' + sys.argv[1])\n") files = {} for name in ('remote_worker_bootstrap.py', 'worker_cli.py'): path = os.path.join(app_root, name) files[name] = { 'path': f'app/{name}', 'sha256': worker_package.sha256_file(path), } manifest = { 'schema': 3, 'protocol_version': 2, 'app_root': 'app', 'files': files, } with open(os.path.join(root, 'worker-package.json'), 'w', encoding='utf-8') as handle: json.dump(manifest, handle) command = [sys.executable, '-I', '-S', '-B', bootstrap, '--', 'fixture'] completed = subprocess.run( command, capture_output=True, text=True, timeout=30, check=False, ) self.assertEqual(completed.returncode, 0, completed.stderr) self.assertEqual(completed.stdout.strip(), 'bootstrap-ok:fixture') for field, value in (('schema', 2), ('protocol_version', 1)): incompatible = dict(manifest) incompatible[field] = value with open( os.path.join(root, 'worker-package.json'), 'w', encoding='utf-8', ) as handle: json.dump(incompatible, handle) rejected = subprocess.run( command, capture_output=True, text=True, timeout=30, check=False, ) self.assertNotEqual(rejected.returncode, 0) self.assertNotIn('bootstrap-ok', rejected.stdout) with open(os.path.join(root, 'worker-package.json'), 'w', encoding='utf-8') as handle: json.dump(manifest, handle) with open(client, 'a', encoding='utf-8') as handle: handle.write("print('must-not-run')\n") rejected = subprocess.run( command, capture_output=True, text=True, timeout=30, check=False, ) self.assertNotEqual(rejected.returncode, 0) self.assertNotIn('must-not-run', rejected.stdout) def test_worker_docker_entrypoint_exposes_cli_and_defaults_to_foreground_run(self): dockerfile = open( os.path.join(os.path.dirname(APP_DIR), 'Dockerfile'), encoding='utf-8', ).read() self.assertIn( 'ENTRYPOINT ["/usr/bin/tini", "--", "/usr/local/bin/python3", "-u", ' '"-I", "-S", "-B", "/opt/truf-worker/app/remote_worker_bootstrap.py", "--"]', dockerfile, ) worker_section = dockerfile.split('FROM worker-native-dependencies AS worker', 1)[1] self.assertIn('CMD ["run"]', worker_section.split('FROM python-base AS native-dependencies', 1)[0]) def test_packaged_worker_verifier_allows_signal_drain_and_checks_final_receipt(self): verifier = open( os.path.join(os.path.dirname(APP_DIR), 'docker', 'verify_packaged_workers.py'), encoding='utf-8', ).read() self.assertIn("'--stop-timeout', '45'", verifier) self.assertGreaterEqual(verifier.count("'container', 'stop', '--time', '45'"), 3) self.assertNotIn("'container', 'stop', '--time', '15'", verifier) self.assertIn("'linux_clean_shutdown_receipt'", verifier) if __name__ == '__main__': unittest.main()