Sprinter-SDCC/tests/sdbg/test_dap.py

"""DAP MVP сообщает только реализованные возможности и точные адреса."""
from pathlib import Path
import json
import subprocess
import sys
import unittest

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / 'toolchain'))
from sdbg.dap import DapEngine
from sdbg.session import SessionError


class Rpc:
    def __init__(self):
        self.calls = []
        self.location = {
            'status': 'mapped', 'pc': 0x8100, 'link_address': 0x8100,
            'function': {'name': 'main', 'bank': None},
            'sources': [{'file': '/src/main.c', 'line': 3}],
        }

    def __call__(self, method, arguments):
        self.calls.append((method, arguments))
        if method == 'status':
            return {'build_id': 'abc', 'event_sequence': 4}
        if method in ('where', 'pause', 'step', 'source_step'):
            return self.location
        if method == 'registers':
            return {'registers': {'PC': 0x8100, 'SP': 0xbffe}}
        if method == 'variables':
            return [{'name': 'total', 'module': None, 'type': 'SI:S',
                     'link_address': 0x8200}]
        if method == 'read_variable':
            return {'value': 42, 'type': 'SI:S', 'link_address': 0x8200}
        if method == 'set_source_breakpoints':
            return {'breakpoints': [{'id': 7, 'verified': True, 'line': 3,
                    'locations': [{'line': 3, 'link_address': 0x8100}]}]}
        if method == 'set_function_breakpoints':
            return {'breakpoints': [{'id': 8, 'verified': True,
                    'locations': [{'link_address': 0x8100}]}]}
        if method == 'continue':
            return {'accepted': True}
        if method == 'events':
            return {'events': [], 'last': 4, 'closed': False}
        raise AssertionError(method)


class DapTests(unittest.TestCase):
    def setUp(self):
        self.rpc = Rpc()
        self.dap = DapEngine(self.rpc)

    def call(self, command, **arguments):
        return self.dap.handle(command, arguments)

    def test_initialize_attach_and_frame(self):
        capabilities, events = self.call('initialize')
        self.assertTrue(capabilities['supportsFunctionBreakpoints'])
        self.assertFalse(capabilities['supportsDisassembleRequest'])
        self.assertEqual(events, [])
        attached, events = self.call('attach')
        self.assertEqual(attached['buildId'], 'abc')
        self.assertEqual(events, [('initialized', {})])
        stack, _ = self.call('stackTrace')
        frame = stack['stackFrames'][0]
        self.assertEqual(frame['source']['path'], '/src/main.c')
        self.assertEqual(frame['instructionPointerReference'], '0x8100')

    def test_breakpoints_variables_and_restricted_evaluate(self):
        self.call('attach')
        body, _ = self.call('setBreakpoints', source={'path': '/src/main.c'},
                            breakpoints=[{'line': 3, 'logMessage': 'total={total}'}])
        self.assertEqual(body['breakpoints'][0]['id'], 7)
        self.assertIn(('set_source_breakpoints', {'file': '/src/main.c',
            'breakpoints': [{'line': 3, 'logMessage': 'total={total}'}]}), self.rpc.calls)
        variables, _ = self.call('variables', variablesReference=2)
        self.assertEqual(variables['variables'][0]['value'], '42')
        evaluated, _ = self.call('evaluate', expression='total')
        self.assertEqual(evaluated['result'], '42')
        with self.assertRaisesRegex(SessionError, 'только имя'):
            self.call('evaluate', expression='total+1')

    def test_source_and_instruction_steps_use_distinct_rpc(self):
        self.call('attach')
        self.call('stepIn')
        self.assertIn(('source_step', {'kind': 'into'}), self.rpc.calls)
        self.call('next')
        self.assertIn(('source_step', {'kind': 'over'}), self.rpc.calls)
        self.call('stepOut')
        self.assertIn(('source_step', {'kind': 'out'}), self.rpc.calls)
        body, events = self.call('stepIn', granularity='instruction')
        self.assertEqual((body, events), ({}, []))
        self.assertIn(('step', {}), self.rpc.calls)

    def test_event_ring_gap_warns_debug_console(self):
        self.dap.event_sequence = 4
        self.dap._rpc_override = lambda method, args: {
            'events': [{'seq': 7, 'event': 'output',
                        'body': {'category': 'console', 'output': 'total=42\n'}}],
            'first': 7, 'last': 7, 'closed': False,
        }
        events, closed = self.dap.poll_events(timeout=0)
        self.assertFalse(closed)
        self.assertEqual(events[0][0], 'output')
        self.assertIn('пропущено 2', events[0][1]['output'])
        self.assertEqual(events[1][1]['output'], 'total=42\n')

    def test_stdio_framing_without_attach(self):
        requests = [
            {'seq': 1, 'type': 'request', 'command': 'initialize', 'arguments': {}},
            {'seq': 2, 'type': 'request', 'command': 'disconnect', 'arguments': {}},
        ]
        data = b''
        for request in requests:
            body = json.dumps(request).encode()
            data += f'Content-Length: {len(body)}\r\n\r\n'.encode() + body
        process = subprocess.run([sys.executable, str(ROOT/'toolchain/sdbg_dap.py')],
                                 input=data, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        self.assertEqual(process.returncode, 0, process.stderr.decode())
        messages = []
        output = process.stdout
        while output:
            header, output = output.split(b'\r\n\r\n', 1)
            length = int(header.split(b':', 1)[1])
            body, output = output[:length], output[length:]
            messages.append(json.loads(body))
        self.assertEqual([item['type'] for item in messages],
                         ['response', 'response'])

if __name__ == '__main__':
    unittest.main()