1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
|
# This file is part of Buildbot. Buildbot is free software: you can
# redistribute it and/or modify it under the terms of the GNU General Public
# License as published by the Free Software Foundation, version 2.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 51
# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# Copyright Buildbot Team Members
from __future__ import annotations
import builtins
import errno
import os
import re
import shutil
import sys
from io import StringIO
from typing import TYPE_CHECKING
from unittest import mock
from twisted.python import log
from twisted.trial.unittest import TestCase
from buildbot_worker.scripts import base
if TYPE_CHECKING:
from typing import Any
from typing import Callable
def nl(s: str | Any) -> str | Any:
"""Convert the given string to the native newline format, assuming it is
already in normal UNIX newline format (\n). Use this to create the
appropriate expectation in an assertEqual"""
if not isinstance(s, str):
return s
return s.replace('\n', os.linesep)
class BasedirMixin:
"""Mix this in and call setUpBasedir and tearDownBasedir to set up
a clean basedir with a name given in self.basedir."""
def setUpBasedir(self) -> None:
self.basedir = "test-basedir"
if os.path.exists(self.basedir):
shutil.rmtree(self.basedir)
def tearDownBasedir(self) -> None:
if os.path.exists(self.basedir):
shutil.rmtree(self.basedir)
class IsWorkerDirMixin:
"""
Mixin for setting up mocked base.isWorkerDir() function
"""
def setupUpIsWorkerDir(self, return_value: bool) -> None:
self.isWorkerDir = mock.Mock(return_value=return_value)
assert isinstance(self, TestCase)
self.patch(base, "isWorkerDir", self.isWorkerDir)
class PatcherMixin:
"""
Mix this in to get a few special-cased patching methods
"""
def patch_os_uname(self, replacement: Callable[[], os.uname_result]) -> None:
# twisted's 'patch' doesn't handle the case where an attribute
# doesn't exist..
assert isinstance(self, TestCase)
if hasattr(os, 'uname'):
self.patch(os, 'uname', replacement)
else:
def cleanup() -> None:
del os.uname
self.addCleanup(cleanup)
os.uname = replacement
class FileIOMixin:
"""
Mixin for patching open(), read() and write() to simulate successful
I/O operations and various I/O errors.
"""
def setUpOpen(self, file_contents: str = "dummy-contents") -> None:
"""
patch open() to return file object with provided contents.
@param file_contents: contents that will be returned by file object's
read() method
"""
assert isinstance(self, TestCase)
# Use mock.mock_open() to create a substitute for
# open().
fakeOpen = mock.mock_open(read_data=file_contents)
# When fakeOpen() is called, it returns a Mock
# that has these methods: read(), write(), __enter__(), __exit__().
# read() will always return the value of the 'file_contents variable.
self.fileobj = fakeOpen()
# patch open() to always return our Mock file object
self.open = mock.Mock(return_value=self.fileobj)
self.patch(builtins, "open", self.open)
def setUpOpenError(
self,
errno: int = errno.ENOENT,
strerror: str = "dummy-msg",
filename: str = "dummy-file",
) -> None:
"""
patch open() to raise IOError
@param errno: exception's errno value
@param strerror: exception's strerror value
@param filename: exception's filename value
"""
assert isinstance(self, TestCase)
# Use mock.mock_open() to create a substitute for
# open().
fakeOpen = mock.mock_open()
# Add side_effect so that calling fakeOpen() will always
# raise an IOError.
fakeOpen.side_effect = OSError(errno, strerror, filename)
self.open = fakeOpen
self.patch(builtins, "open", self.open)
def setUpReadError(
self,
errno: int = errno.EIO,
strerror: str = "dummy-msg",
filename: str = "dummy-file",
) -> None:
"""
patch open() to return a file object that will raise IOError on read()
@param errno: exception's errno value
@param strerror: exception's strerror value
@param filename: exception's filename value
"""
assert isinstance(self, TestCase)
# Use mock.mock_open() to create a substitute for
# open().
fakeOpen = mock.mock_open()
# When fakeOpen() is called, it returns a Mock
# that has these methods: read(), write(), __enter__(), __exit__().
self.fileobj = fakeOpen()
# Add side_effect so that calling read() will always
# raise an IOError.
self.fileobj.read.side_effect = OSError(errno, strerror, filename)
# patch open() to always return our Mock file object
self.open = mock.Mock(return_value=self.fileobj)
self.patch(builtins, "open", self.open)
def setUpWriteError(
self,
errno: int = errno.ENOSPC,
strerror: str = "dummy-msg",
filename: str = "dummy-file",
) -> None:
"""
patch open() to return a file object that will raise IOError on write()
@param errno: exception's errno value
@param strerror: exception's strerror value
@param filename: exception's filename value
"""
assert isinstance(self, TestCase)
# Use mock.mock_open() to create a substitute for
# open().
fakeOpen = mock.mock_open()
# When fakeOpen() is called, it returns a Mock
# that has these methods: read(), write(), __enter__(), __exit__().
self.fileobj = fakeOpen()
# Add side_effect so that calling write() will always
# raise an IOError.
self.fileobj.write.side_effect = OSError(errno, strerror, filename)
# patch open() to always return our Mock file object
self.open = mock.Mock(return_value=self.fileobj)
self.patch(builtins, "open", self.open)
class LoggingMixin:
def setUpLogging(self) -> None:
assert isinstance(self, TestCase)
self._logEvents: list[log.EventDict] = []
log.addObserver(self._logEvents.append)
self.addCleanup(log.removeObserver, self._logEvents.append)
def assertLogged(self, *args: str) -> None:
assert isinstance(self, TestCase)
for regexp in args:
r = re.compile(regexp)
for event in self._logEvents:
msg = log.textFromEventDict(event)
if msg is not None and r.search(msg):
return
self.fail(f"{regexp!r} not matched in log output.\n{self._logEvents} ")
def assertWasQuiet(self) -> None:
assert isinstance(self, TestCase)
self.assertEqual(self._logEvents, [])
class StdoutAssertionsMixin:
"""
Mix this in to be able to assert on stdout during the test
"""
def setUpStdoutAssertions(self) -> None:
assert isinstance(self, TestCase)
self.stdout = StringIO()
self.patch(sys, 'stdout', self.stdout)
def assertWasQuiet(self) -> None:
assert isinstance(self, TestCase)
self.assertEqual(self.stdout.getvalue(), '')
def assertInStdout(self, exp: str) -> None:
assert isinstance(self, TestCase)
self.assertIn(exp, self.stdout.getvalue())
def assertStdoutEqual(self, exp: str, msg: str | None = None) -> None:
assert isinstance(self, TestCase)
self.assertEqual(exp, self.stdout.getvalue(), msg)
def getStdout(self) -> str:
return self.stdout.getvalue().strip()
|
