Source code for pook.engine

from functools import partial
from inspect import isfunction
from .mock import Mock
from .regex import isregex
from .mock_engine import MockEngine
from .exceptions import PookNoMatches, PookExpiredMock

[docs]class Engine(object): """ Engine represents the mock interceptor and matcher engine responsible of triggering interceptors and match outgoing HTTP traffic. Arguments: network (bool, optional): enables/disables real networking mode. Attributes: debug (bool): enables/disables debug mode. active (bool): stores the current engine activation status. networking (bool): stores the current engine networking mode status. mocks (list[pook.Mock]): stores engine mocks. filters (list[function]): stores engine-level mock filter functions. mappers (list[function]): stores engine-level mock mapper functions. interceptors (list[pook.BaseInterceptor]): stores engine-level HTTP traffic interceptors. unmatched_reqs (list[pook.Request]): stores engine-level unmatched outgoing HTTP requests. network_filters (list[function]): stores engine-level real networking mode filters. """ def __init__(self, network=False): # Enables/Disables debug mode. self.debug = True # Store the engine enable/disable status = False # Enables/Disables real networking self.networking = network # Stores mocks self.mocks = [] # Store engine-level global filters self.filters = [] # Store engine-level global mappers self.mappers = [] # Store unmatched requests. self.unmatched_reqs = [] # Store network filters used to determine when a request # should be filtered or not. self.network_filters = [] # Built-in mock engine to be used self.mock_engine = MockEngine(self)
[docs] 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. For mock engine implementation details, see `pook.MockEngine`. 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 not all([hasattr(mock_engine, method) for method in methods]): raise NotImplementedError('engine must implementent the ' 'required methods') # Use the custom mock engine self.mock_engine = mock_engine # Enable mock engine, if needed if self.mock_engine.activate()
[docs] def enable_network(self, *hostnames): """ Enables real networking mode, optionally passing one or multiple hostnames that would be used as filter. If at least one hostname matches with the outgoing traffic, the request will be executed via the real network. Arguments: *hostnames: optional list of host names to enable real network against them. hostname value can be a regular expression. """ def hostname_filter(hostname, req): if isregex(hostname): return hostname.match(req.url.hostname) return req.url.hostname == hostname for hostname in hostnames: self.use_network_filter(partial(hostname_filter, hostname)) self.networking = True
[docs] def disable_network(self): """ Disables real networking mode. """ self.networking = False
[docs] def use_network_filter(self, *fn): """ Adds network filters to determine if certain outgoing unmatched HTTP traffic can stablish real network connections. Arguments: *fn (function): variadic function filter arguments to be used. """ self.network_filters = self.network_filters + fn
[docs] def flush_network_filters(self): """ Flushes registered real networking filters in the current mock engine. """ self.network_filters = []
[docs] def mock(self, url=None, **kw): """ Creates and registers a new HTTP mock in the current engine. Arguments: url (str): request URL to mock. activate (bool): force mock engine activation. Defaults to ``False``. **kw (mixed): variadic keyword arguments for ``Mock`` constructor. Returns: pook.Mock: new mock instance. """ # Activate mock engine, if explicitly requested if kw.get('activate'): kw.pop('activate') self.activate() # Create the new HTTP mock expectation mock = Mock(url=url, **kw) # Expose current engine instance via mock mock._engine = self # Register the mock in the current engine self.add_mock(mock) # Return it for consumer satisfaction return mock
[docs] def add_mock(self, mock): """ Adds a new mock instance to the current engine. Arguments: mock (pook.Mock): mock instance to add. """ self.mocks.append(mock)
[docs] def remove_mock(self, mock): """ Removes a specific mock instance by object reference. Arguments: mock (pook.Mock): mock instance to remove. """ self.mocks = [m for m in self.mocks if m is not mock]
[docs] def flush_mocks(self): """ Flushes the current mocks. """ 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)
[docs] 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. Note: this method is may not be implemented if using a custom mock engine. Arguments: interceptors (pook.interceptors.BaseInterceptor) """ self._engine_proxy('add_interceptor', *interceptors)
[docs] 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._engine_proxy('flush_interceptors')
[docs] 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. Arguments: name (str): interceptor name to disable. Returns: bool: `True` if the interceptor was disabled, otherwise `False`. """ return self._engine_proxy('remove_interceptor', name)
[docs] 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. """ if return None # Activate mock engine self.mock_engine.activate() # Enable engine state = True
[docs] def disable(self): """ Disables interceptors and stops intercepting any outgoing HTTP traffic. """ if not return None # Disable current mock engine self.mock_engine.disable() # Disable engine state = False
[docs] def reset(self): """ Resets and flushes engine state and mocks to defaults. """ # Reset engine Engine.__init__(self, network=self.networking)
[docs] def unmatched_requests(self): """ Returns a ``tuple`` of unmatched requests. Unmatched requests will be registered only if ``networking`` mode has been enabled. Returns: list: unmatched intercepted requests. """ return [mock for mock in self.unmatched_reqs]
[docs] def unmatched(self): """ Returns the total number of unmatched requests intercepted by pook. Unmatched requests will be registered only if ``networking`` mode has been enabled. Returns: int: total number of unmatched requests. """ return len(self.unmatched_requests())
[docs] def isunmatched(self): """ Returns ``True`` if there are unmatched requests. Otherwise ``False``. Unmatched requests will be registered only if ``networking`` mode has been enabled. Returns: bool """ return len(self.unmatched()) > 0
[docs] def pending(self): """ Returns the number of pending mocks to be matched. Returns: int: number of pending mocks. """ return len(self.pending_mocks())
[docs] def pending_mocks(self): """ Returns a ``tuple`` of pending mocks to be matched. Returns: tuple: pending mock instances. """ return [mock for mock in self.mocks if not mock.isdone()]
[docs] def ispending(self): """ Returns the ``True`` if the engine has pending mocks to be matched. Otherwise ``False``. Returns: bool """ return len(self.pending_mocks())
[docs] def isactive(self): """ Returns the current engine enabled/disabled status. Returns: bool: ``True`` if the engine is active. Otherwise ``False``. """ return
[docs] def isdone(self): """ Returns True if all the registered mocks has been triggered. Returns: bool: True is all the registered mocks are gone, otherwise False. """ return all(mock.isdone() for mock in self.mocks)
def _append(self, target, *fns): (target.append(fn) for fn in fns if isfunction(fn))
[docs] def filter(self, *filters): """ Append engine-level HTTP request filter functions. Arguments: filters*: variadic filter functions to be added. """ self._append(self.filters, *filters)
[docs] def map(self, *mappers): """ Append engine-level HTTP request mapper functions. Arguments: filters*: variadic mapper functions to be added. """ self._append(self.mappers, *mappers)
[docs] def should_use_network(self, request): """ Verifies if real networking mode should be used for the given request, passing it to the registered network filters. Arguments: request (pook.Request): outgoing HTTP request to test. Returns: bool """ return (self.networking and all((fn(request) for fn in self.network_filters)))
[docs] def match(self, request): """ Matches a given Request instance contract against the registered mocks. If a mock passes all the matchers, its response will be returned. Arguments: request (pook.Request): Request contract to match. Raises: pook.PookNoMatches: if networking is disabled and no mock matches with the given request contract. Returns: pook.Response: the mock response to be used by the interceptor. """ # Trigger engine-level request filters for test in self.filters: if not test(request, self): return False # Trigger engine-level request mappers for mapper in self.mappers: request = mapper(request, self) if not request: raise ValueError('map function must return a request object') # Store list of mock matching errors for further debugging match_errors = [] # Try to match the request against registered mock definitions for mock in self.mocks[:]: try: # Return the first matched HTTP request mock matches, errors = mock.match(request.copy()) if len(errors): match_errors += errors if matches: return mock except PookExpiredMock: # Remove the mock if already expired self.mocks.remove(mock) # Validate that we have a mock if not self.should_use_network(request): msg = 'pook error!\n\n' msg += ( '=> Cannot match any mock for the ' 'following request:\n{}'.format(request) ) # Compose unmatch error details, if debug mode is enabled if self.debug: err = '\n\n'.join([str(err) for err in match_errors]) if err: msg += '\n\n=> Detailed matching errors:\n{}\n'.format(err) # Raise no matches exception raise PookNoMatches(msg) # Register unmatched request self.unmatched_reqs.append(request)