"""Постоянная сессия: локальный RPC и replace source breakpoints.""" from pathlib import Path import sys import tempfile import threading import unittest 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' 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} 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 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}], '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 = [] def close(self): pass def request(self, command, **arguments): self.calls.append(command) if command in ('step', 'step_over', 'step_out'): self.owner.index = min(self.owner.index + 1, 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') events = rpc_call(server.path, 'events', {'after': 0}) self.assertEqual(events['events'][0]['event'], 'stopped') with self.assertRaisesRegex(SessionError, 'Неизвестный'): rpc_call(server.path, 'unknown') finally: server.server.shutdown() thread.join() server.close() 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_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()