audit_starter_content Namespace Reference

Functions
subprocess.CompletedProcess[str]	run_cli (list[str] args, Path cwd=REPO_ROOT)
	Run one CLI command used to validate a shipped starter artifact.
None	fail (str context, subprocess.CompletedProcess[str] result)
	Raise a compact error that preserves the relevant CLI output.
None	validate_bundle (dict[str, str] bundle, str label)
	Validate one declared case/solver/monitor/post or cluster/study composition.
Optional[str]	_yaml_role (Path path)
	Classify a starter YAML by the same stable top-level shapes used by init.
None	audit_template_copy (str template_name, Path temporary_root)
	Initialize one declared template and verify canonical relocation preserves its content.
int	main ()
	Run the starter-content inventory, composition, and initializer audit.

Variables
	REPO_ROOT = Path(__file__).resolve().parents[2]
str	CONTRACT_PATH = REPO_ROOT / "tests" / "tooling" / "starter_content_contract.json"
str	PICURV = REPO_ROOT / "picurv_cli" / "picurv"
str	RUNTIME_EXECUTION_EXAMPLE = "execution.example.yml"
	file

Detailed Description

Certify that shipped starter templates and reusable configuration profiles stay usable.

Function Documentation

run_cli()

subprocess.CompletedProcess[str] audit_starter_content.run_cli	(	list[str]	args,
		Path	cwd = REPO_ROOT
	)

Run one CLI command used to validate a shipped starter artifact.

Parameters

[in]	args	CLI argument list.
[in]	cwd	Working directory for the command.

Returns

Completed CLI process.

Definition at line 22 of file audit_starter_content.py.

def run_cli(args: list[str], cwd: Path = REPO_ROOT) -> subprocess.CompletedProcess[str]:
    """!
    @brief Run one CLI command used to validate a shipped starter artifact.
    @param[in] args CLI argument list.
    @param[in] cwd Working directory for the command.
    @return Completed CLI process.
    """
    return subprocess.run(
        [sys.executable, str(PICURV), *args], cwd=cwd, text=True, capture_output=True, timeout=90, check=False
    )
 
 

Here is the caller graph for this function:

fail()

None audit_starter_content.fail	(	str	context,
		subprocess.CompletedProcess[str]	result
	)

Raise a compact error that preserves the relevant CLI output.

Parameters

[in]	context	Human-readable operation description.
[in]	result	Completed failing CLI process.

Returns

None.

Definition at line 34 of file audit_starter_content.py.

def fail(context: str, result: subprocess.CompletedProcess[str]) -> None:
    """!
    @brief Raise a compact error that preserves the relevant CLI output.
    @param[in] context Human-readable operation description.
    @param[in] result Completed failing CLI process.
    @return None.
    """
    raise RuntimeError(f"{context} failed:\n{result.stdout}\n{result.stderr}")
 
 

Here is the caller graph for this function:

validate_bundle()

None audit_starter_content.validate_bundle	(	dict[str, str]	bundle,
		str	label
	)

Validate one declared case/solver/monitor/post or cluster/study composition.

Parameters

[in]	bundle	Declared role-to-path mapping.
[in]	label	Human-readable bundle description.

Returns

None.

Definition at line 44 of file audit_starter_content.py.

def validate_bundle(bundle: dict[str, str], label: str) -> None:
    """!
    @brief Validate one declared case/solver/monitor/post or cluster/study composition.
    @param[in] bundle Declared role-to-path mapping.
    @param[in] label Human-readable bundle description.
    @return None.
    """
    args = ["validate"]
    for role, path in bundle.items():
        args.extend([f"--{role}", str(REPO_ROOT / path)])
    result = run_cli(args)
    if result.returncode:
        fail(label, result)
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

_yaml_role()

Optional[str] audit_starter_content._yaml_role ( Path  path )

protected

Classify a starter YAML by the same stable top-level shapes used by init.

Parameters

[in]

path

Source YAML path.

Returns

Canonical configuration role, or None for unclassified YAML.

Definition at line 59 of file audit_starter_content.py.

def _yaml_role(path: Path) -> Optional[str]:
    """!
    @brief Classify a starter YAML by the same stable top-level shapes used by init.
    @param[in] path Source YAML path.
    @return Canonical configuration role, or None for unclassified YAML.
    """
    payload = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
    if not isinstance(payload, dict):
        return None
    keys = set(payload)
    if {"grid", "properties", "run_control"} <= keys:
        return "case"
    if "base_configs" in keys and ("study_type" in keys or "parameters" in keys or "parameter_sets" in keys):
        return "study"
    if "scheduler" in keys and "resources" in keys:
        return "cluster"
    if "source_data" in keys or "eulerian_pipeline" in keys or "lagrangian_pipeline" in keys:
        return "post"
    if "io" in keys and ("logging" in keys or "profiling" in keys or "diagnostics" in keys):
        return "monitor"
    if "momentum_solver" in keys or "poisson_solver" in keys or "operation_mode" in keys:
        return "solver"
    return None
 
 

Here is the caller graph for this function:

audit_template_copy()

None audit_starter_content.audit_template_copy	(	str	template_name,
		Path	temporary_root
	)

Initialize one declared template and verify canonical relocation preserves its content.

Parameters

[in]	template_name	Top-level example directory name.
[in]	temporary_root	Temporary parent directory for initialized cases.

Returns

None.

Definition at line 84 of file audit_starter_content.py.

def audit_template_copy(template_name: str, temporary_root: Path) -> None:
    """!
    @brief Initialize one declared template and verify canonical relocation preserves its content.
    @param[in] template_name Top-level example directory name.
    @param[in] temporary_root Temporary parent directory for initialized cases.
    @return None.
    """
    source = REPO_ROOT / "examples" / template_name
    destination = temporary_root / template_name
    result = run_cli(["init", template_name, "--dest", str(destination)])
    if result.returncode:
        fail(f"picurv init {template_name}", result)
    required_dirs = (
        "config", "config/studies", "inputs", "assets/objects", "assets/sets", "runs", "studies"
    )
    for relative in required_dirs:
        if not (destination / relative).is_dir():
            raise RuntimeError(f"picurv init {template_name} omitted workspace directory {relative}")
    if not (destination / ".picurv-workspace.yml").is_file():
        raise RuntimeError(f"picurv init {template_name} omitted .picurv-workspace.yml")
 
    source_yamls = [path for path in source.rglob("*.yml") if path.name != RUNTIME_EXECUTION_EXAMPLE]
    initialized_yamls = list((destination / "config").rglob("*.yml"))
    if len(initialized_yamls) < len(source_yamls):
        raise RuntimeError(f"picurv init {template_name} lost one or more YAML configurations")
 
    for source_path in source.rglob("*"):
        if not source_path.is_file():
            continue
        relative = source_path.relative_to(source)
        if relative.as_posix() == RUNTIME_EXECUTION_EXAMPLE:
            continue
        if source_path.suffix.lower() in {".yml", ".yaml"}:
            role = _yaml_role(source_path)
            stem = source_path.stem
            candidates = [
                destination / "config" / source_path.name,
                destination / "config" / f"{role}.yml" if role else destination / source_path.name,
                destination / "config" / f"{role}-{stem}{source_path.suffix}" if role else destination / source_path.name,
                destination / "config" / "studies" / source_path.name,
            ]
            if not any(candidate.is_file() for candidate in candidates):
                raise RuntimeError(f"picurv init {template_name} did not relocate YAML {relative}")
            continue
        matches = [path for path in destination.rglob(source_path.name) if path.is_file()]
        if not any(path.read_bytes() == source_path.read_bytes() for path in matches):
            raise RuntimeError(f"picurv init {template_name} did not preserve {relative}")
    if (destination / RUNTIME_EXECUTION_EXAMPLE).exists():
        raise RuntimeError(f"picurv init {template_name} copied site-specific {RUNTIME_EXECUTION_EXAMPLE}")
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

main()

int audit_starter_content.main

(

)

Run the starter-content inventory, composition, and initializer audit.

Returns

Process status code.

Definition at line 135 of file audit_starter_content.py.

def main() -> int:
    """!
    @brief Run the starter-content inventory, composition, and initializer audit.
    @return Process status code.
    """
    contract = json.loads(CONTRACT_PATH.read_text(encoding="utf-8"))
    actual_templates = sorted(path.name for path in (REPO_ROOT / "examples").iterdir() if path.is_dir())
    if actual_templates != sorted(contract["template_directories"]):
        raise RuntimeError(f"Template inventory differs from contract: actual={actual_templates}")
    unknown_reference_templates = set(contract["reference_only_templates"]) - set(actual_templates)
    if unknown_reference_templates:
        raise RuntimeError(f"Reference-only templates are not in the template inventory: {sorted(unknown_reference_templates)}")
 
    actual_config_assets = sorted(str(path.relative_to(REPO_ROOT)) for path in (REPO_ROOT / "config").rglob("*") if path.is_file())
    if actual_config_assets != sorted(contract["config_assets"]):
        raise RuntimeError("Configuration asset inventory differs from the starter-content contract.")
 
    declared_example_yamls = set(contract["auxiliary_example_yamls"])
    for bundle in contract["case_bundles"] + contract["study_bundles"]:
        declared_example_yamls.update(path for path in bundle.values() if path.startswith("examples/"))
    actual_example_yamls = {str(path.relative_to(REPO_ROOT)) for path in (REPO_ROOT / "examples").rglob("*.yml")}
    if actual_example_yamls != declared_example_yamls:
        raise RuntimeError("Example YAML inventory differs from the declared runnable/reference compositions.")
 
    for bundle in contract["case_bundles"]:
        validate_bundle(bundle, f"case bundle {bundle['case']}")
    for bundle in contract["config_role_bundles"]:
        validate_bundle(bundle, f"config role bundle {bundle['solver']}")
    for bundle in contract["study_bundles"]:
        validate_bundle(bundle, f"study bundle {bundle['study']}")
 
    with tempfile.TemporaryDirectory(prefix="picurv-starter-content-") as temporary_directory:
        temporary_root = Path(temporary_directory)
        for template_name in contract["template_directories"]:
            audit_template_copy(template_name, temporary_root)
 
    print("Starter template, example, and configuration audit passed.")
    return 0
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

Variable Documentation

REPO_ROOT

audit_starter_content.REPO_ROOT = Path(__file__).resolve().parents[2]

Definition at line 16 of file audit_starter_content.py.

CONTRACT_PATH

str audit_starter_content.CONTRACT_PATH = REPO_ROOT / "tests" / "tooling" / "starter_content_contract.json"

Definition at line 17 of file audit_starter_content.py.

PICURV

str audit_starter_content.PICURV = REPO_ROOT / "picurv_cli" / "picurv"

Definition at line 18 of file audit_starter_content.py.

RUNTIME_EXECUTION_EXAMPLE

str audit_starter_content.RUNTIME_EXECUTION_EXAMPLE = "execution.example.yml"

Definition at line 19 of file audit_starter_content.py.

file

audit_starter_content.file

Definition at line 179 of file audit_starter_content.py.