import base64 import contextlib import hashlib import http.client import importlib.util import io import json from pathlib import Path import socket import ssl import tempfile import unittest from unittest.mock import MagicMock, patch import zipfile try: import paramiko except ImportError: paramiko = None ROOT = Path(__file__).resolve().parents[2] spec = importlib.util.spec_from_file_location('transfer', ROOT / 'scripts/deploy/transfer.py') deploy = importlib.util.module_from_spec(spec) spec.loader.exec_module(deploy) class FakeSSHException(Exception): pass class FakeAuthenticationException(FakeSSHException): pass class FakeBadHostKeyException(FakeSSHException): pass class MemoryTransfer: def __init__(self): self.files = {'.deploy/worker.php': b'installed', 'shared/.env': b'private', 'shared/storage/upload': b'keep'} self.writes = [] self.fail_package = False def mkdir(self, name, mode=0o700): pass def get(self, name): return self.files.get(name) def put(self, name, stream): if self.fail_package and name.endswith('.zip'): raise ConnectionError('simulated partial transfer') self.files[name] = stream.read() self.writes.append(name) def put_json(self, name, value): self.put(name, io.BytesIO(json.dumps(value).encode())) class TransferTests(unittest.TestCase): def setUp(self): self.progress = io.StringIO() self.enterContext(contextlib.redirect_stdout(self.progress)) self.directory = tempfile.TemporaryDirectory(prefix='conjugate-transfer-test-') self.addCleanup(self.directory.cleanup) self.root = Path(self.directory.name) self.env = dict(DEPLOY_ENVIRONMENT='staging', DEPLOY_OPERATION='deploy', DEPLOY_PATH='/apidev.appconjugate', OVH_TRANSFER_PROTOCOL='ftps', OVH_TRANSFER_HOST='ftp.example.test', OVH_TRANSFER_USER='account', OVH_TRANSFER_PASSWORD='private-password', OVH_PHP_VERSION='7.4') self.remote = MemoryTransfer() def package(self): stage = self.root / 'stage' stage.mkdir() for name in ['artisan', 'app/Controller.php', 'vendor/autoload.php', 'public/index.php', '.env', 'storage/secret.key', 'bootstrap/cache/config.php', 'public/storage/upload', 'app/.env', 'vendor/pkg/.git/config', 'public/.DS_Store', 'public/_deploy.php']: target = stage / name target.parent.mkdir(parents=True, exist_ok=True) target.write_text('fixture') package = self.root / 'release.zip' deploy.pack(stage, package, 'a' * 40, '7.4') self.env['DEPLOY_PACKAGE'] = str(package) return package def run_operation(self): output = io.StringIO() with contextlib.redirect_stdout(output): code = deploy.run(self.remote, deploy.settings(self.env), self.env) return code, output.getvalue() def test_environment_paths_cannot_cross(self): self.env['DEPLOY_PATH'] = '/api.appconjugate' with self.assertRaises(ValueError): deploy.settings(self.env) def test_transfer_path_traversal_rejected(self): for path in ['/../apidev.appconjugate', '/', '/api.appconjugate', '/apidev.appconjugate\n']: with self.subTest(path=path), self.assertRaises(ValueError): deploy.settings(dict(self.env, DEPLOY_PATH=path)) def test_ftp_password_auth_needs_no_ssh_key(self): for protocol in ['ftp', 'ftps']: self.assertEqual(deploy.settings(dict(self.env, OVH_TRANSFER_PROTOCOL=protocol))['password'], 'private-password') def test_sftp_requires_verified_host_keys(self): with self.assertRaisesRegex(ValueError, 'KNOWN_HOSTS'): deploy.settings(dict(self.env, OVH_TRANSFER_PROTOCOL='sftp')) def test_ftps_encrypts_control_and_data_channels(self): with patch.object(deploy.ftplib, 'FTP_TLS') as factory, patch.object(deploy.ftplib, 'FTP') as plain: with deploy.Transfer(deploy.settings(self.env)): pass self.assertEqual(factory.call_args.kwargs['context'].verify_mode, ssl.CERT_REQUIRED) self.assertTrue(factory.call_args.kwargs['context'].check_hostname) factory.return_value.prot_p.assert_called_once() factory.return_value.login.assert_called_once_with('account', 'private-password') plain.assert_not_called() def test_ftps_failure_never_falls_back_to_plain_ftp(self): with patch.object(deploy.ftplib, 'FTP_TLS') as factory, patch.object(deploy.ftplib, 'FTP') as plain: factory.return_value.login.side_effect = ssl.SSLError('bad certificate') with self.assertRaises(ssl.SSLError), deploy.Transfer(deploy.settings(self.env)): pass plain.assert_not_called() def test_sftp_opens_only_file_transfer_subsystem(self): fake = MagicMock() with patch.dict('sys.modules', {'paramiko': fake}): config = deploy.settings(dict(self.env, OVH_TRANSFER_PROTOCOL='sftp', OVH_TRANSFER_KNOWN_HOSTS='verified-key')) with deploy.Transfer(config): pass client = fake.SSHClient.return_value client.open_sftp.assert_called_once() client.exec_command.assert_not_called() client.invoke_shell.assert_not_called() self.assertFalse(client.connect.call_args.kwargs['look_for_keys']) client.set_missing_host_key_policy.assert_called_once_with(fake.RejectPolicy.return_value) def sftp_diagnostic_fixture(self): fake = MagicMock() fake.SSHException = FakeSSHException fake.AuthenticationException = FakeAuthenticationException fake.BadHostKeyException = FakeBadHostKeyException config = deploy.settings(dict(self.env, OVH_TRANSFER_PROTOCOL='sftp', OVH_TRANSFER_KNOWN_HOSTS='fixture')) return fake, fake.SSHClient.return_value, config def test_sftp_missing_host_entry_fails_before_connecting(self): fake, client, config = self.sftp_diagnostic_fixture() client.get_host_keys.return_value.lookup.return_value = None with patch.dict('sys.modules', {'paramiko': fake}): with self.assertRaisesRegex(deploy.TransferError, 'no server key matching'), deploy.Transfer(config): pass client.connect.assert_not_called() client.close.assert_called_once() def test_sftp_malformed_host_keys_never_echo_the_secret(self): fake, client, config = self.sftp_diagnostic_fixture() client.load_host_keys.side_effect = ValueError('private-secret-in-invalid-entry') with patch.dict('sys.modules', {'paramiko': fake}): with self.assertRaises(deploy.TransferError) as raised: with deploy.Transfer(config): pass self.assertIn('Cannot read OVH_TRANSFER_KNOWN_HOSTS', str(raised.exception)) self.assertNotIn('private-secret', str(raised.exception)) client.connect.assert_not_called() def test_sftp_connection_failures_have_safe_specific_diagnostics(self): cases = [(FakeAuthenticationException, 'authentication failed'), (FakeBadHostKeyException, 'does not match the trusted key'), (FakeSSHException, 'handshake failed'), (OSError, 'Cannot reach the SFTP server')] for error_type, message in cases: with self.subTest(error=error_type.__name__): fake, client, config = self.sftp_diagnostic_fixture() client.connect.side_effect = error_type('private-password-in-library-error') with patch.dict('sys.modules', {'paramiko': fake}): with self.assertRaises(deploy.TransferError) as raised: with deploy.Transfer(config): pass self.assertIn(message, str(raised.exception)) self.assertNotIn('private-password', str(raised.exception)) self.assertNotIn('private-password', self.progress.getvalue()) client.open_sftp.assert_not_called() client.close.assert_called_once() client.set_missing_host_key_policy.assert_called_once_with(fake.RejectPolicy.return_value) def test_sftp_subsystem_failure_is_distinct_from_authentication(self): fake, client, config = self.sftp_diagnostic_fixture() client.open_sftp.side_effect = FakeSSHException('private-password-in-library-error') with patch.dict('sys.modules', {'paramiko': fake}): with self.assertRaisesRegex(deploy.TransferError, 'Authenticated, but the SFTP subsystem'), deploy.Transfer(config): pass def test_sftp_remote_path_failure_identifies_the_protocol(self): fake, client, config = self.sftp_diagnostic_fixture() client.open_sftp.return_value.chdir.side_effect = FileNotFoundError('private-path') with patch.dict('sys.modules', {'paramiko': fake}): with self.assertRaisesRegex(deploy.TransferError, 'SFTP, not FTP'), deploy.Transfer(config): pass def test_sftp_nonstandard_port_uses_bracketed_host_identity(self): fake, client, config = self.sftp_diagnostic_fixture() config['port'] = 2222 with patch.dict('sys.modules', {'paramiko': fake}), deploy.Transfer(config): pass client.get_host_keys.return_value.lookup.assert_called_once_with('[ftp.example.test]:2222') @unittest.skipUnless(paramiko, 'Install requirements-sftp.txt to test the real host-key parser') def test_real_sftp_host_key_parser_rejects_incomplete_or_wrong_host_entries(self): key = 'ssh-ed25519 ' + base64.b64encode(b'\x00\x00\x00\x0bssh-ed25519\x00\x00\x00\x20' + bytes(range(32))).decode() valid = 'ftp.example.test ' + key cases = [('ftp.example.test', False), ('SHA256:only-a-fingerprint', False), ('other.example.test ' + key, False), ('ftp.example.test ssh-ed25519 invalid', False), (valid, True), (paramiko.HostKeys.hash_host('ftp.example.test') + ' ' + key, True)] for entry, accepted in cases: with self.subTest(entry=entry): config = deploy.settings(dict(self.env, OVH_TRANSFER_PROTOCOL='sftp', OVH_TRANSFER_KNOWN_HOSTS=entry)) client = paramiko.SSHClient() with patch.object(paramiko, 'SSHClient', return_value=client), \ patch.object(client, 'connect') as connect, patch.object(client, 'open_sftp', return_value=MagicMock()): if accepted: with deploy.Transfer(config): pass connect.assert_called_once() else: with self.assertRaises(deploy.TransferError), deploy.Transfer(config): pass connect.assert_not_called() def test_ftp_publishes_only_after_complete_upload(self): transfer = deploy.Transfer(deploy.settings(self.env)) transfer.ftp = MagicMock() transfer.ftp.storbinary.side_effect = ConnectionError('lost connection') with self.assertRaises(ConnectionError): transfer.put('job.json', io.BytesIO(b'payload')) transfer.ftp.rename.assert_not_called() def test_zip_contains_dependencies_but_no_environment_storage_or_caches(self): with zipfile.ZipFile(self.package()) as package: self.assertEqual(set(package.namelist()), {'artisan', 'app/Controller.php', 'vendor/autoload.php', 'public/index.php', '.deployment.json'}) self.assertEqual(json.loads(package.read('.deployment.json'))['revision'], 'a' * 40) def test_package_rejects_links_outside_build(self): stage = self.root / 'stage' (stage / 'app').mkdir(parents=True) secret = self.root / 'secret' secret.write_text('private') (stage / 'app/leak').symlink_to(secret) with self.assertRaises(ValueError): deploy.pack(stage, self.root / 'bad.zip', 'a' * 40, '7.4') def test_deploy_uploads_package_before_queue_and_preserves_shared_files(self): self.package() code, output = self.run_operation() self.assertEqual(code, 0) self.assertIn('QUEUED', output) self.assertTrue(self.remote.writes[0].endswith('.zip')) self.assertIn('/queue/', self.remote.writes[1]) self.assertEqual(self.remote.files['shared/.env'], b'private') self.assertEqual(self.remote.files['shared/storage/upload'], b'keep') request = json.loads(self.remote.files[self.remote.writes[1]]) self.assertEqual(request['environment'], 'staging') self.assertFalse(request['migrate']) def test_partial_package_does_not_queue_a_job(self): self.package() self.remote.fail_package = True with self.assertRaises(ConnectionError): self.run_operation() self.assertFalse(any('/queue/' in name for name in self.remote.files)) def test_artisan_arguments_are_data_not_shell_code(self): arguments = ['list', '"; $(touch NEVER)'] self.env.update(DEPLOY_OPERATION='artisan', ARTISAN_ARGUMENTS=json.dumps(arguments)) self.run_operation() request = json.loads(self.remote.files[self.remote.writes[0]]) self.assertEqual(request['arguments'], arguments) def test_status_reports_queue_then_failure_and_hides_command_output(self): self.env.update(DEPLOY_OPERATION='artisan') self.run_operation() job = json.loads(self.remote.files[self.remote.writes[0]]) self.env['DEPLOY_OPERATION'] = 'status' self.assertIn('queued', self.run_operation()[1]) self.remote.put_json(f".deploy/results/{job['id']}.json", dict(job, status='failed', output='private-output')) code, output = self.run_operation() self.assertEqual(code, 1) self.assertNotIn('private-output', output) self.env['DEPLOY_SHOW_OUTPUT'] = 'true' self.assertIn('private-output', self.run_operation()[1]) def test_status_can_read_runner_heartbeat_before_any_job(self): self.env['DEPLOY_OPERATION'] = 'status' self.remote.put_json('.deploy/runner-status.json', dict(id='runner', environment='staging', status='ready', php='7.4.33')) self.assertIn('7.4.33', self.run_operation()[1]) def test_status_rejects_job_path_traversal(self): self.env.update(DEPLOY_OPERATION='status', DEPLOY_REQUEST_ID='../../shared/.env') with self.assertRaises(ValueError): self.run_operation() def test_runner_install_does_not_queue_or_deploy(self): self.env['DEPLOY_OPERATION'] = 'install-runner' self.run_operation() self.assertEqual(self.remote.writes, ['.deploy/.htaccess', '.deploy/worker.php']) def https_settings(self, operation='artisan'): self.env.update(DEPLOY_EXECUTION='https', STAGING_RUNNER_TOKEN='b' * 64, DEPLOY_OPERATION=operation) return dict(environment='staging', protocol=1, status='ready', php='7.4.33', zip_available=True, curl_available=True) def test_https_refuses_production_plain_ftp_and_weak_tokens(self): self.https_settings() for overrides in [dict(DEPLOY_ENVIRONMENT='production', DEPLOY_PATH='/api.appconjugate'), dict(OVH_TRANSFER_PROTOCOL='ftp'), dict(STAGING_RUNNER_TOKEN='short')]: with self.subTest(overrides=overrides), self.assertRaises(ValueError): deploy.settings(dict(self.env, **overrides)) def test_https_install_uploads_only_token_digest_and_bootstrap_bridge(self): self.https_settings('install-runner') code, output = self.run_operation() self.assertEqual(code, 0) self.assertIn('runner-public/_deploy.php', self.remote.files) self.assertIn('.deploy/http.php', self.remote.files) self.assertEqual(self.remote.files['.deploy/http-token.sha256'].strip(), hashlib.sha256(b'b' * 64).hexdigest().encode()) self.assertNotIn('b' * 64, output) self.assertFalse(any(b'b' * 64 in value for value in self.remote.files.values())) self.assertFalse(any('/queue/' in name for name in self.remote.writes)) self.assertEqual(self.remote.files['shared/.env'], b'private') def test_https_install_refuses_to_overwrite_custom_public_files(self): self.https_settings('install-runner') self.remote.files['runner-public/_deploy.php'] = b'custom' with self.assertRaises(deploy.TransferError): self.run_operation() self.assertEqual(self.remote.files['runner-public/_deploy.php'], b'custom') self.assertNotIn('.deploy/http-token.sha256', self.remote.files) def test_https_preflight_failure_does_not_queue_anything(self): health = self.https_settings() for report in [dict(health, status='blocked'), dict(health, zip_available=False)]: with patch.object(deploy, 'https_request', return_value=report), self.assertRaises(deploy.TransferError): self.run_operation() self.assertEqual(self.remote.writes, []) def test_https_runs_only_uploaded_id_and_waits_for_success(self): health = self.https_settings() def request(env, action, request_id=None): if action == 'health': return health self.assertEqual(action, 'run') self.assertIn(f'.deploy/queue/{request_id}.json', self.remote.files) self.assertEqual(json.loads(self.remote.files['.deploy/latest.json'])['id'], request_id) return dict(id=request_id, environment='staging', status='succeeded', output='hidden-command-output') with patch.object(deploy, 'https_request', side_effect=request) as client: code, output = self.run_operation() self.assertEqual(code, 0) self.assertEqual(client.call_count, 2) self.assertNotIn('hidden-command-output', output) def test_https_failed_or_unfinished_operation_is_not_workflow_success(self): health = self.https_settings() for status in ['failed', 'interrupted', 'running', 'queued']: with patch.object(deploy, 'https_request', side_effect=[health, dict(status=status)]): self.assertEqual(self.run_operation()[0], 1) def test_https_client_verifies_tls_and_sends_only_id_with_header_token(self): self.https_settings() identifier = '20260908120000-aaaaaaaaaaaa' opener = MagicMock() response = opener.open.return_value.__enter__.return_value response.status = 200 response.headers.get_content_type.return_value = 'application/json' response.read.return_value = json.dumps(dict(environment='staging', id=identifier, status='succeeded')).encode() with patch.object(deploy.urllib.request, 'build_opener', return_value=opener) as build: report = deploy.https_request(self.env, 'run', identifier) request = opener.open.call_args.args[0] self.assertEqual(request.method, 'POST') self.assertEqual(json.loads(request.data), {'id': identifier}) self.assertEqual(request.get_header('X-conjugate-deploy-token'), 'b' * 64) self.assertNotIn('b' * 64, request.full_url) self.assertEqual(report['status'], 'succeeded') context = build.call_args.args[1]._context self.assertTrue(context.check_hostname) self.assertEqual(context.verify_mode, ssl.CERT_REQUIRED) self.assertIsInstance(build.call_args.args[0], deploy.NoRedirect) self.assertIsNone(deploy.NoRedirect().redirect_request(request, None, 302, '', {}, 'http://unsafe.test')) def test_https_client_does_not_retry_timeouts_or_leak_server_errors(self): self.https_settings() for error in [TimeoutError('private-secret'), deploy.urllib.error.HTTPError('https://private.test', 302, 'private-secret', {}, None)]: opener = MagicMock() opener.open.side_effect = error with patch.object(deploy.urllib.request, 'build_opener', return_value=opener): with self.assertRaises(deploy.TransferError) as raised: deploy.https_request(self.env, 'run', '20260908120000-aaaaaaaaaaaa') self.assertNotIn('private-secret', str(raised.exception)) opener.open.assert_called_once() def test_https_health_does_not_queue_jobs(self): health = self.https_settings('health') with patch.object(deploy, 'https_request', return_value=health): self.assertEqual(self.run_operation()[0], 0) self.assertEqual(self.remote.writes, []) def test_https_connection_diagnostics_are_typed_redacted_and_never_retried(self): self.https_settings('health') private = 'private-secret https://private.example/?token=private-secret' cases = [ (ssl.SSLCertVerificationError(1, private), 'tls_certificate'), (ssl.SSLError(1, private), 'tls_error'), (socket.gaierror(socket.EAI_NONAME, private), 'dns_error'), (TimeoutError(private), 'timeout'), (ConnectionResetError(private), 'connection_reset'), (ConnectionRefusedError(private), 'connection_refused'), (ConnectionAbortedError(private), 'connection_aborted'), (http.client.RemoteDisconnected(private), 'connection_closed'), (http.client.IncompleteRead(private.encode(), 100), 'incomplete_response'), (http.client.BadStatusLine(private), 'http_protocol_error'), (OSError(private), 'network_error'), (deploy.urllib.error.URLError(private), 'url_error'), ] for error, category in cases: for wrapped in [False, True]: with self.subTest(category=category, wrapped=wrapped): opener = MagicMock() opener.open.side_effect = deploy.urllib.error.URLError(error) if wrapped else error with patch.object(deploy.urllib.request, 'build_opener', return_value=opener): with self.assertRaises(deploy.TransferError) as raised: deploy.https_request(self.env, 'health') message = str(raised.exception) self.assertIn('health failed [' + category + '; phase=connect]', message) self.assertIn('read-only', message) self.assertNotIn('Execution may still be running', message) self.assertNotIn('private-secret', message) self.assertNotIn('b' * 64, message) opener.open.assert_called_once() def test_https_response_diagnostics_distinguish_invalid_data_without_disclosing_it(self): self.https_settings('health') cases = [ (200, 'text/html', b'private-secret', 'unexpected_content_type', 'response_headers'), (503, 'application/json', b'private-secret', 'http_status_503', 'response_headers'), (200, 'application/json', b'private-secret', 'invalid_json', 'decode'), (200, 'application/json', b'', 'invalid_json', 'decode'), (200, 'application/json', b'\xffprivate-secret', 'invalid_encoding', 'decode'), (200, 'application/json', b'x' * (1024 * 1024 + 1), 'response_too_large', 'response_body'), (200, 'application/json', b'["private-secret"]', 'invalid_response_shape', 'validate'), (200, 'application/json', b'{"environment":"private-secret"}', 'environment_mismatch', 'validate'), ] for status, content_type, raw, category, phase in cases: with self.subTest(category=category, raw_empty=not raw): opener = MagicMock() response = opener.open.return_value.__enter__.return_value response.status = status response.headers.get_content_type.return_value = content_type response.read.return_value = raw with patch.object(deploy.urllib.request, 'build_opener', return_value=opener): with self.assertRaises(deploy.TransferError) as raised: deploy.https_request(self.env, 'health') message = str(raised.exception) self.assertIn('[' + category + '; phase=' + phase + ']', message) self.assertNotIn('private-secret', message) self.assertNotIn('b' * 64, message) opener.open.assert_called_once() def test_https_read_failure_reports_body_phase_not_connection_phase(self): self.https_settings('health') opener = MagicMock() response = opener.open.return_value.__enter__.return_value response.status = 200 response.headers.get_content_type.return_value = 'application/json' response.read.side_effect = http.client.IncompleteRead(b'private-secret', 100) with patch.object(deploy.urllib.request, 'build_opener', return_value=opener): with self.assertRaises(deploy.TransferError) as raised: deploy.https_request(self.env, 'health') self.assertIn('[incomplete_response; phase=response_body]', str(raised.exception)) self.assertNotIn('private-secret', str(raised.exception)) opener.open.assert_called_once() def test_https_http_error_keeps_only_status_code_and_action(self): self.https_settings('health') opener = MagicMock() error = deploy.urllib.error.HTTPError('https://private-secret', 403, 'private-secret', {'private-header': 'private-secret'}, io.BytesIO(b'private-secret')) opener.open.side_effect = error with patch.object(deploy.urllib.request, 'build_opener', return_value=opener): with self.assertRaises(deploy.TransferError) as raised: deploy.https_request(self.env, 'health') message = str(raised.exception) self.assertIn('health failed [http_status_403;', message) self.assertNotIn('private-secret', message) self.assertNotIn('private-header', message) opener.open.assert_called_once() self.assertTrue(error.fp.closed) def test_https_failure_guidance_depends_on_action_and_redacts_invalid_ids(self): self.https_settings() identifier = '20260922114440-5faf7bb28b63' for action, request_id in [('health', None), ('status', identifier), ('run', identifier), ('run', 'private-secret')]: with self.subTest(action=action, valid_id=request_id == identifier): opener = MagicMock() opener.open.side_effect = TimeoutError('private-secret') with patch.object(deploy.urllib.request, 'build_opener', return_value=opener): with self.assertRaises(deploy.TransferError) as raised: deploy.https_request(self.env, action, request_id) message = str(raised.exception) self.assertIn(action + ' failed [timeout;', message) self.assertIn('No automatic retry was made.', message) self.assertNotIn('private-secret', message) if action == 'run': self.assertIn('Execution may still be running or interrupted.', message) self.assertIn('Inspect status', message) if request_id == identifier: self.assertIn(identifier, message) else: self.assertIn('read-only', message) self.assertNotIn('Execution may still be running', message) opener.open.assert_called_once() self.assertEqual(opener.open.call_args.kwargs['timeout'], 210 if action == 'run' else 30) def test_https_mismatched_job_id_is_redacted_and_does_not_retry_post(self): self.https_settings() identifier = '20260922114440-5faf7bb28b63' opener = MagicMock() response = opener.open.return_value.__enter__.return_value response.status = 200 response.headers.get_content_type.return_value = 'application/json' response.read.return_value = b'{"environment":"staging","id":"private-secret"}' with patch.object(deploy.urllib.request, 'build_opener', return_value=opener): with self.assertRaises(deploy.TransferError) as raised: deploy.https_request(self.env, 'run', identifier) message = str(raised.exception) self.assertIn('run failed [request_id_mismatch; phase=validate]', message) self.assertIn(identifier, message) self.assertIn('Execution may still be running or interrupted.', message) self.assertNotIn('private-secret', message) opener.open.assert_called_once() def test_https_network_failure_during_preflight_submits_no_job(self): for operation in ['deploy', 'artisan', 'initialize-key', 'health']: with self.subTest(operation=operation): self.https_settings(operation) opener = MagicMock() opener.open.side_effect = deploy.urllib.error.URLError(ConnectionResetError('private-secret')) with patch.object(deploy.urllib.request, 'build_opener', return_value=opener): with self.assertRaises(deploy.TransferError) as raised: self.run_operation() self.assertIn('health failed [connection_reset;', str(raised.exception)) self.assertNotIn('Execution may still be running', str(raised.exception)) self.assertEqual(self.remote.writes, []) opener.open.assert_called_once() def test_initialize_key_is_staging_only_and_does_not_upload_environment_or_packages(self): health = self.https_settings('initialize-key') self.env['ARTISAN_ARGUMENTS'] = 'not parsed for this operation' with patch.object(deploy, 'https_request', side_effect=[health, dict(status='succeeded', key_status='preserved')]): code, output = self.run_operation() self.assertEqual(code, 0) queued = [name for name in self.remote.writes if '/queue/' in name] self.assertEqual(len(queued), 1) self.assertEqual(json.loads(self.remote.files[queued[0]])['operation'], 'initialize-key') self.assertEqual(self.remote.files['shared/.env'], b'private') self.assertFalse(any(name.endswith('.zip') for name in self.remote.writes)) with self.assertRaises(ValueError): deploy.settings(dict(self.env, DEPLOY_ENVIRONMENT='production', DEPLOY_PATH='/api.appconjugate')) def test_production_still_installs_only_private_cron_worker_without_token(self): self.env.update(DEPLOY_ENVIRONMENT='production', DEPLOY_PATH='/home/account/api.appconjugate', DEPLOY_OPERATION='install-runner', DEPLOY_EXECUTION='cron') with patch.object(deploy, 'https_request') as client: code, output = self.run_operation() self.assertEqual(code, 0) self.assertIn('cron', output) client.assert_not_called() self.assertEqual(self.remote.writes, ['.deploy/.htaccess', '.deploy/worker.php']) def test_https_client_rejects_wrong_identity_and_invalid_responses(self): self.https_settings() identifier = '20260908120000-aaaaaaaaaaaa' opener = MagicMock() response = opener.open.return_value.__enter__.return_value response.status = 200 response.headers.get_content_type.return_value = 'application/json' for raw in [b'not-json', b'{}', b'x' * (1024 * 1024 + 1), json.dumps(dict(environment='production', id=identifier)).encode(), json.dumps(dict(environment='staging', id='wrong-id')).encode()]: response.read.return_value = raw with patch.object(deploy.urllib.request, 'build_opener', return_value=opener): with self.assertRaises(deploy.TransferError): deploy.https_request(self.env, 'run', identifier) if __name__ == '__main__': unittest.main()