"""Autowisp unit-test init."""
from collections.abc import Sequence
from os import path, makedirs, environ
from subprocess import run, PIPE, STDOUT
from shutil import copytree, copy, move, rmtree
from tempfile import TemporaryDirectory
from glob import glob
import atexit
import logging
import platformdirs
import sqlalchemy
from asteval import Interpreter
from astrowisp.tests.utilities import FloatTestCase
_user_data_dir = TemporaryDirectory( # pylint: disable=consider-using-with
prefix="autowisp_tests_"
)
"""Throwaway stand-in for the real user data directory.
Five call sites derive paths from ``platformdirs.user_data_dir("autowisp")``:
the BUI database and ``bui.log`` (``django_project/settings.py``), the
default project home when none is given (``database/interface.py``), and the
``run_pipeline.out`` capture file (``run_pipeline.py`` and
``processing/views.py``). Redirecting the lookup once, here, means no test
can reach the developer's real data by forgetting to override something --
the BUI database in particular holds their project list.
This belongs in the package ``__init__`` rather than ``__main__`` because CI
also runs modules directly (``python -m unittest autowisp.tests.…``), which
never loads ``__main__``. Importing anything under ``autowisp.tests``
executes this first, which matters because ``settings.py`` resolves the
directory at *import* time.
"""
atexit.register(_user_data_dir.cleanup)
[docs]
def _redirected_user_data_dir(*args, **kwargs):
"""Return the throwaway directory whatever application is asked for."""
# pylint: disable=unused-argument
return _user_data_dir.name
platformdirs.user_data_dir = _redirected_user_data_dir
# Redirecting the data directory moves the browser-interface database only
# so long as nothing overrides it. AUTOWISP_BUI_DB_URL does exactly that,
# and would point the suite at whatever server a developer happens to have
# configured -- so drop it, the same way the directory itself is replaced
# rather than merely defaulted.
environ.pop("AUTOWISP_BUI_DB_URL", None)
# Imported after the redirect above, so that anything resolving a user data
# path at import time picks up the throwaway directory.
# pylint: disable=wrong-import-position
from autowisp.database.interface import (
set_project_home,
initialize_cmdline_database,
DB_URL_FNAME,
)
from autowisp.database.user_interface import import_json_to_survey
from autowisp.database.initialize_database import initialize_database
# pylint: enable=wrong-import-position
SERVER_URL_ENV = "AUTOWISP_TEST_DB_URL" # pylint: disable=invalid-name
"""Environment variable naming a MySQL/MariaDB URL to test against.
Unset -- the default, and what a developer gets locally -- puts every
project database in a throwaway SQLite file, as before. Setting it runs the
same tests against a centralised server, which is a supported deployment
but one the suite has never exercised: SQLite compares strings
case-sensitively, does not enforce foreign keys or column widths, and
serialises writers instead of taking row locks.
"""
[docs]
def server_test_url():
"""Return the server URL to test against, or None for SQLite.
Empty counts as unset. A CI matrix that carries the URL as a per-cell
key gives every *other* cell the variable set to an empty string rather
than absent, and those cells must stay on SQLite.
"""
return environ.get(SERVER_URL_ENV) or None
[docs]
def empty_server_database(url):
"""Drop every table in *url*, including ``alembic_version``.
SQLite gets a fresh file per test simply by using a fresh directory. A
server has one database shared by the whole run, so it has to be
emptied between tests instead -- and completely, since a leftover
``alembic_version`` would make the next project look migrated when its
tables are gone.
"""
engine = sqlalchemy.create_engine(url, poolclass=sqlalchemy.pool.NullPool)
try:
with engine.begin() as connection:
connection.execute(sqlalchemy.text("SET FOREIGN_KEY_CHECKS=0"))
for table in sqlalchemy.inspect(engine).get_table_names():
connection.execute(
sqlalchemy.text(f"DROP TABLE IF EXISTS `{table}`")
)
connection.execute(sqlalchemy.text("SET FOREIGN_KEY_CHECKS=1"))
finally:
engine.dispose()
[docs]
class AutoWISPTestCase(FloatTestCase):
"""Base class for AutoWISP tests."""
#: Set False by a test that does not want its processing directory
#: kept even when it fails (nothing does at present). Failure is
#: detected from the test result, so a passing test never has to say
#: anything -- the previous arrangement, where each test opted in by
#: setting ``successful_test``, silently preserved the directory of
#: every test that forgot to.
preserve_failed_processing = True
_logger = logging.getLogger(__name__)
# Stage the cached Gaia catalog FITS (``test_data/MASTERS/Gaia``) into the
# processing directory in setUp so steps that call ``ensure_catalog`` reuse
# them instead of hitting the live Gaia archive. Catalog tests, which are
# meant to exercise the live query, set this False.
stage_catalog_cache = True
# Header keys whose presence and value must NOT be required to
# match between the two files:
#
# - ``EXTEND`` -- structural, not data.
# - ``CALGITID`` -- git commit of the calibration code.
# - ``COMMENT`` -- free-form text.
# - ``M{BIAS,DARK,FLAT}SHA`` -- hash of master files.
# - ``TARGETID``, ``TELSCPID``, ``CAMERAID``, ``OBSSSNID``,
# ``OBSERVER`` -- DB row identifiers written by the pipeline's
# ``add_images_to_db`` step but absent from the per-step CLI
# flow used to produce the test_data fixtures.
_ignore_header_keys = {
"DATASUM",
"CHECKSUM",
"CALGITID",
"COMMENT",
"EXTEND",
"TARGETID",
"TELSCPID",
"CAMERAID",
"OBSSSNID",
"OBSERVER",
"PROJHOME",
} | {f"M{master.upper()}SHA" for master in ("bias", "dark", "flat")}
# Keys whose stored value must be evaluated with ``asteval`` before
# the comparison. Subclasses can extend this set.
_evaluated_header_keys = {"OUTLTHRS"}
preserve_processing_dir = None
[docs]
@classmethod
def set_test_directory(
cls,
test_dirname,
processing_dirname,
failed_test_dirname,
preserve_processing_dir=None,
):
"""Set the directory where data to test against is located."""
cls.test_directory = test_dirname
cls.processing_directory = processing_dirname
cls.failed_test_directory = failed_test_dirname
cls.preserve_processing_dir = preserve_processing_dir
[docs]
def setUp(self):
"""Make sure the data to compare against is defined."""
print(f"Setting up processing in {self.processing_directory!r}")
self.assertTrue(
hasattr(self, "test_directory"), "No test data directory defined!"
)
self.assertTrue(
hasattr(self, "processing_directory"),
"No processing directory defined!",
)
self.assertTrue(
path.exists(self.test_directory),
f"Test directory {self.test_directory} does not exist!",
)
makedirs(self.processing_directory, exist_ok=False)
gaia_cache = path.join(self.test_directory, "MASTERS", "Gaia")
if self.stage_catalog_cache and path.isdir(gaia_cache):
makedirs(
path.join(self.processing_directory, "MASTERS"), exist_ok=True
)
copytree(
gaia_cache,
path.join(self.processing_directory, "MASTERS", "Gaia"),
)
copy(
path.join(self.test_directory, "test.cfg"),
path.join(self.processing_directory, "test.cfg"),
)
self._point_at_test_database()
set_project_home(self.processing_directory)
with open(
path.join(self.test_directory, "survey_instruments.json"),
"r",
encoding="utf-8",
) as survey_json:
import_json_to_survey(survey_json)
[docs]
def _point_at_test_database(self):
"""Send this test's project database to the configured server.
Nothing else in the suite -- and no pipeline or step code -- needs
to know. ``set_project_home`` already reads ``autowisp_db.url``
from the project home when it is there, and every step launched by
:meth:`run_step` runs with that directory as its cwd, so it
resolves the same file. Writing it here therefore redirects the
whole run, subprocesses included.
A no-op when the variable is unset, which is the default.
"""
url = server_test_url()
if url is None:
return
# Before the project is opened: it is about to be created in a
# database the previous test left populated.
empty_server_database(url)
with open(
path.join(self.processing_directory, DB_URL_FNAME),
"w",
encoding="utf-8",
) as url_file:
url_file.write(url)
[docs]
def _test_failed(self):
"""Whether *this* test has just failed or errored.
Read off the result rather than a flag the test sets, so nothing
has to be remembered at the end of every test method. Two things
make the obvious shortcuts wrong: ``_outcome.success`` is still
True here (it is reset per test *part*, and ``tearDown`` is its
own part), and ``result.errors`` / ``result.failures`` accumulate
over the whole run -- so the entries have to be matched against
this test rather than merely counted.
Returns:
bool: True if this test recorded a failure or an error.
"""
result = getattr(getattr(self, "_outcome", None), "result", None)
if result is None:
return False
return any(
test is self
for group in ("errors", "failures")
for test, _ in getattr(result, group, ())
)
[docs]
def tearDown(self):
"""Remove the processing directory."""
print(f"Tearing down processing in {self.processing_directory!r}")
if self.preserve_failed_processing and self._test_failed():
# Preserve every failed test in its own subdirectory (keyed by
# class + method) so a run with several failures keeps all of them
# for post-mortem, rather than each failure overwriting the last.
destination = path.join(
self.failed_test_directory,
f"{type(self).__name__}_{self._testMethodName}",
)
if path.exists(destination):
rmtree(destination, ignore_errors=False)
copytree(self.processing_directory, destination)
if self.preserve_processing_dir is not None:
destination = path.join(
self.preserve_processing_dir,
self._testMethodName + "_processing",
)
if path.exists(destination):
rmtree(destination)
move(self.processing_directory, destination)
else:
rmtree(self.processing_directory)
[docs]
def run_step(self, command):
"""Run a calibration step and check the return code."""
calib_process = run(
command,
cwd=self.processing_directory,
check=False,
stdout=PIPE,
stderr=STDOUT,
timeout=3600,
)
self.assertTrue(
calib_process.returncode == 0,
f"AutoWISP step command:\n{command!r}\n"
f"Started from {self.processing_directory!r} "
f"failed:\n{calib_process.stdout.decode('utf-8')}",
)