Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
self.pact = pact
self.incoming_interactions = interactions
self.outgoing_results = results
server_address = ("", pact.port)
super().__init__(server_address, MockHTTPRequestHandler)
self.interactions = []
self.log = logging.getLogger(__name__ + "." + pact.provider.name)
self.log.addHandler(logging.FileHandler(f"{pact.log_dir}/{pact.provider.name}.log"))
self.log.setLevel(logging.DEBUG)
self.log.propagate = False
class Error(Exception):
pass
class MockHTTPRequestHandler(BaseHTTPRequestHandler, PactRequestHandler):
def __init__(self, request, client_address, server):
self.response_status_code = None
self.response_headers = {}
self.response_body = None
PactRequestHandler.__init__(self, server.pact)
BaseHTTPRequestHandler.__init__(self, request, client_address, server)
def error_result(self, message, content="", status="error", status_code=500):
self.server.outgoing_results.put({"status": status, "reason": message})
self.response_status_code = status_code
self.response_headers = {"Content-Type": "text/plain; charset=utf-8"}
self.response_body = (content or message).encode("utf8")
def run_request(self, method):
try:
self.body = None
def patch(self):
urllib3.poolmanager.pool_classes_by_scheme["http"] = MockConnectionPool
self.patched = True
def clear(self):
urllib3.poolmanager.pool_classes_by_scheme[
"http"
] = urllib3.connectionpool.HTTPConnectionPool
self.patched = False
patcher = MonkeyPatcher()
class MockURLOpenHandler(PactRequestHandler):
def __init__(self, config):
self.interactions = []
super().__init__(config)
patcher.add_service(self)
def terminate(self):
patcher.remove_service(self)
def setup(self, interactions):
self.interactions = interactions
def verify(self):
pass
def __call__(self, method, url, redirect=True, headers=None, body=None, **kw):
self.path = url
def __init__(self, request, client_address, server):
self.response_status_code = None
self.response_headers = {}
self.response_body = None
PactRequestHandler.__init__(self, server.pact)
BaseHTTPRequestHandler.__init__(self, request, client_address, server)