diff --git a/README.md b/README.md
index c8cde93..6ba2d1c 100644
--- a/README.md
+++ b/README.md
@@ -8,8 +8,28 @@ IVOA registry validator: OAI-PMH harvest checks, IVOA four-GET profile, VOResour
Web UI colors and typography follow [ivoa.net](https://ivoa.net); design tokens live in [`assets/static/css/ivoa-theme.css`](assets/static/css/ivoa-theme.css).
+## Tests
+
+Run the tests locally:
+
+Environment setup:
+
+```bash
+python3 -m venv .venv
+.venv/bin/pip install -e ".[dev]"
+source .venv/bin/activate
+```
+
+Execute tests:
+
+```bash
+pytest
+```
+
## Run
+Environment setup:
+
```bash
python3 -m venv .venv
.venv/bin/pip install -e ".[dev]"
@@ -141,14 +161,6 @@ OAI responses are validated in two steps: embedded `description` / `metadata` /
**Developer guide:** [docs/schemas-and-validation-assets.md](docs/schemas-and-validation-assets.md) — directory layout, bundle composition, namespace table, XSLT assets (`assets/validate/`), standards catalog (`assets/standards/`), and how each validation phase uses them.
-## Tests
-
-After installing with `.[dev]` (see **Run** above):
-
-```bash
-pytest
-```
-
On Debian/Ubuntu, install system libraries for `lxml` if needed: `apt-get install libxml2 libxslt1.1`.
## Docker
diff --git a/assets/static/css/validator.css b/assets/static/css/validator.css
index 707dcf7..376a78a 100644
--- a/assets/static/css/validator.css
+++ b/assets/static/css/validator.css
@@ -239,7 +239,8 @@
}
.val-test { margin: 0.3rem 0; padding: 0.35rem 0.45rem; border-radius: 6px; }
.val-test--pass { color: var(--muted); }
- .val-test--warn { background: var(--warn-bg); }
+ .val-test--warn { background: var(--warn-bg); border-left: 3px solid var(--warn-border); }
+ .val-test--rec { background: var(--accent-soft); border-left: 3px solid var(--accent-border); }
.val-test--fail { background: #fdeeee; border-left: 3px solid #c53030; }
.val-test-item { font-size: 0.85em; margin-right: 0.35rem; }
.val-ignored-note {
diff --git a/assets/validate/README.md b/assets/validate/README.md
index 09b575e..d4a8808 100644
--- a/assets/validate/README.md
+++ b/assets/validate/README.md
@@ -6,6 +6,7 @@ Default **`ASSETS_ROOT`**: XSLT 1.0 stylesheets for IVOA registry rules beyond X
|------|------|
| `checkIVOAOAI.xsl` | IVOA harvest profile tests on OAI GET responses (phase 2) |
| `checkVOResource.xsl` | VOResource constraint tests on harvested records (phase 3) |
+| `testsVOResource-v1_0.xsl` | Imported test templates used by `checkVOResource.xsl` and `checkIVOAOAI.xsl` |
| `validationCommon.xsl` | Shared helpers (imported by the check stylesheets) |
Full context: [`docs/schemas-and-validation-assets.md`](../../docs/schemas-and-validation-assets.md).
diff --git a/assets/validate/testsVOResource-v1_0.xsl b/assets/validate/testsVOResource-v1_0.xsl
new file mode 100644
index 0000000..6068600
--- /dev/null
+++ b/assets/validate/testsVOResource-v1_0.xsl
@@ -0,0 +1,576 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ VR3.1b1
+
+ warn
+
+ VOResource record should have a status attribute
+
+
+
+
+
+
+
+ VR3.1b2
+
+ warn
+
+ VOResource record should have an updated attribute
+
+
+
+
+
+
+
+ VR3.1b3
+
+ warn
+
+ VOResource record should have a created attribute
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ SIA7.0a
+
+ rec
+
+ Recommend setting VOResource xsi:type='vs:CatalogService' on service with SimpleImageAccess capability
+
+
+
+
+
+
+
+
+
+ SIA7.0b
+
+
+ SimpleImageAccess capability must include interface with xsi:type='vs:ParamHTTP' and role='std'
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ CS4.0a
+
+ rec
+
+ Recommend setting VOResource xsi:type='vs:CatalogService' on service with ConeSearch capability
+
+
+
+
+
+
+
+
+
+ CS4.0b
+
+
+ ConeSearch capability must include interface with xsi:type='vs:ParamHTTP' and role='std'
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ RI4.0b
+
+ rec
+
+ Recommend setting VOResource xsi:type='vg:Registry' on service with a Search or Harvest capability
+
+
+
+
+
+
+
+
+
+
+ RI4.0c
+
+
+ Harvest capability must include interface with xsi:type='vg:OAIHTTP' and role='std'
+
+
+
+
+
+
+
+
+
+
+ RI4.0c
+
+
+ Search capability must include interface with xsi:type='vr:Webservice' and role='std'
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ SN4.0a
+
+ rec
+
+ Recommend setting VOResource xsi:type='vs:CatalogService' on service with SkyNode capability
+
+
+
+
+
+
+
+
+
+ SN4.0b
+
+
+ SimpleImageAccess capability must include interface with xsi:type='vr:WebService' and role='std'
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ VR3.2.2a
+
+ warn
+
+ Service resource should have at least one capability element
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ VR3.2.2a
+
+ warn
+
+ capability element should have at least one interface element
+
+
+
+
+
+
+
+
+
+
+
+ created attribute
+
+
+
+
+ updated attribute
+
+
+
+
+
+ date element
+
+
+
+
+
+
+
+ date element
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ -
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ :
+
+
+
+
+
+
+
+
+
+
+ VRdate
+
+
+ The date for the
+
+ must be in the Past
+
+
+
+
+
+
+
+
+ T
+
+
+
+
+
+
+
+
+
+
+
+
+ T
+
+
+
+
+
+
+
+
+ -
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ RI3.1.4b
+
+
+ The Registry must not serve records with
+ authorities it does not claim in its
+ Identify response ("managedAuthority").
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ 4.3.2b
+
+
+ A harvesting registry must support version
+ 1.0 of the OAIHTTP interface.
+
+
+
+
+
diff --git a/docs/schemas-and-validation-assets.md b/docs/schemas-and-validation-assets.md
index e12ca50..c0b2c52 100644
--- a/docs/schemas-and-validation-assets.md
+++ b/docs/schemas-and-validation-assets.md
@@ -66,7 +66,7 @@ Standalone VOR upload validation (`POST /api/v1/registry-validate/voresource`) a
## `assets/schemas/` — bundled XSD files
-All files in this directory are resolved **locally** via [`BundledSchemaResolver`](../src/benson/xml/schema_resolver.py). No network fetch is performed during validation.
+All files in this directory are resolved **locally** via [`SchemaResolver`](../src/benson/xml/schema_resolver.py). No network fetch is performed during validation.
### Bundle schemas (composition roots)
@@ -137,10 +137,9 @@ These stylesheets implement IVOA registry **business rules** that XSD alone does
| `checkIVOAOAI.xsl` | Phase 2 | Profile tests on OAI GET responses (`Identify`, `ListMetadataFormats`, `ListSets`, `ListRecords`). Emits `` elements. |
| `checkVOResource.xsl` | Phase 3 | Additional constraints on harvested VOResource records. Emits ``. |
| `validationCommon.xsl` | (imported) | Shared helpers for the check stylesheets |
+| `testsVOResource-v1_0.xsl` | (imported) | VOResource constraint test templates used by the check stylesheets |
-If XSLT processing fails or the stylesheet is missing, phase 2/3 fall back to simpler pass/fail heuristics (HTTP status and absence of OAI error codes).
-
-**Note:** `checkIVOAOAI.xsl` references `testsVOResource-v1_0.xsl` from the legacy Java tree; that file is **not** shipped under `assets/validate/`. Import failures are caught and the code falls back to non-XSLT checks.
+If XSLT processing fails or the stylesheet is missing, phase 2/3 fall back to simpler pass/fail heuristics (HTTP status and absence of OAI error codes for phase 2; generic `VRvalid` pass after XSD for phase 3).
---
diff --git a/src/benson/oai/phase2.py b/src/benson/oai/phase2.py
index b784def..3ff9435 100644
--- a/src/benson/oai/phase2.py
+++ b/src/benson/oai/phase2.py
@@ -104,7 +104,7 @@ async def build_ivoa_harvest_validation(
timeout: float,
builtin_schemas: bool,
settings: Settings,
-) -> tuple[etree._Element, dict[str, str]]:
+) -> tuple[etree._Element, dict[str, str], dict[str, str]]:
root = R.harvest_validation_root(endpoint.rstrip(), show_status)
check_xsl = settings.assets_root / "checkIVOAOAI.xsl"
@@ -129,6 +129,8 @@ async def build_ivoa_harvest_validation(
violations: list[str] = []
if builtin_schemas:
violations.extend(xsd_validate.validate_oai_response_envelope(raw, settings.schema_root))
+ else:
+ violations.extend(xsd_validate.validate_oai_response_declared(raw, settings.schema_root))
try:
parsed = etree.fromstring(raw, etree.XMLParser(no_network=True, resolve_entities=False))
@@ -151,7 +153,11 @@ async def build_ivoa_harvest_validation(
continue
if check_xsl.is_file():
- xsl_params: dict[str, str] = {"expectError": "false"}
+ xsl_params: dict[str, str] = {
+ "expectError": "false",
+ "showStatus": show_status,
+ "baseurl": endpoint.rstrip(),
+ }
if role == "ListRecords" and identify_state:
xsl_params.update(identify_state)
try:
@@ -171,4 +177,4 @@ async def build_ivoa_harvest_validation(
msg = "OK" if ri_ok else _failure_message(role)
tq.append(R.ri_test(ri_ok, msg))
- return root, registration_defaults
+ return root, registration_defaults, identify_state
diff --git a/src/benson/oai/phase3.py b/src/benson/oai/phase3.py
index 438e2a7..4b5b353 100644
--- a/src/benson/oai/phase3.py
+++ b/src/benson/oai/phase3.py
@@ -3,6 +3,7 @@
from __future__ import annotations
from dataclasses import dataclass, field
+from datetime import UTC, datetime
from urllib.parse import quote
@@ -119,39 +120,87 @@ async def harvest_voresource_documents(
return collected, stats
-def validate_one_voresource(
- blob: bytes,
+def _xsd_errors_for_element(
+ el: etree._Element, # noqa: SLF001
+ *,
builtin_schemas: bool,
- settings: Settings
+ settings: Settings,
) -> list[str]:
- """validates a VOResource record from its XML source in blob.
+ if builtin_schemas:
+ return xsd_validate.validate_element_tree(el, settings.schema_root)
+ return xsd_validate.validate_element_tree_declared(el, settings.schema_root)
- This returns a list of error messages; if this is empty, the document
- is valid by our tests.
- Use for settings, use Settings.from_env().
+def validate_one_voresource(
+ blob: bytes,
+ builtin_schemas: bool,
+ settings: Settings,
+) -> list[str]:
+ """Validate a VOResource record from its XML source in ``blob``.
- If blob cannot be parsed, this will raise an etree.XMLSyntaxError.
+ Returns XSD error messages; an empty list means the document passed schema
+ validation. Raises ``etree.XMLSyntaxError`` when ``blob`` is not well-formed.
"""
el = etree.fromstring(
blob,
- etree.XMLParser(no_network=True, resolve_entities=False))
+ etree.XMLParser(no_network=True, resolve_entities=False),
+ )
+ return _xsd_errors_for_element(el, builtin_schemas=builtin_schemas, settings=settings)
- # Standalone-ish rule: XSD always applies (contract §4).
- # During harvest mimic same when builtinSchemas.
- if builtin_schemas:
- errs = xsd_validate.validate_element_tree(el, settings.schema_root)
- else:
- errs = []
- xsl_path = settings.assets_root / "checkVOResource.xsl"
- if xsl_path.is_file():
- try:
- _ = xslt_eval.transform(xsl_path, el)
- except Exception:
- pass
+def _local_tag(el: etree._Element) -> str: # noqa: SLF001
+ tag = el.tag
+ if not isinstance(tag, str):
+ return ""
+ if tag.startswith("{"):
+ return tag.rpartition("}")[2]
+ return tag
- return errs
+
+def _rightnow_param() -> str:
+ return datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S")
+
+
+def _vor_xslt_params(
+ record_id: str,
+ show_status: str,
+ *,
+ extra: dict[str, str] | None = None,
+) -> dict[str, str]:
+ params = {
+ "showStatus": show_status,
+ "queryName": record_id,
+ "role": "resource",
+ "rightnow": _rightnow_param(),
+ }
+ if extra:
+ params.update(extra)
+ return params
+
+
+def _append_xslt_tests(
+ tq: etree._Element, # noqa: SLF001
+ xout: etree._ElementTree,
+) -> list[etree._Element]: # noqa: SLF001
+ appended: list[etree._Element] = []
+ for child in xout.getroot():
+ if _local_tag(child) != "test":
+ continue
+ tq.append(child)
+ appended.append(child)
+ return appended
+
+
+def _record_stats_for_tests(stats: HarvestStats, tests: list[etree._Element]) -> None:
+ if not tests:
+ return
+ statuses = [(t.get("status") or "pass").lower() for t in tests]
+ if any(s == "fail" for s in statuses):
+ stats.nfail += 1
+ elif any(s == "warn" for s in statuses):
+ stats.nwarn += 1
+ else:
+ stats.npass += 1
def validate_voresource_documents(
@@ -160,9 +209,26 @@ def validate_voresource_documents(
*,
builtin_schemas: bool,
settings: Settings,
+ xsl_params: dict[str, str] | None = None,
) -> tuple[etree._Element, HarvestStats]:
+ """Validate VOResource records from their XML sources in ``records``.
+
+ Note, this does not call validate_one_voresource for each record, to preserve
+ the Element (``el``) in the current scope, and to capture
+ exceptions and add it to the test tree, rather than raising them individually.
+
+ Returns a validation root element and a stats object.
+ The stats object contains the number of failed and passed tests.
+ The validation root element is a tree of test elements, each containing a testQuery element.
+ The testQuery element contains a test element for each test that was run.
+ The test element contains the test name, the test status, and the test message.
+ The test message is a string that contains the error message if the test failed.
+ The test status is a string that contains the test status.
+ The test name is a string that contains the test name.
+ """
root = R.vor_validation_root(show_status)
stats = HarvestStats()
+ xsl_path = settings.assets_root / "checkVOResource.xsl"
for rid, blob in records.items():
tq = etree.SubElement(
@@ -176,14 +242,13 @@ def validate_voresource_documents(
tq.set("ivo-id", rid)
try:
- errs = validate_one_voresource(
- blob,
- builtin_schemas,
- settings)
+ el = etree.fromstring(blob, etree.XMLParser(no_network=True, resolve_entities=False))
except etree.XMLSyntaxError as exc:
- stats.nfail += 1
- tq.append(err_test(str(exc)))
- continue
+ stats.nfail += 1
+ tq.append(err_test(str(exc)))
+ continue
+
+ errs = _xsd_errors_for_element(el, builtin_schemas=builtin_schemas, settings=settings)
if errs:
stats.nfail += 1
@@ -194,6 +259,17 @@ def validate_voresource_documents(
tq.append(bad)
continue
+ if xsl_path.is_file():
+ params = _vor_xslt_params(rid, show_status, extra=xsl_params)
+ try:
+ xout = xslt_eval.transform(xsl_path, el, params=params)
+ tests = _append_xslt_tests(tq, xout)
+ if tests:
+ _record_stats_for_tests(stats, tests)
+ continue
+ except Exception:
+ pass
+
stats.npass += 1
tq.append(pass_test())
diff --git a/src/benson/service/harvest_runner.py b/src/benson/service/harvest_runner.py
index 91ef579..9db6886 100644
--- a/src/benson/service/harvest_runner.py
+++ b/src/benson/service/harvest_runner.py
@@ -54,7 +54,7 @@ async def execute_harvest_validation(
status="running",
phase="ivoa",
)
- ivoa_root, identify_defaults = await phase2.build_ivoa_harvest_validation(
+ ivoa_root, identify_defaults, identify_state = await phase2.build_ivoa_harvest_validation(
client,
run.endpoint,
ss,
@@ -97,6 +97,7 @@ async def execute_harvest_validation(
ss,
builtin_schemas=builtin,
settings=settings,
+ xsl_params=identify_state,
)
phase3.append_harvest_failures(vor_root, harvest_stats.failures)
vor_tree = etree.ElementTree(vor_root)
@@ -140,7 +141,7 @@ async def validate_ivoa_only(
settings: Settings,
client: httpx.AsyncClient,
) -> etree._ElementTree:
- root, _defaults = await phase2.build_ivoa_harvest_validation(
+ root, _defaults, _identify_state = await phase2.build_ivoa_harvest_validation(
client,
run.endpoint,
run.show_status,
@@ -184,10 +185,13 @@ def phase3_validate_only(
show_status: str,
builtin_schemas: bool,
settings: Settings,
+ *,
+ xsl_params: dict[str, str] | None = None,
) -> tuple[etree._Element, phase3.HarvestStats]:
return phase3.validate_voresource_documents(
records,
show_status,
builtin_schemas=builtin_schemas,
settings=settings,
+ xsl_params=xsl_params,
)
diff --git a/src/benson/xml/schema_resolver.py b/src/benson/xml/schema_resolver.py
index 6c3c723..afd81c8 100644
--- a/src/benson/xml/schema_resolver.py
+++ b/src/benson/xml/schema_resolver.py
@@ -38,41 +38,54 @@
OAI_BUNDLE = "benson-oai-bundle.xsd"
-class BundledSchemaResolver(etree.Resolver):
- """Map remote schemaLocation URLs to files under SCHEMA_ROOT."""
+def _build_url_map() -> dict[str, str]:
+ url_map = dict(_IMPORT_URL_TO_FILE)
+ for _ns, fname in NAMESPACE_SCHEMA_FILES.items():
+ url_map.setdefault(_ns.rstrip("/"), fname)
+ url_map.setdefault(_ns, fname)
+ return url_map
+
+
+def resolve_location_url(schema_root: Path, url: str) -> Path | None:
+ """Map a declared schemaLocation URL to a local file under SCHEMA_ROOT."""
+ schema_root = schema_root.resolve()
+ url = (url or "").strip()
+ if not url:
+ return None
+ url_map = _build_url_map()
+ candidates = [url, url.rstrip("/")]
+ if url.startswith("file:"):
+ candidates.append(Path(url[5:]).name)
+ for key in candidates:
+ fname = url_map.get(key)
+ if fname:
+ path = schema_root / fname
+ if path.is_file():
+ return path
+ base = Path(url).name
+ if base:
+ path = schema_root / base
+ if path.is_file():
+ return path
+ return None
+
+
+class SchemaResolver(etree.Resolver):
+ """Map schemaLocation URLs to schema files under SCHEMA_ROOT."""
def __init__(self, schema_root: Path) -> None:
self.schema_root = schema_root.resolve()
- self._url_map = dict(_IMPORT_URL_TO_FILE)
- for _ns, fname in NAMESPACE_SCHEMA_FILES.items():
- self._url_map.setdefault(_ns.rstrip("/"), fname)
- self._url_map.setdefault(_ns, fname)
def resolve(self, system_url: str, _public_id: str, context) -> object | None: # noqa: ANN001
- url = (system_url or "").strip()
- if not url:
- return None
- candidates = [url, url.rstrip("/")]
- if url.startswith("file:"):
- candidates.append(Path(url[5:]).name)
- for key in candidates:
- fname = self._url_map.get(key)
- if fname:
- path = self.schema_root / fname
- if path.is_file():
- return self.resolve_filename(str(path), context)
- # basename fallback (e.g. VOResource-v1.0.xsd)
- base = Path(url).name
- if base:
- path = self.schema_root / base
- if path.is_file():
- return self.resolve_filename(str(path), context)
+ path = resolve_location_url(self.schema_root, system_url)
+ if path is not None:
+ return self.resolve_filename(str(path), context)
return None
def _parser_with_resolver(schema_root: Path) -> etree.XMLParser:
parser = etree.XMLParser()
- parser.resolvers.add(BundledSchemaResolver(schema_root))
+ parser.resolvers.add(SchemaResolver(schema_root))
return parser
@@ -94,6 +107,20 @@ def oai_bundle_schema(schema_root_str: str) -> etree.XMLSchema:
return etree.XMLSchema(etree.parse(str(path), parser))
+@lru_cache(maxsize=32)
+def schema_from_location(schema_root_str: str, location: str) -> etree.XMLSchema | None:
+ """Compile XSD from a declared schemaLocation URL (offline resolution only)."""
+ root = Path(schema_root_str)
+ path = resolve_location_url(root, location)
+ if path is None:
+ return None
+ parser = _parser_with_resolver(root)
+ try:
+ return etree.XMLSchema(etree.parse(str(path), parser))
+ except etree.XMLSchemaParseError:
+ return None
+
+
@lru_cache(maxsize=32)
def namespace_schema(schema_root_str: str, namespace: str) -> etree.XMLSchema | None:
"""Single-namespace schema from the catalog, with bundled import resolution."""
diff --git a/src/benson/xml/xsd_validate.py b/src/benson/xml/xsd_validate.py
index 1c936e9..43326ea 100644
--- a/src/benson/xml/xsd_validate.py
+++ b/src/benson/xml/xsd_validate.py
@@ -13,12 +13,14 @@
ivoa_bundle_schema,
namespace_schema,
oai_bundle_schema,
+ schema_from_location,
)
OAI_NS = "http://www.openarchives.org/OAI/2.0/"
OAI_DC_NS = "http://www.openarchives.org/OAI/2.0/oai_dc/"
DC_NS = "http://purl.org/dc/elements/1.1/"
VOR_NS = "http://www.ivoa.net/xml/VOResource/v1.0"
+XSI_NS = "http://www.w3.org/2001/XMLSchema-instance"
_OAI_FOREIGN_HOLDERS = ("description", "metadata", "about")
@@ -50,6 +52,86 @@ def _validate_tree(el: etree._Element, schema: etree.XMLSchema) -> list[str]: #
return _collect_schema_errors(schema)
+def parse_schema_location(attr: str) -> list[tuple[str, str]] | None:
+ """Split xsi:schemaLocation into namespace/schema URL pairs."""
+ tokens = (attr or "").split()
+ if len(tokens) % 2 != 0:
+ return None
+ return [(tokens[i], tokens[i + 1]) for i in range(0, len(tokens), 2)]
+
+
+def declared_schema_url(el: etree._Element) -> str | None: # noqa: SLF001
+ """Return the schema URL paired with the element namespace in xsi:schemaLocation."""
+ ns = _element_namespace(el)
+ if not ns:
+ return None
+ loc_attr = el.get(f"{{{XSI_NS}}}schemaLocation")
+ if not loc_attr:
+ return None
+ pairs = parse_schema_location(loc_attr)
+ if pairs is None:
+ return None
+ ns_norm = ns.rstrip("/")
+ for pair_ns, url in pairs:
+ if pair_ns.rstrip("/") == ns_norm:
+ return url
+ return None
+
+
+def validate_element_tree_declared(el: etree._Element, schema_root: Path) -> list[str]: # noqa: SLF001
+ """Validate a subtree using the schema declared in xsi:schemaLocation."""
+ ns = _element_namespace(el)
+ if not ns:
+ return ["Element has no target namespace"]
+
+ loc_attr = el.get(f"{{{XSI_NS}}}schemaLocation")
+ if not loc_attr:
+ return ["Missing xsi:schemaLocation (declared-schema mode)"]
+
+ pairs = parse_schema_location(loc_attr)
+ if pairs is None:
+ return ["Malformed xsi:schemaLocation (odd number of tokens)"]
+
+ schema_url = declared_schema_url(el)
+ if schema_url is None:
+ return [f"xsi:schemaLocation has no schema for namespace {ns}"]
+
+ root_key = str(schema_root.resolve())
+ schema = schema_from_location(root_key, schema_url)
+ if schema is None:
+ return [f"Cannot resolve declared schema location {schema_url!r}"]
+
+ return _validate_tree(el, schema)
+
+
+def _validate_oai_embedded_payloads_declared(root: etree._Element, schema_root: Path) -> list[str]: # noqa: SLF001
+ errors: list[str] = []
+ for el in iter_oai_foreign_elements(root):
+ ns = _element_namespace(el)
+ if not ns:
+ errors.append(f"Foreign OAI payload {el.tag!r} has no namespace")
+ continue
+ errors.extend(validate_element_tree_declared(el, schema_root))
+ return errors
+
+
+def _validate_oai_envelope_shell_declared(root: etree._Element, schema_root: Path) -> list[str]: # noqa: SLF001
+ schema_url = declared_schema_url(root)
+ if schema_url is None:
+ ns = _element_namespace(root) or OAI_NS
+ if not root.get(f"{{{XSI_NS}}}schemaLocation"):
+ return ["Missing xsi:schemaLocation (declared-schema mode)"]
+ return [f"xsi:schemaLocation has no schema for namespace {ns}"]
+
+ root_key = str(schema_root.resolve())
+ schema = schema_from_location(root_key, schema_url)
+ if schema is None:
+ return [f"Cannot resolve declared schema location {schema_url!r}"]
+
+ stubbed = _stub_oai_foreign_elements(root)
+ return _validate_tree(stubbed, schema)
+
+
def iter_oai_foreign_elements(oai_root: etree._Element) -> list[etree._Element]: # noqa: SLF001
"""Direct element children under oai:description, oai:metadata, and oai:about."""
out: list[etree._Element] = []
@@ -117,6 +199,25 @@ def validate_oai_response_envelope(xml_bytes: bytes, schema_root: Path) -> list[
return errors
+def validate_oai_response_declared(xml_bytes: bytes, schema_root: Path) -> list[str]:
+ """Validate OAI-PMH 2.0 response using declared xsi:schemaLocation on each part."""
+ parser = etree.XMLParser(no_network=True, resolve_entities=False, huge_tree=True)
+ try:
+ root = etree.fromstring(xml_bytes, parser)
+ except etree.XMLSyntaxError as e:
+ return [str(e)]
+ ns = root.nsmap.get(None) or ""
+ if root.tag.startswith("{"):
+ ns = root.tag[1 : root.tag.index("}")]
+ if ns != OAI_NS:
+ return [f"Expected OAI-PMH root, got namespace {ns!r}"]
+
+ errors: list[str] = []
+ errors.extend(_validate_oai_embedded_payloads_declared(root, schema_root))
+ errors.extend(_validate_oai_envelope_shell_declared(root, schema_root))
+ return errors
+
+
def validate_full_doc(xml_bytes: bytes, schema_root: Path) -> list[str]:
"""Validate a full XML document with the appropriate bundled schema (OAI or VOR)."""
parser = etree.XMLParser(no_network=True, resolve_entities=False, huge_tree=True)
diff --git a/tests/test_phase2_ivoa_xslt.py b/tests/test_phase2_ivoa_xslt.py
new file mode 100644
index 0000000..6778e4a
--- /dev/null
+++ b/tests/test_phase2_ivoa_xslt.py
@@ -0,0 +1,112 @@
+"""Phase 2 IVOA XSLT parameter wiring."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from lxml import etree
+
+from benson.config import Settings
+from benson.http.validation_report import render_validation_report
+from benson.xml import xslt_eval
+
+_repo = Path(__file__).resolve().parents[1]
+_CADC_IDENTIFY = (_repo / "tests" / "fixtures" / "cadc_identify.xml").read_bytes()
+_ENDPOINT = "https://ws.cadc-ccda.hia-iha.nrc-cnrc.gc.ca/reg/oai?"
+
+
+def _ivoa_xslt_params(*, show_status: str) -> dict[str, str]:
+ return {
+ "expectError": "false",
+ "showStatus": show_status,
+ "baseurl": _ENDPOINT.rstrip(),
+ }
+
+
+def _identify_tests(show_status: str) -> list[etree._Element]:
+ settings = Settings.from_env()
+ parsed = etree.fromstring(_CADC_IDENTIFY)
+ xsl = settings.assets_root / "checkIVOAOAI.xsl"
+ xout = xslt_eval.transform(xsl, parsed, params=_ivoa_xslt_params(show_status=show_status))
+ return [
+ el
+ for el in xout.getroot().iter()
+ if etree.QName(el).localname == "test"
+ and (el.get("item") or "").startswith("RI3.1.5")
+ ]
+
+
+def test_ivoa_xslt_hides_passing_checks_when_show_omits_pass() -> None:
+ tests = _identify_tests("fail warn rec")
+ assert tests == []
+
+
+def test_ivoa_xslt_marks_failures_and_warnings_with_expected_status() -> None:
+ xml = b"""
+
+ 2026-01-01T00:00:00Z
+ https://example/oai
+
+ Test
+ https://example/oai
+ 2.0
+ a@b.c
+
+
+ T
+ t
+ ivo://example/reg
+ PNa@b.c
+ sdhttp://x
+
+
+
+"""
+ settings = Settings.from_env()
+ xsl = settings.assets_root / "checkIVOAOAI.xsl"
+ xout = xslt_eval.transform(
+ xsl,
+ etree.fromstring(xml),
+ params=_ivoa_xslt_params(show_status="fail warn rec"),
+ )
+ by_item = {
+ el.get("item"): el.get("status")
+ for el in xout.getroot().iter()
+ if etree.QName(el).localname == "test"
+ }
+ assert by_item["RI3.1.5b4"] == "warn"
+ assert "RI3.1.5b1" not in by_item
+
+
+def test_render_identify_does_not_style_passing_profile_checks_as_pass() -> None:
+ tests = _identify_tests("fail warn rec")
+ tq = etree.Element(
+ "testQuery",
+ name="Identify",
+ options="verb=Identify",
+ role="Identify",
+ )
+ for test in tests:
+ tq.append(test)
+ root = etree.Element("HarvestValidation", baseURL=_ENDPOINT, showStatus="fail warn rec")
+ root.append(tq)
+ html = render_validation_report(etree.ElementTree(root))
+ assert "val-test--pass" not in html
+ assert "RI3.1.5b1" not in html
+
+
+def test_render_identify_shows_warning_styling_for_profile_warnings() -> None:
+ xml = b"""
+
+
+ A Harvesting Registry should declare a Harvest capability.
+
+"""
+ html = render_validation_report(etree.ElementTree(etree.fromstring(xml)))
+ assert "val-test--warn" in html
+ assert "val-test--pass" not in html
+ assert "Harvest capability" in html
diff --git a/tests/test_xsd_validate.py b/tests/test_xsd_validate.py
index f67344a..36c7bcc 100644
--- a/tests/test_xsd_validate.py
+++ b/tests/test_xsd_validate.py
@@ -6,13 +6,19 @@
from pathlib import Path
import pytest
+from lxml import etree
_repo = Path(__file__).resolve().parents[1]
os.environ.setdefault("SCHEMA_ROOT", str(_repo / "assets" / "schemas"))
+from benson.oai.phase3 import validate_voresource_documents # noqa: E402
+from benson.config import Settings # noqa: E402
from benson.xml import xsd_validate # noqa: E402
_SCHEMA_ROOT = Path(os.environ["SCHEMA_ROOT"])
+_VOR_NS = "http://www.ivoa.net/xml/VOResource/v1.0"
+_RI_NS = "http://www.ivoa.net/xml/RegistryInterface/v1.0"
+_XSI_NS = "http://www.w3.org/2001/XMLSchema-instance"
@pytest.fixture
@@ -26,11 +32,116 @@ def test_cadc_identify_envelope_passes_builtin_catalog(cadc_identify_bytes: byte
def test_cadc_identify_resource_fragment_passes_ivoa_bundle(cadc_identify_bytes: bytes) -> None:
- from lxml import etree
-
root = etree.fromstring(cadc_identify_bytes)
- ri_ns = "http://www.ivoa.net/xml/RegistryInterface/v1.0"
- resource = root.find(f".//{{{ri_ns}}}Resource")
+ resource = root.find(f".//{{{_RI_NS}}}Resource")
assert resource is not None
errors = xsd_validate.validate_element_tree(resource, _SCHEMA_ROOT)
assert errors == [], errors
+
+
+def test_parse_schema_location_valid() -> None:
+ attr = (
+ "http://www.ivoa.net/xml/VOResource/v1.0 http://www.ivoa.net/xml/VOResource/v1.0 "
+ "http://www.ivoa.net/xml/RegistryInterface/v1.0 http://www.ivoa.net/xml/RegistryInterface/v1.0"
+ )
+ pairs = xsd_validate.parse_schema_location(attr)
+ assert pairs == [
+ ("http://www.ivoa.net/xml/VOResource/v1.0", "http://www.ivoa.net/xml/VOResource/v1.0"),
+ (
+ "http://www.ivoa.net/xml/RegistryInterface/v1.0",
+ "http://www.ivoa.net/xml/RegistryInterface/v1.0",
+ ),
+ ]
+
+
+def test_parse_schema_location_malformed() -> None:
+ assert xsd_validate.parse_schema_location("one two three") is None
+
+
+def test_validate_element_tree_declared_passes_minimal_resource() -> None:
+ xml = f"""
+ Test
+ t
+ ivo://example/test
+ ExampleNa@b.c
+ TestDeschttp://example.org
+"""
+ el = etree.fromstring(xml)
+ errors = xsd_validate.validate_element_tree_declared(el, _SCHEMA_ROOT)
+ assert errors == [], errors
+
+
+def test_validate_element_tree_declared_rejects_extension_without_bundle() -> None:
+ adql = (_repo / "assets" / "standards" / "adql.xml").read_bytes()
+ el = etree.fromstring(adql)
+ errors = xsd_validate.validate_element_tree_declared(el, _SCHEMA_ROOT)
+ assert errors
+
+
+def test_validate_element_tree_declared_missing_schema_location() -> None:
+ el = etree.fromstring(f'')
+ errors = xsd_validate.validate_element_tree_declared(el, _SCHEMA_ROOT)
+ assert errors
+ assert "Missing xsi:schemaLocation" in errors[0]
+
+
+def test_validate_element_tree_declared_unresolvable_url() -> None:
+ el = etree.fromstring(
+ f''
+ )
+ errors = xsd_validate.validate_element_tree_declared(el, _SCHEMA_ROOT)
+ assert errors
+ assert "Cannot resolve declared schema location" in errors[0]
+
+
+def test_declared_vs_bundled_cadc_resource(cadc_identify_bytes: bytes) -> None:
+ root = etree.fromstring(cadc_identify_bytes)
+ resource = root.find(f".//{{{_RI_NS}}}Resource")
+ assert resource is not None
+ assert xsd_validate.validate_element_tree(resource, _SCHEMA_ROOT) == []
+ declared = xsd_validate.validate_element_tree_declared(resource, _SCHEMA_ROOT)
+ assert declared
+ assert "Missing xsi:schemaLocation" in declared[0]
+
+
+def test_validate_oai_response_declared_cadc(cadc_identify_bytes: bytes) -> None:
+ errors = xsd_validate.validate_oai_response_declared(cadc_identify_bytes, _SCHEMA_ROOT)
+ assert errors
+ assert any("Missing xsi:schemaLocation" in e for e in errors)
+
+
+_VALID_DECLARED_RESOURCE = f"""
+ Test
+ t
+ ivo://example/test
+ ExampleNa@b.c
+ TestDeschttp://example.org
+""".encode()
+
+
+def test_validate_voresource_declared_mode_pass_and_fail() -> None:
+ settings = Settings.from_env()
+ bare = {"ivo://example/bare": f''.encode()}
+ declared = {"ivo://example/ok": _VALID_DECLARED_RESOURCE}
+
+ vor_root_bare, stats_bare = validate_voresource_documents(
+ bare, "fail warn rec", builtin_schemas=False, settings=settings
+ )
+ assert stats_bare.nfail == 1
+ bare_tests = vor_root_bare.findall(".//test")
+ assert bare_tests
+ assert any(t.get("status") == "fail" for t in bare_tests)
+
+ vor_root_ok, stats_ok = validate_voresource_documents(
+ declared, "fail warn rec pass", builtin_schemas=False, settings=settings
+ )
+ assert stats_ok.npass == 1
+ assert stats_ok.nfail == 0
+ ok_tests = [t for t in vor_root_ok.findall(".//test") if t.get("item")]
+ assert len(ok_tests) > 1
+ assert not any(t.get("status") == "fail" for t in ok_tests)