import contextlib import importlib.util import io import json import os from pathlib import Path import subprocess import sys import tempfile import unittest from unittest.mock import patch HERE = Path(__file__).resolve().parent def module(name): spec = importlib.util.spec_from_file_location(name, HERE / (name + '.py')) result = importlib.util.module_from_spec(spec) spec.loader.exec_module(result) return result class SourcesTest(unittest.TestCase): def setUp(self): self.m = module('production') self.root = Path(tempfile.mkdtemp()) self.addCleanup(lambda: __import__('shutil').rmtree(self.root)) def private(self, name, text): p = self.root / name p.write_text(text) p.chmod(0o600) return p def test_env_is_literal_and_never_executes(self): p = self.private('input.env', 'KEY="a b#c"\nOTHER=$(touch sentinel)\n') self.assertEqual(self.m.read_env(p), {'KEY': 'a b#c', 'OTHER': '$(touch sentinel)'}) self.assertFalse((self.root / 'sentinel').exists()) def test_permissions_duplicates_and_quotes_fail_without_values(self): for text in ('KEY=private-value\nKEY=another\n', 'KEY="private-value\n'): with self.assertRaises(ValueError) as ctx: self.m.read_env(self.private('input.env', text)) self.assertNotIn('private-value', str(ctx.exception)) p = self.private('input.env', 'KEY=private-value\n') p.chmod(0o664) with self.assertRaises(ValueError): self.m.read_env(p) def test_oss_format_preserves_case_and_required_fields(self): p = self.private('aliyun-oss.env', 'bucket: test-bucket\nEndpoint: oss-cn-beijing.aliyuncs.com\nRegion: cn-beijing\nRAM:\n username: synthetic\n accessKeyId: test-id\n accessKeySecret: test-secret\n') d = self.m.read_oss(p) self.assertEqual(d['bucket'], 'test-bucket') self.assertEqual(d['RAM.accessKeySecret'], 'test-secret') self.assertEqual(d['Region'], 'cn-beijing') # The real source is a colon file, not YAML: RAM children may be flat. p.write_text(p.read_text().replace(' username:', 'username:').replace(' accessKeyId:', 'accessKeyId:').replace(' accessKeySecret:', 'accessKeySecret:')) self.assertEqual(self.m.read_oss(p), d) def test_sources_missing_required_field_do_not_default(self): self.private('saas.env', 'DispatcherUUID=00000000-0000-4000-8000-000000000001\nSaaSBaseURL=https://saas.example.invalid\n') self.private('rabbitmq.env', 'RABBITMQ_HOST=host\n') self.private('aliyun-oss.env', 'bucket: test\n') with self.assertRaises(ValueError): self.m.sources(self.root) class EntryPointTest(unittest.TestCase): def test_help_uses_production_entry_without_private_inputs(self): result = subprocess.run([sys.executable, str(HERE / 'production.py'), '--help'], capture_output=True, text=True) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn('production.py', result.stdout) self.assertNotIn('preprod', result.stdout) def test_deploy_builds_and_uploads_matching_production_artifacts(self): m = module('production') calls = [] payloads = [] def checked(args, **kwargs): calls.append(args) if args[:3] == ['git', 'branch', '--show-current']: return 'main' if args[:3] == ['git', 'rev-parse', 'HEAD']: return 'a' * 40 if args[:2] == ['git', 'ls-remote']: return 'a' * 40 + '\trefs/heads/main' if args[:2] == ['go', 'build']: Path(args[args.index('-o') + 1]).write_bytes(b'synthetic executable') if args[0] == 'ssh' and args[-1].startswith('python3 /'): payloads.append(json.loads(kwargs['input'])) return json.dumps({'ok': True}) return '' with tempfile.TemporaryDirectory() as tmp: known_hosts = Path(tmp) / 'known_hosts' known_hosts.touch() argv = ['production.py', 'deploy', '--known-hosts', str(known_hosts)] with patch.object(sys, 'argv', argv), patch.object(m, 'sources', return_value={'synthetic': True}), patch.object(m, 'checked', side_effect=checked), contextlib.redirect_stdout(io.StringIO()): m.main() builds = [args for args in calls if args[:2] == ['go', 'build']] self.assertEqual([args[-1] for args in builds], ['./cmd/sip-go-agent', './deploys/production']) self.assertEqual([Path(args[args.index('-o') + 1]).name for args in builds], ['sip-go-agent', 'production-probe']) self.assertEqual(set(payloads[0]['hashes']), {'sip-go-agent', 'production-probe'}) self.assertTrue(any(args[0] == 'scp' and args[-1].endswith('/production-remote.py') for args in calls)) self.assertFalse(any('preprod' in arg for args in calls for arg in args)) class SafetyTest(unittest.TestCase): def setUp(self): self.m = module('remote') def test_fixed_paths_not_per_deployment(self): p = self.m.paths(Path('/home/synthetic')) self.assertEqual(str(p['state']), '/home/synthetic/.local/share/go-sip') self.assertEqual(str(p['config']), '/home/synthetic/.config/go-sip') self.assertEqual(str(p['releases']), '/home/synthetic/.local/opt/go-sip/releases') def test_reset_requires_explicit_authorization_and_zero_calls(self): for requested, confirmed, channels in ((False, True, 0), (True, False, 0), (True, True, 1)): with self.assertRaises(ValueError): self.m.require_reset(requested, confirmed, channels) self.m.require_reset(True, True, 0) def test_unavailable_saas_is_deferred_not_success_or_abort(self): status = self.m.dependency_state({'success': False, 'phase': 'sip', 'http_status': 404}) self.assertEqual(status, 'saas_pending') self.assertEqual(self.m.dependency_state({'success': True, 'phase': 'complete'}), 'verified') self.assertEqual(self.m.dependency_state({'success': False, 'phase': 'mq_connect'}), 'mq_failed') for kind in ('tls_failed', 'transport_failed', ''): self.assertEqual(self.m.dependency_state({'success': False, 'phase': 'sip', 'http_status': 0, 'transport_failure': kind}), 'dependency_failed') self.assertEqual(self.m.dependency_state({'success': False, 'phase': 'sip', 'http_status': 0, 'transport_failure': 'unavailable'}), 'saas_pending') def test_role_rendering_overrides_old_mock_business_source(self): source = {'dispatcher_id': 'new-owner', 'saas_url': 'https://real.example.invalid', 'secret': 'new-secret', 'mq_url': 'amqp://synthetic:placeholder@mq.example.invalid/test'} p = self.m.paths(Path('/home/synthetic')) roles = self.m.render_roles({'AGENT_ID': 'agent', 'CELL_ID': 'cell', 'AGENT_SESSION_PATH': '/old/session', 'AGENT_MOCK_SCENARIO_FILE': '/old/scenario', 'AGENT_APPROVED_AI_TASK_MAP_FILE': '/old/ai'}, {'SAAS_BASE_URL': 'http://127.0.0.1:18080', 'DISPATCHER_SECRET_KEY': 'old-secret', 'RABBITMQ_URL': 'amqp://old', 'DISPATCHER_SQLITE_PATH': '/old/db'}, source, p) self.assertEqual(roles['dispatcher']['SAAS_BASE_URL'], source['saas_url']) self.assertEqual(roles['dispatcher']['RABBITMQ_URL'], source['mq_url']) self.assertEqual(roles['agent']['DISPATCHER_ID'], 'new-owner') self.assertEqual(roles['agent']['AGENT_GRPC_LISTEN'], '127.0.0.1:19443') self.assertEqual(roles['agent']['AGENT_SESSION_PATH'], str(p['state'] / 'agent-session.json')) self.assertEqual(roles['dispatcher']['DISPATCHER_SECRET_KEY'], 'new-secret') self.assertEqual(roles['dispatcher']['DISPATCHER_GRPC_LISTEN'], '127.0.0.1:19444') self.assertNotIn('SAAS_SECRET_KEY', roles['dispatcher']) self.assertNotIn('AGENT_APPROVED_AI_TASK_MAP_FILE', roles['agent']) self.assertTrue(roles['dispatcher']['DISPATCHER_SQLITE_PATH'].endswith('/go-sip/dispatcher.sqlite')) self.assertFalse(any('MOCK' in k or 'FIXTURE' in k for d in roles.values() for k in d)) def test_absent_unit_is_not_an_inspection_failure(self): from unittest.mock import patch import subprocess with patch.object(self.m.subprocess, 'run', return_value=subprocess.CompletedProcess([], 4, 'LoadState=not-found\nFragmentPath=\n')): self.assertEqual(self.m.unit_properties('new.service')['LoadState'], 'not-found') with patch.object(self.m.subprocess, 'run', return_value=subprocess.CompletedProcess([], 1, '')): with self.assertRaises(ValueError): self.m.unit_properties('broken.service') def test_legacy_sip_only_is_stopped_but_native_asterisk_preserved(self): from unittest.mock import patch calls = [] def run(args, **kwargs): calls.append(args) if 'list-unit-files' in args: return 'go-sip-nonprod-sip-only-test.service disabled\ngo-sip-nonprod-siponly-test.service disabled\ngo-sip-asterisk.service enabled' return '' with patch.object(self.m, 'run', side_effect=run), patch.object(self.m, 'ari', return_value=[]), patch.object(self.m, 'unit_properties', return_value={'FragmentPath': '/owned/unit'}): units = self.m.stop_for_change(self.m.paths(Path('/home/synthetic')), legacy=True) self.assertEqual(len(units), 2) self.assertTrue(any('go-sip-nonprod-sip-only-test.service' in a and 'stop' in a for a in calls)) self.assertTrue(any('go-sip-nonprod-siponly-test.service' in a and 'stop' in a for a in calls)) self.assertFalse(any('go-sip-asterisk.service' in a for a in calls)) def test_dispatcher_failed_is_never_accepted_as_waiting(self): for state, substate in (('failed', 'failed'), ('activating', 'start'), ('inactive', 'dead')): self.assertFalse(self.m.dispatcher_acceptable({'UnitFileState': 'enabled', 'ActiveState': state, 'SubState': substate}, 'saas_topology_pending')) self.assertTrue(self.m.dispatcher_acceptable({'UnitFileState': 'enabled', 'ActiveState': 'activating', 'SubState': 'auto-restart'}, 'saas_topology_pending')) self.assertFalse(self.m.dispatcher_acceptable({'UnitFileState': 'enabled', 'ActiveState': 'activating', 'SubState': 'auto-restart'}, None)) def test_normal_deployment_never_selects_reset(self): self.assertFalse(self.m.reset_requested({})) self.assertFalse(self.m.reset_requested({'reset_state': False})) self.assertTrue(self.m.reset_requested({'reset_state': True})) if __name__ == '__main__': unittest.main()