Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 8 additions & 5 deletions tests/const/test_const_apptype_split_unit.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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, [])

Expand Down
43 changes: 42 additions & 1 deletion tests/dumpkit/test_dumper_roundtrip_unit.py
Original file line number Diff line number Diff line change
Expand Up @@ -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'
Expand Down Expand Up @@ -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': '<NO_VALUE>',
'nl': '<NULL>', 'tz': 28800.0})
Expand Down
13 changes: 7 additions & 6 deletions tests/foundation/reassembly/test_reassembly_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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)

Expand Down Expand Up @@ -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)
Expand All @@ -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:
Expand Down Expand Up @@ -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)
Expand All @@ -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)
Expand Down
18 changes: 10 additions & 8 deletions tests/foundation/test_extraction.py
Original file line number Diff line number Diff line change
Expand Up @@ -189,16 +189,18 @@ 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)
if isinstance(node, ast.AnnAssign)
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 = {
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -524,15 +526,15 @@ 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

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
Expand Down
8 changes: 5 additions & 3 deletions tests/project/test_documentation_claims.py
Original file line number Diff line number Diff line change
Expand Up @@ -95,21 +95,23 @@ 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')

continues_on_flag = False
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(
Expand Down
31 changes: 27 additions & 4 deletions tests/project/test_module_isolation.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down Expand Up @@ -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,
Expand Down
6 changes: 4 additions & 2 deletions tests/test_final_enforcement.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -585,15 +587,15 @@ 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
if isinstance(stmt, ast.ImportFrom) and stmt.module == 'typing_extensions'
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 '
Expand Down
Loading