"""Постоянная сессия: локальный RPC и replace source breakpoints.""" from pathlib import Path import sys import tempfile import threading import unittest from types import SimpleNamespace ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(ROOT / 'toolchain')) from sdbg.server import SessionController, SessionRpcServer, rpc_call from sdbg.session import SessionError class DummyModel: variables = [] logpoints = [] stale_sources = [] class DummyBridge: def __init__(self): self.calls = [] self.state = 'stopped' self.generation = 1 def close(self): pass def request(self, command, **arguments): self.calls.append((command, arguments)) if command == 'snapshot': return {'state': self.state} if command == 'console_print': return {'printed': True} if command == 'memory': return {'hex': '00' * arguments['length']} if command == 'disassemble_logical': return {'space': 'logical_z80', 'address': arguments['address'], 'length': arguments['length'], 'text': '8100: NOP', 'generation': self.generation} if command == 'key': return {'accepted': True} raise AssertionError(command) class DummySession: def __init__(self): self.bridge = DummyBridge() self.model = DummyModel() self.next_id = 1 self.cleared = [] self.activated = 0 def attach(self): return {'build_id': 'test', 'location': {'status': 'mapped'}} def where(self, snapshot=None): return {'status': 'mapped', 'pc': 0x8100, 'link_address': 0x8100, 'sources': [{'file': '/src/main.c', 'line': 3}]} def refresh(self): return SimpleNamespace(generation=1, bank_pages={}) def break_line(self, filename, line, enabled=True): if line == 99: raise SessionError('нет адреса') identifier = self.next_id self.next_id += 1 return {'id': identifier, 'backend_ids': [identifier], 'conditions': [], 'locations': [{'line': line, 'link_address': 0x8100}], 'enabled': enabled} def break_function(self, name, enabled=True): return self.break_line('/src/main.c', 3, enabled=enabled) def clear_breakpoint(self, identifier): self.cleared.append(identifier) return {'cleared': identifier} def activate_breakpoints(self): self.activated += 1 return {'enabled': self.next_id - 1} def read_variable(self, name, module=None): if name == 'total': return {'value': 42} if name == 'flag': return {'value': 0} raise SessionError('нет переменной') class MacroSession(DummySession): def __init__(self): super().__init__() self.model = type('MacroModel', (), { 'variables': [], 'stale_sources': [], 'logpoints': [ {'tag': 'authored', 'source': '/src/main.c', 'line': 3, 'verified': True, 'message': 'total={total}', 'condition': None, 'module': None, 'link_address': 0x8100}, {'tag': 'conditional', 'source': '/src/main.c', 'line': 4, 'verified': True, 'message': 'never', 'condition': 'flag', 'module': None, 'link_address': 0x8102}, ]})() def attach(self): return {'build_id': 'macro', 'location': {'status': 'mapped'}, 'capabilities': {'console_print': True}} def break_anchor(self, location, enabled=True): identifier = self.next_id self.next_id += 1 return {'id': identifier, 'backend_ids': [identifier], 'locations': [{'link_address': location['link_address']}], 'conditions': [], 'enabled': enabled} class StepBridge: def __init__(self, owner): self.owner = owner self.calls = [] self.step_counts = [] def close(self): pass def request(self, command, **arguments): self.calls.append(command) if command in ('step', 'step_over', 'step_out'): count = arguments.get('count', 1) if command == 'step' else 1 self.step_counts.append(count) self.owner.index = min(self.owner.index + count, len(self.owner.locations) - 1) return {'accepted': True} if command == 'snapshot': return {'state': 'stopped'} raise AssertionError(command) def wait_stopped(self, timeout=5): return {'state': 'stopped'} class StepSession: def __init__(self): self.model = DummyModel() self.locations = [ {'status': 'mapped', 'link_address': 0x8100, 'sources': [{'file': '/src/main.c', 'line': 3}]}, {'status': 'mapped', 'link_address': 0x8102, 'sources': [{'file': '/src/main.c', 'line': 3}]}, {'status': 'mapped', 'link_address': 0x8104, 'sources': [{'file': '/src/main.c', 'line': 4}]}, ] self.index = 0 self.bridge = StepBridge(self) def attach(self): return {'build_id': 'step', 'location': self.locations[0]} def where(self, snapshot=None): return self.locations[self.index] class WaitingBridge: """Машинный over ожидает внешний ввод, но pause должен остаться доступен.""" def __init__(self): self.paused = False self.started = False self.calls = [] def close(self): pass def request(self, command, **arguments): self.calls.append(command) if command == 'snapshot': return {'state': 'running' if self.started and not self.paused else 'stopped'} if command == 'pause': self.paused = True return {'accepted': True} if command == 'step_over': self.started = True return {'accepted': True} raise AssertionError(command) def wait_stopped(self, timeout=5): return {'state': 'stopped'} class ServerTests(unittest.TestCase): @staticmethod def wait_source_step(controller): with controller.changed: if not controller.changed.wait_for(lambda: not controller.running, timeout=2): raise AssertionError('source step не сообщил остановку') def test_rpc_roundtrip_and_events(self): with tempfile.TemporaryDirectory() as directory: controller = SessionController(DummySession()) try: server = SessionRpcServer(Path(directory) / 's.sock', controller) except PermissionError: controller.close() self.skipTest('sandbox запрещает bind Unix socket') thread = threading.Thread(target=server.server.serve_forever, daemon=True) thread.start() try: status = rpc_call(server.path, 'status') self.assertEqual(status['build_id'], 'test') identity = {'session_id': status['session_id'], 'build_id': status['build_id']} events = rpc_call(server.path, 'events', {'after': 0}, **identity) self.assertEqual(events['events'][0]['event'], 'stopped') with self.assertRaisesRegex(SessionError, 'Устаревшая RPC-сессия'): rpc_call(server.path, 'events', {'after': 0}) with self.assertRaisesRegex(SessionError, 'Устаревшая RPC-сессия'): rpc_call(server.path, 'events', {'after': 0}, session_id='old', build_id=status['build_id']) with self.assertRaisesRegex(SessionError, 'Устаревшая RPC-сессия'): rpc_call(server.path, 'events', {'after': 0}, session_id=status['session_id'], build_id='old-build') with self.assertRaisesRegex(SessionError, 'Устаревшая generation'): rpc_call(server.path, 'break_line', {'file': '/src/main.c', 'line': 3}, generation=0, **identity) point = rpc_call(server.path, 'break_line', {'file': '/src/main.c', 'line': 3}, generation=status['generation'], **identity) self.assertEqual(point['id'], 1) with self.assertRaisesRegex(SessionError, 'Неизвестный'): rpc_call(server.path, 'unknown') finally: server.server.shutdown() thread.join() server.close() controller.close() def test_individual_breakpoints_keep_owner_and_source_step_priority(self): session = DummySession() controller = SessionController(session) try: mine = controller.call('break_line', {'file': '/src/main.c', 'line': 3, 'owner': 'mcp:a'}) other = controller.call('break_function', {'name': 'main', 'owner': 'mcp:b'}) controller.breakpoint_info[mine['id']]['guard_conditions'] = ['PG3 == 0x12'] listed = controller.call('list_breakpoints', {})['breakpoints'] self.assertEqual([(item['id'], item['owner']) for item in listed], [(mine['id'], 'mcp:a'), (other['id'], 'mcp:b')]) self.assertEqual(listed[0]['locations'][0]['link_address'], 0x8100) self.assertEqual(listed[0]['locations'][0]['bank_guard'], 'PG3 == 0x12') self.assertTrue(controller._has_stop_breakpoint(session.where())) with self.assertRaisesRegex(SessionError, 'не принадлежит'): controller.call('clear_breakpoint', {'id': other['id'], 'owner': 'mcp:a'}) self.assertEqual(controller.call('clear_owned_breakpoints', {'owner': 'mcp:a'}), {'cleared': [mine['id']]}) self.assertEqual(session.cleared, [mine['id']]) self.assertIn(other['id'], controller.breakpoint_info) controller.call('clear_breakpoint', {'id': other['id'], 'owner': 'mcp:b'}) self.assertFalse(controller.breakpoint_info) finally: controller.close() def test_disassemble_logical_is_bounded_and_labels_bank_mapping(self): controller = SessionController(DummySession()) try: result = controller.call('disassemble_logical', {'address': 0x8100, 'length': 32}) self.assertEqual(result['space'], 'logical_z80') self.assertEqual(result['address'], 0x8100) self.assertEqual(result['bank_pages'], {}) with self.assertRaisesRegex(SessionError, 'Дизассемблирование'): controller.call('disassemble_logical', {'address': 0xffff, 'length': 2}) self.assertEqual(len([call for call in controller.session.bridge.calls if call[0] == 'disassemble_logical']), 1) finally: controller.close() def test_control_lease_blocks_other_client_and_can_be_released_or_expire(self): controller = SessionController(DummySession()) try: first = controller.call('claim_control', {'owner': 'mcp:a'}) self.assertEqual(first['owner'], 'mcp:a') self.assertEqual(controller.call('status', {})['control_owner'], 'mcp:a') with self.assertRaisesRegex(SessionError, 'mcp:a'): controller.call('claim_control', {'owner': 'mcp:b'}) controller.call('renew_control', {'owner': 'mcp:a'}) self.assertFalse(controller.call('renew_control', {'owner': 'mcp:b'})['has_control']) controller.call('release_control', {'owner': 'mcp:b'}) self.assertEqual(controller.call('status', {})['control_owner'], 'mcp:a') controller.call('release_control', {'owner': 'mcp:a'}) self.assertEqual(controller.call('claim_control', {'owner': 'mcp:b'})['owner'], 'mcp:b') controller.control_deadline = 0 self.assertEqual(controller.call('claim_control', {'owner': 'cli'})['owner'], 'cli') self.assertFalse(controller.call('renew_control', {'owner': 'mcp:b'})['has_control']) finally: controller.close() def test_expired_mcp_owner_clears_points_without_touching_dap_points(self): session = DummySession() controller = SessionController(session) try: mine = controller.call('break_line', { 'file': '/src/main.c', 'line': 3, 'owner': 'mcp:lost'}) dap = controller.call('break_function', {'name': 'main', 'owner': 'cli'}) controller.call('claim_control', {'owner': 'mcp:lost'}) controller.owner_deadlines['mcp:lost'] = 0 with controller.lock: controller._reap_owners() self.assertIn(mine['id'], session.cleared) self.assertNotIn(dap['id'], session.cleared) self.assertIsNone(controller.call('status', {})['control_owner']) self.assertEqual(controller.events[-1]['event'], 'owner_expired') finally: controller.close() def test_expired_owner_releases_held_key(self): session = DummySession() controller = SessionController(session) try: with controller.lock: controller.running = True session.bridge.state = 'running' controller.call('input_key', { 'owner': 'mcp:lost', 'tag': ':kbd:ms_naturl', 'mask': 0x400, 'down': True}) self.assertIn(('key', {'tag': ':kbd:ms_naturl', 'mask': 0x400, 'down': True}), session.bridge.calls) controller.owner_deadlines['mcp:lost'] = 0 controller._reap_owners() self.assertIn(('key', {'tag': ':kbd:ms_naturl', 'mask': 0x400, 'down': False}), session.bridge.calls) self.assertFalse(controller.held_inputs) finally: controller.close() def test_read_memory_is_bounded_and_generation_tied(self): session = DummySession() controller = SessionController(session) try: result = controller.call('read_memory', {'address': 0xfffe, 'length': 2}) self.assertEqual(result, {'address': 0xfffe, 'length': 2, 'hex': '0000', 'generation': 1, 'bank_pages': {}}) with self.assertRaisesRegex(SessionError, 'за 64 КБ'): controller.call('read_memory', {'address': 0xffff, 'length': 2}) with self.assertRaisesRegex(SessionError, 'длина 1..256'): controller.call('read_memory', {'address': 0, 'length': 257}) finally: controller.close() def test_screen_and_share_reads_reject_oversized_requests(self): controller = SessionController(DummySession()) try: with self.assertRaisesRegex(SessionError, 'длина 1..4096'): controller.call('read_share', {'tag': ':vram', 'address': 0, 'length': 4097}) with self.assertRaisesRegex(SessionError, 'точный tag'): controller.call('read_share', {'tag': '', 'address': 0, 'length': 1}) with self.assertRaisesRegex(SessionError, '8192 пикселей'): controller.call('read_screen_pixels', {'x': 0, 'y': 0, 'width': 128, 'height': 128}) finally: controller.close() def test_idle_snapshot_detects_invalidation(self): session = DummySession() controller = SessionController(session) try: session.bridge.state = 'invalidated' with controller.changed: self.assertTrue(controller.changed.wait_for(lambda: controller.closed, timeout=2)) self.assertEqual(controller.events[-1]['event'], 'invalidated') self.assertEqual(controller.events[-1]['body']['reason'], 'reset_or_load') finally: controller.close() def test_native_debugger_continue_is_reported(self): session = DummySession() controller = SessionController(session) try: session.bridge.state = 'running' with controller.changed: self.assertTrue(controller.changed.wait_for(lambda: controller.running, timeout=2)) self.assertEqual(controller.events[-1]['event'], 'continued') self.assertEqual(controller.events[-1]['body']['reason'], 'external') session.bridge.state = 'stopped' with controller.changed: self.assertTrue(controller.changed.wait_for(lambda: not controller.running, timeout=2)) self.assertEqual(controller.events[-1]['event'], 'stopped') finally: controller.close() def test_source_breakpoints_replace_and_rollback(self): session = DummySession() controller = SessionController(session) try: first = controller.call('set_source_breakpoints', {'file': 'main.c', 'lines': [3, 5]}) self.assertEqual([item['line'] for item in first['breakpoints']], [3, 5]) self.assertEqual(controller.source_breakpoints[str(Path('main.c').resolve())], [1, 2]) controller.call('set_source_breakpoints', {'file': 'main.c', 'lines': [7]}) self.assertEqual(session.cleared, [1, 2]) with self.assertRaisesRegex(SessionError, 'нет адреса'): controller.call('set_source_breakpoints', {'file': 'main.c', 'lines': [8, 99]}) self.assertIn(4, session.cleared) self.assertEqual(controller.source_breakpoints[str(Path('main.c').resolve())], [3]) finally: controller.close() def test_logpoint_is_bounded_and_does_not_hide_stop(self): controller = SessionController(DummySession()) try: result = controller.call('set_source_breakpoints', { 'file': 'main.c', 'breakpoints': [{'line': 3, 'logMessage': 'total={total} {{ok}}'}]}) identifier = result['breakpoints'][0]['id'] controller.breakpoint_info[identifier]['locations'] = [{'link_address': 0x8100}] self.assertTrue(controller._handle_logpoints({'link_address': 0x8100})) self.assertEqual(controller.events[-1]['body']['output'], 'total=42 {ok}\n') controller.breakpoint_info[99] = { 'kind': 'stop', 'message': None, 'locations': [{'link_address': 0x8100}], 'hits': 0} self.assertFalse(controller._handle_logpoints({'link_address': 0x8100})) with self.assertRaisesRegex(SessionError, 'только подстановки'): controller._validate_log_message('{total+1}') finally: controller.close() def test_authored_macro_logs_auto_arm_mirror_and_preserve_user_stop(self): session = MacroSession() controller = SessionController(session) try: self.assertEqual(session.activated, 1) self.assertEqual([info['tag'] for info in controller.breakpoint_info.values()], ['authored', 'conditional']) self.assertTrue(controller._handle_logpoints({'link_address': 0x8100})) self.assertEqual(controller.events[-1]['body']['output'], 'total=42\n') self.assertEqual(session.bridge.calls[-1], ('console_print', {'text': 'total=42'})) before = len(controller.events) self.assertTrue(controller._handle_logpoints({'link_address': 0x8102})) self.assertEqual(len(controller.events), before) controller.breakpoint_info[99] = { 'kind': 'stop', 'message': None, 'locations': [{'link_address': 0x8100}], 'hits': 0} self.assertFalse(controller._handle_logpoints({'link_address': 0x8100})) self.assertEqual(controller.events[-1]['body']['output'], 'total=42\n') controller.call('set_source_breakpoints', {'file': '/src/main.c', 'lines': [5]}) controller.call('set_source_breakpoints', {'file': '/src/main.c', 'lines': []}) self.assertEqual(session.cleared, [3]) self.assertIn(1, controller.breakpoint_info) finally: controller.close() def test_bounded_event_ring_reports_loss(self): controller = SessionController(DummySession()) try: for number in range(1025): controller._emit('output', {'output': str(number)}) result = controller.call('events', {'after': 0, 'timeout': 0}) self.assertEqual(result['lost'], 2) # entry + первый output self.assertEqual(result['first'], 3) self.assertEqual(len(result['events']), 1024) finally: controller.close() def test_source_step_skips_same_line_and_uses_over_primitive(self): session = StepSession() controller = SessionController(session) try: result = controller.call('source_step', {'kind': 'into'}) self.assertEqual(result, {'accepted': True}) self.wait_source_step(controller) self.assertEqual(controller.events[-1]['body']['location']['sources'][0]['line'], 4) self.assertEqual([name for name in session.bridge.calls if name != 'snapshot'], ['step', 'step']) session.index = 0 session.bridge.calls.clear() result = controller.call('source_step', {'kind': 'over'}) self.assertEqual(result, {'accepted': True}) self.wait_source_step(controller) self.assertEqual([name for name in session.bridge.calls if name != 'snapshot'], ['step_over', 'step_over']) session.index = 0 session.bridge.calls.clear() result = controller.call('source_step', {'kind': 'out'}) self.assertEqual(result, {'accepted': True}) self.wait_source_step(controller) self.assertEqual([name for name in session.bridge.calls if name != 'snapshot'], ['step_out', 'step_over']) finally: controller.close() def test_instruction_step_count_is_bounded_and_forwarded(self): session = StepSession() controller = SessionController(session) try: with self.assertRaisesRegex(SessionError, '1..64'): controller.call('step', {'count': 0, 'owner': 'mcp:test'}) for bad in (65, 1.5, True): with self.assertRaisesRegex(SessionError, '1..64'): controller.call('step', {'count': bad, 'owner': 'mcp:test'}) result = controller.call('step', {'count': 2, 'owner': 'mcp:test'}) self.assertEqual(result['sources'][0]['line'], 4) self.assertEqual(session.bridge.step_counts, [2]) finally: controller.close() def test_source_step_preserves_user_breakpoint_on_same_line(self): session = StepSession() controller = SessionController(session) try: controller.breakpoint_info[1] = { 'kind': 'stop', 'message': None, 'locations': [{'link_address': 0x8102}], 'hits': 0} result = controller.call('source_step', {'kind': 'into'}) self.assertEqual(result, {'accepted': True}) self.wait_source_step(controller) self.assertEqual(controller.events[-1]['body']['reason'], 'breakpoint') finally: controller.close() def test_waiting_step_over_releases_session_lock_for_pause(self): session = StepSession() session.bridge = WaitingBridge() controller = SessionController(session) try: started = controller.call('source_step', {'kind': 'over'}) self.assertEqual(started, {'accepted': True}) self.assertTrue(controller.running) stopped = controller.call('pause', {}) self.assertEqual(stopped['sources'][0]['line'], 3) self.assertFalse(controller.running) self.assertIsNone(controller.source_step) self.assertIn('pause', session.bridge.calls) self.assertEqual(controller.events[-1]['body']['reason'], 'pause') finally: controller.close() if __name__ == '__main__': unittest.main()