manifest.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/testing/manifest.py
 
"""
Purpose
-------
Parse + validate the spec-10 §9 workspace-manifest ``test`` blocks
(Plan 189). The shipping v0.25 base loader
(``oct.core.workspace.load_workspace``) owns the base schema and
deliberately ignores unknown keys; this module layers the §9 test
configuration on top by re-reading the raw manifest for the additive
keys (``subprojects[].test``, top-level ``default_test_profile``)
without touching the base loader.
 
Responsibilities
----------------
- :func:`load_test_manifest` - load the base workspace via the v0.25
  loader, then parse/validate the test blocks into a
  :class:`TestManifest` of :class:`PackageTestConfig` entries.
- Own the Plan-239 P1 ``max_workers`` key: the per-package cap on test
  workers, validated as a positive int, defaulted to
  :data:`DEFAULT_MAX_WORKERS`. Both the pytest side
  (``oct.testing.parallel``) and the §20 stand-in read this one key.
- Map every invalid-manifest condition to the §12.4 taxonomy as data
  (:class:`oct.testing.errors.OctTestError`): a base-loader rejection
  is the documented ``9101``; a test-config content violation is the
  documented ``9102``; a missing manifest is
  ``WORKSPACE_FILE_NOT_FOUND``. NO ``sys.exit`` here - exits are
  mapped by the future CLI slice.
- Pin the TEST profile vocabulary (``strict``/``smoke``/``full``) -
  distinct from the LINT profiles (``proto``/``compact``/``strict``)
  that live in ``subprojects[].profile`` (the §9 vocabulary hazard).
 
Diagnostics
-----------
Domain: OCT-TESTING
Levels:
    L2: lifecycle (load start / end)
    L3: semantic details (per-package parse results)
    L4: deep tracing
 
Contracts
---------
- ``oct.core.workspace`` is a READ-ONLY dependency: base-schema
  semantics are exactly the v0.25 loader's, witnessed by
  ``tests/tests_testing/test_loader_tolerance_witness.py``.
- ``test_roots`` are SUBPROJECT-ROOT-RELATIVE (§9): absolute entries
  and entries escaping the subproject root are configuration errors.
- An ENABLED test block must declare non-empty ``test_roots`` -
  in workspace mode the manifest roots are authoritative (ADR-T-18),
  so an enabled package without roots is a misdeclaration.
- Existence of declared roots is NOT checked here: stale/empty roots
  are collection-time findings (§10 / §12.4 ``EMPTY_TEST_ROOT`` /
  ``UNDECLARED_TEST_ROOT``), owned by the future collection slice.
- ``max_workers`` refuses rather than degrades: a non-positive or
  non-integer value raises ``9102``, because a silently-defaulted cap
  is exactly the kind of configuration drift a run cannot detect.
 
Dependencies
------------
- oct.core.workspace (read-only: base schema + resolved paths)
- oct.core.diagnostics (_dbg)
- oct.testing.errors (the §12.4 taxonomy as data)
"""
 
from __future__ import annotations
 
import json
from dataclasses import dataclass, field
from pathlib import Path
 
from oct.core.diagnostics import _dbg
from oct.core.workspace import (
    WORKSPACE_MANIFEST,
    Subproject,
    Workspace,
    WorkspaceError,
    load_workspace,
)
from oct.testing.errors import (
    MANIFEST_TEST_BLOCK_INVALID,
    OctTestError,
    WORKSPACE_FILE_NOT_FOUND,
    WORKSPACE_MANIFEST_INVALID,
)
 
 
#: The spec-10 §5 TEST profiles (NOT the lint profiles).
TEST_PROFILES: tuple[str, ...] = ("strict", "smoke", "full")
 
#: §9 per-package config table: profile default is the workspace
#: default, else ``strict``.
DEFAULT_TEST_PROFILE: str = "strict"
 
#: Plan 239 P1: the shipped cap on test workers when a package's test
#: block declares no ``max_workers`` (operator ruling 2026-09-06).
#: ``auto`` resolves to ``min(max_workers, os.cpu_count())``.
DEFAULT_MAX_WORKERS: int = 24
 
 
@dataclass(frozen=True)
class PackageTestConfig:
    """The §9 test configuration of one workspace subproject.
 
    Attributes
    ----------
    name
        Subproject name (manifest ``subprojects[].name``).
    path
        Absolute subproject root, resolved by the base loader.
    enabled
        True only when the entry carries ``test.enabled: true``.
    profile
        Effective TEST profile: the block's ``profile`` when set,
        else the manifest's ``default_test_profile``, else
        ``"strict"``.
    test_roots
        Declared subproject-root-relative test roots, in manifest
        order - authoritative in workspace mode (ADR-T-18).
    max_workers
        Plan 239 P1 cap on parallel test workers for this package:
        the block's ``max_workers`` when set, else
        :data:`DEFAULT_MAX_WORKERS`.
    """
 
    name: str
    path: Path
    enabled: bool
    profile: str
    test_roots: tuple[str, ...] = field(default_factory=tuple)
    max_workers: int = DEFAULT_MAX_WORKERS
 
 
@dataclass(frozen=True)
class TestManifest:
    """The §9 test view over a loaded workspace manifest.
 
    Attributes
    ----------
    workspace
        The base :class:`~oct.core.workspace.Workspace` exactly as the
        v0.25 loader produced it.
    default_test_profile
        Top-level ``default_test_profile``, defaulted to ``strict``.
    packages
        One :class:`PackageTestConfig` per subproject, manifest order
        - including packages with no ``test`` block (disabled).
    """
 
    # Not a pytest collectable despite the Test* name (no annotation,
    # so the dataclass machinery ignores it).
    __test__ = False
 
    workspace: Workspace
    default_test_profile: str = DEFAULT_TEST_PROFILE
    packages: tuple[PackageTestConfig, ...] = field(default_factory=tuple)
 
    @property
    def enabled_packages(self) -> tuple[PackageTestConfig, ...]:
        """The test-enabled packages, manifest order (§9 selection)."""
        return tuple(pkg for pkg in self.packages if pkg.enabled)
 
    @property
    def packages_by_name(self) -> dict[str, PackageTestConfig]:
        """Name → package lookup (regenerated per call; N is small)."""
        return {pkg.name: pkg for pkg in self.packages}
 
 
def load_test_manifest(root: Path) -> TestManifest:
    """Load the workspace manifest at *root* with its §9 test view.
 
    Parameters
    ----------
    root
        Workspace root (the directory containing
        ``.option_c_workspace.json``).
 
    Returns
    -------
    :class:`TestManifest`
 
    Raises
    ------
    OctTestError
        ``WORKSPACE_FILE_NOT_FOUND`` when no manifest exists at
        *root*; ``9101`` when the manifest exists but the v0.25 base
        loader rejects it; ``9102`` when the base schema is valid but
        the §9 test configuration is not.
    """
    func = "load_test_manifest"
    _dbg("OCT-TESTING", func, f"root={root}", 2)
 
    manifest_path = Path(root) / WORKSPACE_MANIFEST
    if not manifest_path.is_file():
        raise OctTestError(
            WORKSPACE_FILE_NOT_FOUND,
            f"no {WORKSPACE_MANIFEST} found at {manifest_path}",
        )
 
    try:
        workspace = load_workspace(root)
    except WorkspaceError as exc:
        raise OctTestError(
            WORKSPACE_MANIFEST_INVALID,
            f"workspace manifest rejected by the base loader: {exc}",
        ) from exc
 
    # Re-read the raw manifest for the additive keys the base loader
    # deliberately ignores (its Workspace carries no unknown keys).
    try:
        data = json.loads(manifest_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise OctTestError(
            WORKSPACE_MANIFEST_INVALID,
            f"failed to re-read workspace manifest {manifest_path}: "
            f"{exc}",
        ) from exc
 
    default_profile = _parse_default_profile(data)
    packages = _parse_packages(data, workspace, default_profile)
 
    _dbg(
        "OCT-TESTING", func,
        f"loaded: {len(packages)} package(s), "
        f"{sum(1 for p in packages if p.enabled)} test-enabled, "
        f"default profile {default_profile!r}",
        3,
    )
    return TestManifest(
        workspace=workspace,
        default_test_profile=default_profile,
        packages=tuple(packages),
    )
 
 
def _parse_default_profile(data: dict) -> str:
    """Validate the top-level ``default_test_profile`` key."""
    raw = data.get("default_test_profile")
    if raw is None:
        return DEFAULT_TEST_PROFILE
    if not isinstance(raw, str) or raw not in TEST_PROFILES:
        raise OctTestError(
            MANIFEST_TEST_BLOCK_INVALID,
            f"'default_test_profile' must be one of "
            f"{'/'.join(TEST_PROFILES)} (TEST profiles, not lint "
            f"profiles) or omitted, got {raw!r}",
        )
    return raw
 
 
def _parse_packages(
    data: dict,
    workspace: Workspace,
    default_profile: str,
) -> list[PackageTestConfig]:
    """Build one :class:`PackageTestConfig` per subproject entry."""
    func = "_parse_packages"
    entries = data.get("subprojects", [])
    if len(entries) != len(workspace.subprojects):
        # The manifest changed between the base load and the re-read.
        raise OctTestError(
            WORKSPACE_MANIFEST_INVALID,
            "workspace manifest changed while it was being loaded "
            f"({len(workspace.subprojects)} subprojects at base load, "
            f"{len(entries)} at test-block read)",
        )
 
    packages: list[PackageTestConfig] = []
    for idx, (entry, subproject) in enumerate(
        zip(entries, workspace.subprojects),
    ):
        block = entry.get("test")
        if block is None:
            packages.append(PackageTestConfig(
                name=subproject.name,
                path=subproject.path,
                enabled=False,
                profile=default_profile,
                test_roots=(),
            ))
            continue
        packages.append(_parse_block(
            idx, subproject, block, default_profile,
        ))
    _dbg("OCT-TESTING", func, f"parsed {len(packages)} package(s)", 4)
    return packages
 
 
def _parse_block(
    idx: int,
    subproject: Subproject,
    block: object,
    default_profile: str,
) -> PackageTestConfig:
    """Validate one ``subprojects[idx].test`` block (§9)."""
    where = f"subprojects[{idx}] ({subproject.name}).test"
 
    if not isinstance(block, dict):
        raise OctTestError(
            MANIFEST_TEST_BLOCK_INVALID,
            f"{where} must be an object, got {type(block).__name__}",
        )
 
    enabled = block.get("enabled")
    if not isinstance(enabled, bool):
        raise OctTestError(
            MANIFEST_TEST_BLOCK_INVALID,
            f"{where}.enabled must be present and a boolean, got "
            f"{enabled!r}",
        )
 
    profile_raw = block.get("profile")
    if profile_raw is None:
        profile = default_profile
    elif isinstance(profile_raw, str) and profile_raw in TEST_PROFILES:
        profile = profile_raw
    else:
        raise OctTestError(
            MANIFEST_TEST_BLOCK_INVALID,
            f"{where}.profile must be one of "
            f"{'/'.join(TEST_PROFILES)} (TEST profiles, not lint "
            f"profiles) or omitted, got {profile_raw!r}",
        )
 
    test_roots = _parse_test_roots(where, subproject, block, enabled)
 
    return PackageTestConfig(
        name=subproject.name,
        path=subproject.path,
        enabled=enabled,
        profile=profile,
        test_roots=test_roots,
        max_workers=_parse_max_workers(where, block),
    )
 
 
def _parse_max_workers(where: str, block: dict) -> int:
    """Validate ``max_workers`` (Plan 239 P1): a positive int.
 
    An absent key takes :data:`DEFAULT_MAX_WORKERS`. A non-positive
    or non-integer value is a configuration error, not a silent
    degrade: a cap of 0 or -1 would otherwise read as "no limit" at
    one call site and "serial" at another. ``True``/``False`` are
    ints in Python and are refused explicitly.
    """
    raw = block.get("max_workers")
    if raw is None:
        return DEFAULT_MAX_WORKERS
    if isinstance(raw, bool) or not isinstance(raw, int) or raw < 1:
        raise OctTestError(
            MANIFEST_TEST_BLOCK_INVALID,
            f"{where}.max_workers must be a positive integer or "
            f"omitted, got {raw!r}",
        )
    return raw
 
 
def _parse_test_roots(
    where: str,
    subproject: Subproject,
    block: dict,
    enabled: bool,
) -> tuple[str, ...]:
    """Validate ``test_roots``: relative, inside the subproject root.
 
    An ENABLED block must declare at least one root (ADR-T-18 makes
    the declared roots authoritative in workspace mode); a disabled
    block may omit the key entirely.
    """
    roots_raw = block.get("test_roots")
    if roots_raw is None:
        if enabled:
            raise OctTestError(
                MANIFEST_TEST_BLOCK_INVALID,
                f"{where}.test_roots must be a non-empty list when "
                f"enabled is true (declared roots are authoritative "
                f"in workspace mode)",
            )
        return ()
 
    if not isinstance(roots_raw, list):
        raise OctTestError(
            MANIFEST_TEST_BLOCK_INVALID,
            f"{where}.test_roots must be a list of strings, got "
            f"{type(roots_raw).__name__}",
        )
    if enabled and not roots_raw:
        raise OctTestError(
            MANIFEST_TEST_BLOCK_INVALID,
            f"{where}.test_roots must be non-empty when enabled is "
            f"true",
        )
 
    roots: list[str] = []
    for ridx, entry in enumerate(roots_raw):
        slot = f"{where}.test_roots[{ridx}]"
        if not isinstance(entry, str) or not entry.strip():
            raise OctTestError(
                MANIFEST_TEST_BLOCK_INVALID,
                f"{slot} must be a non-empty string, got {entry!r}",
            )
        if Path(entry).is_absolute():
            raise OctTestError(
                MANIFEST_TEST_BLOCK_INVALID,
                f"{slot} must be relative to the subproject root, "
                f"got absolute path {entry!r}",
            )
        resolved = (subproject.path / entry).resolve()
        try:
            resolved.relative_to(subproject.path)
        except ValueError:
            raise OctTestError(
                MANIFEST_TEST_BLOCK_INVALID,
                f"{slot} {entry!r} escapes the subproject root "
                f"{subproject.path}",
            ) from None
        roots.append(entry)
    return tuple(roots)
 
 
__all__ = [
    "DEFAULT_MAX_WORKERS",
    "DEFAULT_TEST_PROFILE",
    "PackageTestConfig",
    "TEST_PROFILES",
    "TestManifest",
    "load_test_manifest",
]