diff --git a/pcapkit/foundation/engines/dpkt.py b/pcapkit/foundation/engines/dpkt.py index 508f5bdae..37edce47e 100644 --- a/pcapkit/foundation/engines/dpkt.py +++ b/pcapkit/foundation/engines/dpkt.py @@ -458,7 +458,7 @@ def run(self) -> 'None': if ext._flag_v: from pcapkit.toolkit.dpkt import packet2chain # isort:skip ext._vfunc = lambda e, f: print( - f'Frame {e._frnum:>3d}: {packet2chain(f)}' # pylint: disable=protected-access + f'Frame {e._frnum:>3d}: {packet2chain(cast("DPKTPacket", f))}' # pylint: disable=protected-access ) # extract global header diff --git a/pcapkit/foundation/engines/pcap_ct.py b/pcapkit/foundation/engines/pcap_ct.py index 51f42ef19..5e1f81df3 100644 --- a/pcapkit/foundation/engines/pcap_ct.py +++ b/pcapkit/foundation/engines/pcap_ct.py @@ -403,7 +403,7 @@ def run(self) -> 'None': if ext._flag_v: from pcapkit.toolkit.pcap_ct import packet2chain # isort:skip ext._vfunc = lambda e, f: print( - f'Frame {e._frnum:>3d}: {packet2chain(f[1], data_link=self._dlink)}' # pylint: disable=protected-access + f'Frame {e._frnum:>3d}: {packet2chain(cast("RawFrame", f)[1], data_link=self._dlink)}' # pylint: disable=protected-access ) # pylint: disable=logging-fstring-interpolation # extract & analyse file diff --git a/pcapkit/foundation/engines/pypcap.py b/pcapkit/foundation/engines/pypcap.py index 2121d9782..6eebc29fa 100644 --- a/pcapkit/foundation/engines/pypcap.py +++ b/pcapkit/foundation/engines/pypcap.py @@ -316,7 +316,7 @@ def run(self) -> 'None': if ext._flag_v: from pcapkit.toolkit.pypcap import packet2chain # isort:skip ext._vfunc = lambda e, f: print( - f'Frame {e._frnum:>3d}: {packet2chain(f[1], data_link=self._dlink)}' # pylint: disable=protected-access + f'Frame {e._frnum:>3d}: {packet2chain(cast("RawFrame", f)[1], data_link=self._dlink)}' # pylint: disable=protected-access ) # pylint: disable=logging-fstring-interpolation # extract & analyse file diff --git a/pcapkit/foundation/engines/pypcapfile.py b/pcapkit/foundation/engines/pypcapfile.py index 42a28a6f7..75c5a118e 100644 --- a/pcapkit/foundation/engines/pypcapfile.py +++ b/pcapkit/foundation/engines/pypcapfile.py @@ -224,7 +224,7 @@ def run(self) -> 'None': if ext._flag_v: from pcapkit.toolkit.pypcapfile import packet2chain # isort:skip ext._vfunc = lambda e, f: print( - f'Frame {e._frnum:>3d}: {packet2chain(f, data_link=self._dlink)}' # pylint: disable=protected-access + f'Frame {e._frnum:>3d}: {packet2chain(cast("PCAPFilePacket", f), data_link=self._dlink)}' # pylint: disable=protected-access ) # pylint: disable=logging-fstring-interpolation # extract & analyse file diff --git a/pcapkit/foundation/engines/pyshark.py b/pcapkit/foundation/engines/pyshark.py index 2113ce1ef..7a3dbbb31 100644 --- a/pcapkit/foundation/engines/pyshark.py +++ b/pcapkit/foundation/engines/pyshark.py @@ -259,7 +259,7 @@ def run(self) -> 'None': if ext._flag_v: ext._vfunc = lambda e, f: print( f'Frame {e._frnum:>3d}: ' # pylint: disable=protected-access - f'{cast("BaseLayer", _frame_layer(f)).protocols}' + f'{cast("BaseLayer", _frame_layer(cast("PySharkPacket", f))).protocols}' ) # extract global header diff --git a/pcapkit/foundation/engines/scapy.py b/pcapkit/foundation/engines/scapy.py index 4509cc30b..0f406757c 100644 --- a/pcapkit/foundation/engines/scapy.py +++ b/pcapkit/foundation/engines/scapy.py @@ -387,7 +387,7 @@ def run(self) -> 'None': if ext._flag_v: from pcapkit.toolkit.scapy import packet2chain # isort:skip ext._vfunc = lambda e, f: print( - f'Frame {e._frnum:>3d}: {packet2chain(f)}' # pylint: disable=protected-access + f'Frame {e._frnum:>3d}: {packet2chain(cast("ScapyPacket", f))}' # pylint: disable=protected-access ) # extract global header diff --git a/pcapkit/foundation/extraction.py b/pcapkit/foundation/extraction.py index dd4acb10b..b7feaf080 100644 --- a/pcapkit/foundation/extraction.py +++ b/pcapkit/foundation/extraction.py @@ -85,6 +85,9 @@ PCAPFilePacket, tuple[float, bytes]] Protocols = Union[str, ProtocolBase, Type[ProtocolBase]] + # NOTE: every caller discards the callback's result, so it is typed ``Any``: + # any two-argument callback is accepted, whatever it returns, e.g. + # ``lambda e, f: sys.stdout.write(str(f))``. VerboseHandler = Callable[['Extractor', Packet], Any] __all__ = ['Extractor'] @@ -178,7 +181,7 @@ class Extractor(Generic[_P]): _flag_s: 'bool' #: Verbose callback function. - #_vfunc: 'VerboseHandler' + _vfunc: 'VerboseHandler' #: Frame number. _frnum: 'int' @@ -1284,9 +1287,11 @@ def __init__(self, # user-facing output on stdout, not diagnostics -- the frame # chains are a feature of the tool, so they stay on ``print`` # rather than becoming log records a consumer has to configure - # a handler to see (and on a different stream at that). + # a handler to see (and on a different stream at that). Of the + # built-in engines only PCAP and PCAPNG keep this callback, so + # its frame is theirs; the others install their own. self._vfunc = lambda e, f: print( - f'Frame {e._frnum:>3d}: {f.protochain}' # pylint: disable=protected-access + f'Frame {e._frnum:>3d}: {cast("Frame | PCAPNG", f).protochain}' # pylint: disable=protected-access ) else: self._vfunc = lambda e, f: None diff --git a/tests/project/test_extractor_vfunc_doc_comment_1839_unit.py b/tests/project/test_extractor_vfunc_doc_comment_1839_unit.py new file mode 100644 index 000000000..a3c598d3c --- /dev/null +++ b/tests/project/test_extractor_vfunc_doc_comment_1839_unit.py @@ -0,0 +1,174 @@ +# -*- coding: utf-8 -*- +"""Pins the documentation comment and the type of :attr:`Extractor._vfunc`. + +GitHub issue #1839: :file:`docs/source/pcapkit/foundation/extraction.rst` asks +for ``.. autoattribute:: _vfunc``, but the annotation under its ``#:`` comment +had been commented out. Sphinx attaches a ``#:`` comment only to the assignment +or annotation directly below it, so the page rendered a bare name with no type +and no description. + +The annotation is the ``VerboseHandler`` alias, whose signature is pinned too: +the frame parameter is the whole ``Packet`` union, and the return is +:data:`~typing.Any`, because every caller discards it. The comparison is exact, +so it fails on a return of ``None`` or :class:`object` as well. + +Every test reads source text and never imports the package. The one that asks +Sphinx itself skips without it; the others run everywhere. + +""" + +from __future__ import annotations + +import ast +import pathlib +import re +import unittest +from typing import TYPE_CHECKING + +from tests.project.test_docs_ambiguous_xrefs_1775_unit import HAS_SPHINX + +if TYPE_CHECKING: + from typing import Iterable, Iterator, Optional, Union + + Binding = Union[ast.Assign, ast.AnnAssign] + +ROOT = pathlib.Path(__file__).resolve().parents[2] + +#: The module that declares the attribute, by name and by path. +MODULE = 'pcapkit.foundation.extraction' +SOURCE = ROOT / 'pcapkit' / 'foundation' / 'extraction.py' + +#: The documented attribute, and the description its page should render. +CLASS = 'Extractor' +ATTR = '_vfunc' +TEXT = 'Verbose callback function.' + +#: The alias that types the attribute, and the signature it should spell. +ALIAS = 'VerboseHandler' +SIGNATURE = "Callable[['Extractor', Packet], Any]" + +#: A documentation comment line, as :mod:`sphinx.pycode.parser` matches it. +COMMENT = re.compile(r'^\s*#: ?(.*)$') + + +def scope_statements(body: 'Iterable[ast.stmt]') -> 'Iterator[ast.stmt]': + """Yield every statement of a module or class body, including nested blocks. + + An ``if TYPE_CHECKING:`` block declares names of the scope it sits in, so the + walk goes into compound statements; it stops at a function or class, which + opens a new scope. + + """ + for stmt in body: + if isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + continue + yield stmt + for field in ('body', 'orelse', 'finalbody'): + yield from scope_statements(getattr(stmt, field, [])) + for handler in getattr(stmt, 'handlers', []): + yield from scope_statements(handler.body) + + +def bindings(body: 'Iterable[ast.stmt]', name: 'str') -> 'list[Binding]': + """Return every assignment or annotation of ``name`` in a scope, in order.""" + found = [] # type: list[Binding] + for stmt in scope_statements(body): + if isinstance(stmt, ast.AnnAssign): + targets = [stmt.target] # type: list[ast.expr] + elif isinstance(stmt, ast.Assign): + targets = stmt.targets + else: + continue + if any(isinstance(target, ast.Name) and target.id == name for target in targets): + found.append(stmt) + return found + + +def class_body(tree: 'ast.Module', cls: 'str') -> 'list[ast.stmt]': + """Return the body of the class ``cls`` defined anywhere in ``tree``.""" + classes = [node for node in ast.walk(tree) + if isinstance(node, ast.ClassDef) and node.name == cls] + if not classes: + raise AssertionError(f'no class {cls!r} in {SOURCE}') + return classes[0].body + + +def doc_comment(source: 'str', cls: 'str', attr: 'str') -> 'Optional[list[str]]': + """Return the ``#:`` lines Sphinx would attach to ``cls.attr``. + + These are the consecutive ``#:`` lines directly above the first assignment + or annotation of ``attr`` in the body of ``cls``. The result is :obj:`None` + when the class body neither assigns nor annotates ``attr``. + + """ + found = bindings(class_body(ast.parse(source), cls), attr) + if not found: + return None + + lines = [] # type: list[str] + for line in reversed(source.splitlines()[:found[0].lineno - 1]): + match = COMMENT.match(line) + if match is None: + break + lines.append(match.group(1)) + return lines[::-1] + + +class VerboseCallbackDocCommentTests(unittest.TestCase): + """The ``#:`` comment above :attr:`Extractor._vfunc` attaches to it.""" + + source = '' + + @classmethod + def setUpClass(cls) -> 'None': + cls.source = SOURCE.read_text(encoding='utf-8') + + def test_comment_sits_directly_above_the_annotation(self) -> 'None': + comment = doc_comment(self.source, CLASS, ATTR) + self.assertIsNotNone(comment, f'{CLASS} neither annotates nor assigns {ATTR}') + self.assertIn(TEXT, ' '.join(comment or [])) + + @unittest.skipUnless(HAS_SPHINX, 'Sphinx and sphinx-autodoc-typehints are not installed') + def test_sphinx_attaches_the_comment(self) -> 'None': + from sphinx.pycode import ModuleAnalyzer # pylint: disable=import-outside-toplevel + + docs = ModuleAnalyzer.for_string(self.source, MODULE).find_attr_docs() + lines = docs.get((CLASS, ATTR)) + self.assertIsNotNone(lines, f'Sphinx attaches no comment to {CLASS}.{ATTR}') + found = '\n'.join(lines or []).strip() + self.assertIn(TEXT, found) + # The first test reads the comment the same way, so it is measuring this. + self.assertEqual('\n'.join(doc_comment(self.source, CLASS, ATTR) or []).strip(), found) + + +class VerboseHandlerSignatureTests(unittest.TestCase): + """:attr:`Extractor._vfunc` is typed by the precise ``VerboseHandler`` alias.""" + + source = '' + tree = ast.parse('') + + @classmethod + def setUpClass(cls) -> 'None': + cls.source = SOURCE.read_text(encoding='utf-8') + cls.tree = ast.parse(cls.source) + + def test_attribute_is_annotated_with_the_alias(self) -> 'None': + annotations = [ast.literal_eval(stmt.annotation) + for stmt in bindings(class_body(self.tree, CLASS), ATTR) + if isinstance(stmt, ast.AnnAssign)] + self.assertEqual(annotations, [ALIAS]) + + def test_alias_spells_the_precise_signature(self) -> 'None': + found = bindings(self.tree.body, ALIAS) + self.assertEqual(len(found), 1, f'{ALIAS} is not bound exactly once in {SOURCE}') + value = found[0].value + if value is None: + self.fail(f'{ALIAS} is annotated but never assigned') + # Compared as trees, so spacing and quoting do not matter. + expected = ast.parse(SIGNATURE, mode='eval').body + self.assertEqual(ast.dump(value), ast.dump(expected), + f'{ALIAS} = {ast.get_source_segment(self.source, value)}') + + +if __name__ == '__main__': + unittest.main()