diff --git a/tests/const/test_const_apptype_split_unit.py b/tests/const/test_const_apptype_split_unit.py index c516140ba3..86fbc9d985 100644 --- a/tests/const/test_const_apptype_split_unit.py +++ b/tests/const/test_const_apptype_split_unit.py @@ -1072,11 +1072,13 @@ def test_no_lookup_call_site_in_the_library_builds_a_composite_proto(self) -> No path = root / relative with self.subTest(module=relative): self.assertTrue(path.is_file(), path) - tree = ast.parse(path.read_text(encoding='utf-8'), filename=str(path)) + text = path.read_text(encoding='utf-8') + tree = ast.parse(text, filename=str(path)) + # Each node is read back as written, since ``ast.unparse`` is 3.9+. for node in ast.walk(tree): # A ``TransportProtocol`` composite, however it is spelled. if isinstance(node, ast.BinOp) and isinstance(node.op, ast.BitOr): - if 'TransportProtocol' in ast.unparse(node): + if 'TransportProtocol' in ast.get_source_segment(text, node): composites.append((relative, node.lineno)) # What ``proto`` is bound to, for the port field that uses a # local rather than passing the member straight in, and what @@ -1088,9 +1090,9 @@ def test_no_lookup_call_site_in_the_library_builds_a_composite_proto(self) -> No or relative.endswith('numbers.py')): for target in node.targets: if isinstance(target, ast.Name) and target.id == 'proto': - bindings.append((relative, ast.unparse(node.value))) + bindings.append((relative, ast.get_source_segment(text, node.value))) if isinstance(target, ast.Attribute) and target.attr == '_proto': - stored.append((relative, ast.unparse(node.value))) + stored.append((relative, ast.get_source_segment(text, node.value))) if not isinstance(node, ast.Call): continue name = (node.func.attr if isinstance(node.func, ast.Attribute) else @@ -1102,7 +1104,8 @@ def test_no_lookup_call_site_in_the_library_builds_a_composite_proto(self) -> No if name in ('_dispatch', '_make_port') and len(node.args) > 1: arguments.append(node.args[1]) for argument in arguments: - lookups.append('%s %s(proto=%s)' % (relative, name, ast.unparse(argument))) + lookups.append('%s %s(proto=%s)' % (relative, name, + ast.get_source_segment(text, argument))) self.assertEqual(composites, []) diff --git a/tests/dumpkit/test_dumper_roundtrip_unit.py b/tests/dumpkit/test_dumper_roundtrip_unit.py index 0226624405..43607d3079 100644 --- a/tests/dumpkit/test_dumper_roundtrip_unit.py +++ b/tests/dumpkit/test_dumper_roundtrip_unit.py @@ -20,23 +20,64 @@ """ +import contextlib import datetime import importlib.util import ipaddress import json +import logging import os import plistlib import struct import tempfile import unittest import warnings +from typing import TYPE_CHECKING from tests._support import reimport_once_per_class +if TYPE_CHECKING: + from typing import Iterator + RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) +class _Recorder(logging.Handler): + """A handler that keeps the records instead of formatting them.""" + + def __init__(self) -> None: + super().__init__() + self.records = [] # type: list[logging.LogRecord] + + def emit(self, record: 'logging.LogRecord') -> None: + self.records.append(record) + + +@contextlib.contextmanager +def assert_no_logs(test: 'unittest.TestCase', name: 'str', level: 'int') -> 'Iterator[None]': + """:meth:`unittest.TestCase.assertNoLogs`, which is 3.10+, on every supported Python. + + Takes ``name`` over as that method does -- its own handler only, ``level`` as + the threshold, no propagation -- and fails if the block logged anything. + + """ + logger = logging.getLogger(name) + recorder = _Recorder() + handlers, threshold, propagate = logger.handlers[:], logger.level, logger.propagate + logger.handlers = [recorder] + logger.setLevel(level) + logger.propagate = False + try: + yield + finally: + logger.handlers = handlers + logger.setLevel(threshold) + logger.propagate = propagate + test.assertEqual([(record.levelname, record.getMessage()) for record in recorder.records], [], + f'unexpected logs on {name!r}') + + def pcap_file(major: 'int', minor: 'int') -> 'bytes': """A little-endian PCAP file of the given version holding one Ethernet frame.""" frame = bytes.fromhex('0123456789ab' 'fedcba987654' '88b5') + b'payload' @@ -98,7 +139,7 @@ def test_unhandled_value_types_get_an_explicit_rendering(self) -> None: 'nl': NULL, 'tz': datetime.timezone(datetime.timedelta(hours=8)), } - with self.assertNoLogs('pcapkit', level='WARNING'): + with assert_no_logs(self, 'pcapkit', logging.WARNING): got = json.loads(self.dump('JSON', value))['x'] self.assertEqual(got, {'net4': '10.0.0.0/8', 'net6': '::/0', 'nv': '', 'nl': '', 'tz': 28800.0}) diff --git a/tests/foundation/reassembly/test_reassembly_base.py b/tests/foundation/reassembly/test_reassembly_base.py index 3807201f70..5f96e9cd92 100644 --- a/tests/foundation/reassembly/test_reassembly_base.py +++ b/tests/foundation/reassembly/test_reassembly_base.py @@ -3,6 +3,7 @@ import importlib.util import sys import unittest +from typing import Tuple from unittest import mock from tests._support import reimport_once_per_class @@ -36,7 +37,7 @@ class DummyBuffer(Info): class DummyProtocol: pass - class DummyReassembly(ReassemblyBase[DummyPacket, DummyDatagram, tuple[str], DummyBuffer]): + class DummyReassembly(ReassemblyBase[DummyPacket, DummyDatagram, Tuple[str], DummyBuffer]): __protocol_name__ = 'Dummy' __protocol_type__ = DummyProtocol @@ -99,7 +100,7 @@ class DummyDatagram(Info): class DummyBuffer(Info): entries: list[int] - class BaseFallback(ReassemblyBase[DummyPacket, DummyDatagram, tuple[str], DummyBuffer]): + class BaseFallback(ReassemblyBase[DummyPacket, DummyDatagram, Tuple[str], DummyBuffer]): def reassembly(self, info: DummyPacket) -> None: super().reassembly(info) @@ -146,7 +147,7 @@ class DummyBuffer(Info): entries: list[int] with mock.patch('pcapkit.foundation.extraction.Extractor.register_reassembly') as register: - class Explicit(Reassembly[DummyPacket, DummyDatagram, tuple[str], DummyBuffer], + class Explicit(Reassembly[DummyPacket, DummyDatagram, Tuple[str], DummyBuffer], protocol='ExampleProto'): def reassembly(self, info: DummyPacket) -> None: super().reassembly(info) @@ -158,7 +159,7 @@ def submit(self, buf: DummyBuffer, **kwargs: object) -> list[DummyDatagram]: self.assertEqual(Explicit.__callback_fn__, []) with mock.patch('pcapkit.foundation.extraction.Extractor.register_reassembly') as register: - class Default(Reassembly[DummyPacket, DummyDatagram, tuple[str], DummyBuffer]): + class Default(Reassembly[DummyPacket, DummyDatagram, Tuple[str], DummyBuffer]): __protocol_name__ = 'DefaultProto' def reassembly(self, info: DummyPacket) -> None: @@ -209,7 +210,7 @@ class DummyBuffer(Info): with mock.patch('pcapkit.foundation.extraction.Extractor.register_reassembly') as register: with self.assertRaises(UnsupportedCall) as caught: - class Typo(Reassembly[DummyPacket, DummyDatagram, tuple[str], DummyBuffer], + class Typo(Reassembly[DummyPacket, DummyDatagram, Tuple[str], DummyBuffer], reassembly='wrong-keyword-for-reassembly'): def reassembly(self, info: DummyPacket) -> None: super().reassembly(info) @@ -227,7 +228,7 @@ def submit(self, buf: DummyBuffer, **kwargs: object) -> list[DummyDatagram]: expected = UnsupportedCall if sys.version_info >= (3, 11) else TypeError with mock.patch('pcapkit.foundation.extraction.Extractor.register_reassembly') as register: with self.assertRaises(expected): - class Collides(Reassembly[DummyPacket, DummyDatagram, tuple[str], DummyBuffer], + class Collides(Reassembly[DummyPacket, DummyDatagram, Tuple[str], DummyBuffer], name='collides-with-ABCMeta-on-3.10'): def reassembly(self, info: DummyPacket) -> None: super().reassembly(info) diff --git a/tests/foundation/test_extraction.py b/tests/foundation/test_extraction.py index 9ab71b6a58..43d2dce602 100644 --- a/tests/foundation/test_extraction.py +++ b/tests/foundation/test_extraction.py @@ -189,7 +189,8 @@ def test_declared_engine_type_is_satisfied_by_the_built_in_engines(self) -> None # ``_exeng`` is annotated under ``TYPE_CHECKING`` only, so it is not visible # to ``get_type_hints``: read the annotation out of the module source. - tree = ast.parse(inspect.getsource(module)) + text = inspect.getsource(module) + tree = ast.parse(text) klass = next(node for node in tree.body if isinstance(node, ast.ClassDef) and node.name == 'Extractor') found = [node for node in ast.walk(klass) @@ -197,8 +198,9 @@ def test_declared_engine_type_is_satisfied_by_the_built_in_engines(self) -> None and isinstance(node.target, ast.Name) and node.target.id == '_exeng'] self.assertEqual(len(found), 1) node = found[0].annotation - # quoted (``'EngineBase[_P]'``) or bare (``EngineBase[_P]``) spelling - source = node.value if isinstance(node, ast.Constant) else ast.unparse(node) + # quoted (``'EngineBase[_P]'``) or bare (``EngineBase[_P]``) spelling; the + # bare one is read back as written, since ``ast.unparse`` is 3.9+ + source = node.value if isinstance(node, ast.Constant) else ast.get_source_segment(text, node) attribute = eval(source, vars(module)) # pylint: disable=eval-used declared = { @@ -245,7 +247,7 @@ def read_frame(self) -> str: self.addCleanup(lambda: sys.modules.pop('unit_extraction_engine_mod', None)) with mock.patch('pcapkit.foundation.extraction.Extractor.register_reassembly'): - class UnitReassembly(Reassembly[object, object, tuple[str], object], + class UnitReassembly(Reassembly[object, object, typing.Tuple[str], object], protocol='unit-reassembly'): def reassembly(self, info: object) -> None: pass @@ -344,14 +346,14 @@ def run(self) -> None: def read_frame(self) -> str: return 'frame' - class BaseOnlyReassembly(ReassemblyBase[object, object, tuple[str], object]): + class BaseOnlyReassembly(ReassemblyBase[object, object, typing.Tuple[str], object]): def reassembly(self, info: object) -> None: pass def submit(self, buf: object, **kwargs: object) -> list[object]: return [] - class PublicReassembly(Reassembly[object, object, tuple[str], object]): + class PublicReassembly(Reassembly[object, object, typing.Tuple[str], object]): def reassembly(self, info: object) -> None: pass @@ -524,7 +526,7 @@ def test_register_reassembly_identity_guard(self) -> None: from pcapkit.utilities.warnings import RegistryWarning with mock.patch('pcapkit.foundation.extraction.Extractor.register_reassembly'): - class UnitReassemblyIdentityA(Reassembly[object, object, tuple[str], object], + class UnitReassemblyIdentityA(Reassembly[object, object, typing.Tuple[str], object], protocol='unit-reassembly-identity-a'): def reassembly(self, info: object) -> None: pass @@ -532,7 +534,7 @@ def reassembly(self, info: object) -> None: def submit(self, buf: object, **kwargs: object) -> list[object]: return [] - class UnitReassemblyIdentityB(Reassembly[object, object, tuple[str], object], + class UnitReassemblyIdentityB(Reassembly[object, object, typing.Tuple[str], object], protocol='unit-reassembly-identity-b'): def reassembly(self, info: object) -> None: pass diff --git a/tests/project/test_documentation_claims.py b/tests/project/test_documentation_claims.py index b738a94144..3871974cdd 100644 --- a/tests/project/test_documentation_claims.py +++ b/tests/project/test_documentation_claims.py @@ -95,13 +95,15 @@ def test_documented_sense_matches_the_code(self) -> None: f'no_eof is documented as {line!r}, but the flag suppresses ' 'the EOF stop rather than raising') - tree = ast.parse(textwrap.dedent(inspect.getsource(Extractor.record_frames))) + text = textwrap.dedent(inspect.getsource(Extractor.record_frames)) + tree = ast.parse(text) # Find the handler for EOFError and confirm it can `continue` on the flag. + # Each node is read back as written, since ``ast.unparse`` is 3.9+. handlers = [node for node in ast.walk(tree) if isinstance(node, ast.ExceptHandler)] eof_handlers = [ h for h in handlers - if h.type is not None and 'EOFError' in ast.unparse(h.type) + if h.type is not None and 'EOFError' in ast.get_source_segment(text, h.type) ] self.assertTrue(eof_handlers, 'record_frames no longer handles EOFError') @@ -109,7 +111,7 @@ def test_documented_sense_matches_the_code(self) -> None: for handler in eof_handlers: for node in ast.walk(handler): if (isinstance(node, ast.If) - and '_flag_n' in ast.unparse(node.test) + and '_flag_n' in ast.get_source_segment(text, node.test) and any(isinstance(inner, ast.Continue) for inner in node.body)): continues_on_flag = True self.assertTrue( diff --git a/tests/project/test_module_isolation.py b/tests/project/test_module_isolation.py index 1d5a5fa7d1..7fd6433603 100644 --- a/tests/project/test_module_isolation.py +++ b/tests/project/test_module_isolation.py @@ -178,12 +178,27 @@ def run_pytest(selection: 'list[str]') -> 'subprocess.CompletedProcess[str]': """ return subprocess.run( - [sys.executable, '-m', 'pytest', '-p', 'no:cacheprovider', '-q', *selection], + [sys.executable, '-m', 'pytest', '-p', 'no:cacheprovider', '-q', '-rfE', *selection], cwd=str(ROOT), env=child_environ(), capture_output=True, text=True, timeout=600, check=False, ) +def nested_failures(stdout: str) -> 'list[str]': + """The tests a :func:`run_pytest` run's short summary reports as failed or errored. + + Args: + stdout: The run's standard output, which ends with that summary because + :func:`run_pytest` passes ``-rfE``. + + Returns: + Their node ids, without the reason pytest appends after `` - ``. + + """ + return [line.split(' ', 1)[1].split(' - ', 1)[0] + for line in stdout.splitlines() if line.startswith(('FAILED ', 'ERROR '))] + + def run_unittest(modules: 'list[str]') -> 'subprocess.CompletedProcess[str]': """Run the stdlib :mod:`unittest` runner over ``modules`` in a subprocess. @@ -217,13 +232,21 @@ class OrderIndependenceTests(unittest.TestCase): """Each selection that used to fail on collection order now passes.""" def assert_selection_passes(self, label: str, selection: 'list[str]') -> None: - """Run ``selection`` and fail with its output if pytest did not exit 0.""" + """Run ``selection`` and fail with its output if pytest did not exit 0. + + The message names the nested tests that failed rather than blaming + collection order: a test that fails on its own fails here too, and that + is not #660. + + """ finished = run_pytest(selection) + failed = nested_failures(finished.stdout) or ['none named in the summary'] self.assertEqual( finished.returncode, 0, - f'pytest exited {finished.returncode} on {label} -- the result still ' - f'depends on collection order (#660).\n\n' + f'pytest exited {finished.returncode} on {label}; failing: ' + f'{", ".join(failed)}. That is a collection-order dependence (#660) ' + f'only if each of them passes when run on its own.\n\n' f'selection: {" ".join(selection)}\n\n' f'stdout:\n{finished.stdout[-4000:]}\n\nstderr:\n{finished.stderr[-2000:]}') self.assertNotIn('is not subscriptable', finished.stdout, diff --git a/tests/test_final_enforcement.py b/tests/test_final_enforcement.py index 4fa7beb0ae..810134dab2 100644 --- a/tests/test_final_enforcement.py +++ b/tests/test_final_enforcement.py @@ -576,6 +576,8 @@ def test_the_shim_switches_at_the_version_that_added_the_dunder(self) -> None: Derived from the module's own ``if``, not matched against its text, so reformatting the guard does not fail this and moving the boundary does. + The guard's tree is compared with :func:`ast.dump`, which every supported + version has, rather than rendered with :func:`ast.unparse`, which is 3.9+. """ import ast @@ -585,7 +587,7 @@ def test_the_shim_switches_at_the_version_that_added_the_dunder(self) -> None: source = pathlib.Path(compat.__file__).read_text(encoding='utf-8') boundaries = [ - ast.unparse(node.test) + ast.dump(node.test) for node in ast.walk(ast.parse(source)) if isinstance(node, ast.If) for stmt in node.body @@ -593,7 +595,7 @@ def test_the_shim_switches_at_the_version_that_added_the_dunder(self) -> None: and any(alias.name == 'final' for alias in stmt.names) ] self.assertEqual( - boundaries, ['sys.version_info < (3, 11)'], + boundaries, [ast.dump(ast.parse('sys.version_info < (3, 11)', mode='eval').body)], 'pcapkit.utilities.compat must take `final` from typing_extensions below ' '3.11, because that is the version at which typing.final began recording ' '__final__ -- the marker Info.__init_subclass__ and '