PK!@*xpreacher/__init__.py"""Preacher: A Web API Verification Tool.""" __version__ = '0.9.2' __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!ϸjpreacher/app/cli/application.pyfrom __future__ import annotations from preacher.core.scenario import Scenario from preacher.presentation.logging import LoggingPresentation class Application: def __init__( self: Application, base_url: str, view: LoggingPresentation, ) -> 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.logging import LoggingPresentation 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 = LoggingPresentation(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!/jt88 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 from .util import map_on_key _KEY_SCENARIOS = 'scenarios' def compile(obj: Mapping) -> Iterator[Scenario]: """ >>> scenarios = list(compile({})) >>> scenarios [] >>> next(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.assert_called_once_with({}) sentinel.scenario >>> with scenario_patch as scenario_mock: ... next(scenarios) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ...: scenarios[1] """ scenario_objs = obj.get(_KEY_SCENARIOS, []) if not isinstance(scenario_objs, list): raise CompilationError(message='Must be a list', path=[_KEY_SCENARIOS]) return map_on_key(_KEY_SCENARIOS, _compile_scenario, scenario_objs) def _compile_scenario(obj: Any) -> Scenario: if not isinstance(obj, Mapping): raise CompilationError(f'Scenario must be a mapping') return compile_scenario(obj) PK!e{' ' #preacher/compilation/description.py"""Description compilation.""" from collections.abc import Mapping from preacher.core.description import Description from .error import CompilationError from .predicate import compile as compile_predicate from .extraction import compile as compile_extraction from .util import run_on_key, map_on_key _KEY_DESCRIBE = 'describe' _KEY_IT_SHOULD = 'it_should' def compile(obj: Mapping) -> Description: """ >>> from unittest.mock import call, 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, \\ ... predicate_patch as predicate_mock: ... description = compile({ ... 'describe': 'foo', ... 'it_should': 'string', ... }) ... extraction_mock.assert_called_with('foo') ... predicate_mock.assert_called_once_with('string') >>> description.extraction sentinel.extraction >>> description.predicates [sentinel.predicate] >>> with extraction_patch as extraction_mock, \\ ... predicate_patch as predicate_mock: ... description = compile({ ... 'describe': 'foo', ... 'it_should': {'key': 'value'} ... }) ... extraction_mock.assert_called_once_with('foo') ... predicate_mock.assert_called_once_with({'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_should': [{'key1': 'value1'}, {'key2': 'value2'}] ... }) ... extraction_mock.assert_called_once_with({'key': 'value'}) ... predicate_mock.assert_has_calls([ ... call({'key1': 'value1'}), ... call({'key2': 'value2'}), ... ]) >>> description.extraction sentinel.extraction >>> description.predicates [sentinel.predicate, sentinel.predicate] """ extraction_obj = obj.get(_KEY_DESCRIBE) if ( not isinstance(extraction_obj, Mapping) and not isinstance(extraction_obj, str) ): raise CompilationError( message='Description.describe must be a mapping', path=[_KEY_DESCRIBE], ) extraction = run_on_key(_KEY_DESCRIBE, compile_extraction, extraction_obj) predicate_objs = obj.get(_KEY_IT_SHOULD, []) if not isinstance(predicate_objs, list): predicate_objs = [predicate_objs] predicates = list( map_on_key(_KEY_IT_SHOULD, compile_predicate, predicate_objs) ) return Description(extraction=extraction, predicates=predicates) PK!ƅpreacher/compilation/error.py"""Compilation errors.""" from __future__ import annotations from typing import List, Optional class CompilationError(Exception): """Compilation errors.""" def __init__( self: CompilationError, message: str, path: List[str] = [], cause: Optional[Exception] = None, ) -> None: super().__init__(message) self._message = message self._path = path self._cause = cause def of_parent( self: CompilationError, parent_path: List[str], ) -> CompilationError: return CompilationError( message=self._message, path=parent_path + self._path, cause=self._cause, ) def __str__(self: CompilationError): message = super().__str__() if not self._path: return message path = '.'.join(self._path) return f'{message}: {path}' PK!"preacher/compilation/extraction.py"""Extraction compilation.""" from collections.abc import Mapping from typing import Union 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(obj: Union[Mapping, str]) -> Extraction: """ >>> compile({}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ... has 0 >>> compile('.foo')({'foo': 'bar'}) 'bar' >>> compile({'jq': '.foo'})({'foo': 'bar'}) 'bar' """ if isinstance(obj, str): return compile({'jq': obj}) keys = _EXTRACTION_KEYS.intersection(obj.keys()) if len(keys) != 1: raise CompilationError( f'Extraction must have only 1 valid key, but has {len(keys)}' ) key = next(iter(keys)) return _EXTRACTION_MAP[key](obj[key]) PK! preacher/compilation/matcher.py"""Matcher compilation.""" from collections.abc import Mapping from typing import Any import hamcrest from hamcrest.core.matcher import Matcher from .error import CompilationError from .util import run_on_key _STATIC_MATCHER_MAP = { # For objects. 'be_null': hamcrest.is_(hamcrest.none()), 'not_be_null': hamcrest.is_(hamcrest.not_none()), # For collections. 'be_empty': hamcrest.is_(hamcrest.empty()), } _MATCHER_FUNCTION_MAP_TAKING_SINGLE_VALUE = { # For objects. 'equal': lambda expected: hamcrest.is_(hamcrest.equal_to(expected)), 'have_length': hamcrest.has_length, # For numbers. 'be_greater_than': ( lambda value: hamcrest.is_(hamcrest.greater_than(value)) ), 'be_greater_than_or_equal_to': ( lambda value: hamcrest.is_(hamcrest.greater_than_or_equal_to(value)) ), 'be_less_than': ( lambda value: hamcrest.is_(hamcrest.less_than(value)) ), 'be_less_than_or_equal_to': ( lambda value: hamcrest.is_(hamcrest.less_than_or_equal_to(value)) ), # For strings. 'contain_string': hamcrest.contains_string, 'start_with': hamcrest.starts_with, 'end_with': hamcrest.ends_with, 'match_regexp': hamcrest.matches_regexp, } _MATCHER_FUNCTION_MAP_TAKING_SINGLE_MATCHER = { 'be': hamcrest.is_, 'not': hamcrest.not_, 'have_item': hamcrest.has_item, } def _compile_static_matcher(name: str) -> Matcher: """ >>> _compile_static_matcher('invalid_name') Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ... 'invalid_name' >>> matcher = _compile_static_matcher('be_null') >>> assert matcher.matches(None) >>> assert not matcher.matches(False) >>> matcher = _compile_static_matcher('not_be_null') >>> assert not matcher.matches(None) >>> assert matcher.matches('False') >>> matcher = _compile_static_matcher('be_empty') >>> assert not matcher.matches(None) >>> assert not matcher.matches(0) >>> assert matcher.matches('') >>> assert not matcher.matches('A') >>> assert matcher.matches([]) >>> assert not matcher.matches([1]) """ matcher = _STATIC_MATCHER_MAP.get(name) if not matcher: raise CompilationError(f'Invalid matcher: \'{name}\'') return matcher def _compile_taking_value(key: str, value: Any) -> Matcher: """ >>> _compile_taking_value('invalid_key', 0) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ... 'invalid_key' >>> matcher = _compile_taking_value('have_length', 1) >>> assert not matcher.matches(None) >>> assert not matcher.matches('') >>> assert not matcher.matches([]) >>> assert matcher.matches('A') >>> assert matcher.matches([1]) >>> matcher = _compile_taking_value('equal', 1) >>> assert not matcher.matches(0) >>> assert not matcher.matches('1') >>> assert matcher.matches(1) >>> matcher = _compile_taking_value('be_greater_than', 0) >>> assert not matcher.matches(-1) >>> assert not matcher.matches(0) >>> assert matcher.matches(1) >>> matcher = _compile_taking_value('be_greater_than_or_equal_to', 0) >>> assert not matcher.matches(-1) >>> assert matcher.matches(0) >>> assert matcher.matches(1) >>> matcher = _compile_taking_value('be_less_than', 0) >>> assert matcher.matches(-1) >>> assert not matcher.matches(0) >>> assert not matcher.matches(1) >>> matcher = _compile_taking_value('be_less_than_or_equal_to', 0) >>> assert matcher.matches(-1) >>> assert matcher.matches(0) >>> assert not matcher.matches(1) >>> matcher = _compile_taking_value('contain_string', '0') >>> assert not matcher.matches(0) >>> assert not matcher.matches('123') >>> assert matcher.matches('21012') >>> matcher = _compile_taking_value('start_with', 'AB') >>> assert not matcher.matches(0) >>> assert matcher.matches('ABC') >>> assert not matcher.matches('ACB') >>> matcher = _compile_taking_value('end_with', 'BC') >>> assert not matcher.matches(0) >>> assert matcher.matches('ABC') >>> assert not matcher.matches('ACB') >>> matcher = _compile_taking_value('match_regexp', '^A*B$') >>> assert not matcher.matches('ACB') >>> assert matcher.matches('B') >>> matcher.matches(0) # TODO: Should return `False` when given not `str`. Traceback (most recent call last): ... TypeError: ... """ func = _MATCHER_FUNCTION_MAP_TAKING_SINGLE_VALUE.get(key) if not func: raise CompilationError(f'Unrecognized matcher key: \'{key}\'') return func(value) def _compile_taking_single_matcher(key: str, value: Any): """ >>> _compile_taking_single_matcher('invalid_key', '') Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ... 'invalid_key' >>> matcher = _compile_taking_single_matcher('be', 1) >>> assert not matcher.matches(0) >>> assert not matcher.matches('1') >>> assert matcher.matches(1) >>> matcher = _compile_taking_single_matcher('not', 1) >>> assert matcher.matches('A') >>> assert matcher.matches(0) >>> assert not matcher.matches(1) >>> matcher = _compile_taking_single_matcher('not', {'be_greater_than': 0}) >>> assert matcher.matches(-1) >>> assert matcher.matches(0) >>> assert not matcher.matches(1) >>> matcher = _compile_taking_single_matcher('have_item', {'equal': 1}) >>> assert not matcher.matches(None) >>> assert not matcher.matches([]) >>> assert not matcher.matches([0, 'A']) >>> assert matcher.matches([0, 1, 2]) """ func = _MATCHER_FUNCTION_MAP_TAKING_SINGLE_MATCHER.get(key) if not func: raise CompilationError(f'Unrecognized matcher key: \'{key}\'') if isinstance(value, str) or isinstance(value, Mapping): inner = run_on_key(key, compile, value) else: inner = hamcrest.equal_to(value) return func(inner) def compile(obj: Any) -> Matcher: """ >>> from unittest.mock import patch, sentinel >>> compile({}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ... has 0 >>> compile({'key1': 'value1', 'key2': 'value2'}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ... has 2 >>> matcher = compile('invalid_name') >>> assert not matcher.matches('') >>> assert matcher.matches('invalid_name') >>> matcher = compile({'invalid_key': ''}) >>> assert not matcher.matches('') >>> assert matcher.matches({'invalid_key': ''}) >>> with patch( ... f'{__name__}._compile_static_matcher', ... return_value=sentinel.static_matcher, ... ) as matcher_mock: ... compile('be_null') ... matcher_mock.assert_called_with('be_null') sentinel.static_matcher >>> with patch( ... f'{__name__}._compile_taking_value', ... return_value=sentinel.value_matcher, ... ) as matcher_mock: ... compile({'equal': 'value'}) ... matcher_mock.assert_called_with('equal', 'value') sentinel.value_matcher >>> with patch( ... f'{__name__}._compile_taking_single_matcher', ... return_value=sentinel.single_matcher_matcher, ... ) as matcher_mock: ... compile({'not': 'value'}) ... matcher_mock.assert_called_with('not', 'value') sentinel.single_matcher_matcher """ if isinstance(obj, str): if obj in _STATIC_MATCHER_MAP: return _compile_static_matcher(obj) if isinstance(obj, Mapping): if len(obj) != 1: raise CompilationError( f'Must have only 1 element, but has {len(obj)}' ) key, value = next(iter(obj.items())) if key in _MATCHER_FUNCTION_MAP_TAKING_SINGLE_VALUE: return _compile_taking_value(key, value) if key in _MATCHER_FUNCTION_MAP_TAKING_SINGLE_MATCHER: return _compile_taking_single_matcher(key, value) return hamcrest.equal_to(obj) PK! CC!preacher/compilation/predicate.py"""Predicate compilation.""" from typing import Any from preacher.core.predicate import Predicate, of_hamcrest_matcher from .matcher import compile as compile_matcher def compile(obj: Any) -> Predicate: """ >>> from unittest.mock import patch, sentinel >>> with patch( ... f'{__name__}.compile_matcher', ... return_value=sentinel.matcher ... ) as matcher_mock: ... predicate = compile('matcher') ... matcher_mock.assert_called_with('matcher') """ matcher = compile_matcher(obj) return of_hamcrest_matcher(matcher) PK!jkpreacher/compilation/request.py"""Request compilation.""" from collections.abc import Mapping from typing import Union from preacher.core.request import Request from .error import CompilationError _KEY_PATH = 'path' _KEY_PARAMS = 'params' def compile(obj: Union[Mapping, str]) -> Request: """ >>> compile({'path': {'key': 'value'}}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: Request.path ...: path >>> compile({'params': ''}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: Request.params ...: params >>> request = compile('/path') >>> request.path '/path' >>> request.params {} >>> request = compile({}) >>> request.path '' >>> request.params {} >>> request = compile({'path': '/path', 'params': {'key': 'value'}}) >>> request.path '/path' >>> request.params {'key': 'value'} """ if isinstance(obj, str): return compile({_KEY_PATH: obj}) path = obj.get(_KEY_PATH, '') if not isinstance(path, str): raise CompilationError( message=f'Request.{_KEY_PATH} must be a string', path=[_KEY_PATH], ) params = obj.get(_KEY_PARAMS, {}) if not isinstance(params, Mapping): raise CompilationError( message=f'Request.{_KEY_PARAMS} must be a mapping', path=[_KEY_PARAMS], ) return Request(path=path, params=params) PK!`,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 from .util import map_on_key _KEY_STATUS_CODE = 'status_code' _KEY_BODY = 'body' def compile(obj: Mapping) -> ResponseDescription: """ >>> from unittest.mock import call, 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({}) ... predicate_mock.assert_not_called() ... description_mock.assert_not_called() >>> response_description.status_code_predicates [] >>> response_description.body_descriptions [] >>> compile({'body': 'str'}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ...: body >>> compile({'body': ['str']}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: Description ...: body[0] >>> with predicate_patch as predicate_mock, \\ ... description_patch as description_mock: ... response_description = compile({ ... 'status_code': 402, ... 'body': {'key1': 'value1'}} ... ) ... predicate_mock.assert_called_once_with(402) ... description_mock.assert_called_once_with({'key1': 'value1'}) >>> response_description.body_descriptions [sentinel.description] >>> with predicate_patch as predicate_mock, \\ ... description_patch as description_mock: ... response_description = compile({ ... 'status_code': [{'be_greater_than': 0}, {'be_less_than': 400}], ... 'body': [{'key1': 'value1'}, {'key2': 'value2'}], ... }) ... predicate_mock.assert_has_calls([ ... call({'be_greater_than': 0}), ... call({'be_less_than': 400}), ... ]) ... description_mock.assert_has_calls([ ... call({'key1': 'value1'}), ... call({'key2': 'value2'}), ... ]) >>> response_description.body_descriptions [sentinel.description, sentinel.description] """ status_code_predicate_objs = obj.get(_KEY_STATUS_CODE, []) if not isinstance(status_code_predicate_objs, list): status_code_predicate_objs = [status_code_predicate_objs] status_code_predicates = list(map_on_key( key=_KEY_STATUS_CODE, func=compile_predicate, items=status_code_predicate_objs, )) body_description_objs = obj.get(_KEY_BODY, []) if isinstance(body_description_objs, Mapping): body_description_objs = [body_description_objs] if not isinstance(body_description_objs, list): raise CompilationError( message='ResponseDescription.body must be a list or a mapping', path=[_KEY_BODY], ) body_descriptions = list(map_on_key( key=_KEY_BODY, func=_compile_description, items=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('Description must be a mapping') return compile_description(obj) PK!@ȟ 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 from .util import run_on_key _KEY_LABEL = 'label' _KEY_REQUEST = 'request' _KEY_RESPONSE = 'response' 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.assert_called_once_with({}) ... response_description_mock.assert_called_once_with({}) >>> scenario.request sentinel.request >>> scenario.response_description sentinel.response_description >>> compile_scenario({'label': []}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: Scenario.label ...: label >>> compile_scenario({'request': []}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: Scenario.request ...: request >>> with request_patch as request_mock: ... request_mock.side_effect=CompilationError( ... message='message', ... path=['foo'], ... ) ... compile_scenario({}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: message: request.foo >>> with request_patch: ... compile_scenario({'response': 'str'}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: Scenario.response...: response >>> with request_patch as request_mock, \\ ... response_description_patch as response_description_mock: ... response_description_mock.side_effect=CompilationError( ... message='message', ... path=['bar'], ... ) ... scenario = compile_scenario({'request': '/path'}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: message: response.bar >>> with request_patch as request_mock, \\ ... response_description_patch as response_description_mock: ... scenario = compile_scenario({'request': '/path'}) ... request_mock.assert_called_once_with('/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.assert_called_once_with({'path': '/path'}) ... response_description_mock.assert_called_once_with({'key': 'value'}) >>> scenario.label 'label' >>> scenario.request sentinel.request >>> scenario.response_description sentinel.response_description """ label = obj.get(_KEY_LABEL) if label is not None and not isinstance(label, str): raise CompilationError( message=f'Scenario.{_KEY_LABEL} must be a string', path=[_KEY_LABEL], ) request_obj = obj.get(_KEY_REQUEST, {}) if ( not isinstance(request_obj, Mapping) and not isinstance(request_obj, str) ): raise CompilationError( message=f'Scenario.{_KEY_REQUEST} must be a string or a mapping', path=[_KEY_REQUEST], ) request = run_on_key(_KEY_REQUEST, compile_request, request_obj) response_obj = obj.get('response', {}) if not isinstance(response_obj, Mapping): raise CompilationError( message=f'Scenario.{_KEY_RESPONSE} object must be a mapping', path=[_KEY_RESPONSE], ) response_description = run_on_key( _KEY_RESPONSE, compile_response_description, response_obj, ) return Scenario( label=label, request=request, response_description=response_description, ) PK!f;preacher/compilation/util.py"""Utilities for compilations.""" from typing import Callable, Iterable, Iterator, TypeVar from .error import CompilationError T = TypeVar('T') U = TypeVar('U') def run_on_key( key: str, func: Callable[[T], U], arg: T, ) -> U: """ >>> def succeeding_func(arg): ... return arg >>> run_on_key('key', succeeding_func, 1) 1 >>> def failing_func(arg): ... raise CompilationError(message='message', path=['path']) >>> run_on_key('key', failing_func, 1) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: message: key.path """ try: return func(arg) except CompilationError as error: raise error.of_parent([key]) def map_on_key( key: str, func: Callable[[T], U], items: Iterable[T], ) -> Iterator[U]: """ >>> def succeeding_func(arg): ... return arg >>> results = map_on_key('key', succeeding_func, [1, 2, 3]) >>> next(results) 1 >>> next(results) 2 >>> next(results) 3 >>> def failing_func(arg): ... if arg == 2: ... raise CompilationError(message='message', path=['path']) ... return arg >>> results = map_on_key('key', failing_func, [1, 2, 3]) >>> next(results) 1 >>> next(results) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: message: key[1].path >>> next(results) Traceback (most recent call last): ... StopIteration """ for idx, item in enumerate(items): try: yield func(item) except CompilationError as error: raise error.of_parent([f'{key}[{idx}]']) PK!preacher/core/__init__.pyPK!]m preacher/core/description.py"""Description.""" from __future__ import annotations from typing import Any, Callable, List from .status import merge_statuses from .verification import Verification Extraction = Callable[[Any], Any] Predicate = Callable[[Any], Verification] class Description: """ Description. >>> from .status 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 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 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 FAILURE >>> len(verification.children) 3 >>> verification.children[0].status UNSTABLE >>> verification.children[1].status FAILURE >>> verification.children[2].status 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 SUCCESS >>> len(verification.children) 2 >>> verification.children[0].status SUCCESS >>> verification.children[1].status 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!A(,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 .status import Status from .verification import 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 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 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 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! 8%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 .status import Status, merge_statuses from .verification import Verification @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 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].assert_called_once_with(200) >>> description.body_descriptions[0].assert_not_called() >>> verification.status FAILURE >>> verification.body.status FAILURE >>> verification.body.message 'JSONDecodeError: Expecting value: line 1 column 1 (char 0)' >>> from unittest.mock import MagicMock, call >>> 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].assert_called_once_with({}) >>> description.body_descriptions[1].assert_called_once_with({}) >>> verification.status UNSTABLE >>> verification.body.status UNSTABLE >>> verification.body.children[0].status UNSTABLE >>> verification.body.children[1].status 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!b6I 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 .status import Status, merge_statuses from .verification import Verification @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 FAILURE >>> verification.request.status 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 UNSTABLE >>> verification.request.status SUCCESS >>> verification.response.status UNSTABLE >>> verification.response.body.status 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!N  preacher/core/status.py"""Status.""" from __future__ import annotations from collections.abc import Iterable from enum import Enum from functools import reduce, singledispatch class Status(Enum): """ >>> Status.SUCCESS.is_succeeded True >>> Status.SUCCESS.merge(Status.SUCCESS) SUCCESS >>> Status.SUCCESS.merge(Status.UNSTABLE) UNSTABLE >>> Status.SUCCESS.merge(Status.FAILURE) FAILURE >>> Status.UNSTABLE.is_succeeded False >>> Status.UNSTABLE.merge(Status.SUCCESS) UNSTABLE >>> Status.UNSTABLE.merge(Status.UNSTABLE) UNSTABLE >>> Status.UNSTABLE.merge(Status.FAILURE) FAILURE >>> Status.FAILURE.is_succeeded False >>> Status.FAILURE.merge(Status.SUCCESS) FAILURE >>> Status.FAILURE.merge(Status.UNSTABLE) FAILURE >>> Status.FAILURE.merge(Status.FAILURE) FAILURE """ # Numbers stand for the priorities for merging. SUCCESS = 0 UNSTABLE = 1 FAILURE = 2 @property def is_succeeded(self: Status): return self is Status.SUCCESS def merge(self: Status, other: Status): return max(self, other, key=lambda status: status.value) def __str__(self: Status) -> str: return self.name def __repr__(self: Status) -> str: return str(self) @singledispatch def merge_statuses(*args) -> Status: """ >>> merge_statuses(1) Traceback (most recent call last): ... ValueError: (1,) For varargs. >>> merge_statuses(Status.UNSTABLE) UNSTABLE >>> merge_statuses(Status.SUCCESS, Status.FAILURE, Status.UNSTABLE) FAILURE For iterables. >>> merge_statuses([]) SUCCESS >>> merge_statuses([Status.SUCCESS, Status.UNSTABLE, Status.FAILURE]) 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) PK!h\&9\\preacher/core/verification.py"""Verification.""" from __future__ import annotations from dataclasses import dataclass from typing import Collection, Optional from .status import Status @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! preacher/presentation/logging.py"""Logging presentation.""" from __future__ import annotations import contextlib import logging import io from typing import Iterator, Optional from preacher.core.status import Status from preacher.core.verification import 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 LoggingPresentation: def __init__(self: LoggingPresentation, logger: logging.Logger) -> None: self._logger = logger self._indent = '' def show_scenario_verification( self: LoggingPresentation, verification: ScenarioVerification, label: Optional[str] = None, ) -> None: status = verification.status level = _LEVEL_MAP[status] self._log(level, '%s: %s', verification.label, status) with self._nested(): self.show_verification( verification=verification.request, label='Request', ) response = verification.response if response: self.show_response_verification(response) self._log(level, '') def show_response_verification( self: LoggingPresentation, verification: ResponseVerification, label: str = 'Response', ) -> None: status = verification.status level = _LEVEL_MAP[status] self._log(level, f'%s: %s', label, status) 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: LoggingPresentation, verification: Verification, label: str, child_label: str = 'Predicate', ) -> None: status = verification.status level = _LEVEL_MAP[status] self._log(level, f'%s: %s', label, status) 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: LoggingPresentation) -> Iterator[None]: original = self._indent self._indent += '..' yield self._indent = original PK!HD+`3;)preacher-0.9.2.dist-info/entry_points.txtN+I/N.,()*(JMLH-Mɴq zyV PK!T-- preacher-0.9.2.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.9.2.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!Hy4ݪS!preacher-0.9.2.dist-info/METADATAX[s6~ǯt슔t-&47Y;m'O"DB$P6s"J-Ϯ^DknX ~J YE,<%f%h8KrHvۦ,j#|E Wb-f)%jܘZGy&LެDI>5\WH"REsCJ"fK&Z0e" o:)X{y:~vmxBLkXFp.Z Oq ڻ-Jf2zŪi§҇mֆ8 ',|=;=A` B#:;=@a Oӿ w<17WHΞpoN17225 !<_CCgQf/R 9_4 VEC_dʸBGJͧ4XQ-*ޡRL7ma- UsX%$L*)Ly"7h]ϝ7sA& VdDEiwu) U#E*&}dĵU+XyEb*u|-592]K,@ j:${LqاU2{x "ZtfथPx!kTH3Zp*a57Y^msw ٦I0AkCYmW0gBix=2sbE kÊb/a[//j - 4[<_WtlPʾ*u;w;NZdr; :Ҁzέwi7s#k)G.\~DfDfONFG+cRT2p,Yj"v(bj׃47c7]v=Fб!t8 oO#aPwv ٓAmŗT%$VjiY '/֖6>R`?C%xD'yӖp,삝T+_,h{c*W)%k[c$AhQB>D!Wp~ޥegDMfzC#^8_p3O_y G&}ps *N EiVZ5:o8@Kk%K`B7m%y:ctwt ]d `R>`\k83LCRw`A-mPk4/^qh;Όs'zF唎@l+7KoJc;UkZu̅X-(e"O MVr[AA'ո9{d2x cR| ^ͻ~wT;%300v? VAlUSlv6IGށ.귧14"'(?>8q]lˆ#)Pyݏqv&H@Pݓ72FqŇEV|j}5P$=d;2ye=Zdp:Eq_Y\+( zڭ;1 Dݩ}[@0R5EO] xYvA]vH;_X=n S-|>]%axU; M%Ջqp-n0ݽZ=Dmp2ip@q훥rgp~״3$Gv,{A s!</-k%ՄiYJ*y س`4?U%vy83vT&$etBhe K -06NjÂm~/c5d5yavbe[C" xjl42c_oi۝hdr~wjQj3uəL'>BPK!H preacher-0.9.2.dist-info/RECORDɲ6}nEf2` `o( b6ȀqwU'ߤ*$:,Nl5@`} wd*cf~=fcW_Uuqn.I#~X0(VTZu1ٱnis?yKM9!ZڀA|~ joH%ft( ]cX:+xo#,mcꁽ=8R 0̄sΞ*e?o|_߻_.ɢ*T،rYVc64O/ٔ`TOHòyN;ecFʙ]lP.{;~|Qeun3L;C*ggmlc~Z/5JS&@4C5RT˄r"{}Tr6O|6^eFA$&f\ǷNPj0z }^,{F^g)4E~DG3k~LQ)9f]o_9u9؄lJ]VvruJީqs G$ )B4ݔEÁ:W LAy ܷ9Ni#_[HR1ÒI VUubv3Z\nx]EkL0rt znrUE^Tdv?m;ks ^Z*,&&$ ڹ$B8 }{%hŮ.ޒ6}]>%,e5G"dP?Iˬ>.6"6$K'WE{ܞ #O_z;)QSՉ,  T >;S 0v(bc3aD'B.-'M>u+g05b UFJ i'ϮHk:ަϽf]ݽW=cek\ G*hDQu}l_b#+=E3y SC4v*6`/~LY|peb(@W\m7] ~PrIM;{a/|# Q >.`($=z L*@ [ji fh7_ʅ ]Bohl:Ǧik0"aBgϳz<7ʲuEM(g>mbm39;w~kp^v7phZnꈺR#qv{d0v4+ aZ>ʢuPK!@*xpreacher/__init__.pyPK!preacher/app/__init__.pyPK!preacher/app/cli/__init__.pyPK!ϸj!preacher/app/cli/application.pyPK!*Bpreacher/app/cli/main.pyPK!/jt88 i preacher/compilation/__init__.pyPK!e{' ' #preacher/compilation/description.pyPK!ƅGpreacher/compilation/error.pyPK!"#preacher/compilation/extraction.pyPK! &preacher/compilation/matcher.pyPK! CC!Gpreacher/compilation/predicate.pyPK!jkIpreacher/compilation/request.pyPK!`,Opreacher/compilation/response_description.pyPK!@ȟ ^preacher/compilation/scenario.pyPK!f;ppreacher/compilation/util.pyPK!wpreacher/core/__init__.pyPK!]m wpreacher/core/description.pyPK!Lpreacher/core/extraction.pyPK!A(,wpreacher/core/predicate.pyPK!OW͍preacher/core/request.pyPK! 8%preacher/core/response_description.pyPK!b6I preacher/core/scenario.pyPK!N  preacher/core/status.pyPK!h\&9\\9preacher/core/verification.pyPK!!мpreacher/presentation/__init__.pyPK! preacher/presentation/logging.pyPK!HD+`3;)preacher-0.9.2.dist-info/entry_points.txtPK!T-- ~preacher-0.9.2.dist-info/LICENSEPK!HڽTUpreacher-0.9.2.dist-info/WHEELPK!Hy4ݪS!ypreacher-0.9.2.dist-info/METADATAPK!H bpreacher-0.9.2.dist-info/RECORDPK6