Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
class LoggedResult(Result):
def start(self, interaction):
super().start(interaction)
log.info(f"Verifying {interaction}")
def warn(self, message):
log.warning(" " + message)
def fail(self, message, path=None):
self.success = self.FAIL
log.warning(" " + message)
return not message
class PytestResult(Result): # pragma: no cover
def __init__(self, *, level=logging.WARNING):
self.records = []
self.level = level
self.current_consumer = None
def start(self, interaction):
super().start(interaction)
log = logging.getLogger("pactman")
log.handlers = [self]
log.setLevel(logging.DEBUG)
log.propagate = False
self.records[:] = []
def handle(self, record):
self.records.append(record)
from ..verifier.verify import RequestVerifier
class Request:
def __init__(self, method, path, query, headers, body):
self.method = method
self.path = path
self.query = query
self.headers = headers
self.body = body
def json(self):
return self.body
class RecordResult(Result):
def start(self, interaction):
super().start(interaction)
def fail(self, message, path=None):
self.success = self.FAIL
self.reason = message
return not message
class PactVersionConflict(AssertionError):
pass
class PactInteractionMismatch(AssertionError):
pass
success = PASS
def start(self, message):
self.success = self.PASS
def end(self):
pass
def warn(self, message):
raise NotImplementedError()
def fail(self, message, path=None):
raise NotImplementedError() # pragma: no cover
class LoggedResult(Result):
def start(self, interaction):
super().start(interaction)
log.info(f"Verifying {interaction}")
def warn(self, message):
log.warning(" " + message)
def fail(self, message, path=None):
self.success = self.FAIL
log.warning(" " + message)
return not message
class PytestResult(Result): # pragma: no cover
def __init__(self, *, level=logging.WARNING):
self.records = []
raise Failed(message) from None
def warn(self, message):
log.warning(message)
def results_for_terminal(self):
for record in self.records:
if record.levelno > logging.WARN:
yield record.msg, dict(red=True)
elif record.levelno > logging.INFO:
yield record.msg, dict(yellow=True)
else:
yield record.msg, {}
class CaptureResult(Result):
def __init__(self, *, level=logging.WARNING):
self.messages = []
self.level = level
self.current_consumer = None
def start(self, interaction):
super().start(interaction)
log = logging.getLogger("pactman")
log.handlers = [self]
log.setLevel(logging.DEBUG)
self.messages[:] = []
if self.current_consumer != interaction.pact.consumer:
print(f"{Style.BRIGHT}Consumer: {interaction.pact.consumer}")
self.current_consumer = interaction.pact.consumer
print(f'Request: "{interaction.description}" ... ', end="")