test_tap_services.py

from lsst.rsp import get_tap_service
from IPython.display import display, HTML

def print_test(msg, color="black"):
    display(HTML(f'<span style="color: {color}">{msg}</span>'))

def print_error(msg, object_name, exception):
    display(HTML(f'<span style="color: red; font-weight: bold;">Error {msg}: {object_name}</span>'))
    display(HTML(f'<span style="color: red;">{exception}</span>'))

def test_tap_services():
    schema_query = 'SELECT * FROM tap_schema.schemas'
    for tap_service_name in ["tap", "ssotap", "consdbtap"]:
        print_test(f"Testing TAP service: {tap_service_name}")
        try:
            service = get_tap_service(tap_service_name)
        except Exception as e:
            print_error("getting TAP service", tap_service_name, e)
            continue
        try:
            schema_results_table = service.search(schema_query).to_table()
        except Exception as e:
            print_error("fetching schemas for", tap_service_name, e)
            continue
        for schema_name in schema_results_table['schema_name']:
            table_query = f"SELECT * FROM tap_schema.tables WHERE tap_schema.tables.schema_name = '{schema_name}' ORDER BY table_index ASC"
            try:
                table_results_table = service.search(table_query).to_table()
            except Exception as e:
                print_error("getting tables for", schema_name, e)
                continue
            for table_name in table_results_table["table_name"]:
                print_test(f"Testing table: {table_name}")
                test_query = f"SELECT * FROM {table_name} LIMIT 10"
                try:
                    test_results_table = service.search(test_query).to_table()
                    print_test(f"Successfully got {len(test_results_table)} records from {table_name}", color="green") 
                except Exception as e:
                    print_error("selecting records from", table_name, e)
                    continue