PK!lpreacher/__init__.py"""Preacher: A Web API Verification Tool.""" __version__ = '0.9.6' __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!K""preacher/app/cli/application.pyfrom __future__ import annotations from multiprocessing import Pool from typing import Callable, Iterable, Iterator import ruamel.yaml as yaml from preacher.core.scenario_running import ScenarioResult, run_scenario from preacher.compilation.scenario import ScenarioCompiler from preacher.presentation.logging import LoggingPresentation MapFunction = Callable[ [ Callable[[str], ScenarioResult], Iterable[str] ], Iterator[ScenarioResult] ] class Application: def __init__( self: Application, view: LoggingPresentation, base_url: str, retry: int = 0, ) -> None: self._view = view self._base_url = base_url self._retry = retry self._scenario_compiler = ScenarioCompiler() self._is_succeeded = True @property def is_succeeded(self: Application) -> bool: return self._is_succeeded def run( self: Application, config_paths: Iterable[str], map_func: MapFunction = map, ) -> None: results = map_func(self._run_each, config_paths) for result in results: self._is_succeeded &= result.status.is_succeeded self._view.show_scenario_result(result) def run_concurrently( self: Application, config_paths: Iterable[str], concurrency: int, ) -> None: with Pool(concurrency) as pool: self.run(config_paths, map_func=pool.imap) def _run_each(self: Application, config_path: str) -> ScenarioResult: with open(config_path) as config_file: config = yaml.safe_load(config_file) scenario = self._scenario_compiler.compile(config) return run_scenario( scenario, base_url=self._base_url, retry=self._retry, ) PK!3$Ipreacher/app/cli/main.py"""Preacher CLI.""" from __future__ import annotations import argparse import logging import sys from preacher import __version__ as VERSION from preacher.presentation.logging import LoggingPresentation from .application import Application HANDLER = logging.StreamHandler() LOGGER = logging.getLogger(__name__) LOGGER.addHandler(HANDLER) LOGGING_LEVEL_MAP = { 'skipped': logging.DEBUG, 'success': logging.INFO, 'unstable': logging.WARN, 'failure': logging.ERROR, } def zero_or_positive_int(value: str) -> int: int_value = int(value) if int_value < 0: raise argparse.ArgumentTypeError( f"must be positive or 0, given {int_value}" ) return int_value 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( '-l', '--level', choices=LOGGING_LEVEL_MAP.keys(), help='show only above or equal to this level', default='success', ) parser.add_argument( '-r', '--retry', type=zero_or_positive_int, help='max retry count', default=0, ) parser.add_argument( '-c', '--scenario-concurrency', type=int, help='concurrency for scenarios', default=1, ) return parser.parse_args() def main() -> None: """Main.""" args = parse_args() level = LOGGING_LEVEL_MAP[args.level] HANDLER.setLevel(level) LOGGER.setLevel(level) view = LoggingPresentation(LOGGER) base_url = args.url retry = args.retry app = Application(view=view, base_url=base_url, retry=retry) config_paths = args.conf scenario_concurrency = args.scenario_concurrency app.run_concurrently(config_paths, concurrency=scenario_concurrency) if not app.is_succeeded: sys.exit(1) PK! preacher/compilation/__init__.pyPK!c--preacher/compilation/case.py"""Case compilation.""" from __future__ import annotations from collections.abc import Mapping from typing import Any, Optional, Union from preacher.core.case import Case from .error import CompilationError from .request import RequestCompiler from .response_description import ResponseDescriptionCompiler from .util import run_on_key _KEY_LABEL = 'label' _KEY_REQUEST = 'request' _KEY_RESPONSE = 'response' class CaseCompiler: """ >>> from unittest.mock import MagicMock, sentinel >>> def default_request_compiler() -> RequestCompiler: ... return MagicMock( ... spec=RequestCompiler, ... compile=MagicMock(return_value=sentinel.request), ... ) >>> def default_response_compiler() -> ResponseDescriptionCompiler: ... return MagicMock( ... spec=ResponseDescriptionCompiler, ... compile=MagicMock(return_value=sentinel.response_description), ... ) When given an empty object, then generates a default case. >>> request_compiler = default_request_compiler() >>> response_compiler = default_response_compiler() >>> compiler = CaseCompiler(request_compiler, response_compiler) >>> case = compiler.compile({}) >>> case.request sentinel.request >>> case.response_description sentinel.response_description >>> request_compiler.compile.assert_called_once_with({}) >>> response_compiler.compile.assert_called_once_with({}) When given a not string label, then raises a compilation error. >>> compiler = CaseCompiler() >>> compiler.compile({'label': []}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ...: label When given an invalid type request, then raises a compilation error. >>> compiler = CaseCompiler() >>> compiler.compile({'request': []}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ...: request When a request compilation fails, then raises a compilation error. >>> request_compiler = MagicMock( ... spec=RequestCompiler, ... compile=MagicMock( ... side_effect=CompilationError(message='message', path=['foo']) ... ), ... ) >>> compiler = CaseCompiler(request_compiler) >>> compiler.compile({}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: message: request.foo When given an invalid type response description, then raises a compilation error. >>> request_compiler = default_request_compiler() >>> compiler = CaseCompiler(request_compiler) >>> compiler.compile({'response': 'str'}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ...: response When a response description compilation fails, then raises a compilation error. >>> request_compiler = default_request_compiler() >>> response_compiler = MagicMock( ... spec=ResponseDescriptionCompiler, ... compile=MagicMock( ... side_effect=CompilationError(message='message', path=['bar']), ... ), ... ) >>> compiler = CaseCompiler(request_compiler, response_compiler) >>> compiler.compile({'request': '/path'}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: message: response.bar Creates a case only with a request. >>> request_compiler = default_request_compiler() >>> response_compiler = default_response_compiler() >>> compiler = CaseCompiler(request_compiler, response_compiler) >>> case = compiler.compile({'request': '/path'}) >>> case.label >>> case.request sentinel.request >>> request_compiler.compile.assert_called_once_with('/path') Creates a case. >>> request_compiler = default_request_compiler() >>> response_compiler = default_response_compiler() >>> compiler = CaseCompiler(request_compiler, response_compiler) >>> case = compiler.compile({ ... 'label': 'label', ... 'request': {'path': '/path'}, ... 'response': {'key': 'value'}, ... }) >>> case.label 'label' >>> case.request sentinel.request >>> case.response_description sentinel.response_description >>> request_compiler.compile.assert_called_once_with({'path': '/path'}) >>> response_compiler.compile.assert_called_once_with({'key': 'value'}) When given invalid default request, then raises a compilation error. >>> case_compiler = CaseCompiler() >>> case_compiler.of_default({'request': []}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ...: request Accepts default values. >>> request_compiler = MagicMock( ... RequestCompiler, ... of_default=MagicMock(return_value=sentinel.foo), ... ) >>> response_compiler = default_response_compiler() >>> compiler = CaseCompiler(request_compiler, response_compiler) >>> default_compiler = compiler.of_default({}) >>> default_compiler.request_compiler sentinel.foo """ def __init__( self: CaseCompiler, request_compiler: Optional[RequestCompiler] = None, response_compiler: Optional[ResponseDescriptionCompiler] = None ) -> None: self._request_compiler = request_compiler or RequestCompiler() self._response_compiler = ( response_compiler or ResponseDescriptionCompiler() ) @property def request_compiler(self: CaseCompiler) -> RequestCompiler: return self._request_compiler def compile(self: CaseCompiler, obj: Mapping) -> Case: label = obj.get(_KEY_LABEL) if label is not None and not isinstance(label, str): raise CompilationError( message=f'Case.{_KEY_LABEL} must be a string', path=[_KEY_LABEL], ) request = run_on_key( _KEY_REQUEST, self._request_compiler.compile, _extract_request(obj) ) response_description = run_on_key( _KEY_RESPONSE, self._response_compiler.compile, _extract_response(obj), ) return Case( label=label, request=request, response_description=response_description, ) def of_default(self: CaseCompiler, obj: Mapping) -> CaseCompiler: request_compiler = run_on_key( _KEY_REQUEST, self._request_compiler.of_default, _extract_request(obj), ) return CaseCompiler(request_compiler=request_compiler) def _extract_request(obj: Any) -> Union[Mapping, str]: target = obj.get(_KEY_REQUEST, {}) if not isinstance(target, Mapping) and not isinstance(target, str): raise CompilationError( message='must be a string or a mapping', path=[_KEY_REQUEST], ) return target def _extract_response(obj: Any) -> Mapping: target = obj.get('response', {}) if not isinstance(target, Mapping): raise CompilationError('must be a mapping', path=[_KEY_RESPONSE]) return target PK!U8kk#preacher/compilation/description.py"""Description compilation.""" from __future__ import annotations from collections.abc import Mapping from typing import Optional from preacher.core.description import Description from .error import CompilationError from .predicate import PredicateCompiler from .extraction import ExtractionCompiler from .util import run_on_key, map_on_key _KEY_DESCRIBE = 'describe' _KEY_SHOULD = 'should' class DescriptionCompiler: """ >>> from unittest.mock import MagicMock, call, sentinel >>> def default_extraction_compiler() -> ExtractionCompiler: ... return MagicMock( ... spec=ExtractionCompiler, ... compile=MagicMock(return_value=sentinel.extraction), ... ) >>> def default_predicate_compiler() -> PredicateCompiler: ... return MagicMock( ... spec=PredicateCompiler, ... compile=MagicMock(return_value=sentinel.predicate), ... ) >>> compiler = DescriptionCompiler() >>> compiler.compile({}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: Description.describe ... >>> extraction_compiler = default_extraction_compiler() >>> predicate_compiler = default_predicate_compiler() >>> compiler = DescriptionCompiler( ... extraction_compiler=extraction_compiler, ... predicate_compiler=predicate_compiler, ... ) >>> description = compiler.compile({ ... 'describe': 'foo', ... 'should': 'string', ... }) >>> description.extraction sentinel.extraction >>> description.predicates [sentinel.predicate] >>> extraction_compiler.compile.assert_called_with('foo') >>> predicate_compiler.compile.assert_called_once_with('string') >>> extraction_compiler = default_extraction_compiler() >>> predicate_compiler = default_predicate_compiler() >>> compiler = DescriptionCompiler( ... extraction_compiler=extraction_compiler, ... predicate_compiler=predicate_compiler, ... ) >>> description = compiler.compile({ ... 'describe': 'foo', ... 'should': {'key': 'value'} ... }) >>> description.extraction sentinel.extraction >>> description.predicates [sentinel.predicate] >>> extraction_compiler.compile.assert_called_once_with('foo') >>> predicate_compiler.compile.assert_called_once_with({'key': 'value'}) >>> extraction_compiler = default_extraction_compiler() >>> predicate_compiler = default_predicate_compiler() >>> compiler = DescriptionCompiler( ... extraction_compiler=extraction_compiler, ... predicate_compiler=predicate_compiler, ... ) >>> description = compiler.compile({ ... 'describe': {'key': 'value'}, ... 'should': [{'key1': 'value1'}, {'key2': 'value2'}] ... }) >>> description.extraction sentinel.extraction >>> description.predicates [sentinel.predicate, sentinel.predicate] >>> extraction_compiler.compile.assert_called_once_with({'key': 'value'}) >>> predicate_compiler.compile.assert_has_calls([ ... call({'key1': 'value1'}), ... call({'key2': 'value2'}), ... ]) """ def __init__( self: DescriptionCompiler, extraction_compiler: Optional[ExtractionCompiler] = None, predicate_compiler: Optional[PredicateCompiler] = None, ) -> None: self._extraction_compiler = extraction_compiler or ExtractionCompiler() self._predicate_compiler = predicate_compiler or PredicateCompiler() def compile(self: DescriptionCompiler, obj: Mapping): 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 or a string', path=[_KEY_DESCRIBE], ) extraction = run_on_key( _KEY_DESCRIBE, self._extraction_compiler.compile, extraction_obj ) predicate_objs = obj.get(_KEY_SHOULD, []) if not isinstance(predicate_objs, list): predicate_objs = [predicate_objs] predicates = list(map_on_key( _KEY_SHOULD, self._predicate_compiler.compile, 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!R""preacher/compilation/extraction.py"""Extraction compilation.""" from __future__ import annotations 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()) class ExtractionCompiler: def compile( self: ExtractionCompiler, obj: Union[Mapping, str], ) -> Extraction: 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]) 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' """ compiler = ExtractionCompiler() return compiler.compile(obj) 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!~%O!preacher/compilation/predicate.py"""Predicate compilation.""" from __future__ import annotations from typing import Any from preacher.core.description import Predicate from preacher.core.predicate import MatcherPredicate from .matcher import compile as compile_matcher class PredicateCompiler: """ >>> from unittest.mock import patch, sentinel >>> compiler = PredicateCompiler() >>> with patch( ... f'{__name__}.compile_matcher', ... return_value=sentinel.matcher ... ) as matcher_mock: ... predicate = compiler.compile('matcher') ... matcher_mock.assert_called_with('matcher') """ def compile(self: PredicateCompiler, obj: Any) -> Predicate: matcher = compile_matcher(obj) return MatcherPredicate(matcher) PK!ߌpreacher/compilation/request.py"""Request compilation.""" from __future__ import annotations from collections.abc import Mapping from dataclasses import dataclass from typing import Optional, Union from preacher.core.request import Request from .error import CompilationError from .util import or_default _KEY_PATH = 'path' _KEY_PARAMS = 'params' @dataclass(frozen=True) class _Compiled: path: Optional[str] = None params: Optional[Mapping] = None def to_request( self: _Compiled, default_path: str, default_params: Mapping, ) -> Request: return Request( path=or_default(self.path, default_path), params=or_default(self.params, default_params), ) def _compile(obj: Union[Mapping, str]) -> _Compiled: if isinstance(obj, str): return _compile({_KEY_PATH: obj}) path = obj.get(_KEY_PATH) if path is not None and not isinstance(path, str): raise CompilationError('Must be a string', path=[_KEY_PATH]) params = obj.get(_KEY_PARAMS) if params is not None and not isinstance(params, Mapping): raise CompilationError('Must be a mapping', path=[_KEY_PARAMS]) return _Compiled(path=path, params=params) class RequestCompiler: """ When given not a string path, then raises a compiration error. >>> compiler = RequestCompiler() >>> compiler.compile({'path': {'key': 'value'}}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ...: path >>> compiler.of_default({'path': {'key': 'value'}}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ...: path When given not a mapping parameters, then raises a compilation error. >>> compiler = RequestCompiler() >>> compiler.compile({'params': ''}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ...: params >>> compiler.of_default({'params': ''}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ...: params When given an empty mapping, then returns the dafault mapping.. >>> compiler = RequestCompiler() >>> request = compiler.compile({}) >>> request.path '' >>> request.params {} When given a string, then returns a request of the path. >>> compiler = RequestCompiler() >>> request = compiler.compile('/path') >>> request.path '/path' >>> request.params {} >>> compiler = compiler.of_default('/default-path') >>> request = compiler.compile({'params': {'k': 'v'}}) >>> request.path '/default-path' >>> request.params {'k': 'v'} When given a filled mapping, then returns the request of it. >>> compiler = RequestCompiler() >>> request = compiler.compile( ... {'path': '/path', 'params': {'key': 'value'}} ... ) >>> request.path '/path' >>> request.params {'key': 'value'} >>> compiler = compiler.of_default({ ... 'path': '/default-path', ... 'params': {'k': 'v'}, ... }) >>> request = compiler.compile({}) >>> request.path '/default-path' >>> request.params {'k': 'v'} >>> request = compiler.compile('/path') >>> request.path '/path' >>> request.params {'k': 'v'} >>> request = compiler.compile( ... {'path': '/path', 'params': {'key': 'value'}} ... ) >>> request.path '/path' >>> request.params {'key': 'value'} """ def __init__( self: RequestCompiler, path: str = '', params: Mapping = {}, ) -> None: self._path = path self._params = params def compile(self: RequestCompiler, obj: Union[Mapping, str]) -> Request: compiled = _compile(obj) return compiled.to_request( default_path=self._path, default_params=self._params, ) def of_default( self: RequestCompiler, obj: Union[Mapping, str], ) -> RequestCompiler: compiled = _compile(obj) return RequestCompiler( path=or_default(compiled.path, self._path), params=or_default(compiled.params, self._params), ) PK!*_22,preacher/compilation/response_description.py"""Response description compilations.""" from __future__ import annotations from collections.abc import Mapping from typing import Any, Optional from preacher.core.description import Description from preacher.core.response_description import ResponseDescription from .error import CompilationError from .description import DescriptionCompiler from .predicate import PredicateCompiler from .util import map_on_key _KEY_STATUS_CODE = 'status_code' _KEY_BODY = 'body' class ResponseDescriptionCompiler: """ >>> from unittest.mock import MagicMock, call, patch, sentinel >>> def default_predicate_compiler() -> PredicateCompiler: ... return MagicMock( ... spec=PredicateCompiler, ... compile=MagicMock(return_value=sentinel.predicate), ... ) >>> def default_description_compiler() -> DescriptionCompiler: ... return MagicMock( ... spec=DescriptionCompiler, ... compile=MagicMock(return_value=sentinel.description), ... ) >>> predicate_compiler = default_predicate_compiler() >>> description_compiler = default_description_compiler() >>> compiler = ResponseDescriptionCompiler( ... predicate_compiler=predicate_compiler, ... description_compiler=description_compiler, ... ) >>> response_description = compiler.compile({}) >>> response_description.status_code_predicates [] >>> response_description.body_descriptions [] >>> predicate_compiler.compile.assert_not_called() >>> description_compiler.compile.assert_not_called() >>> compiler = ResponseDescriptionCompiler() >>> compiler.compile({'body': 'str'}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ...: body >>> compiler = ResponseDescriptionCompiler() >>> compiler.compile({'body': ['str']}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: Description ...: body[0] >>> predicate_compiler = default_predicate_compiler() >>> description_compiler = default_description_compiler() >>> compiler = ResponseDescriptionCompiler( ... predicate_compiler=predicate_compiler, ... description_compiler=description_compiler, ... ) >>> response_description = compiler.compile({ ... 'status_code': 402, ... 'body': {'key1': 'value1'}} ... ) >>> response_description.status_code_predicates [sentinel.predicate] >>> response_description.body_descriptions [sentinel.description] >>> predicate_compiler.compile.assert_called_once_with(402) >>> description_compiler.compile.assert_called_once_with( ... {'key1': 'value1'} ... ) >>> predicate_compiler = default_predicate_compiler() >>> description_compiler = default_description_compiler() >>> compiler = ResponseDescriptionCompiler( ... predicate_compiler=predicate_compiler, ... description_compiler=description_compiler, ... ) >>> response_description = compiler.compile({ ... 'status_code': [{'be_greater_than': 0}, {'be_less_than': 400}], ... 'body': [{'key1': 'value1'}, {'key2': 'value2'}], ... }) >>> response_description.status_code_predicates [sentinel.predicate, sentinel.predicate] >>> response_description.body_descriptions [sentinel.description, sentinel.description] >>> predicate_compiler.compile.assert_has_calls([ ... call({'be_greater_than': 0}), ... call({'be_less_than': 400}), ... ]) >>> description_compiler.compile.assert_has_calls([ ... call({'key1': 'value1'}), ... call({'key2': 'value2'}), ... ]) """ def __init__( self: ResponseDescriptionCompiler, predicate_compiler: Optional[PredicateCompiler] = None, description_compiler: Optional[DescriptionCompiler] = None, ) -> None: self._predicate_compiler = predicate_compiler or PredicateCompiler() self._description_compiler = ( description_compiler or DescriptionCompiler( predicate_compiler=self._predicate_compiler ) ) def compile( self: ResponseDescriptionCompiler, obj: Mapping, ) -> ResponseDescription: 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=self._predicate_compiler.compile, 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=self._compile_description, items=body_description_objs, )) return ResponseDescription( status_code_predicates=status_code_predicates, body_descriptions=body_descriptions, ) def _compile_description( self: ResponseDescriptionCompiler, obj: Any, ) -> Description: if not isinstance(obj, Mapping): raise CompilationError('Description must be a mapping') return self._description_compiler.compile(obj) PK!Ӯo preacher/compilation/scenario.py"""Scenario compilation.""" from __future__ import annotations from collections.abc import Mapping from functools import partial from typing import Any, Optional from preacher.core.scenario import Scenario from preacher.core.case import Case from .error import CompilationError from .case import CaseCompiler from .util import map_on_key, run_on_key _KEY_LABEL = 'label' _KEY_DEFAULT = 'default' _KEY_CASES = 'cases' class ScenarioCompiler: """ When given an empty object, then generates an empty scenario. >>> from unittest.mock import MagicMock, call, patch, sentinel >>> case_compiler = MagicMock(CaseCompiler) >>> compiler = ScenarioCompiler(case_compiler=case_compiler) >>> scenario = compiler.compile({}) >>> scenario.label >>> list(scenario.cases()) [] >>> case_compiler.of_default.assert_called_once_with({}) When given not an object, then raises a compilation error. >>> ScenarioCompiler().compile({'cases': ''}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ...: cases When given a not string label, then raises a compilation error. >>> ScenarioCompiler().compile({'label': []}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ...: label When given not mapping default, then raises a compilation error. >>> ScenarioCompiler().compile({'default': ''}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ...: default When given a not mapping case, then raises a compilation error. >>> ScenarioCompiler().compile({'cases': ['']}) Traceback (most recent call last): ... preacher.compilation.error.CompilationError: ...: cases[0] Generates an iterator of cases. >>> default_case_compiler = MagicMock( ... CaseCompiler, ... compile=MagicMock(return_value=sentinel.case), ... ) >>> case_compiler = MagicMock( ... CaseCompiler, ... of_default=MagicMock(return_value=default_case_compiler), ... ) >>> compiler = ScenarioCompiler(case_compiler=case_compiler) >>> scenario = compiler.compile({ ... 'label': 'label', ... 'default': {'a': 'b'}, ... 'cases': [{}, {'k': 'v'}], ... }) >>> scenario.label 'label' >>> list(scenario.cases()) [sentinel.case, sentinel.case] >>> case_compiler.of_default.assert_called_once_with({'a': 'b'}) >>> default_case_compiler.compile.assert_has_calls([ ... call({}), ... call({'k': 'v'})], ... ) """ def __init__( self: ScenarioCompiler, case_compiler: Optional[CaseCompiler] = None, ) -> None: self._case_compiler = case_compiler or CaseCompiler() def compile(self: ScenarioCompiler, obj: Mapping) -> Scenario: label = obj.get(_KEY_LABEL) if label is not None and not isinstance(label, str): raise CompilationError( message='Must be a string', path=[_KEY_LABEL], ) default = obj.get(_KEY_DEFAULT, {}) if not isinstance(default, Mapping): raise CompilationError( message='Must be a mapping', path=[_KEY_DEFAULT], ) case_compiler = run_on_key( _KEY_DEFAULT, self._case_compiler.of_default, default, ) case_objs = obj.get(_KEY_CASES, []) if not isinstance(case_objs, list): raise CompilationError(message='Must be a list', path=[_KEY_CASES]) cases = map_on_key( _KEY_CASES, partial(_compile_case, case_compiler), case_objs, ) return Scenario(label=label, cases=list(cases)) def _compile_case(case_compiler: CaseCompiler, obj: Any) -> Case: if not isinstance(obj, Mapping): raise CompilationError(f'Case must be a mapping') return case_compiler.compile(obj) PK!O\ ??preacher/compilation/util.py"""Utilities for compilations.""" from typing import Callable, Iterable, Iterator, Optional, 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}]']) def or_default(value: Optional[T], default_value: T) -> T: if value is None: return default_value return value PK!preacher/core/__init__.pyPK!k\%__preacher/core/case.py"""Test case.""" 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 CaseResult: status: Status request: Verification response: Optional[ResponseVerification] = None label: Optional[str] = None class Case: """ >>> from unittest.mock import MagicMock >>> case = Case( ... request=MagicMock(Request, side_effect=RuntimeError('message')), ... response_description=MagicMock(ResponseDescription), ... ) >>> case.label >>> case('base-url', -1) Traceback (most recent call last): ... RuntimeError: ... -1 >>> verification = case('base-url') >>> verification.label >>> verification.status FAILURE >>> verification.request.status FAILURE >>> verification.request.message 'RuntimeError: message' >>> case.request.call_args call('base-url') >>> case.response_description.call_count 0 >>> from .request import Response >>> inner_response = MagicMock(Response, status_code=402, body='body') >>> case = Case( ... label='Response should be unstable', ... request=MagicMock(Request, return_value=inner_response), ... response_description=MagicMock( ... spec=ResponseDescription, ... return_value=ResponseVerification( ... status=Status.UNSTABLE, ... status_code=Verification.succeed(), ... body=Verification(status=Status.UNSTABLE) ... ), ... ), ... ) >>> verification = case(base_url='base-url', retry=1) >>> verification.label 'Response should be unstable' >>> verification.status UNSTABLE >>> verification.request.status SUCCESS >>> verification.response.status UNSTABLE >>> verification.response.body.status UNSTABLE >>> case.response_description.call_args call(body='body', status_code=402) Retries. >>> case = Case( ... label='Succeeds', ... request=MagicMock( ... spec=Request, ... side_effect=[RuntimeError(), inner_response, inner_response], ... ), ... response_description=MagicMock( ... spec=ResponseDescription, ... side_effect=[ ... ResponseVerification( ... status=Status.UNSTABLE, ... status_code=Verification.succeed(), ... body=Verification(status=Status.UNSTABLE), ... ), ... ResponseVerification( ... status=Status.SUCCESS, ... status_code=Verification.succeed(), ... body=Verification.succeed(), ... ), ... ] ... ), ... ) >>> verification = case(base_url='base-url', retry=2) >>> verification.status SUCCESS >>> assert case.request.call_count == 3 >>> assert case.response_description.call_count == 2 """ def __init__( self: Case, request: Request, response_description: ResponseDescription, label: Optional[str] = None, ) -> None: self._label = label self._request = request self._response_description = response_description def __call__(self: Case, base_url: str, retry: int = 0) -> CaseResult: if retry < 0: raise RuntimeError( f'Retry count must be positive or 0, given {retry}' ) for _ in range(1 + retry): result = self._run(base_url) if result.status.is_succeeded: break return result def _run(self: Case, base_url: str) -> CaseResult: try: response = self._request(base_url) except Exception as error: return CaseResult( 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 CaseResult( status=status, request=request_verification, response=response_verification, label=self._label, ) @property def label(self: Case) -> Optional[str]: return self._label @property def request(self: Case) -> Request: return self._request @property def response_description(self: Case) -> ResponseDescription: return self._response_description PK!F\=} } 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 is skipped. >>> description = Description( ... extraction=MagicMock(return_value='target'), ... predicates=[], ... ) >>> verification = description('described') >>> verification.status SKIPPED >>> 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!preacher/core/predicate.py"""Predicate.""" from __future__ import annotations from typing import Any from hamcrest import assert_that from hamcrest.core.matcher import Matcher from .status import Status from .verification import Verification class MatcherPredicate: """ >>> from unittest.mock import MagicMock, patch >>> matcher = MagicMock(Matcher) >>> predicate = MatcherPredicate(matcher) >>> with patch( ... f'{__name__}.assert_that', ... side_effect=RuntimeError('message') ... ) as assert_that: ... 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: ... 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: ... verification = predicate(1) ... assert_that.assert_called_with(1, matcher) >>> verification.status SUCCESS """ def __init__(self: MatcherPredicate, matcher: Matcher) -> None: self._matcher = matcher def __call__(self: MatcherPredicate, actual: Any) -> Verification: try: assert_that(actual, self._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() PK!qpreacher/core/request.py"""Request.""" from __future__ import annotations from collections.abc import Mapping from dataclasses import dataclass import requests from preacher import __version__ as _VERSION _DEFAULT_HEADERS = { 'User-Agent': f'Preacher {_VERSION}' } @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', headers={'User-Agent': ...}, 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, headers=_DEFAULT_HEADERS, 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!Ek%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: """ When given no descriptions, then skips. >>> description = ResponseDescription( ... status_code_predicates=[], ... body_descriptions=[], ... ) >>> verification = description(status_code=200, body='') >>> verification.status_code.status SKIPPED >>> verification.body.status SKIPPED >>> verification.status SKIPPED When given invalid body, then marks as failure. >>> 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.skipped() 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!mp~preacher/core/scenario.py"""Scenario""" from __future__ import annotations from typing import Iterator, List, Optional from .case import Case class Scenario: """ When given no cases, then skips. >>> scenario = Scenario() >>> scenario.label >>> list(scenario.cases()) [] When given a label, then returns it. >>> scenario = Scenario(label='label') >>> scenario.label 'label' When given cases, then iterates them. >>> from unittest.mock import sentinel >>> scenario = Scenario(cases=[sentinel.case1, sentinel.case2]) >>> cases = scenario.cases() >>> next(cases) sentinel.case1 >>> next(cases) sentinel.case2 """ def __init__( self: Scenario, label: Optional[str] = None, cases: List[Case] = [], ) -> None: self._label = label self._cases = cases @property def label(self: Scenario) -> Optional[str]: return self._label def cases(self: Scenario) -> Iterator[Case]: return iter(self._cases) PK!q!preacher/core/scenario_running.py"""Scenario running helpers.""" from __future__ import annotations from dataclasses import dataclass from typing import List, Optional from .case import CaseResult from .scenario import Scenario from .status import Status, merge_statuses @dataclass class ScenarioResult: label: Optional[str] status: Status case_results: List[CaseResult] def run_scenario( scenario: Scenario, base_url: str, retry: int = 0, ) -> ScenarioResult: """ When given empty scenario, then runs and returns its result as skipped. >>> from unittest.mock import MagicMock, sentinel >>> scenario = MagicMock( ... spec=Scenario, ... label=None, ... cases=MagicMock(return_value=iter([])) ... ) >>> result = run_scenario(scenario, base_url='base_url') >>> result.label >>> result.status SKIPPED >>> result.case_results [] When given filled scenario, then runs and returns its result. >>> sentinel.case_result1.status = Status.UNSTABLE >>> case1 = MagicMock(return_value=sentinel.case_result1) >>> sentinel.case_result2.status = Status.SUCCESS >>> case2 = MagicMock(return_value=sentinel.case_result2) >>> scenario = MagicMock( ... spec=Scenario, ... label='label', ... cases=MagicMock(return_value=iter([case1, case2])) ... ) >>> result = run_scenario(scenario, base_url='base_url', retry=3) >>> result.label 'label' >>> result.status UNSTABLE >>> result.case_results [sentinel.case_result1, sentinel.case_result2] >>> case1.assert_called_once_with(base_url='base_url', retry=3) >>> case2.assert_called_once_with(base_url='base_url', retry=3) """ case_results = [ case(base_url=base_url, retry=retry) for case in scenario.cases() ] status = merge_statuses(result.status for result in case_results) return ScenarioResult( label=scenario.label, status=status, case_results=case_results, ) PK!7ߐo 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.SKIPPED.is_succeeded True >>> Status.SKIPPED.merge(Status.SUCCESS) SUCCESS >>> Status.SKIPPED.merge(Status.UNSTABLE) UNSTABLE >>> Status.SKIPPED.merge(Status.FAILURE) FAILURE >>> Status.SUCCESS.is_succeeded True >>> Status.SUCCESS.merge(Status.SKIPPED) SUCCESS >>> 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.SKIPPED) UNSTABLE >>> 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.SKIPPED) FAILURE >>> 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. SKIPPED = 0 SUCCESS = 1 UNSTABLE = 2 FAILURE = 3 @property def is_succeeded(self: Status): return self.value <= Status.SUCCESS.value 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([]) SKIPPED >>> 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.SKIPPED) PK!FWɅ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 skipped() -> Verification: return Verification(status=Status.SKIPPED) @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!M preacher/presentation/logging.py"""Logging presentation.""" from __future__ import annotations import contextlib import logging import io from typing import Iterator from preacher.core.case import CaseResult from preacher.core.response_description import ResponseVerification from preacher.core.scenario_running import ScenarioResult from preacher.core.status import Status from preacher.core.verification import Verification _LEVEL_MAP = { Status.SKIPPED: logging.DEBUG, 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_result( self: LoggingPresentation, scenario_result: ScenarioResult, ) -> None: status = scenario_result.status level = _LEVEL_MAP[status] label = scenario_result.label or 'Not labeled scenario' self._log(level, '%s: %s', label, status) with self._nested(): for case_result in scenario_result.case_results: self.show_case_result(case_result) self._log(level, '') def show_case_result( self: LoggingPresentation, case_result: CaseResult, ) -> None: status = case_result.status level = _LEVEL_MAP[status] label = case_result.label or 'Not labeled case' self._log(level, '%s: %s', label, status) with self._nested(): self.show_verification( verification=case_result.request, label='Request', ) response = case_result.response if response: self.show_response_verification(response) 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.6.dist-info/entry_points.txtN+I/N.,()*(JMLH-Mɴq zyV PK!T-- preacher-0.9.6.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.6.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!H!preacher-0.9.6.dist-info/METADATAX[s6~ǯtlEvzrlj&3vuv$B$D! e=/z 뇃憥̰UDS+yDkYsE;)Kڈ.=_tÕXF"$odɃe`(7| 70-e{W"ɢ1Tk5 xDQ| $wP\[h//^ο Oȫi r0}(>VrS|{%3RTbU@Z(8 '/nP,|9;9!h'tvz@5.– ߆C <17WHNW22Ƕ5 !^?BKgQf/R 9_4_d/2Le SpݨmaU6Blc.RJi&~ƔD<.wCֹ`_U+2S[ӮQV[*SBtVXl<=qmJszFsv1rlE 6ሀd@fA4d+}+CٳXYK;ZQE(w0q'>6ҠKBpt;;5k UN 1Gtv,an:M؊ӛ9_K9ue78V4z q}r2Zɴ dñeo HQJ!( '1+yӽdFF&pLmy{ >? 4prgq_fP@K؝Rdi.^Z[و~كU 0 eH}Ug[B= nEdAcK4NH1ZtN $Ut yW#U@<}A\ '&=83:N , + R f 2FJC~oCt#bS~d~ .㸋:+SGv _î8qS-:97qϣ?A2t'%V*<,r01Vʙpz#13! ͻrgqZ`څqLϨ?hZʍS[#M^:r5뚕.0u0"tA޳AZVֱo˸@'w]*IY@Ǽ{[ڬ+YM`r՗`À[߲!dy@W(uuyz }èm?>=$q&O&p NBx{7џ}E0#!'Cbd a1"+3= 'AHzvY7`s8deArں1/ݢAu6P!* ufemT<Ѳi~'gs|08KrQw ׊-8(CNHM{(4\Ww A[aC1FDjżs>1Cvũk_]`iGox3QT{(?K; G'9< b/.Qp]<-ݟ?0Q-L3i [,xV[7SƇAa C~/ee>jLx"NDx1{ 76qGKL!įn©FPm3mhFޯ8{̱l L?kO?I4eOb12夝aX1$e*ty 0l/B-Ȣi8!D.Zጉ0VW}:!}8.y;jt k))30P{k/cBӅ6^U6"K[ʲ,7ۮ2K"KcL9NDz?y"\!F ǥm6\gZ<" Ds4/ӓk$Xe^βĠJ㞫VdEFzk6_/L䰠{]#6Ēn5!j2Ôk.Ͽ6TJ]ڂQJSOڢ-<.%š-}GZN3XG[we;O6~8oB)NbhiP{$Z, 3/hOG/*nLF?'{uc}VЃ փ%ʵur׮cψ]Ov 0wZo,/z#MZrɠK`m@Q?1sb>v~T9ؿ; >ɏ:MuKIp2A`gGPK!lpreacher/__init__.pyPK!preacher/app/__init__.pyPK!preacher/app/cli/__init__.pyPK!K""!preacher/app/cli/application.pyPK!3$I preacher/app/cli/main.pyPK! 9preacher/compilation/__init__.pyPK!c--wpreacher/compilation/case.pyPK!U8kk#.preacher/compilation/description.pyPK!ƅ@preacher/compilation/error.pyPK!R""UDpreacher/compilation/extraction.pyPK! AIpreacher/compilation/matcher.pyPK!~%O!dipreacher/compilation/predicate.pyPK!ߌlpreacher/compilation/request.pyPK!*_22,_}preacher/compilation/response_description.pyPK!Ӯo ۓpreacher/compilation/scenario.pyPK!O\ ??ˣpreacher/compilation/util.pyPK!Dpreacher/core/__init__.pyPK!k\%__{preacher/core/case.pyPK!F\=} }  preacher/core/description.pyPK!Lpreacher/core/extraction.pyPK!preacher/core/predicate.pyPK!qpreacher/core/request.pyPK!Ek%preacher/core/response_description.pyPK!mp~Lpreacher/core/scenario.pyPK!q!preacher/core/scenario_running.pyPK!7ߐo preacher/core/status.pyPK!FWɅ{preacher/core/verification.pyPK!!{preacher/presentation/__init__.pyPK!M preacher/presentation/logging.pyPK!HD+`3;)preacher-0.9.6.dist-info/entry_points.txtPK!T-- !preacher-0.9.6.dist-info/LICENSEPK!HڽTUpreacher-0.9.6.dist-info/WHEELPK!H!preacher-0.9.6.dist-info/METADATAPK!H }w: J!preacher-0.9.6.dist-info/RECORDPK"" &