INFRA-1009 - Refactoring

tests/conftest.py

@@ -1,32 +1,34 @@
 #!/usr/bin/env python
 # -*- coding: utf-8
 import itertools
+
 import pytest
 
 pytest_plugins = ['helpers_namespace']
 
 
 @pytest.helpers.register
-def assert_any_call_with_useful_error_message(mockk, uri, *args):
+def assert_any_call(mockk, first, *args):
     __tracebackhide__ = True
 
     def _assertion():
-        mockk.assert_any_call(uri, *args)
+        mockk.assert_any_call(first, *args)
 
-    _add_useful_error_message(_assertion, mockk, uri, args)
+    _add_useful_error_message(_assertion, mockk, first, args)
 
 
 @pytest.helpers.register
-def assert_no_calls(mockk):
+def assert_no_calls(mockk, uri=None):
     __tracebackhide__ = True
 
     def _assertion():
-        assert mockk.call_count == 0
+        calls = [call[0] for call in mockk.call_args_list if (uri is None or call[0][0] == uri)]
+        assert len(calls) == 0
 
     _add_useful_error_message(_assertion, mockk, None, None)
 
 
-def _add_useful_error_message(assertion, mockk, uri, args):
+def _add_useful_error_message(assertion, mockk, first, args):
     """
     If an AssertionError is raised in the assert, find any other calls on mock where the first parameter is uri and
     append those calls to the AssertionErrors message to more easily find the cause of the test failure.
@@ -35,9 +37,9 @@ def _add_useful_error_message(assertion, mockk, uri, args):
     try:
         assertion()
     except AssertionError as ae:
-        other_calls = [call[0] for call in mockk.call_args_list if (uri is None or call[0][0] == uri)]
+        other_calls = [call[0] for call in mockk.call_args_list if (first is None or call[0][0] == first)]
         if other_calls:
-            extra_info = '\n\nURI {} got the following other calls:\n{}\n'.format(uri, '\n'.join(
+            extra_info = '\n\nURI {} got the following other calls:\n{}\n'.format(first, '\n'.join(
                 _format_call(call) for call in other_calls))
             if len(other_calls) == 1 and len(other_calls[0]) == 2 and args is not None:
                 extra_info += _add_argument_diff(other_calls[0][1], args[0])
@@ -52,21 +54,21 @@ def _add_argument_diff(actual, expected, indent=0, acc=None):
         acc = []
         first = True
     if type(actual) != type(expected):
-        acc.append("{}{!r} {} {!r}".format(" "*indent*2, actual, "==" if actual == expected else "!=", expected))
+        acc.append("{}{!r} {} {!r}".format(" " * indent * 2, actual, "==" if actual == expected else "!=", expected))
     elif isinstance(actual, dict):
-        for k in set(actual.keys()+expected.keys()):
-            acc.append("{}{}:".format(" "*indent*2, k))
+        for k in set(actual.keys() + expected.keys()):
+            acc.append("{}{}:".format(" " * indent * 2, k))
             a = actual.get(k)
             e = expected.get(k)
             if a != e:
-                _add_argument_diff(a, e, indent+1, acc)
+                _add_argument_diff(a, e, indent + 1, acc)
     elif isinstance(actual, list):
         for a, e in itertools.izip_longest(actual, expected):
-            acc.append("{}-".format(" "*indent*2))
+            acc.append("{}-".format(" " * indent * 2))
             if a != e:
-                _add_argument_diff(a, e, indent+1, acc)
+                _add_argument_diff(a, e, indent + 1, acc)
     else:
-        acc.append("{}{!r} {} {!r}".format(" "*indent*2, actual, "==" if actual == expected else "!=", expected))
+        acc.append("{}{!r} {} {!r}".format(" " * indent * 2, actual, "==" if actual == expected else "!=", expected))
     if first:
         return "\n".join(acc)
 

tests/k8s/test_service.py

@@ -40,7 +40,7 @@ def test_service_created_if_not_exists(self, post, get):
         service = create_default_service()
         service.save()
         assert service._new
-        pytest.helpers.assert_any_call_with_useful_error_message(post, SERVICES_URI, service.as_dict())
+        pytest.helpers.assert_any_call(post, SERVICES_URI, service.as_dict())
 
     @mock.patch('k8s.client.Client.get')
     @mock.patch('k8s.client.Client.put')
@@ -83,14 +83,14 @@ def test_get_or_create_service_not_new(self, put, get):
         assert from_api.metadata.labels
         assert from_api.metadata.name == service.metadata.name
         from_api.save()
-        pytest.helpers.assert_any_call_with_useful_error_message(put, SERVICES_URI + SERVICE_NAME, from_api.as_dict())
+        pytest.helpers.assert_any_call(put, SERVICES_URI + SERVICE_NAME, from_api.as_dict())
 
     @mock.patch('k8s.client.Client.delete')
     def test_service_deleted(self, delete):
         Service.delete(SERVICE_NAME, SERVICE_NAMESPACE)
 
         # call delete with service_name
-        pytest.helpers.assert_any_call_with_useful_error_message(delete, (SERVICES_URI + SERVICE_NAME))
+        pytest.helpers.assert_any_call(delete, (SERVICES_URI + SERVICE_NAME))
 
 
 def create_default_service():