Files
open-cmis-tck/runners/opencmis_tck.py

117 lines
3.2 KiB
Python

#!/usr/bin/env python3
"""OpenCMIS TCK wrapper boundary.
This wrapper intentionally stops before invoking Apache Chemistry. Its current
job is to prove the command-runner contract, verify local Java/Maven posture, and
return structured evidence that the actual TCK execution remains pending.
"""
from __future__ import annotations
import argparse
import json
import shutil
import subprocess
from pathlib import Path
from typing import Any
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--context", required=True)
args = parser.parse_args()
context = _load_context(Path(args.context))
selected_group = context["step"].get("check_group")
dependency_results = {
"java": _probe_command(["java", "-version"]),
"maven": _probe_command(["mvn", "-version"]),
}
missing = [
name
for name, result in dependency_results.items()
if not result["available"]
]
if missing:
_emit(
{
"result": "blocked",
"observations": [
"OpenCMIS TCK execution skipped because required local dependencies are missing: "
+ ", ".join(missing)
+ "."
],
"facts": {
"blocked_reason": "missing_dependency",
"selected_check_group": selected_group,
"dependencies": dependency_results,
},
"artifact_refs": [],
}
)
return 0
_emit(
{
"result": "blocked",
"observations": [
"Java and Maven are available, but the Apache Chemistry OpenCMIS TCK invocation is not configured yet."
],
"facts": {
"blocked_reason": "tck_invocation_not_configured",
"selected_check_group": selected_group,
"dependencies": dependency_results,
"next_step": "Resolve the Maven artifact, classpath, TCK group mapping, and raw artifact capture contract.",
},
"artifact_refs": [],
}
)
return 0
def _load_context(path: Path) -> dict[str, Any]:
with path.open("r", encoding="utf-8") as handle:
value = json.load(handle)
if not isinstance(value, dict):
raise ValueError("context must be a JSON object")
return value
def _probe_command(command: list[str]) -> dict[str, Any]:
executable = shutil.which(command[0])
if executable is None:
return {
"available": False,
"path": None,
"returncode": None,
"version_output": None,
}
completed = subprocess.run(
command,
capture_output=True,
text=True,
timeout=10,
check=False,
)
output = "\n".join(
part.strip()
for part in [completed.stdout, completed.stderr]
if part.strip()
)
return {
"available": completed.returncode == 0,
"path": executable,
"returncode": completed.returncode,
"version_output": output[:2000],
}
def _emit(value: dict[str, Any]) -> None:
print(json.dumps(value, indent=2, sort_keys=True))
if __name__ == "__main__":
raise SystemExit(main())