from typing import Optional
from testoot.base import TestootStorage, TestootSerializer, \
CanonizePolicy, TestootContext, FileType, Comparator
[docs]class BaseTestoot:
"""Main logic object. Can test and canonize data."""
[docs] def __init__(self, storage: TestootStorage, serializer: TestootSerializer,
canonize_policy: CanonizePolicy,
comparator: Optional[Comparator] = None):
"""Constructor
:param storage: storage instance
:param serializer: serializer instance
:param canonize_policy: controls behavior when we met result
test conflict.
- :py:class:`.testoot.pub.NoCanonizePolicy` always raises
an error in assert.
- :py:class:`.testoot.pub.AskCanonizePolicy`
with `--canonize` pytest flag asks user approval for
canonizing. If user refuses it skips for later and raises
an error in assert then.
:param comparator: comparison for objects
"""
self._storage: TestootStorage = storage
self._serializer: TestootSerializer = serializer
self._canonize_policy: CanonizePolicy = canonize_policy
self._comparator: Comparator = comparator
[docs] def test(self, obj: any, context: TestootContext,
suffix: Optional[str] = None,
file_type_hint: Optional[FileType] = None,
comparator: Optional[Comparator] = None,
serializer: Optional[TestootSerializer] = None):
"""Tests object.
:param obj: test object
:param context: test context
:param suffix: test suffix for making a few tests
in one context
:param file_type_hint: override serializer hint for file
:param comparator: custom comparator override
:param serializer: custom serializer override
:return:
"""
serializer = self._get_serializer(serializer, context=context)
if file_type_hint is None:
file_type_hint = serializer.file_type_hint
storage_name = context.get_storage_name(suffix=suffix,
file_type_hint=file_type_hint)
with self._storage.open_read(storage_name,
mode=serializer.mode) as rstream:
canon_obj = (serializer.load(rstream) if rstream is not None
else None)
self._do_test(
test_obj=obj,
canon_obj=canon_obj,
comparator=comparator,
storage_name=storage_name,
context=context,
)
return True
[docs] def test_filename(self, filename: str, *, context: TestootContext,
comparator: Optional[Comparator] = None,
serializer: Optional[TestootSerializer] = None):
"""Tests file generated by software.
:param filename: test filename
:param context: test context
:param comparator: custom comparator override
:param serializer: custom serializer override
:return:
"""
serializer = self._get_serializer(serializer, context=context)
from pathlib import Path
# FIXME: this internal
set_filename = Path(filename).relative_to(self._storage.root_dir)
storage_name = context.get_storage_name_from_filename(set_filename)
with self._storage.open_read(storage_name,
mode=serializer.mode) as rstream:
canon_obj = (serializer.load(rstream) if rstream is not None
else None)
with open(filename, 'rb') as test_stream:
test_obj = serializer.load(test_stream)
self._do_test(
test_obj=test_obj,
canon_obj=canon_obj,
comparator=comparator,
storage_name=storage_name,
context=context,
)
return True
@property
def storage(self):
return self._storage
@property
def canonize_policy(self):
return self._canonize_policy
[docs] def clone(self, *, storage: Optional[TestootStorage] = None,
serializer: Optional[TestootSerializer] = None,
canonize_policy: Optional[CanonizePolicy] = None,
comparator: Optional[Comparator] = None):
return type(self)(
storage=storage or self._storage,
serializer=serializer or self._serializer,
canonize_policy=canonize_policy or self._canonize_policy,
comparator=comparator or self._comparator,
)
def _canonize(self, obj: any, *, storage_name: str,
serializer: TestootSerializer):
"""Canonizes result of test
:param obj: test object
:param storage_name: storage name
:return:
"""
with self._storage.open_write(storage_name,
mode=serializer.mode) as wstream:
serializer.dump(obj, wstream)
def _do_test(self, *, test_obj: any, canon_obj: any,
storage_name: str, context: TestootContext,
comparator: Optional[Comparator] = None,
serializer: Optional[TestootSerializer] = None):
comparator = self._get_comparator(comparator, context=context)
serializer = self._get_serializer(serializer, context=context)
if canon_obj is None:
return self._canonize(test_obj, storage_name=storage_name,
serializer=serializer)
try:
comparator.compare(test_obj, canon_obj)
except Exception as e:
test_result = context.create_test_result(test_obj=test_obj,
canon_obj=canon_obj,
exc=e)
# TODO: refactor this strange logic
# Asks query parameters to canonize and check it again
do_canonize = context.ask_canonize()
if do_canonize:
do_canonize = self._canonize_policy.ask_canonize(
test_result=test_result,
)
if not do_canonize:
raise
return self._canonize(test_obj, storage_name=storage_name,
serializer=serializer)
def _get_comparator(self, comparator: Optional[Comparator], *,
context: TestootContext) -> Comparator:
return next(filter(lambda x: x is not None, (
comparator, context.get_comparator(), self._comparator)))
def _get_serializer(self, serializer: Optional[TestootSerializer], *,
context: TestootContext) \
-> TestootSerializer:
return next(filter(lambda x: x is not None, (
serializer, context.get_serializer(), self._serializer)))