Add intentional unit tests for basic._set_cwd and common.dict_merge (#70283)

* Add unit tests for basic._set_cwd

* incidental coverage for dict_merge

* add test for async stderr inclusion
pull/70326/head
jctanner 4 years ago committed by GitHub
parent 29169ae847
commit b019029bf3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

@ -0,0 +1,195 @@
# -*- coding: utf-8 -*-
# Copyright (c) 2018 Ansible Project
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
# Make coding more python3-ish
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import json
import os
import shutil
import tempfile
import pytest
from units.compat.mock import patch, MagicMock
from ansible.module_utils._text import to_bytes
from ansible.module_utils import basic
class TestAnsibleModuleSetCwd:
def test_set_cwd(self, monkeypatch):
'''make sure /tmp is used'''
def mock_getcwd():
return '/tmp'
def mock_access(path, perm):
return True
def mock_chdir(path):
pass
monkeypatch.setattr(os, 'getcwd', mock_getcwd)
monkeypatch.setattr(os, 'access', mock_access)
monkeypatch.setattr(basic, '_ANSIBLE_ARGS', to_bytes(json.dumps({'ANSIBLE_MODULE_ARGS': {}})))
with patch('time.time', return_value=42):
am = basic.AnsibleModule(argument_spec={})
result = am._set_cwd()
assert result == '/tmp'
def test_set_cwd_unreadable_use_self_tmpdir(self, monkeypatch):
'''pwd is not readable, use instance's tmpdir property'''
def mock_getcwd():
return '/tmp'
def mock_access(path, perm):
if path == '/tmp' and perm == 4:
return False
return True
def mock_expandvars(var):
if var == '$HOME':
return '/home/foobar'
return var
def mock_gettempdir():
return '/tmp/testdir'
def mock_chdir(path):
if path == '/tmp':
raise Exception()
return
monkeypatch.setattr(os, 'getcwd', mock_getcwd)
monkeypatch.setattr(os, 'chdir', mock_chdir)
monkeypatch.setattr(os, 'access', mock_access)
monkeypatch.setattr(os.path, 'expandvars', mock_expandvars)
monkeypatch.setattr(basic, '_ANSIBLE_ARGS', to_bytes(json.dumps({'ANSIBLE_MODULE_ARGS': {}})))
with patch('time.time', return_value=42):
am = basic.AnsibleModule(argument_spec={})
am._tmpdir = '/tmp2'
result = am._set_cwd()
assert result == am._tmpdir
def test_set_cwd_unreadable_use_home(self, monkeypatch):
'''cwd and instance tmpdir are unreadable, use home'''
def mock_getcwd():
return '/tmp'
def mock_access(path, perm):
if path in ['/tmp', '/tmp2'] and perm == 4:
return False
return True
def mock_expandvars(var):
if var == '$HOME':
return '/home/foobar'
return var
def mock_gettempdir():
return '/tmp/testdir'
def mock_chdir(path):
if path == '/tmp':
raise Exception()
return
monkeypatch.setattr(os, 'getcwd', mock_getcwd)
monkeypatch.setattr(os, 'chdir', mock_chdir)
monkeypatch.setattr(os, 'access', mock_access)
monkeypatch.setattr(os.path, 'expandvars', mock_expandvars)
monkeypatch.setattr(basic, '_ANSIBLE_ARGS', to_bytes(json.dumps({'ANSIBLE_MODULE_ARGS': {}})))
with patch('time.time', return_value=42):
am = basic.AnsibleModule(argument_spec={})
am._tmpdir = '/tmp2'
result = am._set_cwd()
assert result == '/home/foobar'
def test_set_cwd_unreadable_use_gettempdir(self, monkeypatch):
'''fallback to tempfile.gettempdir'''
thisdir = None
def mock_getcwd():
return '/tmp'
def mock_access(path, perm):
if path in ['/tmp', '/tmp2', '/home/foobar'] and perm == 4:
return False
return True
def mock_expandvars(var):
if var == '$HOME':
return '/home/foobar'
return var
def mock_gettempdir():
return '/tmp3'
def mock_chdir(path):
if path == '/tmp':
raise Exception()
thisdir = path
monkeypatch.setattr(os, 'getcwd', mock_getcwd)
monkeypatch.setattr(os, 'chdir', mock_chdir)
monkeypatch.setattr(os, 'access', mock_access)
monkeypatch.setattr(os.path, 'expandvars', mock_expandvars)
monkeypatch.setattr(basic, '_ANSIBLE_ARGS', to_bytes(json.dumps({'ANSIBLE_MODULE_ARGS': {}})))
with patch('time.time', return_value=42):
am = basic.AnsibleModule(argument_spec={})
am._tmpdir = '/tmp2'
monkeypatch.setattr(tempfile, 'gettempdir', mock_gettempdir)
result = am._set_cwd()
assert result == '/tmp3'
def test_set_cwd_unreadable_use_None(self, monkeypatch):
'''all paths are unreable, should return None and not an exception'''
def mock_getcwd():
return '/tmp'
def mock_access(path, perm):
if path in ['/tmp', '/tmp2', '/tmp3', '/home/foobar'] and perm == 4:
return False
return True
def mock_expandvars(var):
if var == '$HOME':
return '/home/foobar'
return var
def mock_gettempdir():
return '/tmp3'
def mock_chdir(path):
if path == '/tmp':
raise Exception()
monkeypatch.setattr(os, 'getcwd', mock_getcwd)
monkeypatch.setattr(os, 'chdir', mock_chdir)
monkeypatch.setattr(os, 'access', mock_access)
monkeypatch.setattr(os.path, 'expandvars', mock_expandvars)
monkeypatch.setattr(basic, '_ANSIBLE_ARGS', to_bytes(json.dumps({'ANSIBLE_MODULE_ARGS': {}})))
with patch('time.time', return_value=42):
am = basic.AnsibleModule(argument_spec={})
am._tmpdir = '/tmp2'
monkeypatch.setattr(tempfile, 'gettempdir', mock_gettempdir)
result = am._set_cwd()
assert result is None

@ -117,3 +117,19 @@ class DictMergeTestCase(unittest.TestCase):
self.assertTrue('b2' in result)
self.assertTrue(result['b3'])
self.assertTrue(result['b4'])
class AzureIncidentalTestCase(unittest.TestCase):
def test_dict_merge_invalid_dict(self):
''' if b is not a dict, return b '''
res = dict_merge({}, None)
self.assertEqual(res, None)
def test_merge_sub_dicts(self):
'''merge sub dicts '''
a = {'a': {'a1': 1}}
b = {'a': {'b1': 2}}
c = {'a': {'a1': 1, 'b1': 2}}
res = dict_merge(a, b)
self.assertEqual(res, c)

@ -0,0 +1,57 @@
# Copyright (c) 2017 Ansible Project
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import os
import json
import shutil
import tempfile
import pytest
from units.compat.mock import patch, MagicMock
from ansible.modules import async_wrapper
from pprint import pprint
class TestAsyncWrapper:
def test_run_module(self, monkeypatch):
def mock_get_interpreter(module_path):
return ['/usr/bin/python']
module_result = {'rc': 0}
module_lines = [
'#!/usr/bin/python',
'import sys',
'sys.stderr.write("stderr stuff")',
"print('%s')" % json.dumps(module_result)
]
module_data = '\n'.join(module_lines) + '\n'
module_data = module_data.encode('utf-8')
workdir = tempfile.mkdtemp()
fh, fn = tempfile.mkstemp(dir=workdir)
with open(fn, 'wb') as f:
f.write(module_data)
command = fn
jobid = 0
jobpath = os.path.join(os.path.dirname(command), 'job')
monkeypatch.setattr(async_wrapper, '_get_interpreter', mock_get_interpreter)
res = async_wrapper._run_module(command, jobid, jobpath)
with open(os.path.join(workdir, 'job'), 'r') as f:
jres = json.loads(f.read())
shutil.rmtree(workdir)
assert jres.get('rc') == 0
assert jres.get('stderr') == 'stderr stuff'
Loading…
Cancel
Save