PK!preacher/__init__.py"""Preacher: A Web API Verification Tool.""" __version__ = '0.0.0' __author__ = 'Yu Mochizuki' __author_email__ = 'ymoch.dev@gmail.com' NAME = 'preacher' DESCRIPTION = 'A Web API verification tool.' URL = 'https://github.com/ymoch/preacher' LICENSE = 'MIT' def __dummy__() -> None: """ Dummy function so that tests cover this module. >>> __dummy__() """ pass PK!preacher/app/__init__.pyPK!preacher/app/cli/__init__.pyPK!_3preacher/app/cli/application.pyfrom __future__ import annotations from preacher.core.scenario import Scenario from preacher.presentation.logger import LoggerPresentation class Application: def __init__( self: Application, base_url: str, view: LoggerPresentation, ) -> None: self._view = view self._base_url = base_url self._is_succeeded = True @property def is_succeeded(self: Application) -> bool: return self._is_succeeded def consume_scenario(self: Application, scenario: Scenario) -> None: verification = scenario(base_url=self._base_url) self._is_succeeded &= verification.status.is_succeeded self._view.show_scenario_verification(verification, 'Response') PK!_@preacher/app/cli/main.py"""Preacher CLI.""" from __future__ import annotations import argparse import logging import sys import ruamel.yaml as yaml from preacher import __version__ as VERSION from preacher.compilation import compile from preacher.presentation.logger import LoggerPresentation from .application import Application HANDLER = logging.StreamHandler() HANDLER.setLevel(logging.WARN) LOGGER = logging.getLogger(__name__) LOGGER.addHandler(HANDLER) LOGGER.setLevel(logging.INFO) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument( 'conf', nargs='+', help='config file paths' ) parser.add_argument( '-v', '--version', action='version', version=VERSION, ) parser.add_argument( '-u', '--url', metavar='url', help='specify the base URL', default='http://localhost:5000', ) parser.add_argument( '-q', '--quiet', action='store_true', help='show details on the console only when an any issue occurs', ) return parser.parse_args() def main() -> None: """Main.""" args = parse_args() logging_level = logging.INFO should_be_quiet = args.quiet if should_be_quiet: logging_level = logging.WARN HANDLER.setLevel(logging_level) base_url = args.url view = LoggerPresentation(LOGGER) app = Application(base_url=base_url, view=view) config_paths = args.conf for config_path in config_paths: with open(config_path) as config_file: config = yaml.safe_load(config_file) scenarios = compile(config) for scenario in scenarios: app.consume_scenario(scenario) if not app.is_succeeded: sys.exit(1) PK!2 preacher/compilation/__init__.py"""Compilation.""" from __future__ import annotations from collections.abc import Mapping from typing import Any, Iterator from preacher.core.scenario import Scenario from .error import CompilationError from .scenario import compile_scenario def compile(obj: Mapping) -> Iterator[Scenario]: """ >>> scenarios = list(compile({})) >>> scenarios [] >>> scenarios = compile({'scenarios': ''}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: scenarios ... >>> from unittest.mock import sentinel, patch >>> scenario_patch = patch( ... f'{__name__}.compile_scenario', ... return_value=sentinel.scenario, ... ) >>> scenarios = compile({'scenarios': [{}, '']}) >>> with scenario_patch as scenario_mock: ... next(scenarios) ... scenario_mock.call_args sentinel.scenario call({}) >>> with scenario_patch as scenario_mock: ... next(scenarios) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: Scenario ... """ scenario_objs = obj.get('scenarios', []) if not isinstance(scenario_objs, list): raise CompilationError(f'scenarios must be a list: {scenario_objs}') return map(_compile_scenario, scenario_objs) def _compile_scenario(obj: Any) -> Scenario: if not isinstance(obj, Mapping): raise CompilationError(f'Scenario must be a mapping: {obj}') return compile_scenario(obj) PK!;L{ #preacher/compilation/description.py"""Description compilation.""" from collections.abc import Mapping from typing import Any from preacher.core.description import Description, Predicate from .error import CompilationError from .predicate import compile as compile_predicate from .extraction import compile as compile_extraction def compile(obj: Mapping) -> Description: """ >>> from unittest.mock import patch, sentinel >>> extraction_patch = patch( ... f'{__name__}.compile_extraction', ... return_value=sentinel.extraction, ... ) >>> predicate_patch = patch( ... f'{__name__}.compile_predicate', ... return_value=sentinel.predicate, ... ) >>> compile({}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: Description.describe ... >>> with extraction_patch as extraction_mock: ... compile({'describe': 'foo', 'it': 'str'}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: Description.it ... >>> with extraction_patch as extraction_mock: ... compile({'describe': {}, 'it': [None]}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: Predicate ... >>> with extraction_patch as extraction_mock, \\ ... predicate_patch as predicate_mock: ... description = compile({ ... 'describe': 'foo', ... 'it': {'key': 'value'} ... }) ... extraction_mock.call_args ... predicate_mock.call_args_list call({'jq': 'foo'}) [call({'key': 'value'})] >>> description.extraction sentinel.extraction >>> description.predicates [sentinel.predicate] >>> with extraction_patch as extraction_mock, \\ ... predicate_patch as predicate_mock: ... description = compile({ ... 'describe': {'key': 'value'}, ... 'it': [{'key1': 'value1'}, {'key2': 'value2'}] ... }) ... extraction_mock.call_args ... predicate_mock.call_args_list call({'key': 'value'}) [call({'key1': 'value1'}), call({'key2': 'value2'})] >>> description.extraction sentinel.extraction >>> description.predicates [sentinel.predicate, sentinel.predicate] """ extraction_obj = obj.get('describe') if isinstance(extraction_obj, str): extraction_obj = {'jq': extraction_obj} if not isinstance(extraction_obj, Mapping): raise CompilationError( f'Description.describe must be a mapping: {extraction_obj}' ) extraction = compile_extraction(extraction_obj) predicate_objs = obj.get('it', []) if isinstance(predicate_objs, Mapping): predicate_objs = [predicate_objs] if not isinstance(predicate_objs, list): raise CompilationError( f'Description.it must be a list or a mapping: {predicate_objs}' ) return Description( extraction=extraction, predicates=[_compile_predicate(obj) for obj in predicate_objs] ) def _compile_predicate(obj: Any) -> Predicate: if not isinstance(obj, str) and not isinstance(obj, Mapping): raise CompilationError( f'Predicate must be a string or a mapping: {obj}' ) return compile_predicate(obj) PK!(xffpreacher/compilation/error.py"""Compilation errors.""" class CompilationError(Exception): """Compilation errors.""" pass PK!ۋBB"preacher/compilation/extraction.py"""Extraction compilation.""" from collections.abc import Mapping from preacher.core.extraction import Extraction, with_jq from .error import CompilationError _EXTRACTION_MAP = { 'jq': with_jq, } _EXTRACTION_KEYS = frozenset(_EXTRACTION_MAP.keys()) def compile(extraction_obj: Mapping) -> Extraction: """ >>> compile({}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ... has 0 >>> compile({'jq': '.foo'})({'foo': 'bar'}) 'bar' """ keys = _EXTRACTION_KEYS.intersection(extraction_obj.keys()) if len(keys) != 1: raise CompilationError( 'Description object must have only 1 valid extraction key' f', but has {len(keys)}' ) key = next(iter(keys)) return _EXTRACTION_MAP[key](extraction_obj[key]) PK!R:'!preacher/compilation/predicate.py"""Predicate compilation.""" from collections.abc import Mapping from typing import Union import hamcrest from preacher.core.predicate import Predicate, of_hamcrest_matcher from .error import CompilationError _STATIC_MATCHER_MAP = { # For objects. 'is_null': hamcrest.is_(hamcrest.none()), 'is_not_null': hamcrest.is_(hamcrest.not_none()), # For collections. 'is_empty': hamcrest.is_(hamcrest.empty()), } _VALUE_MATCHER_FUNCTION_MAP = { # For objects. 'is': lambda expected: hamcrest.is_(expected), 'equals_to': lambda expected: hamcrest.is_(hamcrest.equal_to(expected)), 'has_length': lambda expected: hamcrest.has_length(expected), # For numbers. 'is_greater_than': ( lambda value: hamcrest.is_(hamcrest.greater_than(value)) ), 'is_greater_than_or_equal_to': ( lambda value: hamcrest.is_(hamcrest.greater_than_or_equal_to(value)) ), 'is_less_than': ( lambda value: hamcrest.is_(hamcrest.less_than(value)) ), 'is_less_than_or_equal_to': ( lambda value: hamcrest.is_(hamcrest.less_than_or_equal_to(value)) ), # For strings. 'contains_string': lambda value: hamcrest.contains_string(value), 'starts_with': lambda value: hamcrest.starts_with(value), 'ends_with': lambda value: hamcrest.ends_with(value), 'matches_regexp': lambda value: hamcrest.matches_regexp(value), } _PREDICATE_KEYS = frozenset(_VALUE_MATCHER_FUNCTION_MAP.keys()) def compile(obj: Union[str, Mapping]) -> Predicate: """ >>> compile('invalid_key') Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ... 'invalid_key' >>> predicate = compile('is_null') >>> predicate(None).status.name 'SUCCESS' >>> predicate('SUCCESS').status.name 'UNSTABLE' >>> predicate = compile('is_not_null') >>> predicate(None).status.name 'UNSTABLE' >>> predicate('UNSTABLE').status.name 'SUCCESS' >>> predicate = compile('is_empty') >>> predicate(None).status.name 'UNSTABLE' >>> predicate(0).status.name 'UNSTABLE' >>> predicate('').status.name 'SUCCESS' >>> predicate([]).status.name 'SUCCESS' >>> predicate('A').status.name 'UNSTABLE' >>> predicate([1]).status.name 'UNSTABLE' >>> compile({}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ... has 0 >>> compile({'equals_to': 0, 'not': {'equal_to': 1}}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ... has 2 >>> predicate = compile({'has_length': 1}) >>> predicate(None).status.name 'UNSTABLE' >>> predicate('').status.name 'UNSTABLE' >>> predicate([]).status.name 'UNSTABLE' >>> predicate('A').status.name 'SUCCESS' >>> predicate([1]).status.name 'SUCCESS' >>> compile({'invalid_key': 0}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ... 'invalid_key' >>> predicate = compile({'is': 1}) >>> predicate(0).status.name 'UNSTABLE' >>> predicate('1').status.name 'UNSTABLE' >>> predicate(1).status.name 'SUCCESS' >>> predicate = compile({'equals_to': 1}) >>> predicate(0).status.name 'UNSTABLE' >>> predicate('1').status.name 'UNSTABLE' >>> predicate(1).status.name 'SUCCESS' >>> predicate = compile({'is_greater_than': 0}) >>> predicate(-1).status.name 'UNSTABLE' >>> predicate(0).status.name 'UNSTABLE' >>> predicate(1).status.name 'SUCCESS' >>> predicate = compile({'is_greater_than_or_equal_to': 0}) >>> predicate(-1).status.name 'UNSTABLE' >>> predicate(0).status.name 'SUCCESS' >>> predicate(1).status.name 'SUCCESS' >>> predicate = compile({'is_less_than': 0}) >>> predicate(-1).status.name 'SUCCESS' >>> predicate(0).status.name 'UNSTABLE' >>> predicate(1).status.name 'UNSTABLE' >>> predicate = compile({'is_less_than_or_equal_to': 0}) >>> predicate(-1).status.name 'SUCCESS' >>> predicate(0).status.name 'SUCCESS' >>> predicate(1).status.name 'UNSTABLE' >>> predicate = compile({'contains_string': '0'}) >>> predicate(0).status.name 'UNSTABLE' >>> predicate('012').status.name 'SUCCESS' >>> predicate('123').status.name 'UNSTABLE' >>> predicate = compile({'starts_with': 'AB'}) >>> predicate(0).status.name 'UNSTABLE' >>> predicate('ABC').status.name 'SUCCESS' >>> predicate('ACB').status.name 'UNSTABLE' >>> predicate = compile({'ends_with': 'BC'}) >>> predicate(0).status.name 'UNSTABLE' >>> predicate('ABC').status.name 'SUCCESS' >>> predicate('ACB').status.name 'UNSTABLE' >>> predicate = compile({'matches_regexp': '^A*B$'}) >>> predicate('B').status.name 'SUCCESS' >>> predicate('ACB').status.name 'UNSTABLE' TODO: Should return `False` when the value type is not `str`. >>> predicate(0).status.name 'FAILURE' """ if isinstance(obj, str): matcher = _STATIC_MATCHER_MAP.get(obj) if not matcher: raise CompilationError( f'Invalid predicate: \'{obj}\'' ) return of_hamcrest_matcher(matcher) if len(obj) != 1: raise CompilationError( 'Predicate must have only 1 element' f', but has {len(obj)}' ) key, value = next(iter(obj.items())) if key in _VALUE_MATCHER_FUNCTION_MAP: matcher = _VALUE_MATCHER_FUNCTION_MAP[key](value) return of_hamcrest_matcher(matcher) raise CompilationError(f'Unrecognized predicate key: \'{key}\'') PK!5GAApreacher/compilation/request.py"""Request compilation.""" from collections.abc import Mapping from preacher.core.request import Request from .error import CompilationError def compile(obj: Mapping) -> Request: """ >>> compile({'path': {'key': 'value'}}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: Request.path ... >>> compile({'params': ''}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: Request.params ... >>> request = compile({}) >>> request.path '' >>> request.params {} >>> request = compile({'path': '/path', 'params': {'key': 'value'}}) >>> request.path '/path' >>> request.params {'key': 'value'} """ path = obj.get('path', '') if not isinstance(path, str): raise CompilationError(f'Request.path must be a string: {path}') params = obj.get('params', {}) if not isinstance(params, Mapping): raise CompilationError(f'Request.params must be a mapping: {params}') return Request(path=path, params=params) PK!C+,preacher/compilation/response_description.py"""Response description compilations.""" from collections.abc import Mapping from typing import Any from preacher.core.description import Description from preacher.core.response_description import ResponseDescription from .error import CompilationError from .description import compile as compile_description from .predicate import compile as compile_predicate def compile(obj: Mapping) -> ResponseDescription: """ >>> from unittest.mock import patch, sentinel >>> predicate_patch = patch( ... f'{__name__}.compile_predicate', ... return_value=sentinel.predicate, ... ) >>> description_patch = patch( ... f'{__name__}.compile_description', ... return_value=sentinel.description, ... ) >>> with predicate_patch as predicate_mock, \\ ... description_patch as description_mock: ... response_description = compile({}) ... assert predicate_mock.call_count == 0 ... assert description_mock.call_count == 0 >>> response_description.status_code_predicates [] >>> response_description.body_descriptions [] >>> compile({'body': 'str'}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ResponseDescription.body ... >>> compile({'body': ['str']}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: Description ... >>> with predicate_patch as predicate_mock, \\ ... description_patch as description_mock: ... response_description = compile({ ... 'status_code': 402, ... 'body': {'key1': 'value1'}} ... ) ... predicate_mock.call_args_list ... description_mock.call_args_list [call({'equals_to': 402})] [call({'key1': 'value1'})] >>> response_description.body_descriptions [sentinel.description] >>> with predicate_patch as predicate_mock, \\ ... description_patch as description_mock: ... response_description = compile({ ... 'status_code': [{'is_greater_than': 0}, {'is_less_than': 400}], ... 'body': [{'key1': 'value1'}, {'key2': 'value2'}], ... }) ... predicate_mock.call_args_list ... description_mock.call_args_list [call({'is_greater_than': 0}), call({'is_less_than': 400})] [call({'key1': 'value1'}), call({'key2': 'value2'})] >>> response_description.body_descriptions [sentinel.description, sentinel.description] """ status_code_predicate_objs = obj.get('status_code', []) if isinstance(status_code_predicate_objs, int): status_code_predicate_objs = {'equals_to': status_code_predicate_objs} if not isinstance(status_code_predicate_objs, list): status_code_predicate_objs = [status_code_predicate_objs] status_code_predicates = [ compile_predicate(obj) for obj in status_code_predicate_objs ] body_description_objs = obj.get('body', []) if isinstance(body_description_objs, Mapping): body_description_objs = [body_description_objs] if not isinstance(body_description_objs, list): raise CompilationError( 'ResponseDescription.body must be a list or a mapping' f': {body_description_objs}' ) body_descriptions = [ _compile_description(body_description_obj) for body_description_obj in body_description_objs ] return ResponseDescription( status_code_predicates=status_code_predicates, body_descriptions=body_descriptions, ) def _compile_description(obj: Any) -> Description: if not isinstance(obj, Mapping): raise CompilationError( f'Description must be a mapping: {obj}' ) return compile_description(obj) PK!.o; ; preacher/compilation/scenario.py"""Scenario compilation.""" from collections.abc import Mapping from preacher.core.scenario import Scenario from .error import CompilationError from .request import compile as compile_request from .response_description import compile as compile_response_description def compile_scenario(obj: Mapping) -> Scenario: """ >>> from unittest.mock import patch, sentinel >>> request_patch = patch( ... f'{__name__}.compile_request', ... return_value=sentinel.request, ... ) >>> response_description_patch = patch( ... f'{__name__}.compile_response_description', ... return_value=sentinel.response_description ... ) >>> with request_patch as request_mock, \\ ... response_description_patch as response_description_mock: ... scenario = compile_scenario({}) ... request_mock.call_args ... response_description_mock.call_args call({}) call({}) >>> scenario.request sentinel.request >>> scenario.response_description sentinel.response_description >>> compile_scenario({'label': []}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: Scenario.label ... >>> compile_scenario({'request': []}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: Scenario.request ... >>> with request_patch: ... compile_scenario({'response': 'str'}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: Scenario.response ... >>> with request_patch as request_mock, \\ ... response_description_patch as response_description_mock: ... scenario = compile_scenario({'request': '/path'}) ... request_mock.call_args call({'path': '/path'}) >>> scenario.label >>> scenario.request sentinel.request >>> with request_patch as request_mock, \\ ... response_description_patch as response_description_mock: ... scenario = compile_scenario({ ... 'label': 'label', ... 'request': {'path': '/path'}, ... 'response': {'key': 'value'}, ... }) ... request_mock.call_args ... response_description_mock.call_args call({'path': '/path'}) call({'key': 'value'}) >>> scenario.label 'label' >>> scenario.request sentinel.request >>> scenario.response_description sentinel.response_description """ label = obj.get('label') if label is not None and not isinstance(label, str): raise CompilationError(f'Scenario.label must be a string: {label}') request_obj = obj.get('request', {}) if isinstance(request_obj, str): request_obj = {'path': request_obj} if not isinstance(request_obj, Mapping): raise CompilationError( f'Scenario.request must be a string or a mapping: {request_obj}' ) request = compile_request(request_obj) response_obj = obj.get('response', {}) if not isinstance(response_obj, Mapping): raise CompilationError( f'Scenario.response object must be a mapping: {response_obj}' ) response_description = compile_response_description(response_obj) return Scenario( label=label, request=request, response_description=response_description, ) PK!preacher/core/__init__.pyPK!|. preacher/core/description.py"""Description.""" from __future__ import annotations from typing import Any, Callable, List from .verification import ( Verification, merge_statuses, ) Extraction = Callable[[Any], Any] Predicate = Callable[[Any], Verification] class Description: """ Description. >>> from .verification import Status >>> from unittest.mock import MagicMock When extraction fails, then description fails. >>> description = Description( ... extraction=MagicMock(side_effect=Exception('message')), ... predicates=[] ... ) >>> verification = description('described') >>> verification.status.name 'FAILURE' >>> verification.message 'Exception: message' When given no predicates, then describes that any described value is valid. >>> description = Description( ... extraction=MagicMock(return_value='target'), ... predicates=[], ... ) >>> verification = description('described') >>> verification.status.name 'SUCCESS' >>> len(verification.children) 0 When given at least one predicates that returns false, then describes that it is invalid. >>> description = Description( ... extraction=MagicMock(return_value='target'), ... predicates=[ ... MagicMock(return_value=Verification(Status.UNSTABLE)), ... MagicMock(return_value=Verification(Status.FAILURE)), ... MagicMock(return_value=Verification(Status.SUCCESS)), ... ] ... ) >>> verification = description('described') >>> verification.status.name 'FAILURE' >>> len(verification.children) 3 >>> verification.children[0].status.name 'UNSTABLE' >>> verification.children[1].status.name 'FAILURE' >>> verification.children[2].status.name 'SUCCESS' When given only predicates that returns true, then describes that it is valid. >>> description = Description( ... extraction=MagicMock(return_value='target'), ... predicates=[ ... MagicMock(return_value=Verification(Status.SUCCESS)), ... MagicMock(return_value=Verification(Status.SUCCESS)), ... ] ... ) >>> verification = description('described') >>> verification.status.name 'SUCCESS' >>> len(verification.children) 2 >>> verification.children[0].status.name 'SUCCESS' >>> verification.children[1].status.name 'SUCCESS' """ def __init__( self: Description, extraction: Extraction, predicates: List[Predicate], ): self._extraction = extraction self._predicates = predicates def __call__(self: Description, value: Any) -> Verification: try: verified_value = self._extraction(value) except Exception as error: return Verification.of_error(error) verifications = [pred(verified_value) for pred in self._predicates] status = merge_statuses(v.status for v in verifications) return Verification(status, children=verifications) @property def extraction(self: Description) -> Extraction: return self._extraction @property def predicates(self: Description) -> List[Predicate]: return self._predicates PK!Lpreacher/core/extraction.py"""Extraction.""" from .description import Extraction from pyjq import compile as jq_compile def with_jq(query: str) -> Extraction: """ Returns a extractor of given `jq`. >>> extract = with_jq('.foo') >>> extract({'not_foo': 'bar'}) >>> extract({'foo': 'bar'}) 'bar' >>> extract({'foo': ['bar', 'baz', 1, 2]}) ['bar', 'baz', 1, 2] """ return jq_compile(query).first PK!Y  preacher/core/predicate.py"""Predicate.""" from typing import Any from hamcrest import assert_that from hamcrest.core.matcher import Matcher from .description import Predicate from .verification import Status, Verification def of_hamcrest_matcher(matcher: Matcher) -> Predicate: """ Make a predicate from a Hamcrest matcher. >>> from unittest.mock import MagicMock, patch >>> matcher = MagicMock(Matcher) >>> with patch( ... f'{__name__}.assert_that', ... side_effect=RuntimeError('message') ... ) as assert_that: ... predicate = of_hamcrest_matcher(matcher) ... verification = predicate(0) ... assert_that.assert_called_with(0, matcher) >>> verification.status.name 'FAILURE' >>> verification.message 'RuntimeError: message' >>> with patch( ... f'{__name__}.assert_that', ... side_effect=AssertionError(' message\\n') ... ) as assert_that: ... predicate = of_hamcrest_matcher(matcher) ... verification = predicate(0) ... assert_that.assert_called_with(0, matcher) >>> verification.status.name 'UNSTABLE' >>> verification.message 'message' >>> with patch(f'{__name__}.assert_that') as assert_that: ... predicate = of_hamcrest_matcher(matcher) ... verification = predicate(1) ... assert_that.assert_called_with(1, matcher) >>> verification.status.name 'SUCCESS' """ def _test(actual: Any) -> Verification: try: assert_that(actual, matcher) except AssertionError as error: message = str(error).strip() return Verification(status=Status.UNSTABLE, message=message) except Exception as error: return Verification.of_error(error) return Verification.succeed() return _test PK!OWpreacher/core/request.py"""Request.""" from __future__ import annotations from collections.abc import Mapping from dataclasses import dataclass import requests @dataclass class Response: status_code: int headers: Mapping body: str class Request: """ >>> request = Request(path='/path', params={'key': 'value'}) >>> request.path '/path' >>> request.params {'key': 'value'} >>> from unittest.mock import MagicMock, patch >>> inner_response = MagicMock( ... requests.Response, ... status_code=402, ... headers={'header-key': 'header-value'}, ... text='text', ... ) >>> with patch('requests.get', return_value=inner_response) as mock: ... response = request('base-url') ... mock.call_args call('base-url/path', params={'key': 'value'}) >>> response.status_code 402 >>> response.headers {'header-key': 'header-value'} >>> response.body 'text' """ def __init__(self, path: str, params: Mapping) -> None: self._path = path self._params = params def __call__(self, base_url: str) -> Response: res = requests.get( base_url + self._path, params=self._params, ) return Response( status_code=res.status_code, headers=res.headers, body=res.text, ) @property def path(self: Request) -> str: return self._path @property def params(self: Request) -> Mapping: return self._params PK! O%preacher/core/response_description.py"""Response descriptions.""" from __future__ import annotations import json from dataclasses import dataclass from typing import List from .description import Description, Predicate from .verification import Status, Verification, merge_statuses @dataclass class ResponseVerification: status: Status status_code: Verification body: Verification class ResponseDescription: """ >>> description = ResponseDescription( ... status_code_predicates=[], ... body_descriptions=[], ... ) >>> verification = description(status_code=200, body='') >>> verification.status.name 'SUCCESS' >>> from unittest.mock import MagicMock >>> description = ResponseDescription( ... status_code_predicates=[ ... MagicMock(return_value=Verification.succeed()), ... ], ... body_descriptions=[ ... MagicMock(return_value=Verification.succeed()), ... ], ... ) >>> verification = description(status_code=200, body='invalid-format') >>> description.status_code_predicates[0].call_args_list [call(200)] >>> description.body_descriptions[0].call_count 0 >>> verification.status.name 'FAILURE' >>> verification.body.status.name 'FAILURE' >>> verification.body.message 'JSONDecodeError: Expecting value: line 1 column 1 (char 0)' >>> from unittest.mock import MagicMock >>> description = ResponseDescription( ... status_code_predicates=[], ... body_descriptions=[ ... MagicMock(return_value=Verification(status=Status.UNSTABLE)), ... MagicMock(return_value=Verification.succeed()), ... ], ... ) >>> verification = description(status_code=200, body='{}') >>> description.body_descriptions[0].call_args call({}) >>> description.body_descriptions[1].call_args call({}) >>> verification.status.name 'UNSTABLE' >>> verification.body.status.name 'UNSTABLE' >>> verification.body.children[0].status.name 'UNSTABLE' >>> verification.body.children[1].status.name 'SUCCESS' """ def __init__( self: ResponseDescription, status_code_predicates: List[Predicate], body_descriptions: List[Description], ) -> None: self._status_code_predicates = status_code_predicates self._body_descriptions = body_descriptions def __call__( self: ResponseDescription, status_code: int, body: str, ) -> ResponseVerification: status_code_verification = self._verify_status_code(status_code) try: body_verification = self._verify_body(body) except Exception as error: body_verification = Verification.of_error(error) status = merge_statuses( status_code_verification.status, body_verification.status, ) return ResponseVerification( status=status, status_code=status_code_verification, body=body_verification, ) @property def status_code_predicates(self: ResponseDescription) -> List[Predicate]: return self._status_code_predicates @property def body_descriptions(self: ResponseDescription) -> List[Description]: return self._body_descriptions def _verify_status_code( self: ResponseDescription, code: int, ) -> Verification: children = [pred(code) for pred in self._status_code_predicates] status = merge_statuses(v.status for v in children) return Verification(status=status, children=children) def _verify_body(self: ResponseDescription, body: str) -> Verification: if not self._body_descriptions: return Verification.succeed() data = json.loads(body) verifications = [ describe(data) for describe in self._body_descriptions ] status = merge_statuses(v.status for v in verifications) return Verification(status=status, children=verifications) PK!if preacher/core/scenario.py"""Scenario.""" from __future__ import annotations from dataclasses import dataclass from typing import Optional from .request import Request from .response_description import ( ResponseDescription, ResponseVerification, ) from .verification import ( Status, Verification, merge_statuses, ) @dataclass class ScenarioVerification: status: Status request: Verification response: Optional[ResponseVerification] = None label: Optional[str] = None class Scenario: """ >>> from unittest.mock import MagicMock >>> scenario = Scenario( ... request=MagicMock(Request, side_effect=RuntimeError('message')), ... response_description=MagicMock(ResponseDescription), ... ) >>> scenario.label >>> verification = scenario('base-url') >>> scenario.request.call_args call('base-url') >>> scenario.response_description.call_count 0 >>> verification.label >>> verification.status.name 'FAILURE' >>> verification.request.status.name 'FAILURE' >>> verification.request.message 'RuntimeError: message' >>> from .request import Response >>> inner_response = MagicMock(Response, status_code=402, body='body') >>> scenario = Scenario( ... label='Response should be unstable', ... request=MagicMock(Request, return_value=inner_response), ... response_description=MagicMock( ... ResponseDescription, ... return_value=ResponseVerification( ... status=Status.UNSTABLE, ... status_code=Verification.succeed(), ... body=Verification(status=Status.UNSTABLE) ... ), ... ), ... ) >>> verification = scenario('base-url') >>> scenario.response_description.call_args call(body='body', status_code=402) >>> verification.label 'Response should be unstable' >>> verification.status.name 'UNSTABLE' >>> verification.request.status.name 'SUCCESS' >>> verification.response.status.name 'UNSTABLE' >>> verification.response.body.status.name 'UNSTABLE' """ def __init__( self: Scenario, request: Request, response_description: ResponseDescription, label: Optional[str] = None, ) -> None: self._label = label self._request = request self._response_description = response_description def __call__(self: Scenario, base_url: str) -> ScenarioVerification: try: response = self._request(base_url) except Exception as error: return ScenarioVerification( status=Status.FAILURE, request=Verification.of_error(error), ) request_verification = Verification.succeed() response_verification = self._response_description( status_code=response.status_code, body=response.body, ) status = merge_statuses([ request_verification.status, response_verification.status, ]) return ScenarioVerification( status=status, request=request_verification, response=response_verification, label=self._label, ) @property def label(self: Scenario) -> Optional[str]: return self._label @property def request(self: Scenario) -> Request: return self._request @property def response_description(self: Scenario) -> ResponseDescription: return self._response_description PK!a! preacher/core/verification.py"""Verification.""" from __future__ import annotations from collections.abc import Iterable from dataclasses import dataclass from enum import Enum from functools import reduce, singledispatch from typing import Collection, Optional class Status(Enum): """ >>> Status.SUCCESS.is_succeeded True >>> Status.SUCCESS.merge(Status.SUCCESS).name 'SUCCESS' >>> Status.SUCCESS.merge(Status.UNSTABLE).name 'UNSTABLE' >>> Status.SUCCESS.merge(Status.FAILURE).name 'FAILURE' >>> Status.UNSTABLE.is_succeeded False >>> Status.UNSTABLE.merge(Status.SUCCESS).name 'UNSTABLE' >>> Status.UNSTABLE.merge(Status.UNSTABLE).name 'UNSTABLE' >>> Status.UNSTABLE.merge(Status.FAILURE).name 'FAILURE' >>> Status.FAILURE.is_succeeded False >>> Status.FAILURE.merge(Status.SUCCESS).name 'FAILURE' >>> Status.FAILURE.merge(Status.UNSTABLE).name 'FAILURE' >>> Status.FAILURE.merge(Status.FAILURE).name 'FAILURE' """ # Numbers stand for the priorities for merging. SUCCESS = 0 UNSTABLE = 1 FAILURE = 2 @property def is_succeeded(self): return self is Status.SUCCESS def merge(self, other: Status): return max(self, other, key=lambda status: status.value) @singledispatch def merge_statuses(*args) -> Status: """ >>> merge_statuses(1) Traceback (most recent call last): ... ValueError: (1,) For varargs. >>> merge_statuses(Status.UNSTABLE).name 'UNSTABLE' >>> merge_statuses(Status.SUCCESS, Status.FAILURE, Status.UNSTABLE).name 'FAILURE' For iterables. >>> merge_statuses([]).name 'SUCCESS' >>> merge_statuses([Status.SUCCESS, Status.UNSTABLE, Status.FAILURE]).name 'FAILURE' """ raise ValueError(str(args)) @merge_statuses.register def _merge_statuses_for_varargs(*statuses: Status): return merge_statuses(statuses) @merge_statuses.register def _merge_statuses_for_iterable(statuses: Iterable): return reduce(lambda lhs, rhs: lhs.merge(rhs), statuses, Status.SUCCESS) @dataclass class Verification: status: Status message: Optional[str] = None children: Collection[Verification] = tuple() @staticmethod def succeed() -> Verification: return Verification(status=Status.SUCCESS) @staticmethod def of_error(error: Exception) -> Verification: return Verification( status=Status.FAILURE, message=f'{error.__class__.__name__}: {error}', ) PK!!preacher/presentation/__init__.pyPK!B preacher/presentation/logger.py"""Logger presentation.""" from __future__ import annotations import contextlib import logging import io from typing import Iterator, Optional from preacher.core.verification import Status, Verification from preacher.core.response_description import ResponseVerification from preacher.core.scenario import ScenarioVerification _LEVEL_MAP = { Status.SUCCESS: logging.INFO, Status.UNSTABLE: logging.WARN, Status.FAILURE: logging.ERROR, } class LoggerPresentation: def __init__(self: LoggerPresentation, logger: logging.Logger) -> None: self._logger = logger self._indent = '' def show_scenario_verification( self: LoggerPresentation, verification: ScenarioVerification, label: Optional[str] = None, ) -> None: status = verification.status level = _LEVEL_MAP[status] self._log(level, 'Label: %s', verification.label) self._log(level, 'Status: %s', status.name) with self._nested(): self.show_verification( verification=verification.request, label='Request', ) response = verification.response if response: self.show_response_verification(response) def show_response_verification( self: LoggerPresentation, verification: ResponseVerification, label: str = 'Response', ) -> None: status = verification.status level = _LEVEL_MAP[status] self._log(level, f'%s: %s', label, status.name) with self._nested(): self.show_verification( verification=verification.status_code, label='Status Code', ) self.show_verification( verification=verification.body, label='Body', child_label='Description', ) def show_verification( self: LoggerPresentation, verification: Verification, label: str, child_label: str = 'Predicate', ) -> None: status = verification.status level = _LEVEL_MAP[status] self._log(level, f'%s: %s', label, status.name) message = verification.message if message: with self._nested(): self._multi_line_message(level, message) with self._nested(): for idx, child in enumerate(verification.children): self.show_verification(child, f'{child_label} {idx + 1}') def _log(self, level: int, message: str, *args) -> None: self._logger.log(level, self._indent + message, *args) def _multi_line_message(self, level: int, message: str) -> None: for line in io.StringIO(message): self._log(level, line.rstrip()) @contextlib.contextmanager def _nested(self: LoggerPresentation) -> Iterator[None]: original = self._indent self._indent += '..' yield self._indent = original PK!HD+`3;)preacher-0.1.0.dist-info/entry_points.txtN+I/N.,()*(JMLH-Mɴq zyV PK!T-- preacher-0.1.0.dist-info/LICENSEMIT License Copyright (c) 2019 Yu Mochizuki Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. PK!HڽTUpreacher-0.1.0.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!HCbrŦj!preacher-0.1.0.dist-info/METADATAW[s~_>$i PӦn9';%Vwv(\p)X 'ߤJ ?OϢ oY! i'i|®۪Ko/>iԭʄ;˘ӕCP\c\狦1z>X>QU_:oT?Jo/qZߺ0r/8r_X`|"d㼈n(됙]! ǰ{/f/ B 'P-ʡ;QD5;ToGd_.~%mӦ=#HPD,n䃛?D:^4| c*]9#6F4vsERu$N\83kMA<}'wɭ"oUg #nS[Kk/?C$|H~NoCLϭ%}r2!BɌo3+N}㩶( EiN'|%̡ .ך|* a"b(}>4:>4eIfogop3+IMȯtG}:gih%_XNAQ5B3׎4 ]WC/qgX ,}>{>Y1zPX6 ΎF74O:1"I܊t\IW~m^o 8 σ# =5)Z}7rɘiʱH?;`3΂!i;N Mh8mC8|`;aK$fG3a%?)Ccl ƭ>pH A;]ޭt[}l0/u|QB_fں-M?Y5doۢ:3=cuX'-&ʎGb4K2N`^0`أѹOi>4{D.i>h JD([RqՋ{|QPK!HqIB preacher-0.1.0.dist-info/RECORDKӪ6}? E"(aD7A?}mgNǣL]aK'M$ݯQ*4DWs,^:п kNX:]? 5k!$0iYqӼVVi㑧{K`[U:sǥ?jI-Pаr=yOFaeM.9w\?XFcigѳ'_Iq!ezrU/Rf0e`R *\߻m*ku(RPmDBY" O:ԼE)OrNs$սO檓AZ :cdg+98M{v]= hl6bzJQZ>z{9q;Vwh/CSR1f鳱1P<'' !: ç8.3|G-#pKQ1v*NQ&X'{)p;v͖9LURQ M]0zNTp,2rk홻vvŖx8\~qYx?I*PӀ|Pb|kPM\EG<*ɽ<AvQm#rmKbUKX ){A ,K!"iN$2(ŌD c IgݼJ!{rlTZ{&ċV*1`[!!z_s5Mhh)bYE$@Ue"{#TD/JQ=Hvd˺n1zotqmoVWfb=^S瘅0%:^tך^aN?7cިfќx*> y>reRtAcA->7ʼnc;*Su/ij˻:؉2kcxrq"R8N9[Ԩa0/kA`lrܵe9&zX j>ϾK*lҙSRv ^g 7vXujFy&5E1אUT5[Ie "toRsq3j) h]ݶPDJ5ԔT{e1,|PK!preacher/__init__.pyPK!preacher/app/__init__.pyPK!preacher/app/cli/__init__.pyPK!_3!preacher/app/cli/application.pyPK!_@?preacher/app/cli/main.pyPK!2 c preacher/compilation/__init__.pyPK!;L{ #preacher/compilation/description.pyPK!(xffpreacher/compilation/error.pyPK!ۋBB"N preacher/compilation/extraction.pyPK!R:'!#preacher/compilation/predicate.pyPK!5GAA:preacher/compilation/request.pyPK!C+,?preacher/compilation/response_description.pyPK!.o; ; 4Npreacher/compilation/scenario.pyPK![preacher/core/__init__.pyPK!|. [preacher/core/description.pyPK!Lhpreacher/core/extraction.pyPK!Y  jpreacher/core/predicate.pyPK!OWrpreacher/core/request.pyPK! O%Hxpreacher/core/response_description.pyPK!if Zpreacher/core/scenario.pyPK!a! zpreacher/core/verification.pyPK!!preacher/presentation/__init__.pyPK!B Ԡpreacher/presentation/logger.pyPK!HD+`3;)preacher-0.1.0.dist-info/entry_points.txtPK!T-- 7preacher-0.1.0.dist-info/LICENSEPK!HڽTUpreacher-0.1.0.dist-info/WHEELPK!HCbrŦj!2preacher-0.1.0.dist-info/METADATAPK!HqIB preacher-0.1.0.dist-info/RECORDPKY