"""Define the interfaces used by ExternalIntegration self-tests.
"""
from .util.http import IntegrationException
import json
import logging
import traceback
from .util.opds_writer import AtomFeed
from .util.datetime_helpers import utc_now
[docs]class SelfTestResult(object):
"""The result of running a single self-test.
HasSelfTest.run_self_tests() returns a list of these
"""
def __init__(self, name):
# Name of the test.
self.name = name
# Set to True when the test runs without raising an exception.
self.success = False
# The exception raised, if any.
self.exception = None
# The return value of the test method, assuming it ran to
# completion.
self.result = None
# Start time of the test.
self.start = utc_now()
# End time of the test.
self.end = None
# Collection associated with the test
self.collection = None
@property
def to_dict(self):
"""Convert this SelfTestResult to a dictionary for use in
JSON serialization.
"""
# Time formatting method
f = AtomFeed._strftime
if self.exception:
exception = { "class": self.exception.__class__.__name__,
"message": str(self.exception),
"debug_message" : self.debug_message }
else:
exception = None
value = dict(
name=self.name, success=self.success,
duration=self.duration,
exception=exception,
)
if self.start:
value['start'] = f(self.start)
if self.end:
value['end'] = f(self.end)
if self.collection:
value['collection'] = self.collection.name
# String results will be displayed in a fixed-width font.
# Lists of strings will be hidden behind an expandable toggle.
# Other return values have no defined method of display.
if isinstance(self.result, str) or isinstance(self.result, list):
value['result'] = self.result
else:
value['result'] = None
return value
def __repr__(self):
if self.exception:
if isinstance(self.exception, IntegrationException):
exception = " exception=%r debug=%r" % (
str(self.exception),
self.debug_message
)
else:
exception = " exception=%r" % self.exception
else:
exception = ""
if self.collection:
collection = " collection=%r" % self.collection.name
else:
collection = ""
return "<SelfTestResult: name=%r%s duration=%.2fsec success=%r%s result=%r>" % (
self.name, collection, self.duration, self.success,
exception, self.result
)
@property
def duration(self):
"""How long the test took to run."""
if not self.start or not self.end:
return 0
return (self.end-self.start).total_seconds()
@property
def debug_message(self):
"""The debug message associated with the Exception, if any."""
if not self.exception:
return None
return getattr(self.exception, 'debug_message', None)
[docs]class HasSelfTests(object):
"""An object capable of verifying its own setup by running a
series of self-tests.
"""
# Self-test results are stored in a ConfigurationSetting with this name,
# associated with the appropriate ExternalIntegration.
SELF_TEST_RESULTS_SETTING = 'self_test_results'
[docs] @classmethod
def run_self_tests(cls, _db, constructor_method=None, *args, **kwargs):
"""Instantiate this class and call _run_self_tests on it.
:param _db: A database connection. Will be passed into `_run_self_tests`.
This connection may need to be used again
in args, if the constructor needs it.
:param constructor_method: Method to use to instantiate the
class, if different from the default constructor.
:param args: Positional arguments to pass into the constructor.
:param kwargs: Keyword arguments to pass into the constructor.
:return: A 2-tuple (results_dict, results_list) `results_dict`
is a JSON-serializable dictionary describing the results of
the self-test. `results_list` is a list of SelfTestResult
objects.
"""
from .external_search import ExternalSearchIndex
constructor_method = constructor_method or cls
start = utc_now()
result = SelfTestResult("Initial setup.")
instance = None
integration = None
results = []
# Treat the construction of the integration code as its own
# test.
try:
instance = constructor_method(*args, **kwargs)
result.success = True
result.result = instance
except Exception as e:
result.exception = e
result.success = False
finally:
result.end = utc_now()
results.append(result)
if instance:
try:
for result in instance._run_self_tests(_db):
results.append(result)
except Exception as e:
# This should only happen when there's a bug in the
# self-test method itself.
failure = instance.test_failure(
"Uncaught exception in the self-test method itself.", e
)
results.append(failure)
end = utc_now()
# Format the results in a useful way.
value = dict(
start=AtomFeed._strftime(start),
end=AtomFeed._strftime(end),
duration = (end-start).total_seconds(),
results = [x.to_dict for x in results]
)
# Store the formatted results in the database, if we can find
# a place to store them.
if instance and isinstance(instance, ExternalSearchIndex):
integration = instance.search_integration(_db)
for idx, result in enumerate(value.get("results")):
if isinstance(results[idx].result, list):
result["result"] = results[idx].result
elif instance:
integration = instance.external_integration(_db)
if integration:
integration.setting(
cls.SELF_TEST_RESULTS_SETTING
).value = json.dumps(value)
return value, results
[docs] @classmethod
def prior_test_results(cls, _db, constructor_method=None, *args, **kwargs):
"""Retrieve the last set of test results from the database.
The arguments here are the same as the arguments to run_self_tests.
"""
constructor_method = constructor_method or cls
integration = None
instance = constructor_method(*args, **kwargs)
from .external_search import ExternalSearchIndex
if isinstance(instance, ExternalSearchIndex):
integration = instance.search_integration(_db)
else:
integration = instance.external_integration(_db)
if integration:
return integration.setting(cls.SELF_TEST_RESULTS_SETTING).json_value or "No results yet"
[docs] def external_integration(self, _db):
"""Locate the ExternalIntegration associated with this object.
The status of the self-tests will be stored as a ConfigurationSetting
on this ExternalIntegration.
By default, there is no way to get from an object to its
ExternalIntegration, and self-test status will not be stored.
"""
logger = logging.getLogger("Self-test system")
logger.error(
"No ExternalIntegration was found. Self-test results will not be stored."
)
return None
def _run_self_tests(self, _db):
"""Run a series of self-tests.
:return: A list of SelfTestResult objects.
"""
raise NotImplementedError()
[docs] def run_test(self, name, method, *args, **kwargs):
"""Run a test method, record any exception that happens, and keep
track of how long the test takes to run.
:param name: The name of the test to be run.
:param method: A method to call to run the test.
:param args: Positional arguments to `method`.
:param kwargs: Keyword arguments to `method`.
:return: A filled-in SelfTestResult.
"""
result = SelfTestResult(name)
try:
return_value = method(*args, **kwargs)
result.success = True
result.result = return_value
except Exception as e:
result.exception = e
result.success = False
result.result = None
finally:
if not result.end:
result.end = utc_now()
return result
[docs] @classmethod
def test_failure(cls, name, message, debug_message=None):
"""Create a SelfTestResult for a known failure.
This is useful when you can't even get the data necessary to
run a test method.
"""
result = SelfTestResult(name)
result.end = result.start
result.success = False
if isinstance(message, Exception):
exception = message
message = str(exception)
if not debug_message:
debug_message = traceback.format_exc()
exception = IntegrationException(message, debug_message)
result.exception = exception
return result