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