--- /dev/null
+# -*- coding: utf-8 -*-
+##
+# Copyright 2018 University of Bristol - High Performance Networks Research
+# Group
+# All Rights Reserved.
+#
+# Contributors: Anderson Bravalheri, Dimitrios Gkounis, Abubakar Siddique
+# Muqaddas, Navdeep Uniyal, Reza Nejabati and Dimitra Simeonidou
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+# For those usages not covered by the Apache License, Version 2.0 please
+# contact with: <highperformance-networks@bristol.ac.uk>
+#
+# Neither the name of the University of Bristol nor the names of its
+# contributors may be used to endorse or promote products derived from
+# this software without specific prior written permission.
+#
+# This work has been performed in the context of DCMS UK 5G Testbeds
+# & Trials Programme and in the framework of the Metro-Haul project -
+# funded by the European Commission under Grant number 761727 through the
+# Horizon 2020 and 5G-PPP programmes.
+##
+
+import logging
+import unittest
+from collections import defaultdict
+
+from io import StringIO
+
+from unittest.mock import MagicMock, patch
+
+logger = logging.getLogger()
+
+
+class TestCaseWithLogging(unittest.TestCase):
+ """Attach a special handler to the root logger, capturing the logs in a
+ internal buffer (caplog property).
+
+ To retrieve the logs, do::
+
+ self.caplog.getvalue()
+ """
+ def setUp(self):
+ super(TestCaseWithLogging, self).setUp()
+ self.logger = logging.getLogger()
+ self.caplog = StringIO()
+ self.log_handler = logging.StreamHandler(self.caplog)
+ self.logger.addHandler(self.log_handler)
+ self.logger.setLevel(logging.NOTSET)
+
+ def tearDown(self):
+ super(TestCaseWithLogging, self).tearDown()
+ self.log_handler.close()
+ self.logger.removeHandler(self.log_handler)
+
+
+def mock_imports(modules, preserve=()):
+ """Given a list of modules, mock everything, unless listed in the preserve
+ argument.
+ """
+ # Ensure iterable
+ if isinstance(modules, str):
+ modules = (modules,)
+ if isinstance(preserve, str):
+ preserve = (preserve,)
+
+ # First expand the list, since children modules needs their parent also
+ # mocked most of the time.
+ # Example: ['Crypto.PublicKey'] => ['Crypto', 'Crypto.PublicKey']
+ all_modules = []
+ for name in modules:
+ parts = name.split('.')
+ compound_name = []
+ for part in parts:
+ compound_name.append(part)
+ all_modules.append('.'.join(compound_name))
+
+ all_modules = set(m for m in all_modules if m not in preserve)
+ for module in all_modules:
+ logger.info('Mocking module `%s`', module)
+
+ mocks = {module: MagicMock() for module in all_modules}
+
+ return patch.dict('sys.modules', **mocks)
+
+
+def mock_dict(**kwargs):
+ """Create a dict that always respond something.
+
+ Arguments:
+ **kwargs: certain items that should be set in the created object
+ """
+ response = defaultdict(MagicMock)
+ for k, v in kwargs.items():
+ response[k] = v
+
+ return response
+
+
+def mock_object(**kwargs):
+ """Create an object that always respond something.
+
+ Arguments:
+ **kwargs: certain attributes that should be set in the created object
+ """
+ response = MagicMock()
+ for k, v in kwargs.items():
+ setattr(response, k, v)
+
+ return response
projects / osm / RO.git / blobdiff