318 lines
13 KiB
Python
318 lines
13 KiB
Python
"""Постоянная сессия: локальный 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 = []
|
|
def close(self): pass
|
|
def request(self, command, **arguments):
|
|
self.calls.append((command, arguments))
|
|
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 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.calls = []
|
|
|
|
def close(self): pass
|
|
|
|
def request(self, command, **arguments):
|
|
self.calls.append(command)
|
|
if command == 'snapshot':
|
|
return {'state': 'stopped' if self.paused else 'running'}
|
|
if command == 'pause':
|
|
self.paused = True
|
|
return {'accepted': True}
|
|
if command == 'step_over':
|
|
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_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()
|