Source code for jwql.tests.test_database_interface

#! /usr/bin/env python

"""Tests for the ``database_interface.py`` module.

Authors
-------

    - Joe Filippazzo
    - Matthew Bourque

Use
---

    These tests can be run via the command line (omit the ``-s`` to
    suppress verbose output to stdout):
    ::

        pytest -s database_interface.py
"""

import datetime
import os
import pytest
import random
import string

from sqlalchemy import inspect

from jwql.database import database_interface as di
from jwql.tests.resources import has_test_db
from jwql.utils.utils import get_config
from jwql.utils.constants import ON_GITHUB_ACTIONS


[docs] @pytest.mark.skipif(ON_GITHUB_ACTIONS, reason='Requires access to development database server.') def test_all_tables_exist(): """Test that the table ORMs defined in ``database_interface`` actually exist as tables in the database""" # Get list of table ORMs from database_interface table_orms = [] database_interface_attributes = di.__dict__.keys() for attribute in database_interface_attributes: table_object = getattr(di, attribute) try: table_orms.append(table_object.__tablename__) except AttributeError: pass # Not all attributes of database_interface are table ORMs # Get list of tables that are actually in the database existing_tables = inspect(di.engine).get_table_names() # Ensure that the ORMs defined in database_interface actually exist # as tables in the database for table in table_orms: if table == 'nirspec_ta_stats': continue assert table in existing_tables
[docs] def test_anomaly_orm_factory(): """Test that the ``anomaly_orm_factory`` function successfully creates an ORM and contains the appropriate columns""" test_table_name = 'test_anomaly_table' TestAnomalyTable = di.anomaly_orm_factory(test_table_name) table_attributes = TestAnomalyTable.__dict__.keys() assert str(TestAnomalyTable) == "<class 'jwql.database.database_interface.{}'>"\ .format(test_table_name) for item in ['id', 'rootname', 'flag_date', 'user']: assert item in table_attributes
[docs] @pytest.mark.skipif(not has_test_db(), reason='Modifies test database.') def test_anomaly_records(): """Test to see that new records can be entered""" # Add some data random_rootname = ''.join(random.SystemRandom().choice( string.ascii_lowercase + string.ascii_uppercase + string.digits) for _ in range(10)) di.session.add(di.FGSAnomaly(rootname=random_rootname, flag_date=datetime.datetime.today(), user='test', ghost=True)) di.session.commit() # Test the ghosts column ghosts = di.session.query(di.FGSAnomaly)\ .filter(di.FGSAnomaly.rootname == random_rootname) assert bool(ghosts.data_frame.iloc[0]['ghost']) is True # clean up ghosts.delete() di.session.commit() assert di.session.query(di.FGSAnomaly)\ .filter(di.FGSAnomaly.rootname == random_rootname).count() == 0
[docs] @pytest.mark.skipif(ON_GITHUB_ACTIONS, reason='Requires database configuration.') def test_load_connections(): """Test to see that a connection to the database can be established""" session, base, engine, meta = di.load_connection(get_config()['connection_string']) assert str(type(session)) == "<class 'sqlalchemy.orm.session.Session'>" assert str(type(base)) == "<class 'sqlalchemy.orm.decl_api.DeclarativeMeta'>" assert str(type(engine)) == "<class 'sqlalchemy.engine.base.Engine'>" assert str(type(meta)) == "<class 'sqlalchemy.sql.schema.MetaData'>" session.close()
[docs] def test_monitor_orm_factory(): """Test that the ``monitor_orm_factory`` function successfully creates an ORM and contains the appropriate columns""" test_table_name = 'instrument_test_monitor_table' # Create temporary table definitions file test_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'database', 'monitor_table_definitions', 'instrument') test_filename = os.path.join(test_dir, '{}.txt'.format(test_table_name)) if not os.path.isdir(test_dir): os.mkdir(test_dir) with open(test_filename, 'w') as f: f.write('TEST_COLUMN, string') # Create the test table ORM TestMonitorTable = di.monitor_orm_factory(test_table_name) table_attributes = TestMonitorTable.__dict__.keys() # Ensure the ORM exists and contains appropriate columns assert str(TestMonitorTable) == "<class 'jwql.database.database_interface.{}'>"\ .format(test_table_name) for column in ['id', 'entry_date', 'test_column']: assert column in table_attributes # Remove test files and directories if os.path.isfile(test_filename): os.remove(test_filename) if os.path.isdir(test_dir): os.rmdir(test_dir)