""" Tests for :py:mod:`line_profiler.autoprofile.eager_preimports`. Notes ----- Most of the features are already covered by the doctests. """ import subprocess import sys from contextlib import ExitStack from functools import partial from pathlib import Path from operator import methodcaller from runpy import run_path from tempfile import TemporaryDirectory from textwrap import dedent from types import SimpleNamespace from typing import Collection, Generator, Sequence, Type, Optional, Union from uuid import uuid4 from warnings import catch_warnings import pytest try: import flake8 # noqa: F401 except ImportError: HAS_FLAKE8 = False else: HAS_FLAKE8 = True from line_profiler.autoprofile.eager_preimports import ( split_dotted_path, resolve_profiling_targets, write_eager_import_module) def write(path: Path, content: Optional[str] = None) -> None: path.parent.mkdir(parents=True, exist_ok=True) if content is None: path.touch() else: path.write_text(dedent(content).strip('\n')) def gen_names(name) -> Generator[str, None, None]: while True: yield '_'.join([name, *str(uuid4()).split('-')]) @pytest.fixture def preserve_sys_state() -> None: old_path = sys.path.copy() old_modules = sys.modules.copy() try: yield finally: sys.path.clear() sys.path[:] = old_path sys.modules.clear() sys.modules.update(old_modules) @pytest.fixture def sample_package(preserve_sys_state: None, tmp_path: Path) -> str: """ Write a normal package and put it in :py:data:`sys.path`. When we're done, reset :py:data:`sys.path` and `sys.modules`. """ module_name = next(name for name in gen_names('my_sample_pkg') if name not in sys.modules) new_path = tmp_path / '_modules' write(new_path / module_name / '__init__.py') write(new_path / module_name / 'foo' / '__init__.py') write(new_path / module_name / 'foo' / 'bar.py') write(new_path / module_name / 'foo' / 'baz.py', """ ''' This is a bad module. ''' raise AssertionError """) write(new_path / module_name / 'foobar.py') # Cleanup managed with `preserve_sys_state()` sys.path.insert(0, str(new_path)) yield module_name @pytest.fixture def sample_namespace_package( preserve_sys_state: None, tmp_path_factory: pytest.TempPathFactory) -> str: """ Write a namespace package and put it in :py:data:`sys.path`. When we're done, reset :py:data:`sys.path` and `sys.modules`. """ module_name = next(name for name in gen_names('my_sample_namespace_pkg') if name not in sys.modules) new_paths = [tmp_path_factory.mktemp('_modules-', numbered=True) for _ in range(3)] for submod, new_path in zip(['one', 'two', 'three'], new_paths): write(new_path / module_name / (submod + '.py')) # Cleanup managed with `preserve_sys_state()` sys.path.insert(0, str(new_path)) yield module_name @pytest.mark.parametrize( ('adder', 'xc'), [('foo; bar', ValueError), (1, TypeError), ('(foo\n .bar)', ValueError)]) def test_write_eager_import_module_wrong_adder( adder: str, xc: Type[Exception]) -> None: """ Test passing an erroneous ``adder`` to :py:meth:`~.write_eager_import_module()`. """ with pytest.raises(xc): write_eager_import_module(['foo'], adder=adder) @pytest.mark.skipif(not HAS_FLAKE8, reason='no `flake8`') def test_written_module_pep8_compliance(sample_package: str): """ Test that the module written by :py:meth:`~.write_eager_import_module()` passes linting by :py:mod:`flake8`. """ with TemporaryDirectory() as tmpdir: module = Path(tmpdir) / 'module.py' with module.open(mode='w') as fobj: write_eager_import_module( [sample_package + '.foobar'], recurse=[sample_package + '.foo'], stream=fobj) print(module.read_text()) (subprocess .run([sys.executable, '-m', 'flake8', '--extend-ignore=E501', # Allow long lines module]) .check_returncode()) @pytest.mark.parametrize( ('dotted_paths', 'recurse', 'warnings', 'error'), [(['__MODULE__.foobar'], ['__MODULE__.foo'], # `foo.baz` is indirectly included, so its raising an error # shouldn't cause the script to error out [{'target cannot', '__MODULE__.foo.baz'}], None), # We don't recurse down `__MODULE__.foo`, so that doesn't give a # warning; but `__MODULE__.baz` cannot be imported because it # doesn't exist (['__MODULE__.foo', '__MODULE__.baz'], False, [{'target cannot', '__MODULE__.baz'}], None), # If we do recurse however, `__MODULE__.foo.baz` also ends up in # the warning # (also there's a `__MODULE___foo` which doesn't exist, about which # the warning is issued during module generation) (['__MODULE__' + '_foo', '__MODULE__', '__MODULE__.baz'], True, [{'target cannot', '__MODULE__' + '_foo'}, # Fails at write {'targets cannot', # Fails at import '__MODULE__.foo.baz', '__MODULE__.baz'}], None), # And if the problematic module is an explicit target, raise the # error (['__MODULE__', '__MODULE__.foo.baz'], False, [], AssertionError)]) def test_written_module_error_handling( sample_package: str, dotted_paths: Collection[str], recurse: Union[Collection[str], bool], warnings: Sequence[Collection[str]], error: Union[Type[Exception], None]): """ Test that the module written by :py:meth:`~.write_eager_import_module()` gracefully handles errors for implicitly included modules. """ replace = methodcaller('replace', '__MODULE__', sample_package) dotted_paths = [replace(target) for target in dotted_paths] if recurse not in (True, False): recurse = [replace(target) for target in recurse] warnings = [{replace(fragment) for fragment in fragments} for fragments in warnings] with TemporaryDirectory() as tmpdir: module = Path(tmpdir) / 'module.py' with ExitStack() as stack: enter = stack.enter_context # Set up the warning capturing early so that we catch both # warnings at module-generation time and execution time captured_warnings = enter(catch_warnings(record=True)) with module.open(mode='w') as fobj: write_eager_import_module( dotted_paths, recurse=recurse, stream=fobj) print(module.read_text()) if error is not None: enter(pytest.raises(error)) # Just use a dummy object, no need to instantiate a profiler prof = SimpleNamespace( add_imported_function_or_module=lambda *_, **__: 0) run_path(str(module), {'profile': prof}, 'module') assert len(captured_warnings) == len(warnings) for warning, fragments in zip(captured_warnings, warnings): for fragment in fragments: assert fragment in str(warning.message) def test_split_dotted_path_staticity() -> None: """ Test `split_dotted_path()` with different values for `static`. """ split = partial(split_dotted_path, 'os.path.abspath') # Static analysis has no idea of `os.path` members since `os.path` # is dynamically imported from e.g. `posixpath` or `ntpath` assert split(static=True) == ('os', 'path.abspath') # The import system knows of the already-imported module `os.path` assert split(static=False) == ('os.path', 'abspath') def test_resolve_profiling_targets_staticity( sample_namespace_package: str) -> None: """ Test subpackage/-module discovery with `resolve_profiling_targets()` with different values for `static`. """ all_targets = ({f'{sample_namespace_package}.{submod}' for submod in ['one', 'two', 'three']} | {sample_namespace_package}) # Static analysis can't handle namespace packages resolve = partial( resolve_profiling_targets, [sample_namespace_package], recurse=True) static_result = resolve(static=True) assert set(static_result.targets) < all_targets, static_result # The import system successfully retrieves all submodules dyn_result = resolve(static=False) assert set(dyn_result.targets) == all_targets, dyn_result