105 lines
5.1 KiB
Python
105 lines
5.1 KiB
Python
from dataclasses import replace
|
|
from io import BytesIO
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from logic_analyzer import dslogic
|
|
from logic_analyzer.files import ImportCancelled
|
|
|
|
|
|
class DSLogicCoreTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.options = dslogic.DSLogicSettings('dsview:2.6', channels=(0, 9), sample_rate=1000000)
|
|
self.metadata = dict(channels=[0,9], samples=70, sample_rate=1000000, format='cross64-le')
|
|
|
|
def test_cross_words_preserve_physical_channels_boundary_edges_and_tail(self):
|
|
raw = struct.pack('<QQQQ', 1 << 63, (1 << 64)-1, 1, 0)
|
|
capture = dslogic.read_cross(BytesIO(raw), self.metadata, self.options)
|
|
self.assertEqual(capture.end, 70 / 1000000)
|
|
self.assertEqual([c.name for c in capture.channels], ['D0', 'D9'])
|
|
self.assertEqual([c.initial for c in capture.channels], [0, 1])
|
|
self.assertEqual([list(c.edges) for c in capture.channels], [[63e-6, 65e-6], [64e-6]])
|
|
|
|
def test_padding_is_not_imported_as_samples(self):
|
|
meta = dict(self.metadata, samples=3)
|
|
capture = dslogic.read_cross(BytesIO(struct.pack('<QQ', 1<<63, 0)), meta, self.options)
|
|
self.assertEqual(capture.end, 3e-6)
|
|
self.assertEqual(len(capture.channels[0].edges), 0)
|
|
|
|
def test_empty_truncated_wrong_mapping_and_cancel(self):
|
|
for raw in (b'', b'\0'*31):
|
|
with self.assertRaises(ValueError):
|
|
dslogic.read_cross(BytesIO(raw), self.metadata, self.options)
|
|
with self.assertRaises(ValueError):
|
|
dslogic.read_cross(BytesIO(b'\0'*32), dict(self.metadata, channels=[0,1]), self.options)
|
|
with self.assertRaises(ImportCancelled):
|
|
dslogic.read_cross(BytesIO(b'\0'*32), self.metadata, self.options, lambda: True)
|
|
|
|
def test_validation_rejects_old_driver_and_invalid_parameters(self):
|
|
for changes in (dict(device_id='dreamsourcelab-dslogic'), dict(channels=()),
|
|
dict(channels=(0,0)), dict(duration=float('nan')), dict(sample_rate=0)):
|
|
with self.assertRaises(ValueError):
|
|
replace(self.options, **changes).validate()
|
|
|
|
def test_host_does_not_inherit_frozen_gui_bootloader_state(self):
|
|
with patch.dict(os.environ, {'_MEIPASS2': 'gui-runtime', '_PYI_APPLICATION_HOME_DIR': 'gui-runtime'}):
|
|
with patch.object(dslogic.subprocess, 'Popen') as launch:
|
|
dslogic._popen(['host', 'request'])
|
|
environment = launch.call_args.kwargs['env']
|
|
self.assertNotIn('_MEIPASS2', environment)
|
|
self.assertNotIn('_PYI_APPLICATION_HOME_DIR', environment)
|
|
|
|
def launch(self, code):
|
|
self.processes = []
|
|
def popen(arguments, **kwargs):
|
|
process = subprocess.Popen([sys.executable, '-c', code, arguments[1]], **kwargs)
|
|
self.processes.append(process)
|
|
return process
|
|
self.addCleanup(lambda: [p.kill() for p in self.processes if p.poll() is None])
|
|
return patch.object(dslogic, '_popen', side_effect=popen)
|
|
|
|
def test_real_process_scan(self):
|
|
code = "import sys,json; from pathlib import Path; p=Path(sys.argv[1]); (p/'result.json').write_text(json.dumps({'devices':[{'device_id':'dsview:2.6','name':'DSLogic Plus','channels':[0,9]}]}))"
|
|
with self.launch(code):
|
|
devices = dslogic.list_devices()
|
|
self.assertEqual(devices[0].channels, (0,9))
|
|
self.assertEqual(devices[0].device_id, 'dsview:2.6')
|
|
|
|
def test_stop_is_graceful_and_imports_result(self):
|
|
code = '''import sys,time,json,struct
|
|
from pathlib import Path
|
|
p=Path(sys.argv[1]); (p/'ready').touch()
|
|
while not (p/'stop').exists(): time.sleep(.01)
|
|
(p/'capture.bin').write_bytes(struct.pack('<QQ',6,0))
|
|
(p/'result.json').write_text(json.dumps(dict(channels=[0,9],samples=3,sample_rate=1000000,format='cross64-le')))
|
|
'''
|
|
with self.launch(code):
|
|
capture = dslogic.acquire(self.options, lambda: True, lambda: False)
|
|
self.assertEqual(capture.end, 3e-6)
|
|
self.assertEqual(self.processes[0].returncode, 0)
|
|
|
|
def test_cancel_reaps_process_and_discards_result(self):
|
|
start = time.monotonic()
|
|
with self.launch('import time; time.sleep(30)'):
|
|
with self.assertRaises(ImportCancelled):
|
|
dslogic.acquire(self.options, lambda: False, lambda: time.monotonic()-start > .1)
|
|
self.assertIsNotNone(self.processes[0].poll())
|
|
|
|
def test_native_failure_does_not_hang(self):
|
|
with self.launch('import sys; sys.stderr.write("native failure"); sys.exit(7)'):
|
|
with self.assertRaisesRegex(RuntimeError, '7'):
|
|
dslogic.acquire(self.options, lambda: False, lambda: False)
|
|
|
|
def test_error_result_is_not_a_capture(self):
|
|
code = "import sys,json; from pathlib import Path; (Path(sys.argv[1])/'result.json').write_text(json.dumps({'error':'USB disconnected'})); sys.exit(1)"
|
|
with self.launch(code):
|
|
with self.assertRaisesRegex(RuntimeError, 'USB disconnected'):
|
|
dslogic.acquire(self.options, lambda: False, lambda: False)
|