/usr/lib/python3/dist-packages/plainbox/impl/testing_utils.py is in python3-plainbox 0.25-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 | # This file is part of Checkbox.
#
# Copyright 2012 Canonical Ltd.
# Written by:
# Zygmunt Krynicki <zygmunt.krynicki@canonical.com>
#
# Checkbox is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 3,
# as published by the Free Software Foundation.
#
# Checkbox 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 Checkbox. If not, see <http://www.gnu.org/licenses/>.
"""
:mod:`plainbox.impl.testing_utils` -- plainbox specific test tools
==================================================================
.. warning::
THIS MODULE DOES NOT HAVE STABLE PUBLIC API
"""
from functools import wraps
from gzip import GzipFile
from io import TextIOWrapper
from tempfile import NamedTemporaryFile
import warnings
from plainbox.impl.job import JobDefinition
from plainbox.impl.result import IOLogRecordWriter
from plainbox.impl.result import MemoryJobResult
from plainbox.impl.secure.origin import Origin
from plainbox.vendor.mock import Mock
def MockJobDefinition(id, *args, **kwargs):
"""
Mock for JobDefinition class
"""
job = Mock(*args, name="job-with-id:{}".format(id),
spec_set=JobDefinition, **kwargs)
job.id = id
return job
def make_io_log(io_log, io_log_dir):
"""
Make the io logs serialization to json and return the saved file pathname
WARNING: The caller has to remove the file once done with it!
"""
with NamedTemporaryFile(
delete=False, suffix='.record.gz', dir=io_log_dir) as byte_stream, \
GzipFile(fileobj=byte_stream, mode='wb') as gzip_stream, \
TextIOWrapper(gzip_stream, encoding='UTF-8') as text_stream:
writer = IOLogRecordWriter(text_stream)
for record in io_log:
writer.write_record(record)
return byte_stream.name
# Deprecated, use JobDefinition() directly
def make_job(id, plugin="dummy", requires=None, depends=None, **kwargs):
"""
Make and return a dummy JobDefinition instance
"""
data = {'id': id}
if plugin is not None:
data['plugin'] = plugin
if requires is not None:
data['requires'] = requires
if depends is not None:
data['depends'] = depends
# Add any custom key-value properties
data.update(kwargs)
return JobDefinition(data, Origin.get_caller_origin())
def make_job_result(outcome="dummy"):
"""
Make and return a dummy JobResult instance
"""
return MemoryJobResult({
'outcome': outcome
})
def suppress_warnings(func):
"""
Suppress all warnings from the decorated function
"""
@wraps(func)
def decorator(*args, **kwargs):
with warnings.catch_warnings():
warnings.simplefilter("ignore")
return func(*args, **kwargs)
return decorator
|