117 lines
6.2 KiB
Python
117 lines
6.2 KiB
Python
import importlib.util
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
|
|
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('preprod')
|
|
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 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')
|
|
|
|
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_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()
|