Remove vumi.tests.utils.Mocking and replace its uses with normal patches

vumi/blinkenlights/tests/test_metrics.py

@@ -2,7 +2,7 @@
 from twisted.internet import reactor
 from twisted.internet.defer import inlineCallbacks, Deferred
 from vumi.blinkenlights import metrics
-from vumi.tests.utils import get_stubbed_worker, get_stubbed_channel, mocking
+from vumi.tests.utils import get_stubbed_worker, get_stubbed_channel
 from vumi.message import Message
 from vumi.service import Worker
 
@@ -236,18 +236,26 @@ def test_inc_and_poll(self):
 
 
 class TestTimer(TestCase, CheckValuesMixin):
+
+    def patch_time(self, starting_value):
+        def fake_time():
+            return self._fake_time
+        self.patch(time, 'time', fake_time)
+        self._fake_time = starting_value
+
+    def incr_fake_time(self, value):
+        self._fake_time += value
+
     def test_start_and_stop(self):
         timer = metrics.Timer("foo")
         timer.manage("prefix.")
-        with mocking(time.time) as mockt:
-            mockt.return_value = 12345.0
-            timer.start()
-            try:
-                mockt.return_value += 0.1  # feign sleep
-            finally:
-                timer.stop()
-            self.check_poll_func(timer, 1, lambda x: 0.09 < x < 0.11)
-            self.check_poll(timer, [])
+
+        self.patch_time(12345.0)
+        timer.start()
+        self.incr_fake_time(0.1)
+        timer.stop()
+        self.check_poll_func(timer, 1, lambda x: 0.09 < x < 0.11)
+        self.check_poll(timer, [])
 
     def test_already_started(self):
         timer = metrics.Timer("foo")
@@ -258,24 +266,22 @@ def test_already_started(self):
     def test_context_manager(self):
         timer = metrics.Timer("foo")
         timer.manage("prefix.")
-        with mocking(time.time) as mockt:
-            mockt.return_value = 12345.0
-            with timer:
-                mockt.return_value += 0.1  # feign sleep
-            self.check_poll_func(timer, 1, lambda x: 0.09 < x < 0.11)
-            self.check_poll(timer, [])
+        self.patch_time(12345.0)
+        with timer:
+            self.incr_fake_time(0.1)  # feign sleep
+        self.check_poll_func(timer, 1, lambda x: 0.09 < x < 0.11)
+        self.check_poll(timer, [])
 
     def test_accumulate_times(self):
         timer = metrics.Timer("foo")
         timer.manage("prefix.")
-        with mocking(time.time) as mockt:
-            mockt.return_value = 12345.0
-            with timer:
-                mockt.return_value += 0.1  # feign sleep
-            with timer:
-                mockt.return_value += 0.1  # feign sleep
-            self.check_poll_func(timer, 2, lambda x: 0.09 < x < 0.11)
-            self.check_poll(timer, [])
+        self.patch_time(12345.0)
+        with timer:
+            self.incr_fake_time(0.1)  # feign sleep
+        with timer:
+            self.incr_fake_time(0.1)  # feign sleep
+        self.check_poll_func(timer, 2, lambda x: 0.09 < x < 0.11)
+        self.check_poll(timer, [])
 
 
 class TestMetricsConsumer(TestCase):

vumi/tests/test_sentry.py

@@ -12,7 +12,7 @@
 from twisted.python.failure import Failure
 from twisted.python.log import LogPublisher
 
-from vumi.tests.utils import MockHttpServer, LogCatcher, import_skip, mocking
+from vumi.tests.utils import MockHttpServer, LogCatcher, import_skip
 from vumi.sentry import (quiet_get_page, SentryLogObserver, vumi_raven_client,
                          SentryLoggerService)
 
@@ -171,16 +171,22 @@ def mk_sentry_dsn(self):
         return dsn
 
     def parse_call(self, sentry_call):
-        postdata = sentry_call.kwargs['postdata']
+        args, kwargs = sentry_call
+        postdata = kwargs['postdata']
         return json.loads(base64.b64decode(postdata).decode('zlib'))
 
     def test_vumi_raven_client_capture_message(self):
+        import vumi.sentry
         dsn = self.mk_sentry_dsn()
-        mock_page = mocking(quiet_get_page)
-        mock_page.return_value = Deferred()
-        with mock_page:
-            client = vumi_raven_client(dsn)
-            client.captureMessage("my message")
-        [sentry_call] = mock_page.history
+        call_history = []
+
+        def fake_get_page(*args, **kw):
+            call_history.append((args, kw))
+            return Deferred()
+
+        self.patch(vumi.sentry, 'quiet_get_page', fake_get_page)
+        client = vumi_raven_client(dsn)
+        client.captureMessage("my message")
+        [sentry_call] = call_history
         sentry_data = self.parse_call(sentry_call)
         self.assertEqual(sentry_data['message'], "my message")

vumi/tests/test_testutils.py

@@ -1,7 +1,7 @@
 from twisted.trial.unittest import TestCase
 
 from vumi.service import Worker
-from vumi.tests.utils import get_stubbed_worker, Mocking, LogCatcher
+from vumi.tests.utils import get_stubbed_worker, LogCatcher
 from vumi.tests.fake_amqp import FakeAMQClient
 from vumi import log
 
@@ -11,18 +11,6 @@ def poke(self):
         return "poke"
 
 
-class MockingHistoryItemTestCase(TestCase):
-    def test_basic_item(self):
-        item = Mocking.HistoryItem(("a", "b"), {"c": 1})
-        self.assertEqual(item.args, ("a", "b"))
-        self.assertEqual(item.kwargs, {"c": 1})
-
-    def test_repr(self):
-        item = Mocking.HistoryItem(("a", "b"), {"c": 1})
-        self.assertEqual(repr(item), "<'HistoryItem' object at %r"
-                         " [args: ('a', 'b'), kw: {'c': 1}]>" % id(item))
-
-
 class UtilsTestCase(TestCase):
 
     def test_get_stubbed_worker(self):

vumi/tests/utils.py

@@ -16,8 +16,7 @@
 from twisted.python import log
 from twisted.python.monkey import MonkeyPatcher
 
-from vumi.utils import (vumi_resource_path, import_module, flatten_generator,
-                        LogFilterSite)
+from vumi.utils import vumi_resource_path, flatten_generator, LogFilterSite
 from vumi.service import get_spec, Worker, WorkerCreator
 from vumi.message import TransportUserMessage, TransportEvent
 from vumi.tests.fake_amqp import FakeAMQPBroker, FakeAMQClient
@@ -57,54 +56,6 @@ def __eq__(self, other):
         return self.regex.match(other)
 
 
-class Mocking(object):
-
-    class HistoryItem(object):
-        def __init__(self, args, kwargs):
-            self.args = args
-            self.kwargs = kwargs
-
-        def __repr__(self):
-            return '<%r object at %r [args: %r, kw: %r]>' % (
-                self.__class__.__name__, id(self), self.args, self.kwargs)
-
-    def __init__(self, function):
-        """Mock a function"""
-        self.function = function
-        self.called = 0
-        self.history = []
-        self.return_value = None
-
-    def __enter__(self):
-        """Overwrite whatever module the function is part of"""
-        self.mod = import_module(self.function.__module__)
-        setattr(self.mod, self.function.__name__, self)
-        return self
-
-    def __exit__(self, *exc_info):
-        """Reset to whatever the function was originally when done"""
-        setattr(self.mod, self.function.__name__, self.function)
-
-    def __call__(self, *args, **kwargs):
-        """Return the return value when called, store the args & kwargs
-        for testing later, called is a counter and evaluates to True
-        if ever called."""
-        self.args = args
-        self.kwargs = kwargs
-        self.called += 1
-        self.history.append(self.HistoryItem(args, kwargs))
-        return self.return_value
-
-    def to_return(self, *args):
-        """Specify the return value"""
-        self.return_value = args if len(args) > 1 else list(args).pop()
-        return self
-
-
-def mocking(fn):
-    return Mocking(fn)
-
-
 def fake_amq_message(dictionary, delivery_tag='delivery_tag'):
     Content = namedtuple('Content', ['body'])
     Message = namedtuple('Message', ['content', 'delivery_tag'])