Add intentional unit tests for basic._set_cwd and common.dict_merge

test/units/module_utils/basic/test_set_cwd.py

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

test/units/module_utils/common/test_dict_transformations.py

@@ -117,3 +117,19 @@ def test_dict_merge(self):
         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)

test/units/modules/test_async_wrapper.py

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