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
|
|
|
|