Merge pull request #59 from CaptTrews/master

Updated from network content collector

Reviewed-by: https://github.com/apps/ansible-zuul

6 files changed, 423 insertions, 0 deletions

diff --git a/tests/unit/mock/__init__.py b/tests/unit/mock/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/tests/unit/mock/__init__.py
diff --git a/tests/unit/mock/loader.py b/tests/unit/mock/loader.py
new file mode 100644
index 0000000..c21188e
--- /dev/null
+++ b/tests/unit/mock/loader.py
@@ -0,0 +1,116 @@
+# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
+#
+# This file is part of Ansible
+#
+# Ansible 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, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible 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 Ansible.  If not, see <http://www.gnu.org/licenses/>.
+
+# Make coding more python3-ish
+from __future__ import absolute_import, division, print_function
+
+__metaclass__ = type
+
+import os
+
+from ansible.errors import AnsibleParserError
+from ansible.parsing.dataloader import DataLoader
+from ansible.module_utils._text import to_bytes, to_text
+
+
+class DictDataLoader(DataLoader):
+    def __init__(self, file_mapping=None):
+        file_mapping = {} if file_mapping is None else file_mapping
+        assert type(file_mapping) == dict
+
+        super(DictDataLoader, self).__init__()
+
+        self._file_mapping = file_mapping
+        self._build_known_directories()
+        self._vault_secrets = None
+
+    def load_from_file(self, path, cache=True, unsafe=False):
+        path = to_text(path)
+        if path in self._file_mapping:
+            return self.load(self._file_mapping[path], path)
+        return None
+
+    # TODO: the real _get_file_contents returns a bytestring, so we actually convert the
+    #       unicode/text it's created with to utf-8
+    def _get_file_contents(self, path):
+        path = to_text(path)
+        if path in self._file_mapping:
+            return (to_bytes(self._file_mapping[path]), False)
+        else:
+            raise AnsibleParserError("file not found: %s" % path)
+
+    def path_exists(self, path):
+        path = to_text(path)
+        return path in self._file_mapping or path in self._known_directories
+
+    def is_file(self, path):
+        path = to_text(path)
+        return path in self._file_mapping
+
+    def is_directory(self, path):
+        path = to_text(path)
+        return path in self._known_directories
+
+    def list_directory(self, path):
+        ret = []
+        path = to_text(path)
+        for x in list(self._file_mapping.keys()) + self._known_directories:
+            if x.startswith(path):
+                if os.path.dirname(x) == path:
+                    ret.append(os.path.basename(x))
+        return ret
+
+    def is_executable(self, path):
+        # FIXME: figure out a way to make paths return true for this
+        return False
+
+    def _add_known_directory(self, directory):
+        if directory not in self._known_directories:
+            self._known_directories.append(directory)
+
+    def _build_known_directories(self):
+        self._known_directories = []
+        for path in self._file_mapping:
+            dirname = os.path.dirname(path)
+            while dirname not in ("/", ""):
+                self._add_known_directory(dirname)
+                dirname = os.path.dirname(dirname)
+
+    def push(self, path, content):
+        rebuild_dirs = False
+        if path not in self._file_mapping:
+            rebuild_dirs = True
+
+        self._file_mapping[path] = content
+
+        if rebuild_dirs:
+            self._build_known_directories()
+
+    def pop(self, path):
+        if path in self._file_mapping:
+            del self._file_mapping[path]
+            self._build_known_directories()
+
+    def clear(self):
+        self._file_mapping = dict()
+        self._known_directories = []
+
+    def get_basedir(self):
+        return os.getcwd()
+
+    def set_vault_secrets(self, vault_secrets):
+        self._vault_secrets = vault_secrets
diff --git a/tests/unit/mock/path.py b/tests/unit/mock/path.py
new file mode 100644
index 0000000..aea8ba1
--- /dev/null
+++ b/tests/unit/mock/path.py
@@ -0,0 +1,7 @@
+from ansible_collections.vyos.vyos.tests.unit.compat.mock import MagicMock
+from ansible.utils.path import unfrackpath
+
+
+mock_unfrackpath_noop = MagicMock(
+    spec_set=unfrackpath, side_effect=lambda x, *args, **kwargs: x
+)
diff --git a/tests/unit/mock/procenv.py b/tests/unit/mock/procenv.py
new file mode 100644
index 0000000..1587949
--- /dev/null
+++ b/tests/unit/mock/procenv.py
@@ -0,0 +1,94 @@
+# (c) 2016, Matt Davis <mdavis@ansible.com>
+# (c) 2016, Toshio Kuratomi <tkuratomi@ansible.com>
+#
+# This file is part of Ansible
+#
+# Ansible 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, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible 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 Ansible.  If not, see <http://www.gnu.org/licenses/>.
+
+# Make coding more python3-ish
+from __future__ import absolute_import, division, print_function
+
+__metaclass__ = type
+
+import sys
+import json
+
+from contextlib import contextmanager
+from io import BytesIO, StringIO
+from ansible_collections.vyos.vyos.tests.unit.compat import unittest
+from ansible.module_utils.six import PY3
+from ansible.module_utils._text import to_bytes
+
+
+@contextmanager
+def swap_stdin_and_argv(stdin_data="", argv_data=tuple()):
+    """
+    context manager that temporarily masks the test runner's values for stdin and argv
+    """
+    real_stdin = sys.stdin
+    real_argv = sys.argv
+
+    if PY3:
+        fake_stream = StringIO(stdin_data)
+        fake_stream.buffer = BytesIO(to_bytes(stdin_data))
+    else:
+        fake_stream = BytesIO(to_bytes(stdin_data))
+
+    try:
+        sys.stdin = fake_stream
+        sys.argv = argv_data
+
+        yield
+    finally:
+        sys.stdin = real_stdin
+        sys.argv = real_argv
+
+
+@contextmanager
+def swap_stdout():
+    """
+    context manager that temporarily replaces stdout for tests that need to verify output
+    """
+    old_stdout = sys.stdout
+
+    if PY3:
+        fake_stream = StringIO()
+    else:
+        fake_stream = BytesIO()
+
+    try:
+        sys.stdout = fake_stream
+
+        yield fake_stream
+    finally:
+        sys.stdout = old_stdout
+
+
+class ModuleTestCase(unittest.TestCase):
+    def setUp(self, module_args=None):
+        if module_args is None:
+            module_args = {
+                "_ansible_remote_tmp": "/tmp",
+                "_ansible_keep_remote_files": False,
+            }
+
+        args = json.dumps(dict(ANSIBLE_MODULE_ARGS=module_args))
+
+        # unittest doesn't have a clean place to use a context manager, so we have to enter/exit manually
+        self.stdin_swap = swap_stdin_and_argv(stdin_data=args)
+        self.stdin_swap.__enter__()
+
+    def tearDown(self):
+        # unittest doesn't have a clean place to use a context manager, so we have to enter/exit manually
+        self.stdin_swap.__exit__(None, None, None)
diff --git a/tests/unit/mock/vault_helper.py b/tests/unit/mock/vault_helper.py
new file mode 100644
index 0000000..b34ae13
--- /dev/null
+++ b/tests/unit/mock/vault_helper.py
@@ -0,0 +1,42 @@
+# Ansible 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, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible 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 Ansible.  If not, see <http://www.gnu.org/licenses/>.
+
+# Make coding more python3-ish
+from __future__ import absolute_import, division, print_function
+
+__metaclass__ = type
+
+from ansible.module_utils._text import to_bytes
+
+from ansible.parsing.vault import VaultSecret
+
+
+class TextVaultSecret(VaultSecret):
+    """A secret piece of text. ie, a password. Tracks text encoding.
+
+    The text encoding of the text may not be the default text encoding so
+    we keep track of the encoding so we encode it to the same bytes."""
+
+    def __init__(self, text, encoding=None, errors=None, _bytes=None):
+        super(TextVaultSecret, self).__init__()
+        self.text = text
+        self.encoding = encoding or "utf-8"
+        self._bytes = _bytes
+        self.errors = errors or "strict"
+
+    @property
+    def bytes(self):
+        """The text encoded with encoding, unless we specifically set _bytes."""
+        return self._bytes or to_bytes(
+            self.text, encoding=self.encoding, errors=self.errors
+        )
diff --git a/tests/unit/mock/yaml_helper.py b/tests/unit/mock/yaml_helper.py
new file mode 100644
index 0000000..1a945f1
--- /dev/null
+++ b/tests/unit/mock/yaml_helper.py
@@ -0,0 +1,164 @@
+import io
+import yaml
+
+from ansible.module_utils.six import PY3
+from ansible.parsing.yaml.loader import AnsibleLoader
+from ansible.parsing.yaml.dumper import AnsibleDumper
+
+
+class YamlTestUtils(object):
+    """Mixin class to combine with a unittest.TestCase subclass."""
+
+    def _loader(self, stream):
+        """Vault related tests will want to override this.
+
+        Vault cases should setup a AnsibleLoader that has the vault password."""
+        return AnsibleLoader(stream)
+
+    def _dump_stream(self, obj, stream, dumper=None):
+        """Dump to a py2-unicode or py3-string stream."""
+        if PY3:
+            return yaml.dump(obj, stream, Dumper=dumper)
+        else:
+            return yaml.dump(obj, stream, Dumper=dumper, encoding=None)
+
+    def _dump_string(self, obj, dumper=None):
+        """Dump to a py2-unicode or py3-string"""
+        if PY3:
+            return yaml.dump(obj, Dumper=dumper)
+        else:
+            return yaml.dump(obj, Dumper=dumper, encoding=None)
+
+    def _dump_load_cycle(self, obj):
+        # Each pass though a dump or load revs the 'generation'
+        # obj to yaml string
+        string_from_object_dump = self._dump_string(obj, dumper=AnsibleDumper)
+
+        # wrap a stream/file like StringIO around that yaml
+        stream_from_object_dump = io.StringIO(string_from_object_dump)
+        loader = self._loader(stream_from_object_dump)
+        # load the yaml stream to create a new instance of the object (gen 2)
+        obj_2 = loader.get_data()
+
+        # dump the gen 2 objects directory to strings
+        string_from_object_dump_2 = self._dump_string(
+            obj_2, dumper=AnsibleDumper
+        )
+
+        # The gen 1 and gen 2 yaml strings
+        self.assertEqual(string_from_object_dump, string_from_object_dump_2)
+        # the gen 1 (orig) and gen 2 py object
+        self.assertEqual(obj, obj_2)
+
+        # again! gen 3... load strings into py objects
+        stream_3 = io.StringIO(string_from_object_dump_2)
+        loader_3 = self._loader(stream_3)
+        obj_3 = loader_3.get_data()
+
+        string_from_object_dump_3 = self._dump_string(
+            obj_3, dumper=AnsibleDumper
+        )
+
+        self.assertEqual(obj, obj_3)
+        # should be transitive, but...
+        self.assertEqual(obj_2, obj_3)
+        self.assertEqual(string_from_object_dump, string_from_object_dump_3)
+
+    def _old_dump_load_cycle(self, obj):
+        """Dump the passed in object to yaml, load it back up, dump again, compare."""
+        stream = io.StringIO()
+
+        yaml_string = self._dump_string(obj, dumper=AnsibleDumper)
+        self._dump_stream(obj, stream, dumper=AnsibleDumper)
+
+        yaml_string_from_stream = stream.getvalue()
+
+        # reset stream
+        stream.seek(0)
+
+        loader = self._loader(stream)
+        # loader = AnsibleLoader(stream, vault_password=self.vault_password)
+        obj_from_stream = loader.get_data()
+
+        stream_from_string = io.StringIO(yaml_string)
+        loader2 = self._loader(stream_from_string)
+        # loader2 = AnsibleLoader(stream_from_string, vault_password=self.vault_password)
+        obj_from_string = loader2.get_data()
+
+        stream_obj_from_stream = io.StringIO()
+        stream_obj_from_string = io.StringIO()
+
+        if PY3:
+            yaml.dump(
+                obj_from_stream, stream_obj_from_stream, Dumper=AnsibleDumper
+            )
+            yaml.dump(
+                obj_from_stream, stream_obj_from_string, Dumper=AnsibleDumper
+            )
+        else:
+            yaml.dump(
+                obj_from_stream,
+                stream_obj_from_stream,
+                Dumper=AnsibleDumper,
+                encoding=None,
+            )
+            yaml.dump(
+                obj_from_stream,
+                stream_obj_from_string,
+                Dumper=AnsibleDumper,
+                encoding=None,
+            )
+
+        yaml_string_stream_obj_from_stream = stream_obj_from_stream.getvalue()
+        yaml_string_stream_obj_from_string = stream_obj_from_string.getvalue()
+
+        stream_obj_from_stream.seek(0)
+        stream_obj_from_string.seek(0)
+
+        if PY3:
+            yaml_string_obj_from_stream = yaml.dump(
+                obj_from_stream, Dumper=AnsibleDumper
+            )
+            yaml_string_obj_from_string = yaml.dump(
+                obj_from_string, Dumper=AnsibleDumper
+            )
+        else:
+            yaml_string_obj_from_stream = yaml.dump(
+                obj_from_stream, Dumper=AnsibleDumper, encoding=None
+            )
+            yaml_string_obj_from_string = yaml.dump(
+                obj_from_string, Dumper=AnsibleDumper, encoding=None
+            )
+
+        assert yaml_string == yaml_string_obj_from_stream
+        assert (
+            yaml_string
+            == yaml_string_obj_from_stream
+            == yaml_string_obj_from_string
+        )
+        assert (
+            yaml_string
+            == yaml_string_obj_from_stream
+            == yaml_string_obj_from_string
+            == yaml_string_stream_obj_from_stream
+            == yaml_string_stream_obj_from_string
+        )
+        assert obj == obj_from_stream
+        assert obj == obj_from_string
+        assert obj == yaml_string_obj_from_stream
+        assert obj == yaml_string_obj_from_string
+        assert (
+            obj
+            == obj_from_stream
+            == obj_from_string
+            == yaml_string_obj_from_stream
+            == yaml_string_obj_from_string
+        )
+        return {
+            "obj": obj,
+            "yaml_string": yaml_string,
+            "yaml_string_from_stream": yaml_string_from_stream,
+            "obj_from_stream": obj_from_stream,
+            "obj_from_string": obj_from_string,
+            "yaml_string_obj_from_string": yaml_string_obj_from_string,
+        }