tests.Capture.stdout() - Code Metrics - Inspection of "fix: cleaning up prev 2 commits; fixes #34" - dmyersturnbull/tyrannosaurus - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( 342148...df6f61 )

by Douglas

created 2021-02-07 03:00 UTC

tests.Capture.stdout() A

↳ Parent: tests

Complexity

Conditions

Size

Total Lines	3
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	3
nop	1
dl	0
loc	3
rs	10
c	0
b	0
f	0

"""
Utilities for tests.

Original source: https://github.com/dmyersturnbull/tyrannosaurus
Copyright 2020–2021 Douglas Myers-Turnbull
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
"""
# NOTE: If you modify this file, you should indicate your license and copyright as well.
from __future__ import annotations
import contextlib
import io
import logging
import os
import random
import shutil
import stat
import tempfile
import time
from datetime import datetime
from pathlib import Path, PurePath
from typing import Generator, Union
from warnings import warn


# Keeps created temp files; turn on for debugging
KEEP = False
# Separate logging in the main package vs. inside test functions
logger_name = Path(__file__).parent.parent.name.upper() + ".TEST"
_logger = logging.getLogger(logger_name)


class Capture(contextlib.ExitStack):
    def __init__(self):
        super().__init__()
        self._stdout = io.StringIO()
        self._stderr = io.StringIO()

    @property
    def stdout(self) -> str:
        return self._stdout.getvalue()

    @property
    def stderr(self) -> str:
        return self._stderr.getvalue()

    def __enter__(self):
        _logger.info("Capturing stdout and stderr")
        super().__enter__()
        self._stdout_context = self.enter_context(contextlib.redirect_stdout(self._stdout))
        # If the next line failed, the stdout context wouldn't exit
        # But this line is very unlikely to fail in practice
        self._stderr_context = self.enter_context(contextlib.redirect_stderr(self._stderr))
        return self

    def __exit__(self, *exc):
        _logger.info("Finished capturing stdout and stderr")
        # The ExitStack handles everything
        super().__exit__(*exc)


class TestResources:
    """
    A static singleton with utilities for filesystem operations in tests.
    Use ``TestResources.resource`` to get a file under ``tests/resources/``.

    Initializes a temporary directory with ``tempfile.TemporaryDirectory``
    and populates it with a single subdirectory, ``TestResources.global_temp_dir``.
    Temp directories for independent tests can be created underneath using
    ``TestResources.temp_dir``.
    """

    logger = _logger

    _start_dt = datetime.now()
    _start_ns = time.monotonic_ns()
    if KEEP:
        _tempfile_dir = None
        _temp_dir = Path("tmp-test-data")
    else:
        _tempfile_dir = tempfile.TemporaryDirectory()
        _temp_dir = Path(_tempfile_dir.name)

    logger.info(f"Set up main temp dir {_temp_dir.absolute()}")

    @classmethod
    @contextlib.contextmanager
    def capture(cls) -> Capture:
        """
        Context manager that captures stdout and stderr in a ``Capture`` object that contains both.
        Useful for testing code that prints to stdout and/or stderr.

        Yields:
            A ``Capture`` instance, which contains ``.stdout`` and ``.stderr``
        """
        with Capture() as cap:
            yield cap

    @classmethod
    def resource(cls, *nodes: Union[PurePath, str]) -> Path:
        """
        Gets a path of a test resource file under ``resources/``.

        Arguments:
            nodes: Path nodes under the ``resources/`` dir

        Returns:
            The Path ``resources``/``<node-1>``/``<node-2>``/.../``<node-n>``
        """
        return Path(Path(__file__).parent, "resources", *nodes).resolve()

    @classmethod
    @contextlib.contextmanager
    def temp_dir(
        cls,
        copy_resource: Union[None, str, Path] = None,
        force_delete: bool = True,
    ) -> Generator[Path, None, None]:
        """
        Context manager.
        Creates a new temporary directory underneath ``global_temp_dir``.
        Note that it deletes the directory if it already exists,
        then deletes (if the path exists) when the context closes.

        Arguments:
            copy_resource: Copy from underneath the resource dir
            force_delete: If necessary, change the permissions to delete

        Yields:
            The created directory as a ``pathlib.Path``
        """
        path = TestResources._temp_dir / ("%0x" % random.getrandbits(64))
        if path.exists():
            cls._delete_tree(path, surefire=force_delete)
        if copy_resource is None:
            path.mkdir(parents=True)
            TestResources.logger.info(f"Created empty temp dir {path}")
        else:
            path.parent.mkdir(parents=True, exist_ok=True)
            copy_path = Path(Path(__file__).parent) / "resources" / copy_resource
            shutil.copytree(str(copy_path), str(path))
            TestResources.logger.info(f"Copied {copy_resource} to temp {path}")
        yield path
        # note the global dir is only cleaned by tempfile on exit
        if not KEEP:
            cls._delete_tree(path, surefire=force_delete)

    @classmethod
    def global_temp_dir(cls) -> Path:
        """
        The global temporary directory, which is underneath ``tempfile.TemporaryDirectory``.
        The parent directory will be destroyed, along with all of its components,
        as specified by ``tempfile``.
        """
        return cls._temp_dir

    @classmethod
    def start_datetime(cls) -> datetime:
        """
        The datetime that ``tests/__init__.py`` was imported.
        """
        return cls._start_dt

    @classmethod
    def start_monotonic_ns(cls) -> int:
        """
        The nanosecond value of the ``time.monotonic`` clock at which ``tests/__init__.py`` was imported.
        """
        return cls._start_ns

    @classmethod
    def destroy(cls) -> None:
        """
        Deletes the full tempdir tree.
        """
        if not KEEP:
            cls._tempfile_dir.cleanup()

    @classmethod
    def _delete_tree(cls, path: Path, surefire: bool = False) -> None:
        def on_rm_error(func, pth, exc_info):
            # from: https://stackoverflow.com/questions/4829043/how-to-remove-read-only-attrib-directory-with-python-in-windows
            os.chmod(pth, stat.S_IWRITE)
            os.unlink(pth)

        kwargs = dict(onerror=on_rm_error) if surefire else {}
        if path.exists():
            try:
                shutil.rmtree(str(path), **kwargs)
            except OSError:
                warn(f"Testing dir {path} could not be deleted")


__all__ = ["TestResources"]


1			"""
2			Utilities for tests.
3
4			Original source: https://github.com/dmyersturnbull/tyrannosaurus
5			Copyright 2020–2021 Douglas Myers-Turnbull
6			Licensed under the Apache License, Version 2.0 (the "License");
7			you may not use this file except in compliance with the License.
8			You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
9			"""
10			# NOTE: If you modify this file, you should indicate your license and copyright as well.
11			from __future__ import annotations
12			import contextlib
13			import io
14			import logging
15			import os
16			import random
17			import shutil
18			import stat
19			import tempfile
20			import time
21			from datetime import datetime
22			from pathlib import Path, PurePath
23			from typing import Generator, Union
24			from warnings import warn
25
26
27			# Keeps created temp files; turn on for debugging
28			KEEP = False
29			# Separate logging in the main package vs. inside test functions
30			logger_name = Path(__file__).parent.parent.name.upper() + ".TEST"
31			_logger = logging.getLogger(logger_name)
32
33
34			class Capture(contextlib.ExitStack):
35			def __init__(self):
36			super().__init__()
37			self._stdout = io.StringIO()
38			self._stderr = io.StringIO()
39
40			@property
41			def stdout(self) -> str:
42			return self._stdout.getvalue()
43
44			@property
45			def stderr(self) -> str:
46			return self._stderr.getvalue()
47
48			def __enter__(self):
49			_logger.info("Capturing stdout and stderr")
50			super().__enter__()
51			self._stdout_context = self.enter_context(contextlib.redirect_stdout(self._stdout))
52			# If the next line failed, the stdout context wouldn't exit
53			# But this line is very unlikely to fail in practice
54			self._stderr_context = self.enter_context(contextlib.redirect_stderr(self._stderr))
55			return self
56
57			def __exit__(self, *exc):
58			_logger.info("Finished capturing stdout and stderr")
59			# The ExitStack handles everything
60			super().__exit__(*exc)
61
62
63			class TestResources:
64			"""
65			A static singleton with utilities for filesystem operations in tests.
66			Use ``TestResources.resource`` to get a file under ``tests/resources/``.
67
68			Initializes a temporary directory with ``tempfile.TemporaryDirectory``
69			and populates it with a single subdirectory, ``TestResources.global_temp_dir``.
70			Temp directories for independent tests can be created underneath using
71			``TestResources.temp_dir``.
72			"""
73
74			logger = _logger
75
76			_start_dt = datetime.now()
77			_start_ns = time.monotonic_ns()
78			if KEEP:
79			_tempfile_dir = None
80			_temp_dir = Path("tmp-test-data")
81			else:
82			_tempfile_dir = tempfile.TemporaryDirectory()
83			_temp_dir = Path(_tempfile_dir.name)
84
85			logger.info(f"Set up main temp dir {_temp_dir.absolute()}")
86
87			@classmethod
88			@contextlib.contextmanager
89			def capture(cls) -> Capture:
90			"""
91			Context manager that captures stdout and stderr in a ``Capture`` object that contains both.
92			Useful for testing code that prints to stdout and/or stderr.
93
94			Yields:
95			A ``Capture`` instance, which contains ``.stdout`` and ``.stderr``
96			"""
97			with Capture() as cap:
98			yield cap
99
100			@classmethod
101			def resource(cls, *nodes: Union[PurePath, str]) -> Path:
102			"""
103			Gets a path of a test resource file under ``resources/``.
104
105			Arguments:
106			nodes: Path nodes under the ``resources/`` dir
107
108			Returns:
109			The Path ``resources``/``<node-1>``/``<node-2>``/.../``<node-n>``
110			"""
111			return Path(Path(__file__).parent, "resources", *nodes).resolve()
112
113			@classmethod
114			@contextlib.contextmanager
115			def temp_dir(
116			cls,
117			copy_resource: Union[None, str, Path] = None,
118			force_delete: bool = True,
119			) -> Generator[Path, None, None]:
120			"""
121			Context manager.
122			Creates a new temporary directory underneath ``global_temp_dir``.
123			Note that it deletes the directory if it already exists,
124			then deletes (if the path exists) when the context closes.
125
126			Arguments:
127			copy_resource: Copy from underneath the resource dir
128			force_delete: If necessary, change the permissions to delete
129
130			Yields:
131			The created directory as a ``pathlib.Path``
132			"""
133			path = TestResources._temp_dir / ("%0x" % random.getrandbits(64))
134			if path.exists():
135			cls._delete_tree(path, surefire=force_delete)
136			if copy_resource is None:
137			path.mkdir(parents=True)
138			TestResources.logger.info(f"Created empty temp dir {path}")
139			else:
140			path.parent.mkdir(parents=True, exist_ok=True)
141			copy_path = Path(Path(__file__).parent) / "resources" / copy_resource
142			shutil.copytree(str(copy_path), str(path))
143			TestResources.logger.info(f"Copied {copy_resource} to temp {path}")
144			yield path
145			# note the global dir is only cleaned by tempfile on exit
146			if not KEEP:
147			cls._delete_tree(path, surefire=force_delete)
148
149			@classmethod
150			def global_temp_dir(cls) -> Path:
151			"""
152			The global temporary directory, which is underneath ``tempfile.TemporaryDirectory``.
153			The parent directory will be destroyed, along with all of its components,
154			as specified by ``tempfile``.
155			"""
156			return cls._temp_dir
157
158			@classmethod
159			def start_datetime(cls) -> datetime:
160			"""
161			The datetime that ``tests/__init__.py`` was imported.
162			"""
163			return cls._start_dt
164
165			@classmethod
166			def start_monotonic_ns(cls) -> int:
167			"""
168			The nanosecond value of the ``time.monotonic`` clock at which ``tests/__init__.py`` was imported.
169			"""
170			return cls._start_ns
171
172			@classmethod
173			def destroy(cls) -> None:
174			"""
175			Deletes the full tempdir tree.
176			"""
177			if not KEEP:
178			cls._tempfile_dir.cleanup()
179
180			@classmethod
181			def _delete_tree(cls, path: Path, surefire: bool = False) -> None:
182			def on_rm_error(func, pth, exc_info):
183			# from: https://stackoverflow.com/questions/4829043/how-to-remove-read-only-attrib-directory-with-python-in-windows
184			os.chmod(pth, stat.S_IWRITE)
185			os.unlink(pth)
186
187			kwargs = dict(onerror=on_rm_error) if surefire else {}
188			if path.exists():
189			try:
190			shutil.rmtree(str(path), **kwargs)
191			except OSError:
192			warn(f"Testing dir {path} could not be deleted")
193
194
195			__all__ = ["TestResources"]
196

dmyersturnbull / tyrannosaurus

Push — main ( 342148...df6f61 )

tests.Capture.stdout() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like