feat(#33): add support for user defined custom mock engine

README.rst

@@ -1,7 +1,7 @@
 pook |Build Status| |PyPI| |Coverage Status| |Documentation Status| |Stability| |Quality| |Versions|
 ====================================================================================================
 
-Versatile, expressive and hackable utility library for HTTP traffic mocking and expectations in `Python`_.
+Versatile, expressive and hackable utility library for HTTP traffic mocking and expectations made easy in `Python`_.
 
 Heavily inspired by `gock`_.
 
@@ -29,6 +29,7 @@ Features
 -  Simulated raised error exceptions.
 -  Network delay simulation (only available for ``aiohttp``).
 -  Pluggable and hackable API.
+-  Customizable HTTP traffic mock interceptor engine.
 -  Ideal for painless test doubles.
 -  Does not support WebSocket traffic mocking.
 -  Works with Python +2.7 and +3.0 (including PyPy).

pook/api.py

@@ -8,6 +8,7 @@
 from .request import Request  # noqa
 from .response import Response  # noqa
 from .matcher import MatcherEngine  # noqa
+from .mock_engine import MockEngine  # noqa
 
 # Public API symbols to export
 __all__ = (
@@ -17,7 +18,8 @@
     'delete', 'options', 'pending', 'ispending',
     'pending_mocks', 'unmatched_requests', 'isunmatched',
     'unmatched', 'isactive', 'isdone', 'regex',
-    'Engine', 'Mock', 'Request', 'Response', 'MatcherEngine'
+    'Engine', 'Mock', 'Request', 'Response',
+    'MatcherEngine', 'MockEngine'
 )
 
 # Default singleton mock engine to be used

pook/engine.py

@@ -2,7 +2,7 @@
 from inspect import isfunction
 from .mock import Mock
 from .regex import isregex
-from .interceptors import interceptors
+from .mock_engine import MockEngine
 from .exceptions import PookNoMatches, PookExpiredMock
 
 
@@ -42,15 +42,42 @@ def __init__(self, network=False):
         self.filters = []
         # Store engine-level global mappers
         self.mappers = []
-        # Store HTTP traffic interceptors
-        self.interceptors = []
         # Store unmatched requests.
         self.unmatched_reqs = []
         # Store network filters used to determine when a request
         # should be filtered or not.
         self.network_filters = []
-        # Register built-in interceptors
-        self.add_interceptor(*interceptors)
+        # Built-in mock engine to be used
+        self.mock_engine = MockEngine(self)
+
+    def set_mock_engine(self, engine):
+        """
+        Sets a custom mock engine, replacing the built-in one.
+
+        This is particularly useful if you want to replace the built-in
+        HTTP traffic mock interceptor engine with your custom one.
+
+        Arguments:
+            engine (pook.MockEngine): custom mock engine to use.
+        """
+        if not engine:
+            raise TypeError('engine must be a valid object')
+
+        # Instantiate mock engine
+        mock_engine = engine(self)
+
+        # Validate minimum viable interface
+        methods = ('activate', 'disable')
+        if all([hasattr(mock_engine, method) for method in methods]):
+            raise NotImplementedError('engine must implementent the '
+                                      'required methods')
+
+        # Disable previous mock engine, if needed
+        if self.active:
+            self.disable()
+
+        # Use the custom mock engine
+        self.mock_engine = mock_engine
 
     def enable_network(self, *hostnames):
         """
@@ -134,6 +161,15 @@ def flush_mocks(self):
         """
         self.mocks = []
 
+    def _engine_proxy(self, method, *args, **kw):
+        engine_method = getattr(self.mock_engine, method, None)
+
+        if not engine_method:
+            raise NotImplementedError('current mock engine does not implements'
+                                      ' required "{}" method'.format(method))
+
+        return engine_method(self.mock_engine, *args, **kw)
+
     def add_interceptor(self, *interceptors):
         """
         Adds one or multiple HTTP traffic interceptors to the current
@@ -142,30 +178,39 @@ def add_interceptor(self, *interceptors):
         Interceptors are typically HTTP client specific wrapper classes that
         implements the pook interceptor interface.
 
+        Note: this method is may not be implemented if using a custom mock
+        engine.
+
         Arguments:
             interceptors (pook.interceptors.BaseInterceptor)
         """
-        for interceptor in interceptors:
-            self.interceptors.append(interceptor(self))
+        self._engine_proxy('add_interceptor', *interceptors)
 
     def flush_interceptors(self):
         """
         Flushes registered interceptors in the current mocking engine.
 
         This method is low-level. Only call it if you know what you are doing.
+
+        Note: this method is may not be implemented if using a custom mock
+        engine.
         """
-        self.interceptors = []
+        self._engine_proxy('flush_interceptors')
+
+    def remove_interceptor(self, name):
+        """
+        Removes a specific interceptor by name.
+
+        Note: this method is may not be implemented if using a custom mock
+        engine.
 
-    def disable_interceptor(self, name):
-        for index, interceptor in enumerate(self.interceptors):
-            matches = (
-                type(interceptor).__name__ == name or
-                getattr(interceptor, 'name') == name
-            )
-            if matches:
-                self.interceptors.pop(index)
-                return True
-        return False
+        Arguments:
+            name (str): interceptor name to disable.
+
+        Returns:
+            bool: `True` if the interceptor was disabled, otherwise `False`.
+        """
+        return self._engine_proxy('remove_interceptor', name)
 
     def activate(self):
         """
@@ -178,7 +223,9 @@ def activate(self):
         if self.active:
             return None
 
-        [interceptor.activate() for interceptor in self.interceptors]
+        # Activate mock engine
+        self.mock_engine.activate()
+        # Enable engine state
         self.active = True
 
     def disable(self):
@@ -188,14 +235,9 @@ def disable(self):
         if not self.active:
             return None
 
-        # Restore HTTP interceptors
-        for interceptor in self.interceptors:
-            try:
-                interceptor.disable()
-            except RuntimeError:
-                pass  # explicitely ignore runtime patch errors
-
-        # Restore engine state
+        # Disable current mock engine
+        self.mock_engine.disable()
+        # Disable engine state
         self.active = False
 
     def reset(self):

pook/mock_engine.py

@@ -0,0 +1,99 @@
+from .interceptors import interceptors
+
+
+class MockEngine(object):
+    """
+    MockEngine implements the built-in `pook` mock engine based on HTTP
+    interceptors strategy.
+
+    Mock engines must implementent the following methods:
+
+    - `engine.__init__(self, engine)`
+    - `engine.activate(self)`
+    - `engine.disable(self)`
+
+    Mock engines can optionally implement the follow methods:
+
+    - `engine.add_interceptors(self, *interceptors)`
+    - `engine.flush_interceptors(self)`
+    - `engine.disable_interceptor(self, name) -> bool`
+
+    Arguments:
+        engine (pook.Engine): injected pook engine to be used.
+
+    Attributes:
+        engine (pook.Engine): stores pook engine to be used.
+        interceptors (list[pook.BaseInterceptor]): stores engine-level HTTP
+            traffic interceptors.
+    """
+
+    def __init__(self, engine):
+        # Store pook engine
+        self.engine = engine
+        # Store HTTP client interceptors
+        self.interceptors = []
+        # Self-register built-in interceptors
+        self.add_interceptor(*interceptors)
+
+    def add_interceptor(self, *interceptors):
+        """
+        Adds one or multiple HTTP traffic interceptors to the current
+        mocking engine.
+
+        Interceptors are typically HTTP client specific wrapper classes that
+        implements the pook interceptor interface.
+
+        Arguments:
+            interceptors (pook.interceptors.BaseInterceptor)
+        """
+        for interceptor in interceptors:
+            self.interceptors.append(interceptor(self.engine))
+
+    def flush_interceptors(self):
+        """
+        Flushes registered interceptors in the current mocking engine.
+
+        This method is low-level. Only call it if you know what you are doing.
+        """
+        self.interceptors = []
+
+    def remove_interceptor(self, name):
+        """
+        Removes a specific interceptor by name.
+
+        Arguments:
+            name (str): interceptor name to disable.
+
+        Returns:
+            bool: `True` if the interceptor was disabled, otherwise `False`.
+        """
+        for index, interceptor in enumerate(self.interceptors):
+            matches = (
+                type(interceptor).__name__ == name or
+                getattr(interceptor, 'name') == name
+            )
+            if matches:
+                self.interceptors.pop(index)
+                return True
+        return False
+
+    def activate(self):
+        """
+        Activates the registered interceptors in the mocking engine.
+
+        This means any HTTP traffic captures by those interceptors will
+        trigger the HTTP mock matching engine in order to determine if a given
+        HTTP transaction should be mocked out or not.
+        """
+        [interceptor.activate() for interceptor in self.interceptors]
+
+    def disable(self):
+        """
+        Disables interceptors and stops intercepting any outgoing HTTP traffic.
+        """
+        # Restore HTTP interceptors
+        for interceptor in self.interceptors:
+            try:
+                interceptor.disable()
+            except RuntimeError:
+                pass  # explicitely ignore runtime patch errors

tests/unit/mock_engine_test.py

@@ -0,0 +1,53 @@
+# -*- coding: utf-8 -*-
+
+import pytest
+from pook import MockEngine, Engine
+from pook.interceptors import BaseInterceptor
+
+
+class Interceptor(BaseInterceptor):
+    def activate(self):
+        self.active = True
+
+    def disable(self):
+        self.active = False
+
+
+@pytest.fixture
+def engine():
+    return MockEngine(Engine())
+
+
+def test_mock_engine_instance(engine):
+    assert isinstance(engine.engine, Engine)
+    assert isinstance(engine.interceptors, list)
+    assert len(engine.interceptors) >= 2
+
+
+def test_mock_engine_flush(engine):
+    assert len(engine.interceptors) == 2
+    engine.flush_interceptors()
+    assert len(engine.interceptors) == 0
+
+
+def test_mock_engine_interceptors(engine):
+    engine.flush_interceptors()
+    engine.add_interceptor(Interceptor)
+    assert len(engine.interceptors) == 1
+    assert isinstance(engine.interceptors[0], Interceptor)
+
+    engine.remove_interceptor('Interceptor')
+    assert len(engine.interceptors) == 0
+
+
+def test_mock_engine_status(engine):
+    engine.flush_interceptors()
+    engine.add_interceptor(Interceptor)
+    assert len(engine.interceptors) == 1
+
+    interceptor = engine.interceptors[0]
+    engine.activate()
+    assert interceptor.active
+
+    engine.disable()
+    assert not interceptor.active