valitydev/salt
14
0
Fork 0
mirror of https://github.com/valitydev/salt.git synced 2024-11-07 08:58:59 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Replace the rest of mock_open with a class

Browse Source
This commit is contained in:
Erik Johnson 2018-06-17 16:42:43 -05:00
parent 75307a47c5
commit 4e67955572
No known key found for this signature in database
GPG Key ID: 5E5583C437808F3F
4 changed files with 114 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

138
tests/support/mock.py
View File

@ -185,22 +185,16 @@ class MockFH(object):
pass
# reimplement mock_open to support multiple filehandles
def mock_open(read_data=''):
class MockOpen(object):
'''
A helper function to create a mock to replace the use of `open`. It works
for "open" called directly or used as a context manager.
The "mock" argument is the mock object to configure. If "None" (the
default) then a "MagicMock" will be created for you, with the API limited
to methods or attributes available on standard file handles.
This class can be used to mock the use of "open()".
"read_data" is a string representing the contents of the file to be read.
By default, this is an empty string.
Optionally, "read_data" can be a dictionary mapping fnmatch.fnmatch()
patterns to strings. This allows the mocked filehandle to serve content for
more than one file path.
patterns to strings (or optionally, exceptions). This allows the mocked
filehandle to serve content for more than one file path.
.. code-block:: python
@ -222,39 +216,76 @@ def mock_open(read_data=''):
If the file path being opened does not match any of the glob expressions,
an IOError will be raised to simulate the file not existing.
Glob expressions will be attempted in iteration order, so if a file path
matches more than one glob expression it will match whichever is iterated
first. If a specifc iteration order is desired (and you are not running
Python >= 3.6), consider passing "read_data" as an OrderedDict.
Passing "read_data" as a string is equivalent to passing it with a glob
expression of "*".
expression of "*". That is to say, the below two invocations are
equivalent:
.. code-block:: python
mock_open(read_data='foo\n')
mock_open(read_data={'*': 'foo\n'})
Instead of a string representing file contents, "read_data" can map to an
exception, and that exception will be raised if a file matching that
pattern is opened:
.. code-block:: python
data = {
'/etc/*': IOError(errno.EACCES, 'Permission denied'),
'*': 'Hello world!\n',
}
with patch('salt.utils.files.fopen', mock_open(read_data=data):
do stuff
The above would raise an exception if any files within /etc are opened, but
would produce a mocked filehandle if any other file is opened.
Expressions will be attempted in dictionary iteration order (the exception
being "*" which is tried last), so if a file path matches more than one
fnmatch expression then the first match "wins". If your use case calls for
overlapping expressions, then an OrderedDict can be used to ensure that the
desired matching behavior occurs:
.. code-block:: python
data = OrderedDict()
data['/etc/foo.conf'] = 'Permission granted!'
data['/etc/*'] = IOError(errno.EACCES, 'Permission denied')
data['*'] = '*': 'Hello world!\n'
with patch('salt.utils.files.fopen', mock_open(read_data=data):
do stuff
'''
# Normalize read_data, Python 2 filehandles should never produce unicode
# types on read.
if not isinstance(read_data, dict):
read_data = {'*': read_data}
def __init__(self, read_data=''):
# Normalize read_data, Python 2 filehandles should never produce unicode
# types on read.
if not isinstance(read_data, dict):
read_data = {'*': read_data}
if six.PY2:
# .__class__() used here to preserve the dict class in the event that
# an OrderedDict was used.
new_read_data = read_data.__class__()
for key, val in six.iteritems(read_data):
try:
val = salt.utils.stringutils.to_str(val)
except TypeError:
if not isinstance(val, BaseException):
raise
new_read_data[key] = val
if six.PY2:
# .__class__() used here to preserve the dict class in the event that
# an OrderedDict was used.
new_read_data = read_data.__class__()
for key, val in six.iteritems(read_data):
try:
val = salt.utils.stringutils.to_str(val)
except TypeError:
if not isinstance(val, BaseException):
raise
new_read_data[key] = val
read_data = new_read_data
del new_read_data
read_data = new_read_data
del new_read_data
mock = MagicMock(name='open', spec=open)
mock.handles = {}
self.read_data = read_data
self.filehandles = {}
def _fopen_side_effect(name, *args, **kwargs):
for pat in read_data:
def __call__(self, name, *args, **kwargs):
'''
Match the file being opened to the patterns in the read_data and spawn
a mocked filehandle with the corresponding file contents.
'''
for pat in self.read_data:
if pat == '*':
continue
if fnmatch.fnmatch(name, pat):
@ -264,7 +295,7 @@ def mock_open(read_data=''):
# No non-glob match in read_data, fall back to '*'
matched_pattern = '*'
try:
file_contents = read_data[matched_pattern]
file_contents = self.read_data[matched_pattern]
try:
# Raise the exception if the matched file contents are an
# instance of an exception class.
@ -274,12 +305,37 @@ def mock_open(read_data=''):
# mocked filehandle.
pass
ret = MockFH(name, file_contents)
mock.handles.setdefault(name, []).append(ret)
self.filehandles.setdefault(name, []).append(ret)
return ret
except KeyError:
# No matching glob in read_data, treat this as a file that does
# not exist and raise the appropriate exception.
raise IOError(errno.ENOENT, 'No such file or directory', name)
mock.side_effect = _fopen_side_effect
return mock
def write_calls(self, path=None):
'''
Returns the contents passed to all .write() calls. Use `path` to narrow
the results to files matching a given pattern.
'''
ret = []
for filename, handles in six.iteritems(self.filehandles):
if path is None or fnmatch.fnmatch(filename, path):
for fh_ in handles:
ret.extend(fh_.write_calls)
return ret
def writelines_calls(self, path=None):
'''
Returns the contents passed to all .writelines() calls. Use `path` to
narrow the results to files matching a given pattern.
'''
ret = []
for filename, handles in six.iteritems(self.filehandles):
if path is None or fnmatch.fnmatch(filename, path):
for fh_ in handles:
ret.extend(fh_.writelines_calls)
return ret
# reimplement mock_open to support multiple filehandles
mock_open = MockOpen

24
tests/unit/modules/test_file.py
View File

@ -1022,7 +1022,7 @@ class FilemodLineTests(TestCase, LoaderModuleMockMixin):
patch('salt.utils.files.fopen', mock_open(read_data=file_content)), \
patch('salt.utils.atomicfile.atomic_open', mock_open()) as atomic_open_mock:
filemod.line(name, content=cfg_content, after='- /srv/salt', mode='insert')
handles = atomic_open_mock.handles[name]
handles = atomic_open_mock.filehandles[name]
# We should only have opened the file once
open_count = len(handles)
assert open_count == 1, open_count
@ -1072,7 +1072,7 @@ class FilemodLineTests(TestCase, LoaderModuleMockMixin):
patch('salt.utils.atomicfile.atomic_open',
mock_open()) as atomic_open_mock:
filemod.line(name, content=cfg_content, after=after_line, mode='insert', indent=False)
handles = atomic_open_mock.handles[name]
handles = atomic_open_mock.filehandles[name]
# We should only have opened the file once
open_count = len(handles)
assert open_count == 1, open_count
@ -1146,7 +1146,7 @@ class FilemodLineTests(TestCase, LoaderModuleMockMixin):
patch('salt.utils.atomicfile.atomic_open',
mock_open()) as atomic_open_mock:
filemod.line(name, content=cfg_content, before=before_line, mode='insert')
handles = atomic_open_mock.handles[name]
handles = atomic_open_mock.filehandles[name]
# We should only have opened the file once
open_count = len(handles)
assert open_count == 1, open_count
@ -1191,7 +1191,7 @@ class FilemodLineTests(TestCase, LoaderModuleMockMixin):
patch('salt.utils.atomicfile.atomic_open',
mock_open()) as atomic_open_mock:
filemod.line(name, content=cfg_content, before=b_line, after=a_line, mode='insert')
handles = atomic_open_mock.handles[name]
handles = atomic_open_mock.filehandles[name]
# We should only have opened the file once
open_count = len(handles)
assert open_count == 1, open_count
@ -1231,7 +1231,7 @@ class FilemodLineTests(TestCase, LoaderModuleMockMixin):
patch('salt.utils.atomicfile.atomic_open',
mock_open()) as atomic_open_mock:
filemod.line(name, content=cfg_content, location='start', mode='insert')
handles = atomic_open_mock.handles[name]
handles = atomic_open_mock.filehandles[name]
# We should only have opened the file once
open_count = len(handles)
assert open_count == 1, open_count
@ -1271,7 +1271,7 @@ class FilemodLineTests(TestCase, LoaderModuleMockMixin):
patch('salt.utils.atomicfile.atomic_open',
mock_open()) as atomic_open_mock:
filemod.line(name, content=cfg_content, location='end', mode='insert')
handles = atomic_open_mock.handles[name]
handles = atomic_open_mock.filehandles[name]
# We should only have opened the file once
open_count = len(handles)
assert open_count == 1, open_count
@ -1309,7 +1309,7 @@ class FilemodLineTests(TestCase, LoaderModuleMockMixin):
patch('salt.utils.atomicfile.atomic_open',
mock_open()) as atomic_open_mock:
filemod.line(name, content=cfg_content, before='exit 0', mode='ensure')
handles = atomic_open_mock.handles[name]
handles = atomic_open_mock.filehandles[name]
# We should only have opened the file once
open_count = len(handles)
assert open_count == 1, open_count
@ -1345,7 +1345,7 @@ class FilemodLineTests(TestCase, LoaderModuleMockMixin):
patch('salt.utils.atomicfile.atomic_open',
mock_open()) as atomic_open_mock:
filemod.line(name, content=cfg_content, after='/etc/init.d/someservice restart', mode='ensure')
handles = atomic_open_mock.handles[name]
handles = atomic_open_mock.filehandles[name]
# We should only have opened the file once
open_count = len(handles)
assert open_count == 1, open_count
@ -1381,7 +1381,7 @@ class FilemodLineTests(TestCase, LoaderModuleMockMixin):
patch('salt.utils.atomicfile.atomic_open',
mock_open()) as atomic_open_mock:
filemod.line(name, content=cfg_content, after=_after, before=_before, mode='ensure')
handles = atomic_open_mock.handles[name]
handles = atomic_open_mock.filehandles[name]
# We should only have opened the file once
open_count = len(handles)
assert open_count == 1, open_count
@ -1418,7 +1418,7 @@ class FilemodLineTests(TestCase, LoaderModuleMockMixin):
mock_open()) as atomic_open_mock:
result = filemod.line('foo', content=cfg_content, after=_after, before=_before, mode='ensure')
# We should not have opened the file
assert not atomic_open_mock.handles
assert not atomic_open_mock.filehandles
# No changes should have been made
assert result is False
@ -1475,7 +1475,7 @@ class FilemodLineTests(TestCase, LoaderModuleMockMixin):
patch('salt.utils.files.fopen', files_fopen), \
patch('salt.utils.atomicfile.atomic_open', mock_open()) as atomic_open_mock:
filemod.line(name, content=content, mode='delete')
handles = atomic_open_mock.handles[name]
handles = atomic_open_mock.filehandles[name]
# We should only have opened the file once
open_count = len(handles)
assert open_count == 1, open_count
@ -1515,7 +1515,7 @@ class FilemodLineTests(TestCase, LoaderModuleMockMixin):
patch('salt.utils.files.fopen', files_fopen), \
patch('salt.utils.atomicfile.atomic_open', mock_open()) as atomic_open_mock:
filemod.line(name, content='- /srv/natrium-chloride', match=match, mode='replace')
handles = atomic_open_mock.handles[name]
handles = atomic_open_mock.filehandles[name]
# We should only have opened the file once
open_count = len(handles)
assert open_count == 1, open_count

4
tests/unit/modules/test_linux_sysctl.py
View File

@ -121,9 +121,7 @@ class LinuxSysctlTestCase(TestCase, LoaderModuleMockMixin):
{'salt.utils.systemd.booted': True,
'salt.utils.systemd.version': 232}):
linux_sysctl.persist('net.ipv4.ip_forward', 1, config=config)
writes = []
for fh_ in m_open.handles[config]:
writes.extend(fh_.write_calls)
writes = m_open.write_calls()
assert writes == [
'#\n# Kernel sysctl configuration\n#\n'
], writes

14
tests/unit/modules/test_mac_sysctl.py
View File

@ -88,11 +88,9 @@ class DarwinSysctlTestCase(TestCase, LoaderModuleMockMixin):
patch('os.path.isfile', isfile_mock):
mac_sysctl.persist('net.inet.icmp.icmplim', 50, config=config)
# We only should have opened the one file
num_handles = len(m_open.handles)
num_handles = len(m_open.filehandles)
assert num_handles == 1, num_handles
writes = []
for fh_ in m_open.handles[config]:
writes.extend(fh_.write_calls)
writes = m_open.write_calls()
# We should have called .write() only once, with the expected
# content
num_writes = len(writes)
@ -118,11 +116,7 @@ class DarwinSysctlTestCase(TestCase, LoaderModuleMockMixin):
patch('os.path.isfile', isfile_mock):
mac_sysctl.persist('net.inet.icmp.icmplim', 50, config=config)
# We only should have opened the one file
num_handles = len(m_open.handles)
num_handles = len(m_open.filehandles)
assert num_handles == 1, num_handles
writes = []
# We should have called .writelines() only once, with the expected
# content
for fh_ in m_open.handles[config]:
writes.extend(fh_.writelines_calls)
writes = m_open.writelines_calls()
assert writes == writelines_calls, writes