diff --git a/Makefile b/Makefile index 7c74c007..ac540f15 100644 --- a/Makefile +++ b/Makefile @@ -74,6 +74,10 @@ audit: deps bandit: deps $(BIN)/bandit -r app || true +test: deps + $(UV) pip install --python $(BIN)/python -e ".[dev]" + $(BIN)/python -m pytest test/ -v + # Full validation bundle lint: clean format ruff pylint audit bandit diff --git a/app/demo_adapter.py b/app/demo_adapter.py index 4564f859..10372665 100644 --- a/app/demo_adapter.py +++ b/app/demo_adapter.py @@ -15,7 +15,6 @@ import uuid from fastapi import HTTPException -from fastapi.encoders import jsonable_encoder from pydantic import BaseModel from .routers.account import facility_adapter as account_adapter @@ -520,8 +519,8 @@ async def list_sites( sites = [s for s in sites if s.last_modified > ms] o = offset or 0 - l = limit or len(sites) - return sites[o : o + l] + page_limit = limit or len(sites) + return sites[o : o + page_limit] async def get_site(self: "DemoAdapter", site_id: str, modified_since: str | None = None) -> facility_models.Site: site = next((s for s in self.sites if s.id == site_id), None) @@ -547,7 +546,7 @@ async def get_resources( description: str | None = None, group: str | None = None, modified_since: datetime.datetime | None = None, - resource_type: status_models.ResourceType | None = None, + resource_type: status_models.ResourceTypeValue | None = None, current_status: status_models.Status | None = None, capability: Capability | None = None, site_id: str | None = None, diff --git a/app/routers/account/models.py b/app/routers/account/models.py index e4c6d4f7..f8e80f92 100644 --- a/app/routers/account/models.py +++ b/app/routers/account/models.py @@ -4,7 +4,7 @@ from ...request_context import get_url_prefix from ...types.base import IRIBaseModel -from ...types.scalars import AllocationUnit +from ...types.scalars import AllocationUnit, AllocationUnitValue class Project(IRIBaseModel): @@ -34,7 +34,7 @@ class AllocationEntry(IRIBaseModel): allocation: float = Field(..., description="Total allocation amount granted.", example=100000.0) # how much this allocation can spend usage: float = Field(..., description="Amount of allocation consumed.", example=52342.5) # how much this allocation has spent - unit: AllocationUnit = Field(..., description="Unit of the allocation (e.g., node_hours, bytes).", example="node_hours") + unit: AllocationUnitValue = Field(..., description="DOE IRI URN for the allocation unit.", example=AllocationUnit.node_hours) class ProjectAllocation(IRIBaseModel): diff --git a/app/routers/filesystem/models.py b/app/routers/filesystem/models.py index fc36ebca..79d5c271 100644 --- a/app/routers/filesystem/models.py +++ b/app/routers/filesystem/models.py @@ -9,13 +9,7 @@ from enum import Enum from pydantic import Field, AliasChoices, BaseModel - -class CompressionType(str, Enum): - """Defines the type of compression to be used for compressing or extracting files.""" - none = "none" - bzip2 = "bzip2" - gzip = "gzip" - xz = "xz" +from ...types.scalars import CompressionType, CompressionTypeValue class ContentUnit(str, Enum): @@ -202,7 +196,7 @@ class PostCompressRequest(FilesystemRequestBase): target_path: str = Field(..., description="Path to the compressed file", example="/home/user/file.tar.gz") match_pattern: str|None = Field(default=None, description="Regex pattern to filter files to compress", example=".*\\.txt$") dereference: bool = Field(default=False, description="If set to `true`, it follows symbolic links and archive the files they point to instead of the links themselves.", example=True) - compression: CompressionType = Field(default="gzip", description="Defines the type of compression to be used. By default gzip is used.", example="gzip") + compression: CompressionTypeValue = Field(default=CompressionType.gzip, description="DOE IRI URN for the compression type. Legacy short tokens are accepted only as input compatibility aliases and are normalized.", example=CompressionType.gzip) model_config = { "json_schema_extra": { "examples": [ @@ -211,7 +205,7 @@ class PostCompressRequest(FilesystemRequestBase): "target_path": "/home/user/file.tar.gz", "match_pattern": "*./[ab].*\\.txt", "dereference": "true", - "compression": "none", + "compression": CompressionType.none, } ] } @@ -226,14 +220,14 @@ class PostExtractResponse(BaseModel): class PostExtractRequest(FilesystemRequestBase): """Represents a request to extract a compressed file.""" target_path: str = Field(..., description="Path to the directory where to extract the compressed file", example="/home/user/dir") - compression: CompressionType = Field(default="gzip", description="Defines the type of compression to be used. By default gzip is used.", example="gzip") + compression: CompressionTypeValue = Field(default=CompressionType.gzip, description="DOE IRI URN for the compression type. Legacy short tokens are accepted only as input compatibility aliases and are normalized.", example=CompressionType.gzip) model_config = { "json_schema_extra": { "examples": [ { "source_path": "/home/user/dir/file.tar.gz", "target_path": "/home/user/dir", - "compression": "none", + "compression": CompressionType.none, } ] } diff --git a/app/routers/status/facility_adapter.py b/app/routers/status/facility_adapter.py index d5dd8dde..274d4b2f 100644 --- a/app/routers/status/facility_adapter.py +++ b/app/routers/status/facility_adapter.py @@ -21,7 +21,7 @@ async def get_resources( description: str | None = None, group: str | None = None, modified_since: datetime.datetime | None = None, - resource_type: status_models.ResourceType | None = None, + resource_type: status_models.ResourceTypeValue | None = None, current_status: status_models.Status | None = None, capability: Capability | None = None, site_id: str | None = None, diff --git a/app/routers/status/models.py b/app/routers/status/models.py index a8999939..6e1884ba 100644 --- a/app/routers/status/models.py +++ b/app/routers/status/models.py @@ -6,6 +6,7 @@ from ...request_context import get_url_prefix from ...types.base import NamedObject +from ...types.scalars import ResourceType, ResourceTypeValue, urn_has_complete_prefix, validate_doe_iri_urn class Status(enum.Enum): @@ -16,18 +17,8 @@ class Status(enum.Enum): unknown = "unknown" -class ResourceType(enum.Enum): - """Represents the type of a resource.""" - website = "website" - service = "service" - compute = "compute" - system = "system" - storage = "storage" - network = "network" - unknown = "unknown" - - -class Endpoint(enum.Enum): +class Endpoint(str, enum.Enum): + """Router endpoint a resource supports (used internally to route compute/filesystem requests).""" compute = "compute" filesystem = "filesystem" @@ -42,7 +33,7 @@ def _self_path(self) -> str: capability_ids: list[str] = Field(default_factory=list, exclude=True) group: str|None = Field(default=None, description="Logical grouping of the resource", example="frontend") current_status: Status|None = Field(default=None, description="The current status comes from the status of the last event for this resource", example="up") - resource_type: ResourceType = Field(..., description="Type of the resource", example="service") + resource_type: ResourceTypeValue = Field(..., description="DOE IRI URN for the resource type", example=ResourceType.service) supported_endpoints: list[Endpoint] = Field(default_factory=list, description="a list of endpoints where this resource can be used") @computed_field(description="URI of the site where this resource is located") @@ -63,9 +54,10 @@ def find(cls, items, name=None, description=None, modified_since=None, group=Non if group: items = [item for item in items if item.group == group] if resource_type: - if isinstance(resource_type, str): - resource_type = ResourceType(resource_type) - items = [item for item in items if item.resource_type == resource_type] + # resource_type may be a ResourceType enum (which is a str subclass) or a raw URN string. + # Do not call str() on a str(Enum) — it returns the repr, not the value. + rt_urn = validate_doe_iri_urn(resource_type.value if hasattr(resource_type, "value") else resource_type) + items = [item for item in items if urn_has_complete_prefix(rt_urn, item.resource_type)] if current_status: items = [item for item in items if item.current_status == current_status] if capability: diff --git a/app/routers/status/status.py b/app/routers/status/status.py index ce5958e4..7f43689a 100644 --- a/app/routers/status/status.py +++ b/app/routers/status/status.py @@ -3,7 +3,7 @@ from fastapi import Depends, HTTPException, Query, Request from ...types.http import forbidExtraQueryParams -from ...types.scalars import AllocationUnit, StrictDateTime +from ...types.scalars import AllocationUnitValue, StrictDateTime, DOE_IRI_URN_MIN_LENGTH, DOE_IRI_URN_SCHEMA_PATTERN from .. import iri_router from ..error_handlers import DEFAULT_RESPONSES from ..iri_meta import iri_meta_dict @@ -33,9 +33,15 @@ async def get_resources( offset: int = Query(default=0, ge=0), limit: int = Query(default=100, ge=0, le=1000), modified_since: StrictDateTime = Query(default=None), - resource_type: models.ResourceType = Query(default=None), + resource_type: models.ResourceTypeValue = Query( + default=None, + min_length=DOE_IRI_URN_MIN_LENGTH, + pattern=DOE_IRI_URN_SCHEMA_PATTERN, + description="DOE IRI resource type URN (urn:doe-iri::). Facility-local extensions accepted.", + examples=[models.ResourceType.compute, models.ResourceType.storage, models.ResourceType.service], + ), current_status: models.Status = Query(default=None), - capability: List[AllocationUnit] = Query(default=None, min_length=1), + capability: List[AllocationUnitValue] = Query(default=None, min_length=1), _forbid=Depends(forbidExtraQueryParams("name", "description", "group", "offset", "limit", "modified_since", "resource_type", "current_status", "capability", multiParams={"capability"})), ) -> list[models.Resource]: return await router.adapter.get_resources( diff --git a/app/types/models.py b/app/types/models.py index 113e5dc5..67bf36bd 100644 --- a/app/types/models.py +++ b/app/types/models.py @@ -3,7 +3,7 @@ from pydantic import Field from .base import NamedObject -from .scalars import AllocationUnit, StrictDateTime +from .scalars import AllocationUnit, AllocationUnitValue, StrictDateTime class Capability(NamedObject): @@ -20,4 +20,4 @@ def _self_path(self) -> str: last_modified: StrictDateTime|None = Field(default=None, description="ISO 8601 timestamp when this object was last modified.", example="2026-02-21T12:00:00Z") - units: list[AllocationUnit] = Field(..., description="Allocation units supported by this capability", example=["node_hours"]) + units: list[AllocationUnitValue] = Field(..., description="Allocation units supported by this capability", example=[AllocationUnit.node_hours]) diff --git a/app/types/scalars.py b/app/types/scalars.py index 365be066..5cee9d9c 100644 --- a/app/types/scalars.py +++ b/app/types/scalars.py @@ -2,8 +2,11 @@ # pylint: disable=unused-argument import datetime -import enum +import re +from enum import Enum +from typing import Annotated +from pydantic import BeforeValidator, WithJsonSchema from pydantic_core import core_schema @@ -83,10 +86,137 @@ def __get_pydantic_json_schema__(cls, schema, handler): # ----------------------------------------------------------------------- -# AllocationUnit: an enum for allocation units -class AllocationUnit(enum.Enum): - """Units for allocation""" +# DOE IRI URN validation - node_hours = "node_hours" - bytes = "bytes" - inodes = "inodes" +DOE_IRI_URN_PREFIX = "urn:doe-iri:" +_DOMAIN = r"[A-Za-z0-9][A-Za-z0-9-]{0,31}" +_SEGMENT_CHAR = r"(?:[A-Za-z0-9._~-]|%[0-9A-Fa-f]{2}|[!$&'()*+,;=@]|/)" +_DOMAIN_SPECIFIC_SEGMENT = rf"{_SEGMENT_CHAR}+" +_DOMAIN_SPECIFIC_STRING = rf"{_DOMAIN_SPECIFIC_SEGMENT}(?::{_DOMAIN_SPECIFIC_SEGMENT})*" +DOE_IRI_URN_PATTERN = re.compile(rf"^{DOE_IRI_URN_PREFIX}(?P{_DOMAIN}):(?P{_DOMAIN_SPECIFIC_STRING})$") +# General URN pattern and minimum length — use these for query parameters that accept any domain. +DOE_IRI_URN_SCHEMA_PATTERN = rf"^{DOE_IRI_URN_PREFIX}{_DOMAIN}:{_DOMAIN_SPECIFIC_STRING}$" +DOE_IRI_URN_MIN_LENGTH = len(DOE_IRI_URN_PREFIX) + 1 + 1 + 1 # prefix + 1 domain char + colon + 1 nss char + + +def validate_doe_iri_urn(value: str) -> str: + """Validate a DOE IRI URN string. Raises ValueError on failure.""" + if not isinstance(value, str) or not value.strip(): + raise ValueError("Invalid DOE IRI URN. Expected a non-empty string.") + candidate = value.strip() + if not DOE_IRI_URN_PATTERN.fullmatch(candidate): + raise ValueError("Invalid DOE IRI URN. Expected format urn:doe-iri::.") + return candidate + + +def _validate_urn_domain(value: str, domain: str, label: str) -> str: + """Validate a DOE IRI URN and enforce that its domain matches the expected value.""" + urn = validate_doe_iri_urn(value) + actual_domain = urn.split(":", 3)[2] + if actual_domain != domain: + raise ValueError(f"Invalid {label}. Expected domain '{domain}', got '{actual_domain}'.") + return urn + + +def doe_iri_domain_urn_schema_pattern(domain: str) -> str: + """Return the JSON schema pattern for DOE IRI URNs in one domain.""" + return rf"^{DOE_IRI_URN_PREFIX}{domain}:{_DOMAIN_SPECIFIC_STRING}$" + + +def doe_iri_domain_urn_min_length(domain: str) -> int: + """Return the minimum length for DOE IRI URNs in one domain.""" + return len(f"{DOE_IRI_URN_PREFIX}{domain}:") + 1 + + +def _domain_urn_schema(domain: str, description: str, examples: list[str]) -> dict[str, object]: + return { + "type": "string", + "minLength": doe_iri_domain_urn_min_length(domain), + "pattern": doe_iri_domain_urn_schema_pattern(domain), + "description": description, + "examples": examples, + } + + +def urn_has_complete_prefix(parent_urn: str, candidate_urn: str) -> bool: + """Return True when parent_urn is an exact or parent segment match of candidate_urn.""" + parent_segments = validate_doe_iri_urn(parent_urn).split(":") + candidate_segments = validate_doe_iri_urn(candidate_urn).split(":") + if len(parent_segments) > len(candidate_segments): + return False + return candidate_segments[: len(parent_segments)] == parent_segments + + +# ----------------------------------------------------------------------- +# Canonical enum types + + +class ResourceType(str, Enum): + """Canonical DOE IRI resource type URNs (spec §3.1). + + Note: `service` lives in the `service` domain per spec, not `resource`. + ResourceTypeValue accepts any valid DOE IRI URN to allow facility extensions. + """ + website = "urn:doe-iri:resource:website" + service = "urn:doe-iri:service:generic" + compute = "urn:doe-iri:resource:compute" + system = "urn:doe-iri:resource:system" + storage = "urn:doe-iri:resource:storage" + network = "urn:doe-iri:resource:network" + unknown = "urn:doe-iri:resource:unknown" + + +class AllocationUnit(str, Enum): + """Canonical DOE IRI allocation-unit URNs (spec §3.2).""" + node_hours = "urn:doe-iri:allocation:compute:node-hours" + bytes = "urn:doe-iri:allocation:storage:bytes" + inodes = "urn:doe-iri:allocation:storage:inodes" + + +class CompressionType(str, Enum): + """Canonical DOE IRI compression URNs (spec §3.3).""" + none = "urn:doe-iri:compression:none" + bzip2 = "urn:doe-iri:compression:bzip2" + gzip = "urn:doe-iri:compression:gzip" + xz = "urn:doe-iri:compression:xz" + + +# ----------------------------------------------------------------------- +# Pydantic annotated field types + +# ResourceTypeValue accepts any valid DOE IRI URN. +# No domain constraint: `service` lives in the `service` domain (spec §3.1), +# and facilities may use their own domains for local extensions (spec §5). +ResourceTypeValue = Annotated[ + str, + BeforeValidator(validate_doe_iri_urn), + WithJsonSchema({ + "type": "string", + "description": "DOE IRI resource type URN (urn:doe-iri::). Facility-local extensions accepted.", + "examples": [ResourceType.compute, ResourceType.storage, ResourceType.service], + }), +] + +AllocationUnitValue = Annotated[ + str, + BeforeValidator(lambda v: _validate_urn_domain(v, "allocation", "allocation unit")), + WithJsonSchema( + _domain_urn_schema( + "allocation", + "DOE IRI allocation-unit URN.", + [AllocationUnit.node_hours, AllocationUnit.bytes], + ) + ), +] + +CompressionTypeValue = Annotated[ + str, + BeforeValidator(lambda v: _validate_urn_domain(v, "compression", "compression type")), + WithJsonSchema( + _domain_urn_schema( + "compression", + "DOE IRI compression URN.", + [CompressionType.gzip, CompressionType.none], + ) + ), +] diff --git a/pyproject.toml b/pyproject.toml index 4e34d7e4..49f51e80 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,6 +12,9 @@ dependencies = [ "globus-sdk>=4.3.1", "typer>=0.24.1", ] +[project.optional-dependencies] +dev = ["pytest>=9"] + [tool.ruff] line-length = 200 exclude = [".venv", "__pycache__", "build", "dist"] diff --git a/test/test_urn_types.py b/test/test_urn_types.py new file mode 100644 index 00000000..51926eb7 --- /dev/null +++ b/test/test_urn_types.py @@ -0,0 +1,229 @@ +#!/usr/bin/env python3 +"""DOE IRI URN type regression tests.""" + +import unittest + +from pydantic import TypeAdapter, ValidationError + +from app.routers.filesystem import models as filesystem_models +from app.routers.status import models as status_models +from app.types.scalars import ( + AllocationUnit, + AllocationUnitValue, + CompressionType, + CompressionTypeValue, + ResourceType, + ResourceTypeValue, + validate_doe_iri_urn, + urn_has_complete_prefix, +) + + +def _resource(**kw): + defaults = dict( + id="r-1", + site_id="site-1", + capability_ids=[], + name="R", + description="desc", + last_modified="2026-05-12T12:00:00Z", + current_status=status_models.Status.up, + ) + return status_models.Resource(**(defaults | kw)) + + +class EnumBehaviorTests(unittest.TestCase): + """str(Enum) contracts: value equality, membership, iteration.""" + + def test_resource_type_is_str(self): + self.assertIsInstance(ResourceType.compute, str) + self.assertEqual(ResourceType.compute, "urn:doe-iri:resource:compute") + + def test_allocation_unit_is_str(self): + self.assertIsInstance(AllocationUnit.node_hours, str) + self.assertEqual(AllocationUnit.node_hours, "urn:doe-iri:allocation:compute:node-hours") + + def test_compression_type_is_str(self): + self.assertIsInstance(CompressionType.gzip, str) + self.assertEqual(CompressionType.gzip, "urn:doe-iri:compression:gzip") + + def test_enum_lookup_by_value(self): + self.assertIs(ResourceType("urn:doe-iri:resource:storage"), ResourceType.storage) + self.assertIs(AllocationUnit("urn:doe-iri:allocation:storage:inodes"), AllocationUnit.inodes) + self.assertIs(CompressionType("urn:doe-iri:compression:bzip2"), CompressionType.bzip2) + + def test_resource_type_members(self): + urns = {m.value for m in ResourceType} + self.assertIn("urn:doe-iri:resource:compute", urns) + self.assertIn("urn:doe-iri:resource:storage", urns) + self.assertIn("urn:doe-iri:service:generic", urns) + + def test_service_lives_in_service_domain(self): + """spec §3.1: legacy 'service' enum → urn:doe-iri:service:generic.""" + self.assertEqual(ResourceType.service, "urn:doe-iri:service:generic") + self.assertTrue(ResourceType.service.startswith("urn:doe-iri:service:")) + + def test_all_allocation_units_in_allocation_domain(self): + for member in AllocationUnit: + self.assertTrue(member.value.startswith("urn:doe-iri:allocation:"), member.value) + + def test_all_compression_types_in_compression_domain(self): + for member in CompressionType: + self.assertTrue(member.value.startswith("urn:doe-iri:compression:"), member.value) + + +class UrnValidatorTests(unittest.TestCase): + """validate_doe_iri_urn and urn_has_complete_prefix.""" + + def test_valid_urn_passes(self): + self.assertEqual(validate_doe_iri_urn("urn:doe-iri:resource:compute"), "urn:doe-iri:resource:compute") + + def test_domain_specific_string_allows_rfc8141_slash(self): + self.assertEqual( + validate_doe_iri_urn("urn:doe-iri:resource:facility-code/scanner"), + "urn:doe-iri:resource:facility-code/scanner", + ) + + def test_empty_hierarchy_segments_rejected(self): + for bad in [ + "urn:doe-iri:resource::xrootd", + "urn:doe-iri:resource:storage::xrootd", + "urn:doe-iri:resource:storage:", + "urn:doe-iri:resource::", + ]: + with self.subTest(value=bad): + with self.assertRaises(ValueError): + validate_doe_iri_urn(bad) + + def test_prefix_matching_requires_complete_segments(self): + self.assertFalse( + urn_has_complete_prefix( + "urn:doe-iri:resource:stor", + "urn:doe-iri:resource:storage:filesystem:scratch", + ) + ) + + def test_prefix_matching_exact(self): + self.assertTrue( + urn_has_complete_prefix( + "urn:doe-iri:resource:storage", + "urn:doe-iri:resource:storage", + ) + ) + + def test_prefix_matching_parent(self): + self.assertTrue( + urn_has_complete_prefix( + "urn:doe-iri:resource:storage", + "urn:doe-iri:resource:storage:filesystem:scratch", + ) + ) + + +class ResourceTypeFieldTests(unittest.TestCase): + """ResourceTypeValue: open to any valid DOE IRI URN.""" + + def test_canonical_enum_value_accepted(self): + r = _resource(resource_type=ResourceType.compute) + self.assertEqual(r.resource_type, ResourceType.compute) + + def test_raw_canonical_string_accepted(self): + r = _resource(resource_type="urn:doe-iri:resource:compute") + self.assertEqual(r.resource_type, "urn:doe-iri:resource:compute") + + def test_service_urn_accepted_despite_service_domain(self): + """ResourceTypeValue must accept service domain URNs (spec §3.1 maps service → urn:doe-iri:service:generic).""" + r = _resource(resource_type=ResourceType.service) + self.assertEqual(r.resource_type, "urn:doe-iri:service:generic") + + def test_facility_local_extension_accepted(self): + r = _resource(resource_type="urn:doe-iri:resource:xrootd") + self.assertEqual(r.resource_type, "urn:doe-iri:resource:xrootd") + + def test_short_token_rejected(self): + """Legacy short tokens are no longer accepted.""" + with self.assertRaises(Exception): + _resource(resource_type="compute") + + def test_garbage_rejected(self): + with self.assertRaises(Exception): + _resource(resource_type="not-a-urn") + + def test_prefix_find_matches_subtype(self): + parent = _resource(resource_type="urn:doe-iri:resource:storage:filesystem:scratch") + matches = status_models.Resource.find([parent], resource_type=ResourceType.storage) + self.assertEqual([r.id for r in matches], ["r-1"]) + + def test_prefix_find_unregistered_subtype(self): + r = _resource(resource_type="urn:doe-iri:resource:storage:xrootd") + matches = status_models.Resource.find([r], resource_type=ResourceType.storage) + self.assertEqual([i.id for i in matches], ["r-1"]) + + +class AllocationUnitFieldTests(unittest.TestCase): + """AllocationUnitValue: allocation domain enforced.""" + + def test_canonical_value_accepted(self): + ta = TypeAdapter(AllocationUnitValue) + self.assertEqual(ta.validate_python(AllocationUnit.node_hours), AllocationUnit.node_hours) + + def test_wrong_domain_rejected(self): + ta = TypeAdapter(AllocationUnitValue) + with self.assertRaises((ValueError, ValidationError)): + ta.validate_python(ResourceType.storage) + + def test_short_token_rejected(self): + ta = TypeAdapter(AllocationUnitValue) + with self.assertRaises((ValueError, ValidationError)): + ta.validate_python("node-hours") + + +class CompressionTypeFieldTests(unittest.TestCase): + """CompressionTypeValue: compression domain enforced.""" + + def test_canonical_value_accepted(self): + req = filesystem_models.PostCompressRequest( + path="/tmp/src", + target_path="/tmp/out.tar.gz", + compression=CompressionType.gzip, + ) + self.assertEqual(req.compression, CompressionType.gzip) + + def test_wrong_domain_rejected(self): + with self.assertRaises(Exception): + filesystem_models.PostExtractRequest( + path="/tmp/archive.tar", + target_path="/tmp/out", + compression=ResourceType.storage, + ) + + def test_short_token_rejected(self): + with self.assertRaises(Exception): + filesystem_models.PostCompressRequest( + path="/tmp/src", + target_path="/tmp/out.tar.gz", + compression="gzip", + ) + + +class OpenApiSchemaTests(unittest.TestCase): + """JSON schema hints are emitted correctly.""" + + def test_resource_type_schema(self): + schema = TypeAdapter(ResourceTypeValue).json_schema() + self.assertEqual(schema["type"], "string") + self.assertIn("doe-iri", schema["description"]) + + def test_allocation_unit_schema_has_domain_pattern(self): + schema = TypeAdapter(AllocationUnitValue).json_schema() + self.assertIn("allocation", schema["pattern"]) + self.assertEqual(schema["minLength"], len("urn:doe-iri:allocation:") + 1) + + def test_compression_type_schema_has_domain_pattern(self): + schema = TypeAdapter(CompressionTypeValue).json_schema() + self.assertIn("compression", schema["pattern"]) + self.assertRegex(CompressionType.gzip, schema["pattern"]) + + +if __name__ == "__main__": + unittest.main()