Skip to content

Commit

Permalink
add schema tests + fix extension required props
Browse files Browse the repository at this point in the history
  • Loading branch information
korikuzma committed Oct 10, 2024
1 parent 5ebca83 commit 7d79121
Show file tree
Hide file tree
Showing 2 changed files with 141 additions and 1 deletion.
2 changes: 1 addition & 1 deletion src/ga4gh/core/entity_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -141,7 +141,7 @@ class Extension(BaseModel):

name: str = Field(..., description='A name for the Extension. Should be indicative of its meaning and/or the type of information it value represents.')
value: Optional[Union[float, str, bool, Dict[str, Any], List[Any]]] = Field(
None, description='The value of the Extension - can be any primitive or structured object'
..., description='The value of the Extension - can be any primitive or structured object'
)
description: Optional[str] = Field(None, description="A description of the meaning or utility of the Extension, to explain the type of information it is meant to hold.")

Expand Down
140 changes: 140 additions & 0 deletions tests/test_schemas.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,140 @@
"""Test that VRS-Python Pydantic models match VRS and GKS-Common schemas"""

from enum import Enum
import json
from pathlib import Path

import pytest
from pydantic import BaseModel

from ga4gh.core import entity_models, domain_models
from ga4gh.vrs import models as vrs_models


class GKSSchema(str, Enum):
"""Enum for GKS schema"""

VRS = "vrs"
CORE_IM = "core-im"
DOMAIN = "domain-entities"


class GKSSchemaMapping(BaseModel):
"""Model for representing GKS Schema concrete classes, primitives, and schema"""

base_classes: set = set()
concrete_classes: set = set()
primitives: set = set()
schema: dict = dict()


def _update_gks_schema_mapping(
f_path: Path, gks_schema_mapping: GKSSchemaMapping
) -> None:
"""Update ``gks_schema_mapping`` properties
:param f_path: Path to JSON Schema file
:param gks_schema_mapping: GKS schema mapping to update
"""
with f_path.open() as rf:
cls_def = json.load(rf)

spec_class = cls_def["title"]
gks_schema_mapping.schema[spec_class] = cls_def

if "properties" in cls_def:
gks_schema_mapping.concrete_classes.add(spec_class)
elif cls_def.get("type") in {"array", "integer", "string"}:
gks_schema_mapping.primitives.add(spec_class)
else:
gks_schema_mapping.base_classes.add(spec_class)


GKS_SCHEMA_MAPPING = {gks: GKSSchemaMapping() for gks in GKSSchema}
SUBMODULES_DIR = Path(__file__).parents[1] / "submodules" / "vrs"


# Get vrs classes
vrs_mapping = GKS_SCHEMA_MAPPING[GKSSchema.VRS]
for f in (SUBMODULES_DIR / "schema" / "vrs" / "json").glob("*"):
_update_gks_schema_mapping(f, vrs_mapping)


# Get core-im + domain classes
for child in (SUBMODULES_DIR / "submodules" / "gks-common" / "schema").iterdir():
mapping_key = (
GKSSchema.DOMAIN if str(child).endswith(GKSSchema.DOMAIN) else GKSSchema.CORE_IM
)
mapping = GKS_SCHEMA_MAPPING[mapping_key]
for f in (child / "json").glob("*"):
_update_gks_schema_mapping(f, mapping)


@pytest.mark.parametrize(
"gks_schema,pydantic_models",
[
(GKSSchema.VRS, vrs_models),
(GKSSchema.CORE_IM, entity_models),
(GKSSchema.DOMAIN, domain_models),
],
)
def test_schema_models_in_pydantic(gks_schema, pydantic_models):
"""Ensure that each schema model has corresponding Pydantic model"""
mapping = GKS_SCHEMA_MAPPING[gks_schema]
for schema_model in (
mapping.base_classes | mapping.concrete_classes | mapping.primitives
):
assert getattr(pydantic_models, schema_model, False)


@pytest.mark.parametrize(
"gks_schema,pydantic_models",
[
(GKSSchema.VRS, vrs_models),
(GKSSchema.CORE_IM, entity_models),
(GKSSchema.DOMAIN, domain_models),
],
)
def test_schema_class_fields(gks_schema, pydantic_models):
"""Check that each schema model properties exist and are required in corresponding
Pydantic model, and validate required properties
"""
mapping = GKS_SCHEMA_MAPPING[gks_schema]
for schema_model in mapping.concrete_classes:
schema_fields = set(mapping.schema[schema_model]["properties"])
pydantic_model = getattr(pydantic_models, schema_model)
assert set(pydantic_model.model_fields) == schema_fields, schema_model

# Check required fields
required_schema_fields = set(mapping.schema[schema_model]["required"])
required_pydantic_fields = {
field
for field, field_info in pydantic_model.model_fields.items()
if field_info.is_required() or isinstance(field_info.default, str)
}
assert required_pydantic_fields == required_schema_fields, schema_model


def test_ga4gh_keys():
"""Ensure ga4ghDigest keys defined in schema model exist in corresponding Pydantic model"""
vrs_mapping = GKS_SCHEMA_MAPPING[GKSSchema.VRS]
for vrs_class in vrs_mapping.concrete_classes:
if (
vrs_mapping.schema[vrs_class].get("ga4ghDigest", {}).get("keys", None)
is None
):
continue

pydantic_model = getattr(vrs_models, vrs_class)

try:
pydantic_model_digest_keys = pydantic_model.ga4gh.keys
except AttributeError as e:
raise AttributeError(vrs_class) from e

assert set(pydantic_model_digest_keys) == set(
vrs_mapping.schema[vrs_class]["ga4ghDigest"]["keys"]
), vrs_class
assert pydantic_model_digest_keys == sorted(
pydantic_model.ga4gh.keys
), vrs_class

0 comments on commit 7d79121

Please sign in to comment.