48 lines
2.2 KiB
Python
48 lines
2.2 KiB
Python
#!/usr/bin/env python3
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
LOADER = Path(__file__).with_name('load_shared_persona_context.py')
|
|
TOPOLOGY = ROOT / 'routing/zhuyuan-host-topology.json'
|
|
|
|
|
|
class SharedPersonaContextTest(unittest.TestCase):
|
|
def load(self, host):
|
|
value = subprocess.run(
|
|
[sys.executable, str(LOADER), '--host', host, '--intent', '宿主对齐集成测试', '--format', 'json'],
|
|
text=True, capture_output=True, timeout=45, check=True
|
|
)
|
|
return json.loads(value.stdout)
|
|
|
|
def test_topology_roles(self):
|
|
topology = json.loads(TOPOLOGY.read_text())
|
|
self.assertEqual(topology['primary_host'], 'codex')
|
|
self.assertFalse(topology['host_switch_requires_reteaching'])
|
|
self.assertEqual(topology['hosts']['qoder']['redirect_host'], 'qwen')
|
|
self.assertTrue(topology['hosts']['claude']['state'].startswith('RETIRED'))
|
|
for host in ['codex', 'zcode', 'doubao', 'qwen']:
|
|
adapter = Path(topology['hosts'][host]['adapter'].replace('~', str(Path.home()), 1))
|
|
self.assertTrue(adapter.is_file(), f'missing active adapter: {host}: {adapter}')
|
|
self.assertTrue(Path(topology['hosts']['qwen']['doorplate']).is_file())
|
|
self.assertTrue(Path(topology['hosts']['claude']['history']).is_dir())
|
|
|
|
def test_every_host_reads_one_current_brain(self):
|
|
values = {host: self.load(host) for host in ['codex', 'zcode', 'doubao', 'qwen', 'qoder', 'qoderwork', 'claude']}
|
|
revisions = {value['learning_brain']['revision'] for value in values.values()}
|
|
hashes = {value['learning_brain']['cortex_sha256'] for value in values.values()}
|
|
memory_hashes = {value['daily_memory']['day_sha256'] for value in values.values()}
|
|
self.assertEqual(len(revisions), 1)
|
|
self.assertEqual(len(hashes), 1)
|
|
self.assertEqual(len(memory_hashes), 1)
|
|
self.assertEqual(values['qoder']['effective_host'], 'qwen')
|
|
self.assertEqual(values['qoderwork']['effective_host'], 'qwen')
|
|
self.assertTrue(values['claude']['history_only'])
|
|
self.assertFalse(values['claude']['new_cognition_write'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|