"""Explicit offline regressions for the Docker test target. Run as UID 10001 with read-only /opt/truf/app, writable /tmp tmpfs, and Docker --network none: python -I -S -B /opt/truf/tests/container_unit.py. No /data mount, runtime configuration, credentials, PostgreSQL server, external Git source, or provider entrypoint is needed. SQLite fixtures stay in /tmp; native OwnedProcess children and loopback control sockets are intentional. --list and --check-selection inspect source with the stdlib only, including on the host. -k only narrows the reviewed selection; pytest arguments and additional paths/plugins are not accepted. New test files are never discovered. """ import sys sys.dont_write_bytecode = True import argparse import ast from contextlib import ExitStack import os from pathlib import Path import runpy import stat import tempfile from unittest import mock TESTS = Path(__file__).resolve().parent ROOT = TESTS.parent APP = ROOT / 'app' SELECTION = { 'test_db_backend_safety.py': ( 'PostgresTimeoutTests', ), 'test_docker_foundation.py': ( 'PortablePathTests', 'DockerStagingTests::test_native_checkout_default_is_derived_from_this_copy', 'DockerStagingTests::test_python_entrypoints_refuse_before_application_imports', 'DockerStagingTests::test_all_copied_powershell_tools_have_static_refusals', 'DockerStagingTests::test_docker_context_exceptions_are_explicit_source_files_only', ), # Linux metadata and runtime I/O are mocked; filesystem fixtures stay in tmp_path. 'test_container_runtime.py': ( 'test_private_path_accepts_exact_owner_only_mode', 'test_private_path_rejects_nonabsolute_or_parent_traversal_before_stat', 'test_private_path_rejects_symlink_at_every_component', 'test_private_path_fails_closed_on_missing_or_uninspectable_components', 'test_private_path_rejects_wrong_owner_group_type_or_exact_mode', 'test_container_requires_private_image_native_data_and_dedicated_tmpfs', 'test_container_refuses_before_mount_inventory_when_identity_gate_fails', 'test_root_is_permitted_only_for_explicit_provisioning', 'test_container_rejects_nonprivate_required_paths', 'test_container_rejects_missing_shared_or_host_style_storage', 'test_new_private_file_is_exclusive_nofollow_and_durable', 'test_new_private_file_never_overwrites_existing_file', 'test_fresh_provision_generates_private_password_once_without_initializing_pg', 'test_provision_refuses_nonempty_or_partial_volume_without_repair', 'test_provision_lock_conflict_never_creates_layout_or_credentials', 'test_provision_does_not_regenerate_incomplete_marked_volume', 'test_provision_adds_only_managed_files_to_legacy_marked_volume', 'test_provision_refuses_wrong_existing_managed_files_path', 'test_bad_provision_marker_prevents_environment_or_application_imports', 'test_prepare_environment_scrubs_dsn_and_runtime_overrides_before_imports', 'test_prepare_rejects_invalid_generated_password_before_imports', 'test_prepare_refuses_config_escaping_fixed_storage_contract', 'test_only_default_image_config_or_private_data_config_is_accepted', 'test_bootstrap_target_is_the_real_runtime_safety_migration_module', 'test_initialize_publishes_marker_only_after_migration_and_confirmed_stop', 'test_failed_initialization_always_stops_maintenance_and_never_marks_success', 'test_shutdown_during_initialization_never_publishes_early_marker', 'test_partial_initialization_is_not_adopted_or_repaired', 'test_initialized_marker_must_match_bound_cluster_identity', 'test_health_does_not_import_database_or_control_before_valid_marker', 'test_health_requires_active_supervisor_and_explicit_postgres_ready', 'test_health_rejects_missing_or_unready_pipeline_processes_before_db', 'test_health_requires_each_enabled_pipeline_worker', 'test_health_allows_explicitly_disabled_janitor', 'test_health_requires_worker_api_only_when_explicitly_enabled', 'test_strict_health_requires_enabled_worker_api_before_database', 'test_strict_health_probes_exact_unauthorized_worker_endpoint', 'test_strict_worker_probe_fails_closed_before_database', 'test_health_rejects_failed_or_uncertain_managed_source', 'test_health_checks_readonly_schema_cutover_leases_identity_then_storage', 'test_health_fails_closed_and_closes_database_on_each_database_gate', 'test_health_requires_writable_nonfull_persistent_storage', 'test_secret_import_refuses_active_runtime_before_reading_stdin', 'test_secret_import_cluster_lock_conflict_precedes_stopped_check_and_stdin', 'test_secret_import_is_locked_atomic_private_and_never_prints_credentials', 'test_invalid_secret_input_is_bounded_and_does_not_leak_or_replace', 'test_secret_import_validation_error_clears_candidate_traceback_locals', 'test_secret_import_rejects_config_hash_drift_before_temporary_write', 'test_secret_import_rechecks_config_hash_before_replacement', 'test_write_new_clears_secret_payload_from_traceback_locals', 'test_initialize_rejects_config_hash_drift_before_mutation', 'test_secret_import_error_removes_temporary_file_and_preserves_old_credentials', 'test_main_container_gate_precedes_every_action', 'test_main_provision_does_not_enter_nonroot_environment_or_database', 'test_main_import_secrets_uses_preliminary_environment_validation', 'test_main_initialize_binds_validated_config_hash', 'test_main_run_rechecks_config_hash_before_exec', 'test_main_snapshot_requires_approved_digest_and_default_config', 'test_main_snapshot_dispatch_preserves_module_signal_state_without_initialize', 'test_manifest_digest_is_rejected_for_other_actions', 'test_worker_api_requirement_flag_is_restricted_to_health', 'test_strict_status_dispatches_worker_api_requirement', 'test_status_stdout_contains_readiness_but_no_credentials', ), 'test_runtime_document.py': ( 'RuntimeDocumentTests', ), 'test_managed_files.py': ( 'ManagedFileConfigurationTests', 'ManagedFileTraversalTests', 'ManagedFileTraversalLinuxTests', ), 'test_operations_schema.py': ('OperationsSchemaTests',), 'test_operations_control.py': ('OperationsControlTests',), 'test_host_agent_protocol.py': ( 'HostAgentProtocolTests', 'HostAgentClientTests', 'HostAgentServerTests', ), 'test_host_agent_linux.py': ( 'HostAgentLinuxTests', ), 'test_host_agent_apply.py': ( 'HostAgentApplyTests', ), 'test_host_agent_deploy.py': ( 'HostAgentDeployTests', 'HostAgentInstallerTests', ), 'test_host_agent_lifecycle.py': ( 'HostAgentLifecycleTests', ), 'test_host_agent_reconcile.py': ( 'HostAgentReconcileTests', ), 'test_host_agent_runtime.py': ( 'HostAgentRuntimeTests', ), 'test_host_agent_state.py': ( 'HostAgentStateTests', ), 'test_runtime_document_io.py': ( 'RuntimeDocumentIOTests', ), 'test_operations_service.py': ( 'OperationsServiceTests', ), 'test_owned_process.py': ( 'OwnedProcessTests', 'WindowsOwnedProcessTests', 'StaticProcessSafetyTests', ), 'test_owned_process_linux.py': ( 'LinuxIdentityTests', 'StartupHandshakeTests', 'LinuxContainmentTests', 'LinuxOwnedProcessIntegrationTests', ), 'test_owned_process_boundary.py': ( 'OwnedProcessHostBoundaryTests', 'CredentialBoundaryTests', ), 'test_runtime_bootstrap_authority.py': ( 'RuntimeBootstrapAuthorityTests::test_authenticated_docker_shadow_dispatches_exact_entrypoint', 'RuntimeBootstrapAuthorityTests::test_provider_bootstrap_consumes_one_required_separator_before_provider_parse', 'RuntimeBootstrapAuthorityTests::test_provider_bootstrap_rejects_missing_and_duplicate_separators', 'RuntimeBootstrapAuthorityTests::test_authenticated_entrypoint_exception_is_classified_as_runtime_failure', 'RuntimeBootstrapAuthorityTests::test_command_builders_use_exact_isolation_flag_order', 'RuntimeBootstrapAuthorityTests::test_direct_mutating_supervisor_fails_before_config_env_locks_or_children', ), 'test_supervisor_foreground_shutdown.py': ( 'test_signal_callback_is_only_an_idempotent_flag_assignment', 'test_receipt_parity_and_finalization_order', 'test_windows_does_not_register_signals', 'test_mutating_launch_gates_precede_config_locks_children_and_signals', 'test_non_owning_dispatch_does_not_register_signals', 'test_pending_term_cannot_admit_startup', 'test_background_activation_wait_consumes_term', 'test_admission_rejects_pending_term_before_checkpoint', 'test_control_activation_cannot_reopen_pending_shutdown', 'test_background_activation_callback_failure_is_sticky', 'test_term_between_pipeline_starts_admits_no_more_children', 'test_checkpoints_exit_without_polling_or_starting', 'test_term_between_source_polls_rejects_next_poll', 'test_foreground_waits_consume_term_without_input_or_next_refresh', 'test_metadata_publication_failure_leaves_closed_memory_gates', 'test_repeated_term_checkpoint_does_not_republish_or_rearm_retry', 'test_repeated_shutdown_does_not_discard_confirmed_stop_proof', 'test_initial_cleanup_interruption_retains_authority_and_failure', 'test_interrupted_cleanup_lock_acquisition_retains_authority', 'test_partially_published_activation_uses_full_cleanup', 'test_coordinated_shutdown_interruption_closes_gate_before_propagating', 'test_failed_hold_survives_interruption_and_logging_failure', 'test_failed_hold_does_not_release_with_unconfirmed_children', 'test_receipt_write_failure_is_nonzero', 'test_late_cleanup_failures_are_in_receipt', 'test_failure_latched_by_draining_control_worker_is_in_receipt', 'test_terminal_foreground_failure_is_sticky_without_changing_restart_policy', 'test_foreground_source_failure_survives_cleanup_status_reset', 'test_foreground_missing_autostart_is_nonzero', 'test_system_exit_status_is_preserved', 'test_interrupt_message_failure_cannot_publish_success', 'test_authority_drift_failure_remains_sticky', 'test_preactivation_close_failure_is_sticky', 'test_noninteractive_keeps_existing_completion_policy', 'test_stopper_requires_posix_receipt_and_preserves_windows_crosscheck', 'test_unconfirmed_child_defers_all_postgres_actions', 'test_children_precede_postgres_and_stop_close_share_budget', 'test_postgres_wait_consumes_close_budget', ), 'test_supervisor_startup_rollback.py': ( 'test_unconfirmed_launch_rollback_retains_owner_until_stop_retry', 'test_retained_rollback_refuses_start_schedule_and_poll', 'test_confirmed_launch_rollback_clears_owner_and_allows_explicit_retry', 'test_constructor_failure_without_returned_owner_still_closes_log', 'test_rollback_interruption_propagates_without_losing_owner', 'test_log_cleanup_error_does_not_undo_confirmed_exit', 'test_log_pump_with_failed_thread_start_can_close_on_rollback_retry', 'test_rollback_pending_closes_admission_before_failed_hold_publication', 'test_owned_child_uncertain_defers_postgres_until_retained_retry_confirms_exit', 'test_main_retains_locks_and_control_until_startup_rollback_is_confirmed', 'test_locked_stale_metadata_recovery_uses_exact_identity_not_pid_or_timestamp', 'test_foreground_reconciliation_uses_existing_exact_identity_api', 'test_live_or_uncertain_owner_metadata_is_never_replaced', 'test_metadata_bound_to_another_instance_path_is_not_removed', 'test_reconciliation_error_or_interruption_preserves_metadata_for_retry', 'test_changed_instance_during_matching_removal_fails_closed', ), 'test_supervisor_managed_postgres_gate.py': ('SupervisorManagedPostgresGateTests',), 'test_observer_only_coordinated_shutdown.py': ('ObserverOnlyCoordinatedShutdownTests',), 'test_discovery_producer_supervisor.py': ('DiscoveryProducerSupervisorTests',), 'test_distributed_core_profile.py': ( 'test_default_distributed_core_profile_is_exact', 'test_keychecks_are_independent_from_discovery_profile', 'test_core_wrapper_selects_only_discovery_producers', ), 'test_discovery_only_cycle.py': ('DiscoveryOnlyCycleTests',), 'test_discovery_request_budgets.py': ( 'test_each_discovery_page_has_one_shared_attempt_deadline', 'test_discovery_proxy_uses_source_read_timeout_and_at_most_three_attempts', 'test_late_discovery_success_fails_closed_without_advancing_page', ), 'test_supervisor_safety.py': ( 'AuthenticatedControlTests', 'DependencyFailureTests', 'DashboardManagerTests', ), 'test_postgres_runtime.py': ( 'PostgresRuntimePathTests', 'PostgresControllerTests', 'PostgresBackendTests', 'PostgresEntrypointTests', ), 'test_container_security.py': ( 'NativeExecutablePolicyTests', 'WindowsNativePolicyTests', 'ContainerLifecyclePolicyTests', 'LinuxFilesystemSecurityTests', ), 'test_runtime_security.py': ('RuntimeSecurityTests',), 'test_postgres_empty_initialization.py': ( 'NativePostgresPathTests', 'InitializeEmptyTests', 'InitializationEntrypointTests::test_cli_dispatches_to_lock_owning_initializer_after_preflight', 'InitializationEntrypointTests::test_container_refusal_and_full_layout_gate_precede_application_imports', 'InitializationEntrypointTests::test_bounded_maintenance_stop_requires_proof_and_never_closes_a_supplied_backend', 'InitializationEntrypointTests::test_bootstrap_cannot_replace_an_existing_identity', 'MaintenanceEntrypointTests', ), 'test_process_identity_linux.py': ('LinuxProcessIdentityTests',), 'test_container_migration_paths.py': ( 'LegacySpoolPathTests', 'ImmutableNativeHardeningTests', 'LegacyCutoverChecksTests', 'NormalMigrationContractTests', ), 'test_container_provider_portability.py': ('AskpassTests', 'ProviderDeadlineTests'), 'test_container_e2e_helpers.py': ('ContainerE2EHelperTests', 'ContainerE2EFreshVolumeTests'), 'test_container_import.py': ('ContainerImportTests',), 'test_container_import_config.py': ('ContainerImportConfigTests',), 'test_container_projection_recovery.py': ('ProjectionRecoveryTests',), 'test_result_bundle_v2.py': ('ResultBundleV2Tests',), 'test_pipeline_cutover_invariants.py': ('PipelineCutoverInvariantTests',), 'test_custom_provider_detector_compatibility.py': ( 'CustomProviderDetectorPolicyTests', 'CustomProviderDetectorCLITests', ), 'test_scan_execution.py': ('ScanExecutionTests',), 'test_synthetic_llm_pipeline.py': ('SyntheticLLMPipelineTests',), 'test_worker_api.py': ( 'WorkerServiceTests', 'WorkerAPIRouteTests', 'WorkerSlotRecoveryTests', ), 'test_worker_api_runtime.py': ( 'ConfiguredWorkerServiceTests', 'WorkerSupervisorWiringTests', ), 'test_worker_assignment.py': ('RemoteGitAssignmentBuilderTests',), 'test_worker_assignment_runner.py': ( 'WorkerAssignmentRunnerProtocolTests', 'WorkerSlotRunnerTests', ), 'test_worker_runner_handoff_linux.py': ('LinuxWorkerRunnerHandoffTests',), 'test_worker_contracts.py': ('WorkerContractTests',), 'test_worker_cli.py': ('WorkerCLITests',), 'test_worker_local_state.py': ('WorkerLocalStateTests',), 'test_worker_supervisor.py': ('WorkerSupervisorTests',), 'test_worker_package.py': ('WorkerPackageTests',), 'test_multisource_execution_snapshot.py': ('MultisourceExecutionSnapshotTests',), 'test_remote_direct_credentials.py': ( 'test_direct_child_environment_preserves_operator_provider_settings', 'test_direct_scanner_entries_reject_credentials_and_skip_docker_pool', 'test_anonymous_docker_bearer_never_reads_account_pool', 'test_direct_anonymous_docker_auth_failure_is_permanent', 'test_huggingface_discovery_errors_never_include_response_body', 'test_huggingface_missing_repository_is_permanent_and_not_retryable', 'test_direct_provider_access_failures_are_permanent', ), 'test_remote_worker_db.py': ('RemoteWorkerDBTests',), 'test_admin_api.py': ('AdminAPITests',), 'test_edge_deployment.py': ( 'test_updater_persists_canonical_state_and_expires_automatically', 'test_updater_reuses_persisted_applied_digest_without_reloading', 'test_updater_rejects_corrupt_state_before_commands_or_snippet_changes', 'test_updater_rejects_symlinked_snippet_when_supported', 'test_updater_rejects_non_ip_or_unsafe_addresses_without_commands', 'test_updater_renders_deterministically_and_bounds_matcher_lines', 'test_updater_rolls_back_state_and_snippet_when_reload_fails', 'test_updater_rolls_back_without_reload_when_validation_fails', 'test_fail2ban_filter_counts_only_redacted_supplied_bad_credentials', 'test_fail2ban_jail_is_persistent_bounded_and_admin_only', 'test_caddy_routes_and_failure_log_are_closed_and_redacted', 'test_edge_compose_only_opts_runtime_namespace_into_https_publication', 'test_edge_image_and_startup_require_pinned_caddy_hash_and_random_prefix', 'test_edge_e2e_image_inputs_are_exact_and_host_orchestrator_stays_host_only', 'test_production_server_omits_trufflehog_but_worker_and_test_retain_it', 'test_production_runbook_requires_fixed_host_agent_and_runtime_paths', 'test_real_caddy_e2e_crawls_all_admin_routes_and_detail', ), } HELPER_SOURCES = ('owned_process_helper.py', 'container_e2e.py', 'container_import_stop_e2e.py', 'container_projection_recovery_e2e.py', 'parity_helpers.py') def selected_nodes(): """Resolve only the named declarations, without importing any test or app.""" result = [] for filename, selectors in SELECTION.items(): path = TESTS / filename if path.resolve().parent != TESTS or not path.is_file(): raise SystemExit('container-unit: missing or nonlocal selected source: ' + filename) tree = ast.parse(path.read_text(encoding='utf-8'), filename=str(path)) compile(tree, str(path), 'exec') for selector in selectors: body = tree.body declaration = None for name in selector.split('::'): matches = [node for node in body if isinstance(node, (ast.ClassDef, ast.FunctionDef)) and node.name == name] if len(matches) != 1: raise SystemExit('container-unit: missing or ambiguous selection: ' + filename + '::' + selector) declaration = matches[0] body = declaration.body if isinstance(declaration, ast.ClassDef): members = [node.name for node in body if isinstance(node, ast.FunctionDef) and node.name.startswith('test_')] nodes = [filename + '::' + selector + '::' + member for member in members] elif isinstance(declaration, ast.FunctionDef) and declaration.name.startswith('test_'): nodes = [filename + '::' + selector] else: nodes = [] if not nodes: raise SystemExit('container-unit: empty test selection: ' + filename + '::' + selector) result.extend(nodes) if not result or len(result) != len(set(result)): raise SystemExit('container-unit: empty or duplicate selection') for filename in HELPER_SOURCES: path = TESTS / filename if path.resolve().parent != TESTS or not path.is_file(): raise SystemExit('container-unit: missing or nonlocal helper source: ' + filename) compile(ast.parse(path.read_text(encoding='utf-8'), filename=str(path)), str(path), 'exec') return result def main(argv=None): if not (sys.flags.isolated and sys.flags.no_site and sys.flags.dont_write_bytecode) or sys.flags.optimize: raise SystemExit('container-unit requires python -I -S -B without optimization') parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False, exit_on_error=False) modes = parser.add_mutually_exclusive_group() modes.add_argument('--list', action='store_true', help='print exact leaf selectors without importing tests') modes.add_argument('--check-selection', action='store_true', help='validate selected source AST and node IDs only') parser.add_argument('-k', dest='keyword', metavar='EXPRESSION', help='narrow the allowlist using pytest -k') try: args, extra = parser.parse_known_args(argv) except argparse.ArgumentError: parser.error('invalid runner arguments') if extra or (args.keyword is not None and (args.list or args.check_selection)): parser.error('only --list, --check-selection, or -k EXPRESSION is supported') if args.keyword is not None and (not args.keyword.strip() or len(args.keyword) > 512 or any(char in args.keyword for char in '\x00\r\n')): parser.error('-k requires a nonempty single-line expression of at most 512 characters') nodes = selected_nodes() if args.list: for node in nodes: print('tests/' + node) return 0 if args.check_selection: print(f'container-unit: AST OK; {len(SELECTION)} modules, {len(nodes)} test definitions, ' 'no runner skips (parametrizations expand in pytest)') return 0 if sys.platform != 'linux' or os.geteuid() != 10001 or TESTS != Path('/opt/truf/tests'): raise SystemExit('container-unit execution requires Linux UID 10001 at /opt/truf/tests') if not APP.is_dir() or os.access(APP, os.W_OK): raise SystemExit('container-unit requires readable, read-only /opt/truf/app') # A fresh allowlist also removes mixed-case/future credentials, not just known # TRUF_/SCANNER_/SCAN_/TRUFFLEHOG_/KEYCHECK_/PG*/DATABASE_URL/proxy names. os.environ.clear() os.environ.update({ 'PATH': '/usr/local/bin:/usr/bin:/bin', 'LANG': 'C.UTF-8', 'LC_ALL': 'C.UTF-8', 'PYTHONDONTWRITEBYTECODE': '1', 'PYTHONNOUSERSITE': '1', 'PYTHONPATH': '', 'PYTEST_DISABLE_PLUGIN_AUTOLOAD': '1', 'PYTEST_ADDOPTS': '', 'PYTEST_PLUGINS': '', }) os.umask(0o077) with tempfile.TemporaryDirectory(prefix='container-unit-', dir='/tmp') as temporary: private = Path(temporary).resolve() details = private.stat() if private.parent != Path('/tmp') or details.st_uid != 10001 or stat.S_IMODE(details.st_mode) != 0o700: raise SystemExit('container-unit temporary directory is not private beneath /tmp') for name in ('ALLTEMP', 'TEMP', 'TMP', 'TMPDIR', 'HOME'): os.environ[name] = str(private) for name, child in (('XDG_CONFIG_HOME', 'config'), ('XDG_CACHE_HOME', 'cache'), ('XDG_DATA_HOME', 'data')): os.environ[name] = str(private / child) tempfile.tempdir = None with tempfile.NamedTemporaryFile(prefix='probe-') as probe: actual = Path(probe.name).resolve() if (Path(tempfile.gettempdir()).resolve() != private or actual.parent != private or stat.S_IMODE(actual.stat().st_mode) != 0o600): raise SystemExit('container-unit tempfile escaped its private directory') bootstrap = runpy.run_path(str(APP / 'child_bootstrap.py')) bootstrap['_enable_dependency_paths']('supervisor') sys.path.insert(0, str(APP)) import pytest import psycopg import requests import paths # Only scanner's import-time defaults are relocated. Restore paths before # collection so portable-path and security checks exercise real policy. blocked_import = AssertionError('container-unit forbids import-time runtime I/O') with ExitStack() as imports: imports.enter_context(mock.patch.multiple(paths, APP_DIR=str(private / 'app'), CANONICAL_ROOT=str(private))) for target in ('builtins.open', 'os.open', 'os.mkdir', 'os.makedirs', 'sqlite3.connect', 'subprocess.Popen.__init__', 'threading.Thread.start', 'socket.socket.__init__', 'requests.Session.request', 'psycopg.connect'): imports.enter_context(mock.patch(target, side_effect=blocked_import)) import scanner # Filesystem/process mocks have ended: tests exercise real ACLs and locks. os.chdir(ROOT) sys.path.insert(1, str(TESTS)) def loopback_only(event, arguments): if event in ('socket.getaddrinfo', 'socket.gethostbyname', 'socket.gethostbyaddr'): host = arguments[0] elif event in ('socket.connect', 'socket.bind', 'socket.sendto', 'socket.getnameinfo'): address = arguments[0] if event == 'socket.getnameinfo' else arguments[-1] host = address[0] if isinstance(address, tuple) and address else None else: return if host not in ('127.0.0.1', '::1', 'localhost'): raise AssertionError('container-unit permits loopback sockets only') sys.addaudithook(loopback_only) class Scope: def pytest_collection_modifyitems(self, items): for item in items: node = item.nodeid.removeprefix('tests/').split('[', 1)[0] if node not in nodes: raise pytest.UsageError('container-unit collected a test outside its allowlist') options = [ '-c', '/dev/null', '--rootdir', str(ROOT), '--noconftest', '-p', 'no:cacheprovider', '-o', 'addopts=', '--basetemp', str(private / 'pytest'), '--tb=short', '--show-capture=no', '-ra', ] if args.keyword is not None: options.append('-k=' + args.keyword) options.extend(str(TESTS / node) for node in nodes) # Tests may replace these fences with their own scoped mocks, never real # PostgreSQL or provider transports. Native local processes stay real. with mock.patch.object(psycopg, 'connect', side_effect=AssertionError('external PostgreSQL is forbidden')), \ mock.patch.object(requests.Session, 'request', side_effect=AssertionError('provider HTTP is forbidden')): return int(pytest.main(options, plugins=[Scope()])) if __name__ == '__main__': try: code = main() except Exception as error: print('container-unit bootstrap failed (' + type(error).__name__ + '); details withheld', file=sys.stderr) code = 1 raise SystemExit(code)