From 39d331dda2624e2a301e32f5f671f34b0f1e9571 Mon Sep 17 00:00:00 2001 From: Joshua Fraustro Date: Wed, 13 Dec 2023 20:22:27 -0500 Subject: [PATCH] init --- vo_models/vo_models/__init__.py | 0 vo_models/vo_models/vo_models_test.py | 73 ++++++ vo_models/vo_models/vodataservice/__init__.py | 10 + vo_models/vo_models/vodataservice/models.py | 212 ++++++++++++++++ .../vodataservice_models_test.py | 240 ++++++++++++++++++ .../vo_models/vosi/VOSIAvailability-v1.0.xsd | 73 ++++++ vo_models/vo_models/vosi/VOSITables-v1.1.xsd | 46 ++++ vo_models/vo_models/vosi/__init__.py | 7 + vo_models/vo_models/vosi/availability.py | 22 ++ vo_models/vo_models/vosi/availability_test.py | 45 ++++ vo_models/vo_models/vosi/tables.py | 20 ++ vo_models/vo_models/vosi/tables_test.py | 154 +++++++++++ 12 files changed, 902 insertions(+) create mode 100644 vo_models/vo_models/__init__.py create mode 100644 vo_models/vo_models/vo_models_test.py create mode 100644 vo_models/vo_models/vodataservice/__init__.py create mode 100644 vo_models/vo_models/vodataservice/models.py create mode 100644 vo_models/vo_models/vodataservice/vodataservice_models_test.py create mode 100644 vo_models/vo_models/vosi/VOSIAvailability-v1.0.xsd create mode 100644 vo_models/vo_models/vosi/VOSITables-v1.1.xsd create mode 100644 vo_models/vo_models/vosi/__init__.py create mode 100644 vo_models/vo_models/vosi/availability.py create mode 100644 vo_models/vo_models/vosi/availability_test.py create mode 100644 vo_models/vo_models/vosi/tables.py create mode 100644 vo_models/vo_models/vosi/tables_test.py diff --git a/vo_models/vo_models/__init__.py b/vo_models/vo_models/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/vo_models/vo_models/vo_models_test.py b/vo_models/vo_models/vo_models_test.py new file mode 100644 index 0000000..fa2ae71 --- /dev/null +++ b/vo_models/vo_models/vo_models_test.py @@ -0,0 +1,73 @@ +"""Tests for vo-tap models and datatypes""" +from unittest import TestCase + +import xmldiff.actions +import xmldiff.main + + +class VOModelTestBase: + """Base class for VO model tests. + + VO model test classes that inherit from VOModelTestCase should have the following attributes: + + - test_xml: A string containing the XML to be tested. Considered the ground truth. + - test_element: An instance of the model to be tested. Compared to the XML. + - base_model: The model class to be tested. Used to instantiate the model from XML. + + Extra tests can be added to the inheriting class as needed. + """ + + # pylint: disable=no-member + + class VOModelTestCase(TestCase): + """Default tests for VO models. + + Automatically runs the following tests: + - test_read_xml: Test reading XML into model and that all elements are present. + - test_write_xml: Test writing model to XML and no elements are missing. + + These tests ensure that the model can be round-tripped from XML and back again without losing any data. + """ + + def test_read_xml(self): + """Test reading XML into model and that all elements are present. + + Compares the model created from the XML and ensures it matches the test element. + """ + + test_model = self.base_model.from_xml(self.test_xml) + self.assertIsInstance(test_model, self.base_model) + self.assertEqual(test_model, self.test_element) + + def test_write_xml(self): + """Test writing model to XML. + + Compares the XML created from the model and ensures it matches the test XML. + """ + xml = self.test_element.to_xml(skip_empty=True, encoding=str) + assert_equal_xml(xml, self.test_xml) + + +def assert_equal_xml(xml1: bytes | str, xml2: bytes | str, skip_empty=True, skip_defaults=True): + """Test whether two xml strings are equal, skipping empty elements and default attributes + + Args: + xml1 (bytes | str): XML string to compare + xml2 (bytes | str): XML string to compare + skip_empty (bool, optional): Whether to skip the deletion of empty elements. Defaults to True. + skip_defaults (bool, optional): Whether to skip the addition of default attributes. Defaults to True. + """ + diffs = xmldiff.main.diff_texts(xml1, xml2, diff_options={"fast_match": True}) + + if diffs: + for diff in diffs: + if isinstance(diff, xmldiff.actions.DeleteNode): + # Skip raising over the removal of empty elements. We do this by default. + if not skip_empty: + raise AssertionError(f"XML strings are not equal: {diff}") + elif isinstance(diff, xmldiff.actions.InsertAttrib): + # Skip raising over the addition of default attributes. We do this to conform to the standard. + if not skip_defaults: + raise AssertionError(f"XML strings are not equal: {diff}") + else: + raise AssertionError(f"XML strings are not equal: {diff}") diff --git a/vo_models/vo_models/vodataservice/__init__.py b/vo_models/vo_models/vodataservice/__init__.py new file mode 100644 index 0000000..5deafaf --- /dev/null +++ b/vo_models/vo_models/vodataservice/__init__.py @@ -0,0 +1,10 @@ +"""Module containing models and resources for IVOA VODataService objects.""" +from mast.vo_tap.services.vo_models.vodataservice.models import ( + DataType, + FKColumn, + ForeignKey, + Table, + TableParam, + TableSchema, + TableSet, +) diff --git a/vo_models/vo_models/vodataservice/models.py b/vo_models/vo_models/vodataservice/models.py new file mode 100644 index 0000000..08029b3 --- /dev/null +++ b/vo_models/vo_models/vodataservice/models.py @@ -0,0 +1,212 @@ +"""Pydantic-xml models for VODataService types""" +from typing import Any, Optional +# skip the bandit check here since we use escape() here only on XML we generated +from xml.sax.saxutils import escape # nosec B406 + +from asb.core.encoding import ADQL_SQL_KEYWORDS, handle_votable_arraysize +from pydantic import validator +from pydantic_xml import BaseXmlModel, RootXmlModel, attr, element + +# pylint: disable=no-self-argument + +NSMAP = { + "": "http://www.ivoa.net/xml/VODataService/v1.1", + "xs": "http://www.w3.org/2001/XMLSchema", + "vr": "http://www.ivoa.net/xml/VOResource/v1.0", + "vs": "http://www.ivoa.net/xml/VODataService/v1.1", + "stc": "http://www.ivoa.net/xml/STC/stc-v1.30.xsd", + "vm": "http://www.ivoa.net/xml/VOMetadata/v0.1", +} + + +class FKColumn(BaseXmlModel, tag="fkColumn"): + """An individual foreign key column.""" + + from_column: str = element(tag="fromColumn") + target_column: str = element(tag="targetColumn") + + +class ForeignKey(BaseXmlModel, tag="foreignKey"): + """An element containing one or more foreign key columns.""" + + target_table: str = element(tag="targetTable") + fk_column: list[FKColumn] = element(tag="fkColumn") + description: Optional[str] = element(tag="description", default=None) + utype: Optional[str] = element(tag="utype", default=None) + + def __init__(__pydantic_self__, **data: Any) -> None: + # If what we were given is of the form: + # {'target_table': 'target_table', 'from_column': 'from_column', 'target_column': 'target_column'} + # and we don't have an fk_column, make one + # this is a convenience for database calls to TAP_SCHEMA.keys / TAP_SCHEMA.key_columns + if not data.get("fk_column", None): + if data.get("from_column") and data.get("target_column"): + data["fk_column"] = [ + FKColumn( + from_column=data["from_column"], + target_column=data["target_column"], + ) + ] + super().__init__(**data) + + +class DataType(BaseXmlModel, tag="dataType", nsmap={"xsi": "http://www.w3.org/2001/XMLSchema-instance"}): + """A simple element containing a column datatype""" + + type: Optional[str] = attr(name="type", ns="xsi", default="vs:VOTableType") + arraysize: Optional[str] = attr(name="arraysize", default=None) + value: str + + +class TableParam(BaseXmlModel, ns="", tag="column"): + """A column element as returned from TAP_SCHEMA.columns. + + 'TableParam' is the IVOA standard name for this element, but it's basically a column. + """ + + column_name: str = element(tag="name") + description: Optional[str] = element(tag="description", default=None) + unit: Optional[str] = element(tag="unit", default=None) + ucd: Optional[str] = element(tag="ucd", default=None) + utype: Optional[str] = element(tag="utype", default=None) + xtype: Optional[str] = element(tag="xtype", default=None) + datatype: Optional[DataType] = element(tag="dataType", default=None) + flag: Optional[list[str]] = element(tag="flag", default=None) + + def __init__(__pydantic_self__, **data: Any) -> None: + data["datatype"] = __pydantic_self__.__make_datatype_element(data) + data["flag"] = __pydantic_self__.__make_flags(data) + super().__init__(**data) + + # pylint: disable=unused-private-member + def __make_datatype_element(self, col_data) -> DataType: + """Helper to make datatype element from column data when first created""" + if col_data.get("datatype", None): + if isinstance(col_data["datatype"], DataType): + return col_data["datatype"] + col = handle_votable_arraysize(col_data) + + datatype_type = col.get("datatype", None) + datatype_arraysize = col.get("arraysize", None) + + datatype_elem = DataType( + arraysize=datatype_arraysize, + value=datatype_type, + ) + return datatype_elem + # If no datatype provided, default to char(*) + return DataType(value="char", arraysize="*") + + def __make_flags(self, col_data) -> list[str]: + """Set up the flag elements when creating the column. + + These flags are represented in the database as a boolean integer, but in the XML are represented + as XML elements with the name of the flag, i.e. , , . + + """ + if not col_data.get("flag", None): + flag = [flag for flag in ["principal", "indexed", "std"] if col_data.get(flag, None) == 1] + return flag + return col_data["flag"] + + @validator("column_name") + def validate_colname(cls, value): + """Escape the column name if it is an ADQL reserved word + + See: https://www.ivoa.net/documents/ADQL/20180112/PR-ADQL-2.1-20180112.html#tth_sEc2.1.3 + """ + if value.upper() in ADQL_SQL_KEYWORDS: + value = f'"{value}"' + return value + + @validator("description") + def validate_description(cls, value): + """Sanitize bad XML values in the description""" + if value: + value = escape(str(value)) + return value + + +class _TableName(RootXmlModel[str], tag="name"): + """Element containing a table name + + Note: used internally to avoid namespacing issues with pydantic-xml + """ + + +class _TableTitle(RootXmlModel[str], tag="title"): + """Element containing a table title + + Note: used internally to avoid namespacing issues with pydantic-xml + """ + + +class _TableDesc(RootXmlModel[str], tag="description"): + """Element containing a table description + + Note: used internally to avoid namespacing issues with pydantic-xml + """ + + +class _TableUtype(RootXmlModel[str], tag="utype"): + """Element containing a table utype + + Note: used internally to avoid namespacing issues with pydantic-xml + """ + + +class _TableNRows(RootXmlModel[int], tag="nrows"): + """Element containing an integer describing the number of rows in a table + + Note: used internally to avoid namespacing issues with pydantic-xml + """ + + +class Table(BaseXmlModel, tag="table", ns="", skip_empty=True): + """A model representing a single table element. + + The private classes _TableName, _TableTitle, _TableDesc, _TableUtype, and _TableNRows + are used to create elements that do not inherit the default namespace from their parent. + This is necessary when creating a VOSITable, which has a default namespace of 'vosi', + but needs child elements without that prefix, since the elements below are part of the + VODataservice / DALI standard. + """ + + table_type: Optional[str] = attr(name="type") + + table_name: _TableName = element(tag="name", ns="") + title: Optional[_TableTitle] = element(tag="title", ns="", default=None) + description: Optional[_TableDesc] = element(tag="description", ns="", default=None) + utype: Optional[_TableUtype] = element(tag="utype", ns="", default=None) + nrows: Optional[_TableNRows] = element(tag="nrows", gte=0, ns="", default=None) + column: Optional[list[TableParam]] = element(tag="column", ns="", default=None) + foreign_key: Optional[list[ForeignKey]] = element(tag="foreignKey", ns="", default=None) + + def __init__(__pydantic_self__, **data: Any) -> None: + """Escape any keys that are passed in.""" + for key, val in data.items(): + if isinstance(val, str): + data[key] = escape(val) + super().__init__(**data) + + +class TableSchema(BaseXmlModel, tag="schema", ns="", skip_empty=True): + """A model representing a table schema.""" + + schema_name: str = element(tag="name", default="default") + title: Optional[str] = element(tag="title", default=None) + description: Optional[str] = element(tag="description", default=None) + table: Optional[list[Table]] = element(tag="table", default=None) + + def __init__(__pydantic_self__, **data: Any) -> None: + """Escape any keys that are passed in.""" + for key, val in data.items(): + if isinstance(val, str): + data[key] = escape(val) + super().__init__(**data) + + +class TableSet(BaseXmlModel, tag="tableset", skip_empty=True): + """A model representing a tableset, a list of tables.""" + + tableset_schema: list[TableSchema] = element(tag="schema") diff --git a/vo_models/vo_models/vodataservice/vodataservice_models_test.py b/vo_models/vo_models/vodataservice/vodataservice_models_test.py new file mode 100644 index 0000000..cb0fd6c --- /dev/null +++ b/vo_models/vo_models/vodataservice/vodataservice_models_test.py @@ -0,0 +1,240 @@ +"""Tests for VODataService models""" + +from mast.vo_tap.services.vo_models.vo_models_test import VOModelTestBase +from mast.vo_tap.services.vo_models.vodataservice import ( + DataType, + FKColumn, + ForeignKey, + Table, + TableParam, + TableSchema, + TableSet, +) + + +class TestFKColumn(VOModelTestBase.VOModelTestCase): + """Test the FKColumn element model""" + + test_xml = "from_columntarget_column" + test_element = FKColumn(from_column="from_column", target_column="target_column") + base_model = FKColumn + + +class TestForeignKey(VOModelTestBase.VOModelTestCase): + """Test the ForeignKey element model""" + + test_xml = ( + "" + "target_table" + "" + "from_column" + "target_column" + "" + "" + ) + test_element = ForeignKey( + target_table="target_table", + fk_column=[ + FKColumn( + from_column="from_column", + target_column="target_column", + ) + ], + ) + base_model = ForeignKey + + def test_single_dict_read(self): + """Test that we can read a single dict as a ForeignKey + + This can occur for TAP_SCHEMA.keys / TAP_SCHEMA.key_columns db calls, where we don't have a list of + fk_columns, but instead a single dict of the form: + """ + + single_dict = { + "target_table": "target_table", + "from_column": "from_column", + "target_column": "target_column", + } + + # pylint: disable=unsubscriptable-object + foreign_key_element = ForeignKey(**single_dict) + self.assertIsInstance(foreign_key_element, ForeignKey) + self.assertIsInstance(foreign_key_element.fk_column, list) + self.assertEqual(len(foreign_key_element.fk_column), 1) + self.assertEqual(foreign_key_element.target_table, "target_table") + self.assertEqual(foreign_key_element.fk_column[0].from_column, "from_column") + self.assertEqual(foreign_key_element.fk_column[0].target_column, "target_column") + + +class TestDataType(VOModelTestBase.VOModelTestCase): + """Test the DataType element model""" + + test_xml = ( + "string" + ) + test_element = DataType(type="vs:VOTableType", arraysize="*", value="string") + base_model = DataType + + +class TestTableParam(VOModelTestBase.VOModelTestCase): + """Test the TableParam element model""" + + test_xml = ( + "" + "name" + "description" + "unit" + "ucd" + "caom2:Artifact.productType" + "string" + "flag" + "" + ) + test_element = TableParam( + column_name="name", + description="description", + unit="unit", + ucd="ucd", + utype="caom2:Artifact.productType", + datatype=DataType(type="vs:VOTableType", arraysize="*", value="string"), + flag=["flag"], + ) + base_model = TableParam + + +class TestTableElement(VOModelTestBase.VOModelTestCase): + """Test the Table element model""" + + test_xml = ( + "" + "tap_schema.schemas" + "description of schemas in this dataset" + "" + "schema_name" + "Fully qualified schema name" + "char" + "std" + "" + "
" + ) + test_element = Table( + table_name="tap_schema.schemas", + table_type="table", + description="description of schemas in this dataset", + column=[ + TableParam( + column_name="schema_name", + description="Fully qualified schema name", + datatype=DataType(type="vs:VOTableType", arraysize="*", value="char"), + flag=["std"], + ) + ], + ) + base_model = Table + + +class TestSchemaElement(VOModelTestBase.VOModelTestCase): + """Test the TableSchema element model""" + + test_xml = ( + "" + "tap_schema" + "schema information for TAP services" + "" + "tap_schema.schemas" + "description of schemas in this dataset" + "
" + "" + "tap_schema.tables" + "description of tables in this dataset" + "
" + "" + ) + test_element = TableSchema( + schema_name="tap_schema", + description="schema information for TAP services", + table=[ + Table( + table_name="tap_schema.schemas", + table_type="table", + description="description of schemas in this dataset", + ), + Table( + table_name="tap_schema.tables", + table_type="table", + description="description of tables in this dataset", + ), + ], + ) + base_model = TableSchema + + +class TestTableSetElement(VOModelTestBase.VOModelTestCase): + """Test the TableSet element model""" + + test_xml = ( + "" + "" + "tap_schema" + "schema information for TAP services" + "" + "tap_schema.schemas" + "description of schemas in this dataset" + "
" + "" + "tap_schema.tables" + "description of tables in this dataset" + "
" + "" + "" + "dbo" + "ArchiveCatalog Infrastructure, version 1.0" + "" + "dbo.detailedCatalog" + "
" + "" + "dbo.SumMagAper2Cat" + "
" + "" + "" + ) + + test_element = TableSet( + tableset_schema=[ + TableSchema( + schema_name="tap_schema", + description="schema information for TAP services", + table=[ + Table( + table_name="tap_schema.schemas", + table_type="table", + description="description of schemas in this dataset", + ), + Table( + table_name="tap_schema.tables", + table_type="table", + description="description of tables in this dataset", + ), + ], + ), + TableSchema( + schema_name="dbo", + description="ArchiveCatalog Infrastructure, version 1.0", + table=[ + Table( + table_name="dbo.detailedCatalog", + table_type="table", + ), + Table( + table_name="dbo.SumMagAper2Cat", + table_type="table", + ), + ], + ), + ], + ) + base_model = TableSet diff --git a/vo_models/vo_models/vosi/VOSIAvailability-v1.0.xsd b/vo_models/vo_models/vosi/VOSIAvailability-v1.0.xsd new file mode 100644 index 0000000..7c72bc3 --- /dev/null +++ b/vo_models/vo_models/vosi/VOSIAvailability-v1.0.xsd @@ -0,0 +1,73 @@ + + + + + + A schema for formatting availability metadata as returned by an + availability resource defined in the IVOA Support Interfaces + specification (VOSI). + See http://www.ivoa.net/Documents/latest/VOSI.html. + + + + + + + + + + + + + Indicates whether the service is currently available. + + + + + + + + The instant at which the service last became available. + + + + + + + + The instant at which the service is next scheduled to become + unavailable. + + + + + + + + The instant at which the service is scheduled to become available + again after a period of unavailability. + + + + + + + + A textual note concerning availability. + + + + + + + + diff --git a/vo_models/vo_models/vosi/VOSITables-v1.1.xsd b/vo_models/vo_models/vosi/VOSITables-v1.1.xsd new file mode 100644 index 0000000..47388ff --- /dev/null +++ b/vo_models/vo_models/vosi/VOSITables-v1.1.xsd @@ -0,0 +1,46 @@ + + + + + + A schema for formatting table metadata as returned by a + tables resource, defined by the IVOA Support Interfaces + specification (VOSI). + See http://www.ivoa.net/Documents/latest/VOSI.html. + + + + + + + + + + A description of the table metadata supported by the + service associated with a VOSI-enabled resource. + + + + + + + + + A description of a single table supported by the + service associated with a VOSI-enabled resource. + + + + diff --git a/vo_models/vo_models/vosi/__init__.py b/vo_models/vo_models/vosi/__init__.py new file mode 100644 index 0000000..85c73c3 --- /dev/null +++ b/vo_models/vo_models/vosi/__init__.py @@ -0,0 +1,7 @@ +"""Module containing models and resources for IVOA VOSI services. + +VOSI (VO Service Interface) is a set of IVOA standards for describing and accessing VO services, +allowing for a standard way to discover and access VO services. +""" +from mast.vo_tap.services.vo_models.vosi.availability import Availability +from mast.vo_tap.services.vo_models.vosi.tables import VOSITable, VOSITableSet diff --git a/vo_models/vo_models/vosi/availability.py b/vo_models/vo_models/vosi/availability.py new file mode 100644 index 0000000..f457272 --- /dev/null +++ b/vo_models/vo_models/vosi/availability.py @@ -0,0 +1,22 @@ +"""Availability VOSI Schema using Pydantic-XML models""" + +from typing import Optional + +from pydantic_xml import BaseXmlModel, element +from vo_models.xml.voresource.types import UTCTimestamp + +NSMAP = { + "xsd": "http://www.w3.org/2001/XMLSchema", + "xsi": "http://www.w3.org/2001/XMLSchema-instance", + "": "http://www.ivoa.net/xml/VOSIAvailability/v1.0", +} + + +class Availability(BaseXmlModel, tag="availability", nsmap=NSMAP): + """Availability VOSI element""" + + available: bool = element(tag="available") + up_since: Optional[UTCTimestamp] = element(tag="upSince", default=None) + down_at: Optional[UTCTimestamp] = element(tag="downAt", default=None) + back_at: Optional[UTCTimestamp] = element(tag="backAt", default=None) + note: Optional[list[str]] = element(tag="note", default=None) diff --git a/vo_models/vo_models/vosi/availability_test.py b/vo_models/vo_models/vosi/availability_test.py new file mode 100644 index 0000000..f8cdf38 --- /dev/null +++ b/vo_models/vo_models/vosi/availability_test.py @@ -0,0 +1,45 @@ +"""Tests for the VOSI Availability model""" + +# We're only parsing a locally controlled XSD file +from lxml import etree # nosec B410 + +from mast.vo_tap.services.vo_models.vo_models_test import VOModelTestBase +from mast.vo_tap.services.vo_models.vosi import Availability + +AVAIL_NS_HEADER = """xmlns:xsd="http://www.w3.org/2001/XMLSchema" +xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" +xmlns="http://www.ivoa.net/xml/VOSIAvailability/v1.0" +""" + +with open("mast/vo_tap/services/vo_models/vosi/VOSIAvailability-v1.0.xsd", "r") as schema_file: + availability_schema = etree.XMLSchema(file=schema_file) + + +class TestAvailability(VOModelTestBase.VOModelTestCase): + """Test the Availability model""" + + test_xml = ( + f"" + "true" + "2021-01-01T00:00:00.000Z" + "2021-01-01T00:00:00.000Z" + "2021-01-01T00:00:00.000Z" + "Available Mon-Friday" + "We take weekends off" + "" + ) + + test_element = Availability( + available=True, + up_since="2021-01-01T00:00:00.000Z", + down_at="2021-01-01T00:00:00.000Z", + back_at="2021-01-01T00:00:00.000Z", + note=["Available Mon-Friday", "We take weekends off"], + ) + + base_model = Availability + + def test_validate(self): + """Validate the model agains the Availability schema""" + availability_xml = etree.fromstring(self.test_element.to_xml(skip_empty=True, encoding=str)) + availability_schema.assertValid(availability_xml) diff --git a/vo_models/vo_models/vosi/tables.py b/vo_models/vo_models/vosi/tables.py new file mode 100644 index 0000000..9962285 --- /dev/null +++ b/vo_models/vo_models/vosi/tables.py @@ -0,0 +1,20 @@ +"""Pydantic-xml models for the VOSI Tables specification""" + + +from mast.vo_tap.services.vo_models.vodataservice import Table, TableSet + +NSMAP = { + "vosi": "http://www.ivoa.net/xml/VOSITables/v1.0", + "vr": "http://www.ivoa.net/xml/VOResource/v1.0", + "vs": "http://www.ivoa.net/xml/VODataService/v1.1", + "xsd": "http://www.w3.org/2001/XMLSchema", + "xsi": "http://www.w3.org/2001/XMLSchema-instance", +} + + +class VOSITable(Table, tag="table", ns="vosi", nsmap=NSMAP): + """A table element as returned by a VOSI /tables request""" + + +class VOSITableSet(TableSet, tag="tableset", ns="vosi", nsmap=NSMAP): + """A tableset element as returned by a VOSI /tables request""" diff --git a/vo_models/vo_models/vosi/tables_test.py b/vo_models/vo_models/vosi/tables_test.py new file mode 100644 index 0000000..86c3dfa --- /dev/null +++ b/vo_models/vo_models/vosi/tables_test.py @@ -0,0 +1,154 @@ +"""Tests for VOSI-Tables specific pydantic-xml models""" + +from defusedxml import ElementTree as ET +# We're only parsing a locally controlled XSD file +from lxml import etree # nosec B410 + +from mast.vo_tap.services.vo_models.vo_models_test import VOModelTestBase +from mast.vo_tap.services.vo_models.vodataservice import DataType, Table, TableParam, TableSchema +from mast.vo_tap.services.vo_models.vosi import VOSITable, VOSITableSet + +with open("mast/vo_tap/services/vo_models/vosi/VOSITables-v1.1.xsd", "r") as schema_file: + vosi_tables_schema = etree.XMLSchema(file=schema_file) + + +class TestVOSITableElement(VOModelTestBase.VOModelTestCase): + """Test the Table element model as specified in VOSI Tables""" + + test_xml = """ + + tap_schema.schemas + description of schemas in this dataset + + schema_name + Fully qualified schema name + char + std + + + description + Brief description of the schema + char + std + + + """ + test_element = VOSITable( + table_type="table", + table_name="tap_schema.schemas", + title=None, + description="description of schemas in this dataset", + utype=None, + nrows=None, + column=[ + TableParam( + column_name="schema_name", + description="Fully qualified schema name", + unit=None, + ucd=None, + datatype=DataType(type_attr="vs:VOTableType", arraysize="*", value="char"), + flag=["std"], + ), + TableParam( + column_name="description", + description="Brief description of the schema", + unit=None, + ucd=None, + datatype=DataType(type_attr="vs:VOTableType", arraysize="*", value="char"), + flag=["std"], + ), + ], + foreign_key=None, + ) + base_model = VOSITable + + def test_vosi_table_ns(self): + """Test that the Table is specifically namespaced to VOSI""" + vosi_element = self.base_model.from_xml(self.test_xml) + self.assertEqual(vosi_element.__xml_ns__, "vosi") + + vosi_xml = vosi_element.to_xml(skip_empty=True, encoding=str) + self.assertIn("" + "" + "tap_schema" + "schema information for TAP services" + "" + "tap_schema.schemas" + "description of schemas in this dataset" + "
" + "" + "tap_schema.tables" + "description of tables in this dataset" + "
" + "" + "" + ) + + test_element = VOSITableSet( + tableset_schema=[ + TableSchema( + schema_name="tap_schema", + title=None, + description="schema information for TAP services", + table=[ + Table( + table_type="table", + table_name="tap_schema.schemas", + title=None, + description="description of schemas in this dataset", + utype=None, + nrows=None, + column=None, + foreign_key=None, + ), + Table( + table_type="table", + table_name="tap_schema.tables", + title=None, + description="description of tables in this dataset", + utype=None, + nrows=None, + column=None, + foreign_key=None, + ), + ], + ), + ] + ) + + base_model = VOSITableSet + + def test_vositableset_ns(self): + """Test that the TableSet is specifically namespaced to VOSI""" + vosi_element = self.base_model.from_xml(self.test_xml) + self.assertEqual(vosi_element.__xml_ns__, "vosi") + + vosi_xml = vosi_element.to_xml(skip_empty=True, encoding=str) + self.assertIn("