From: Christian Heller Date: Wed, 28 Jan 2026 17:11:02 +0000 (+0100) Subject: Clean up testing output. X-Git-Url: https://plomlompom.com/repos/booking/%7B%7B%20web_path%20%7D%7D/decks/%7B%7Bdb.prefix%7D%7D/edit?a=commitdiff_plain;h=c4eaf8457eff8ff600715611f2f8ddd38d5e0bf3;p=ledgplom Clean up testing output. --- diff --git a/src/ledgplom/testing.py b/src/ledgplom/testing.py index 5a7263b..2f52606 100644 --- a/src/ledgplom/testing.py +++ b/src/ledgplom/testing.py @@ -1,11 +1,12 @@ 'To run tests.' # built-ins -from sys import exit as sys_exit +from sys import exception as sys_exception from pathlib import Path +import threading # so we can overwrite .excepthook from threading import Thread from typing import Optional # requirements.txt -from playwright.sync_api import sync_playwright +from playwright.sync_api import sync_playwright as pw_sync, Error as pw_error # ourselves from ledgplom.http import Server, SERVER_PORT from plomlib.web import PlomHttpServer @@ -18,60 +19,93 @@ _TAGS_EXPANDED = {'', ''} def run_tests(selector: str) -> None: 'Run tests from tests directory.' + abort_early = [False] + + class TestServer(Server): + 'Server variant suppressing redundant error tracebacks.' + _tracebacked_already = set() + + def handle_error(self, request, client_address): + id_exception = id(sys_exception()) + if id_exception in self._tracebacked_already: + return + self._tracebacked_already.add(id_exception) + super().handle_error(request, client_address) + + class SilencedException(Exception): + 'Not to be re-raised by …' + + def abort_early_and_ignore_silenced(exc): + 'Set abort_early[0], only handle further if not SilencedException.' + abort_early[0] = True + if exc.exc_type != SilencedException: + threading.__excepthook__(exc) + + threading.excepthook = abort_early_and_ignore_silenced + paths = tuple(p for p in _PATH_TESTS.iterdir() if p.parts[-1].startswith(selector) or ('.' in selector and selector.split('.')[0] + _EXT_DAT == p.parts[-1])) def run_tests_on_dat(dat_path: Path, server: PlomHttpServer) -> None: - def fail(abort_msg: str, msg_prefix: str, idx: Optional[int]) -> None: for jdx, line in enumerate(lines_expected[:idx], start=1): print(f'{jdx}: [{line}]') print(f'{msg_prefix} FAILED – {abort_msg}') - sys_exit(1) + raise SilencedException - test_name = str(dat_path.parts[-1])[:-len(_EXT_DAT)] - pw = sync_playwright().start() - page = pw.firefox.launch().new_page() - for test_path in [p for p in paths - if p != dat_path - and p.parts[-1].startswith(f'{test_name}.')]: - page.goto(f'http://localhost:{SERVER_PORT}/' - + str(test_path.parts[-1]).split('.', maxsplit=1 - )[1].replace('.', '/')) - lines_rendered = page.content().split('\n') - with test_path.open('r', encoding='utf8') as f: - lines_expected = tuple(line.rstrip('\n') - for line in f.readlines()) - msg_prefix = f'test for {test_path}:' - in_expansion = False - for idx0, line in enumerate(lines_expected): - idx1 = idx0 + 1 - if line in _TAGS_EXPANDED: - lines_rendered[idx0:idx0] = [line] - in_expansion = line[1] != '/' - if in_expansion: - lines_rendered[idx1:idx1+1]\ + try: + pw = pw_sync().start() + page = pw.firefox.launch().new_page() + test_name = str(dat_path.parts[-1])[:-len(_EXT_DAT)] + for test_path in [p for p in paths + if p != dat_path + and p.parts[-1].startswith(f'{test_name}.')]: + print("TESTING:", test_path) + page.goto(f'http://localhost:{SERVER_PORT}/' + + str(test_path.parts[-1] + ).split('.', maxsplit=1)[1].replace('.', '/')) + lines_rendered = page.content().split('\n') + with test_path.open('r', encoding='utf8') as f: + lines_expected = tuple(line.rstrip('\n') + for line in f.readlines()) + msg_prefix = f'TEST FOR {test_path}' + in_expansion = False + for idx0, line in enumerate(lines_expected): + idx1 = idx0 + 1 + if line in _TAGS_EXPANDED: + lines_rendered[idx0:idx0] = [line] + in_expansion = line[1] != '/' + if in_expansion: + lines_rendered[idx1:idx1+1]\ = lines_rendered[idx1].replace('><', '>\n<' ).splitlines() - continue - abort_msg = '' - if idx1 > len(lines_rendered): - abort_msg = 'more lines expected' - elif lines_rendered[idx0] != (line.lstrip() if in_expansion - else line): - abort_msg = ('line differs, found instead: ' - f'[{lines_rendered[idx0]}]') - if abort_msg: - fail(abort_msg, msg_prefix, idx1) - if len(lines_rendered) > idx1: - fail('more lines rendered', msg_prefix, None) - print(f'{msg_prefix} passed') - pw.stop() - server.shutdown() + continue + abort_msg = '' + if idx1 > len(lines_rendered): + abort_msg = 'more lines expected' + elif lines_rendered[idx0] != (line.lstrip() if in_expansion + else line): + abort_msg = ( + 'expected line differs, rendered instead: ' + f'[{lines_rendered[idx0]}]') + if abort_msg: + fail(abort_msg, msg_prefix, idx1) + if len(lines_rendered) > idx1: + fail('more lines rendered', msg_prefix, None) + print(f'{msg_prefix} PASSED') + except pw_error as e: + print("PLAYWRIGHT FAILED:", e) + raise SilencedException from e + finally: + server.shutdown() for dat_path in [p for p in paths if p.parts[-1].endswith(_EXT_DAT)]: - server = Server(dat_path) - Thread(target=run_tests_on_dat, args=(dat_path, server,)).start() + server = TestServer(dat_path) + t = Thread(target=run_tests_on_dat, args=(dat_path, server,)) + t.start() server.serve() + t.join() + if abort_early[0]: + break