|
1 | | -"""Validate manifests.""" |
| 1 | +"""Validate manifests using StrictYAML.""" |
| 2 | + |
| 3 | +from __future__ import annotations |
2 | 4 |
|
3 | 5 | import logging |
| 6 | +from collections.abc import Mapping |
| 7 | +from typing import Any, cast |
| 8 | + |
| 9 | +from strictyaml import StrictYAMLError, YAMLValidationError, load |
| 10 | + |
| 11 | +from dfetch.manifest.schema import MANIFEST_SCHEMA |
4 | 12 |
|
5 | | -import pykwalify |
6 | | -from pykwalify.core import Core, SchemaError |
7 | | -from yaml.scanner import ScannerError |
8 | 13 |
|
9 | | -import dfetch.resources |
| 14 | +def _ensure_unique(seq: list[dict[str, Any]], key: str, context: str) -> None: |
| 15 | + """Ensure values for `key` are unique within a sequence of dicts.""" |
| 16 | + values = [item.get(key) for item in seq if key in item] |
| 17 | + seen = set() |
| 18 | + dups = {v for v in values if (v in seen) or seen.add(v)} # type: ignore |
| 19 | + if dups: |
| 20 | + dup_list = ", ".join(sorted(map(str, dups))) |
| 21 | + raise RuntimeError(f"Duplicate {context}.{key} value(s): {dup_list}") |
10 | 22 |
|
11 | 23 |
|
12 | 24 | def validate(path: str) -> None: |
13 | | - """Validate the given manifest.""" |
14 | | - logging.getLogger(pykwalify.__name__).setLevel(logging.CRITICAL) |
| 25 | + """Validate the given manifest file against the StrictYAML schema. |
15 | 26 |
|
16 | | - with dfetch.resources.schema_path() as schema_path: |
17 | | - try: |
18 | | - validator = Core(source_file=path, schema_files=[str(schema_path)]) |
19 | | - except ScannerError as err: |
20 | | - raise RuntimeError(f"{schema_path} is not a valid YAML file!") from err |
| 27 | + Raises: |
| 28 | + RuntimeError: if the file is not valid YAML or violates the schema/uniqueness constraints. |
| 29 | + """ |
| 30 | + logging.getLogger("strictyaml").setLevel(logging.CRITICAL) |
21 | 31 |
|
22 | 32 | try: |
23 | | - validator.validate(raise_exception=True) |
24 | | - except SchemaError as err: |
25 | | - raise RuntimeError( |
26 | | - str(err.msg) # pyright: ignore[reportAttributeAccessIssue, reportCallIssue] |
27 | | - ) from err |
| 33 | + with open(path, encoding="utf-8") as f: |
| 34 | + loaded_manifest = load(f.read(), schema=MANIFEST_SCHEMA) |
| 35 | + except YAMLValidationError as err: |
| 36 | + # More specific: schema mismatch (missing required keys, wrong types, unknown keys). |
| 37 | + raise RuntimeError(str(err)) from err |
| 38 | + except StrictYAMLError as err: |
| 39 | + # Broader: parsing errors and strictness violations (e.g., forbidden constructs). |
| 40 | + raise RuntimeError(f"{path} is not a valid YAML file: {err}") from err |
| 41 | + |
| 42 | + data: dict[str, Any] = cast(dict[str, Any], loaded_manifest.data) |
| 43 | + manifest: Mapping[str, Any] = data["manifest"] # required |
| 44 | + projects: list[dict[str, Any]] = manifest["projects"] # required |
| 45 | + remotes: list[dict[str, Any]] = manifest.get("remotes", []) or [] # option |
| 46 | + |
| 47 | + # Enforce cross-item uniqueness constraints originally in pykwalify: |
| 48 | + _ensure_unique(remotes, "name", "manifest.remotes") |
| 49 | + _ensure_unique(projects, "name", "manifest.projects") |
| 50 | + _ensure_unique(projects, "dst", "manifest.projects") # only those that have 'dst' |
0 commit comments