diff --git a/pcapkit/const/reg/apptype/__init__.py b/pcapkit/const/reg/apptype/__init__.py index 374b72dce8..f3ad78f61f 100644 --- a/pcapkit/const/reg/apptype/__init__.py +++ b/pcapkit/const/reg/apptype/__init__.py @@ -32,21 +32,156 @@ right registry through :meth:`AppType.get `. +:class:`~pcapkit.const.reg.apptype.tcp.TCP` and +:class:`~pcapkit.const.reg.apptype.udp.UDP` are imported the first time either +is reached -- as an attribute of this package, or through :meth:`AppType.get +` -- rather than with the package +itself, since they are most of its import time (GitHub issue :issue:`1538`). + """ +import importlib +from typing import TYPE_CHECKING + from pcapkit.const.reg.apptype.apptype import AppType, TransportProtocol from pcapkit.const.reg.apptype.dccp import DCCP from pcapkit.const.reg.apptype.sctp import SCTP -from pcapkit.const.reg.apptype.tcp import TCP -from pcapkit.const.reg.apptype.udp import UDP +from pcapkit.corekit.module import ModuleDescriptor + +if TYPE_CHECKING: + from typing import Any + + from pcapkit.const.reg.apptype.tcp import TCP + from pcapkit.const.reg.apptype.udp import UDP __all__ = ['AppType', 'TransportProtocol', 'TCP', 'UDP', 'SCTP', 'DCCP'] +#: Registries imported on first use rather than with this package, as submodule +#: name to class name -- GitHub issue :issue:`1538`. Generated from IANA's +#: registry, the two hold about 49,700 lines of members and were about 30% of +#: ``import pcapkit``, while SCTP's and DCCP's cost a few milliseconds and stay +#: eager. +_LAZY = {'tcp': 'TCP', 'udp': 'UDP'} + + +class _Registries(dict): + """:attr:`AppType.__registries__ `, + importing a lazy registry the first time its value is read. + + Every key is present from the start, in the order the eager imports used to + insert them, so :func:`len`, ``in`` and iteration import nothing. A lazy + registry's value is a :class:`~pcapkit.corekit.module.ModuleDescriptor` + until the first read replaces it with the class. Every way of reading a + value is overridden, :meth:`__iter__` included: overriding it is what makes + ``dict(...)`` and ``{**...}`` go through :meth:`__getitem__` instead of + copying the descriptors in C, so no caller ever sees one. + + """ + + __slots__ = () + + def _resolve(self, key: 'Any', value: 'Any') -> 'Any': + if isinstance(value, ModuleDescriptor): + value = value.klass + dict.__setitem__(self, key, value) + return value + + def _resolve_all(self) -> 'None': + for key, value in list(dict.items(self)): + self._resolve(key, value) + + def __getitem__(self, key: 'Any') -> 'Any': + return self._resolve(key, dict.__getitem__(self, key)) + + def get(self, key: 'Any', default: 'Any' = None) -> 'Any': + value = dict.get(self, key, default) + # NOTE: the second test keeps a caller's own ``default`` from being + # resolved and stored, should it ever be a descriptor itself. + if isinstance(value, ModuleDescriptor) and dict.__contains__(self, key): + value = self._resolve(key, value) + return value + + def __iter__(self) -> 'Any': + return dict.__iter__(self) + + def values(self) -> 'Any': + self._resolve_all() + return dict.values(self) + + def items(self) -> 'Any': + self._resolve_all() + return dict.items(self) + + def copy(self) -> 'dict[Any, Any]': + self._resolve_all() + return dict.copy(self) + + def pop(self, *args: 'Any') -> 'Any': + self._resolve_all() + return dict.pop(self, *args) + + def popitem(self) -> 'Any': + self._resolve_all() + return dict.popitem(self) + + def setdefault(self, key: 'Any', default: 'Any' = None) -> 'Any': + self._resolve_all() + return dict.setdefault(self, key, default) + + def __eq__(self, other: 'object') -> 'bool': + self._resolve_all() + return dict.__eq__(self, other) + + def __ne__(self, other: 'object') -> 'bool': + self._resolve_all() + return dict.__ne__(self, other) + + def __or__(self, other: 'Any') -> 'Any': + self._resolve_all() + return dict.__or__(self, other) + + def __ror__(self, other: 'Any') -> 'Any': + self._resolve_all() + return dict.__ror__(self, other) + + def __repr__(self) -> 'str': + self._resolve_all() + return dict.__repr__(self) + + # NOTE: this is what makes ``AppType.get(port, proto=...)`` work on the base # class, which is where every caller in the library addresses it. It cannot live # in ``apptype.py``: that module is imported *by* all four registries, so it -# cannot import them back. Here is the first point at which all four exist. -AppType.__registries__[TransportProtocol.tcp] = TCP -AppType.__registries__[TransportProtocol.udp] = UDP -AppType.__registries__[TransportProtocol.sctp] = SCTP -AppType.__registries__[TransportProtocol.dccp] = DCCP +# cannot import them back. Here is the first point at which all four exist, or +# for the lazy two, where to import them from. +AppType.__registries__ = _Registries({ + TransportProtocol.tcp: ModuleDescriptor(f'{__name__}.tcp', _LAZY['tcp']), + TransportProtocol.udp: ModuleDescriptor(f'{__name__}.udp', _LAZY['udp']), + TransportProtocol.sctp: SCTP, + TransportProtocol.dccp: DCCP, +}) + + +def __getattr__(name: 'str') -> 'Any': + """Import a :data:`_LAZY` registry, or its submodule, on first access. + + :pep:`562` calls this only for a name missing from this module's globals, + so it runs once per name: importing a submodule binds it here, and the + class is bound here before it is returned. + + """ + if name in _LAZY: + return importlib.import_module(f'{__name__}.{name}') + module = name.lower() + if _LAZY.get(module) == name: + registry = getattr(importlib.import_module(f'{__name__}.{module}'), name) + globals()[name] = registry + return registry + raise AttributeError(f'module {__name__!r} has no attribute {name!r}') + + +def __dir__() -> 'list[str]': + """The names an eager import of all four registries would list.""" + helpers = {'importlib', 'TYPE_CHECKING', 'ModuleDescriptor', '_LAZY', '_Registries', + '__getattr__', '__dir__'} + return sorted({*globals(), *_LAZY, *_LAZY.values()} - helpers) diff --git a/tests/conftest.py b/tests/conftest.py index a8248280ba..e1d5f107c9 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -66,6 +66,11 @@ #: win. _pinned_snapshot = None # type: Optional[dict[str, types.ModuleType]] +#: What :func:`pytest_sessionstart` imports before the pin: the package, plus the +#: two app-type registries it loads on first use rather than with itself (GitHub +#: issue #1538). See that hook for why the registries cannot be left out. +WARM_MODULES = ('pcapkit', 'pcapkit.const.reg.apptype.tcp', 'pcapkit.const.reg.apptype.udp') + def _pin_module_snapshot() -> None: """Capture the :mod:`pcapkit` region as the restore target, once. @@ -149,6 +154,16 @@ def pytest_sessionstart(session: 'pytest.Session') -> 'None': that imports the library and nothing else leaves the region exactly as it found it, so there is nothing to put back. + The two lazily loaded app-type registries in :data:`WARM_MODULES` are + imported here too, and that is for correctness rather than speed. Left out + of the pin, they were dropped from :data:`sys.modules` after whichever test + first loaded them, while the package kept the classes it had already bound + in ``TCP``/``UDP`` and :attr:`AppType.__registries__ + `. The next direct + import of the submodule then built a second ``TCP`` and ``UDP``, and + :mod:`tests.project.test_conventions_doc_claims` counted both generations: + ``127 != 129`` registries on the 3.10-3.13 unit legs of PR #1560. + Failure here is not an error. A checkout without the runtime dependencies installed cannot import the package at all -- which is a supported way to run the parts of the suite that are guarded by ``skipUnless`` -- so this degrades @@ -159,7 +174,8 @@ def pytest_sessionstart(session: 'pytest.Session') -> 'None': """ try: - importlib.import_module('pcapkit') + for name in WARM_MODULES: + importlib.import_module(name) except Exception: # pragma: no cover # pylint: disable=broad-except # Deliberately broad: anything at all going wrong here should cost # nothing but the optimisation. ``BaseException`` is not caught, so an diff --git a/tests/const/test_const_apptype_lazy_1538_unit.py b/tests/const/test_const_apptype_lazy_1538_unit.py new file mode 100644 index 0000000000..b8ef46d9fa --- /dev/null +++ b/tests/const/test_const_apptype_lazy_1538_unit.py @@ -0,0 +1,368 @@ +# -*- coding: utf-8 -*- +"""The TCP and UDP app-type registries load on first use, not with ``pcapkit``. + +GitHub issue #1538. :mod:`pcapkit.const.reg.apptype.tcp` and +:mod:`~pcapkit.const.reg.apptype.udp` are generated from IANA's registry and +hold about 49,700 lines of members between them; ``python -X importtime -c +'import pcapkit'`` put them at 104 ms and 102 ms of a 700 ms import. Nothing on +any import path needs them, so :mod:`pcapkit.const.reg.apptype` now imports them +from a :pep:`562` ``__getattr__`` and from +:attr:`AppType.__registries__ ` +the first time either is read. + +The subprocess checks run in a fresh interpreter, since the question is what a +cold ``import pcapkit`` leaves in :data:`sys.modules` and this process has long +since imported both. Each fails on an eager tree at its first ``loaded()`` +check; what follows it pins that nothing a caller can observe changed apart +from *when* the import happens. :class:`RegistriesMappingTests` and +:class:`PackageGetattrTests` run in-process, on a private import, so that +coverage sees every line. :class:`PinnedSnapshotTests` covers the suite's own +side: :file:`tests/conftest.py` has to pin both registries, or its restore +leaves a second generation of each behind. + +""" + +import json +import os +import pathlib +import subprocess +import sys +import textwrap +import unittest + +from tests._support import reimport_once_per_class + +ROOT = pathlib.Path(__file__).resolve().parents[2] + +PRELUDE = f''' +import json, sys +import pcapkit +assert pcapkit.__file__.startswith({str(ROOT)!r}), pcapkit.__file__ + +def loaded(): + return [name for name in ('tcp', 'udp') + if f'pcapkit.const.reg.apptype.{{name}}' in sys.modules] + +def emit(**kwargs): + print(json.dumps(kwargs)) +''' + +#: ``dir(pcapkit.const.reg.apptype)`` as the eager package answered it. +EAGER_DIR = ['AppType', 'DCCP', 'SCTP', 'TCP', 'TransportProtocol', 'UDP', '__all__', + '__builtins__', '__cached__', '__doc__', '__file__', '__loader__', '__name__', + '__package__', '__path__', '__spec__', 'apptype', 'dccp', 'sctp', 'tcp', 'udp'] + +#: ``repr(AppType.__registries__)`` once populated, eager or lazy. +REGISTRIES_REPR = ("{: , : " + ", : , " + ": }") + + +def run_child(code: 'str') -> 'dict': + """Run :data:`PRELUDE` then ``code`` in a fresh interpreter; return its last + :func:`emit` line.""" + environ = dict(os.environ) + environ['PYTHONPATH'] = os.pathsep.join( + [str(ROOT), environ['PYTHONPATH']] if environ.get('PYTHONPATH') else [str(ROOT)]) + result = subprocess.run([sys.executable, '-c', PRELUDE + textwrap.dedent(code)], + cwd=str(ROOT), env=environ, capture_output=True, text=True, + timeout=300, check=False) + if result.returncode != 0: + raise AssertionError(f'child exited {result.returncode}:\n{result.stderr}') + return json.loads(result.stdout.splitlines()[-1]) + + +class LazyAppTypeImportTests(unittest.TestCase): + """What a cold import loads, and what loads each registry afterwards.""" + + def test_import_pcapkit_leaves_tcp_and_udp_unimported(self) -> 'None': + out = run_child(''' + emit(loaded=loaded(), + eager=[f'pcapkit.const.reg.apptype.{name}' in sys.modules + for name in ('apptype', 'sctp', 'dccp')]) + ''') + self.assertEqual(out['loaded'], []) + self.assertEqual(out['eager'], [True, True, True]) + + def test_attribute_access_loads_only_the_registry_it_names(self) -> 'None': + out = run_child(''' + import pcapkit.const.reg.apptype as pkg + steps = [loaded()] + tcp = pkg.TCP + steps.append(loaded()) + udp = pkg.udp.UDP + steps.append(loaded()) + emit(steps=steps, + same=[tcp is sys.modules['pcapkit.const.reg.apptype.tcp'].TCP, + udp is pkg.UDP, pkg.TCP is pkg.tcp.TCP, 'TCP' in vars(pkg)]) + ''') + self.assertEqual(out['steps'], [[], ['tcp'], ['tcp', 'udp']]) + self.assertEqual(out['same'], [True, True, True, True]) + + def test_port_lookup_loads_only_the_transport_it_asks_for(self) -> 'None': + out = run_child(''' + from pcapkit.const.reg.apptype import AppType, TransportProtocol + steps = [loaded()] + http = AppType.get(80, proto='tcp') + steps.append(loaded()) + dns = AppType.get(53, proto=TransportProtocol.udp) + steps.append(loaded()) + emit(steps=steps, members=[str(http), type(http).__qualname__, + str(dns), type(dns).__qualname__]) + ''') + self.assertEqual(out['steps'], [[], ['tcp'], ['tcp', 'udp']]) + self.assertEqual(out['members'], ['http [80 - tcp] (aliases: www, www-http)', 'TCP', + 'domain [53 - udp]', 'UDP']) + + +class LazyAppTypeContractTests(unittest.TestCase): + """Nothing but the moment of import changed.""" + + def test_dir_and_star_import_are_unchanged(self) -> 'None': + out = run_child(''' + import pcapkit.const.reg.apptype as pkg + listed, before = dir(pkg), loaded() + namespace = {} + exec('from pcapkit.const.reg.apptype import *', namespace) + emit(dir=listed, before=before, after=loaded(), all=pkg.__all__, + star=sorted(name for name in namespace if name != '__builtins__'), + resolved=[namespace[name] is getattr(pkg, name) for name in pkg.__all__]) + ''') + self.assertEqual(out['dir'], EAGER_DIR) + self.assertEqual(out['before'], []) + self.assertEqual(out['star'], sorted(out['all'])) + self.assertEqual(out['resolved'], [True] * 6) + self.assertEqual(out['after'], ['tcp', 'udp']) + + def test_registries_never_expose_a_pending_registry(self) -> 'None': + # NOTE: one interpreter for every view, so each starts from the pending + # state again rather than from whatever the previous view loaded. + out = run_child(''' + import copy + from pcapkit.const.reg.apptype import AppType, TransportProtocol + from pcapkit.corekit.module import ModuleDescriptor + registries = AppType.__registries__ + shape = [len(registries), [int(key) for key in registries], + 1 in registries, TransportProtocol.udp in registries, loaded()] + populated = dict(registries) + views = { + 'getitem': lambda: registries[TransportProtocol.tcp], + 'get': lambda: registries.get(1), + 'dict': lambda: dict(registries), + 'unpack': lambda: {**registries}, + 'values': lambda: list(registries.values()), + 'items': lambda: dict(registries.items()), + 'copy': registries.copy, + 'copy.copy': lambda: dict(copy.copy(registries)), + 'or': lambda: registries | {}, + 'ror': lambda: {} | registries, + 'repr': lambda: registries, + 'eq': lambda: registries == populated, + 'ne': lambda: registries != populated, + } + reprs = {} + for name, view in views.items(): + for proto in (TransportProtocol.tcp, TransportProtocol.udp): + dict.__setitem__(registries, proto, ModuleDescriptor( + f'pcapkit.const.reg.apptype.{proto.name}', proto.name.upper())) + reprs[name] = repr(view()) + emit(shape=shape, reprs=reprs, default=registries.get(0, 'fallback')) + ''') + self.assertEqual(out['shape'], [4, [1, 2, 3, 4], True, True, []]) + for name, text in out['reprs'].items(): + with self.subTest(view=name): + self.assertNotIn('ModuleDescriptor', text) + self.assertEqual(text, {'getitem': "", 'get': "", + 'values': "[, , " + ", ]", + 'eq': 'True', 'ne': 'False'} + .get(name, REGISTRIES_REPR)) + self.assertEqual(out['default'], 'fallback') + + def test_member_pickled_eagerly_unpickles_after_a_lazy_import(self) -> 'None': + payload = run_child(''' + import pickle + from pcapkit.const.reg.apptype.tcp import TCP + emit(payload=pickle.dumps(TCP.http).hex()) + ''')['payload'] + out = run_child(f''' + import copy, pickle + before = loaded() + member = pickle.loads(bytes.fromhex({payload!r})) + from pcapkit.const.reg.apptype import TCP + emit(before=before, after=loaded(), + same=[member is TCP.http, copy.copy(member) is member, + copy.deepcopy(member) is member]) + ''') + self.assertEqual(out['before'], []) + self.assertEqual(out['after'], ['tcp']) + self.assertEqual(out['same'], [True, True, True]) + + def test_port_registered_before_first_use_dispatches_the_same(self) -> 'None': + code = ''' + from pcapkit.foundation.registry import register_tcp + from pcapkit.protocols.transport.tcp import TCP + if {lookup_first}: + import pcapkit.const.reg.apptype + pcapkit.const.reg.apptype.TCP + at_register = loaded() + register_tcp(54321, 'pcapkit.protocols.application.ftp', 'FTP') + raw = TCP(srcport=40000, dstport=54321, payload=b'USER anonymous\\r\\n').data + parsed = TCP(raw, len(raw)) + emit(at_register=at_register, raw=raw.hex(), chain=str(parsed.protochain), + ports=[repr(parsed.info.srcport), repr(parsed.info.dstport)], + rebuilt=[TCP.from_data(parsed.info).data == raw, + TCP.from_data(parsed.info.to_dict()).data == raw]) + ''' + before = run_child(code.format(lookup_first=False)) + after = run_child(code.format(lookup_first=True)) + self.assertEqual(before.pop('at_register'), []) + self.assertEqual(after.pop('at_register'), ['tcp']) + self.assertEqual(before, after) + self.assertEqual(before['chain'], 'TCP:FTP') + self.assertEqual(before['rebuilt'], [True, True]) + + +class RegistriesMappingTests(unittest.TestCase): + """Every method of the lazy mapping, on a private instance in this process.""" + + def setUp(self) -> 'None': + # NOTE: a private import, so that the package starts cold -- the shared + # one has both registries loaded, since tests/conftest.py pins them -- + # and nothing bound here outlives the class. + reimport_once_per_class(self, restore=True) + self.make() + + def make(self) -> 'None': + from pcapkit.const.reg.apptype import _Registries # type: ignore[attr-defined] + from pcapkit.const.reg.apptype import DCCP, SCTP, TCP, UDP, TransportProtocol + from pcapkit.corekit.module import ModuleDescriptor + + self.proto = TransportProtocol + self.classes = [TCP, UDP, SCTP, DCCP] + self.descriptor = ModuleDescriptor + self.registries = _Registries({ + TransportProtocol.tcp: ModuleDescriptor('pcapkit.const.reg.apptype.tcp', 'TCP'), + TransportProtocol.udp: ModuleDescriptor('pcapkit.const.reg.apptype.udp', 'UDP'), + TransportProtocol.sctp: SCTP, + TransportProtocol.dccp: DCCP, + }) + + def assert_pending(self, *protos: 'object') -> 'None': + raw = dict.items(self.registries) + self.assertEqual([key for key, value in raw if isinstance(value, self.descriptor)], + list(protos)) + + def test_reads_resolve_in_place(self) -> 'None': + registries, proto = self.registries, self.proto + self.assertEqual((len(registries), list(registries), 1 in registries), (4, list(proto)[1:], True)) + self.assert_pending(proto.tcp, proto.udp) + self.assertIs(registries.get(2), self.classes[1]) + self.assert_pending(proto.tcp) + self.assertIs(registries[proto.tcp], self.classes[0]) + self.assert_pending() + self.assertEqual(list(registries), list(proto)[1:]) + self.assertIsNone(registries.get(proto.undefined)) + + def test_a_descriptor_default_is_returned_untouched(self) -> 'None': + default = self.descriptor('pcapkit.const.reg.apptype.tcp', 'TCP') + self.assertIs(self.registries.get(0, default), default) + self.assertNotIn(0, self.registries) + + def test_whole_mapping_reads_resolve_everything(self) -> 'None': + proto, classes = self.proto, self.classes + expected = dict(zip(list(proto)[1:], classes)) + views = { + 'values': lambda registries: list(registries.values()) == classes, + 'items': lambda registries: dict(registries.items()) == expected, + 'copy': lambda registries: registries.copy() == expected, + 'dict': lambda registries: dict.__eq__(dict(registries), expected), + 'eq': lambda registries: registries == expected, + 'ne': lambda registries: (registries != expected) is False, + 'or': lambda registries: dict.__eq__(registries | {}, expected), + 'ror': lambda registries: dict.__eq__({} | registries, expected), + 'repr': lambda registries: 'ModuleDescriptor' not in repr(registries), + 'pop': lambda registries: registries.pop(proto.tcp) is classes[0], + 'popitem': lambda registries: registries.popitem() == (proto.dccp, classes[3]), + 'setdefault': lambda registries: registries.setdefault(proto.udp) is classes[1], + } + for name, view in views.items(): + with self.subTest(view=name): + self.make() + self.assertTrue(view(self.registries)) + self.assert_pending() + + +class PackageGetattrTests(unittest.TestCase): + """The :pep:`562` hooks, called in this process so that coverage sees them.""" + + def setUp(self) -> 'None': + reimport_once_per_class(self, restore=True) # see RegistriesMappingTests.setUp + + def test_hooks(self) -> 'None': + import pcapkit.const.reg.apptype as pkg + + self.assertNotIn('TCP', vars(pkg)) + self.assertNotIn('pcapkit.const.reg.apptype.tcp', sys.modules) + getattr_ = vars(pkg)['__getattr__'] + tcp_module = getattr_('tcp') + self.assertIs(tcp_module, sys.modules['pcapkit.const.reg.apptype.tcp']) + self.assertIs(getattr_('TCP'), tcp_module.TCP) + self.assertIs(vars(pkg)['TCP'], tcp_module.TCP) + for name in ('Tcp', 'tCP', 'sctp_', 'nope'): + with self.subTest(name=name), self.assertRaises(AttributeError): + getattr_(name) + self.assertEqual(vars(pkg)['__dir__'](), EAGER_DIR) + + +class SharedImportProbe(unittest.TestCase): + """Two tests on the shared import, with the conftest restore between them. + + The first binds both registries through the package; the second imports + their submodules directly. Unless :data:`tests.conftest.WARM_MODULES` puts + them in the pinned snapshot, the restore drops them in between and the + second test builds another ``TCP`` and ``UDP``. :mod:`unittest` runs the + two in name order, and :class:`PinnedSnapshotTests` runs them in one + :program:`pytest` process, so they meet that restore whatever runs this file. + + """ + + def test_1_bind_through_the_package(self) -> 'None': + import pcapkit.const.reg.apptype as pkg + from pcapkit.const.reg.apptype import AppType + + self.assertIs(type(AppType.get(80, proto='tcp')), pkg.TCP) + self.assertIs(type(AppType.get(53, proto='udp')), pkg.UDP) + + def test_2_import_the_submodules_directly(self) -> 'None': + import pcapkit.const.reg.apptype as pkg + from pcapkit.const.reg.apptype import AppType, TransportProtocol + from pcapkit.const.reg.apptype.tcp import TCP + from pcapkit.const.reg.apptype.udp import UDP + + self.assertIs(TCP, pkg.TCP) + self.assertIs(UDP, pkg.UDP) + self.assertIs(AppType.__registries__[TransportProtocol.tcp], TCP) + self.assertIs(AppType.__registries__[TransportProtocol.udp], UDP) + + +class PinnedSnapshotTests(unittest.TestCase): + """:data:`tests.conftest.WARM_MODULES` keeps one generation of each registry.""" + + def test_the_probe_passes_in_one_pytest_process(self) -> 'None': + environ = dict(os.environ) + environ['PYTHONPATH'] = os.pathsep.join( + [str(ROOT), environ['PYTHONPATH']] if environ.get('PYTHONPATH') else [str(ROOT)]) + for name in ('PYTEST_ADDOPTS', 'PYTEST_CURRENT_TEST', 'PYTEST_XDIST_WORKER', + 'PYTEST_XDIST_WORKER_COUNT', 'PYTEST_XDIST_TESTRUNUID'): + environ.pop(name, None) + probe = f'{pathlib.Path(__file__).resolve().relative_to(ROOT)}::SharedImportProbe' + result = subprocess.run([sys.executable, '-m', 'pytest', '-q', '-p', 'no:cacheprovider', probe], + cwd=str(ROOT), env=environ, capture_output=True, text=True, + timeout=300, check=False) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + self.assertIn('2 passed', result.stdout) + + +if __name__ == '__main__': + unittest.main()