#!/usr/bin/env python3 """Adapter for Apache Chemistry OpenCMIS TCK ConsoleRunner.""" from __future__ import annotations import argparse import json import subprocess from datetime import datetime, timezone from pathlib import Path from typing import Any GROUP_CLASSES = { "repository-type": [ "org.apache.chemistry.opencmis.tck.tests.basics.BasicsTestGroup", "org.apache.chemistry.opencmis.tck.tests.types.TypesTestGroup", ], "object-content": [ "org.apache.chemistry.opencmis.tck.tests.crud.CRUDTestGroup", ], "navigation": [ "org.apache.chemistry.opencmis.tck.tests.filing.FilingTestGroup", ], "query-acl-versioning": [ "org.apache.chemistry.opencmis.tck.tests.query.QueryTestGroup", "org.apache.chemistry.opencmis.tck.tests.control.ControlTestGroup", "org.apache.chemistry.opencmis.tck.tests.versioning.VersioningTestGroup", ], "relationships": [ "org.apache.chemistry.opencmis.tck.tests.crud.CRUDTestGroup", ], "change-log": [ "org.apache.chemistry.opencmis.tck.tests.control.ControlTestGroup", ], "extension-gaps": [], } def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--browser-url", required=True) parser.add_argument("--repository-id", required=True) parser.add_argument("--check-group", required=True) parser.add_argument("--artifact-dir", type=Path, required=True) parser.add_argument("--run-dir", type=Path) parser.add_argument("--extension-path", type=Path, default=Path(__file__).resolve().parents[1]) parser.add_argument("--user") parser.add_argument("--password") parser.add_argument("--maven-executable", default="mvn") parser.add_argument("--timeout-seconds", type=int, default=300) parser.add_argument("--dry-run", action="store_true") args = parser.parse_args() result = run_console_adapter(args) print(json.dumps(result, indent=2, sort_keys=True)) return 0 if result["result"] in {"pass", "skipped"} else 1 def run_console_adapter(args: argparse.Namespace) -> dict[str, Any]: artifact_dir = args.artifact_dir.resolve() artifact_dir.mkdir(parents=True, exist_ok=True) session_path = artifact_dir / "session.properties" groups_path = artifact_dir / "groups.txt" invocation_path = artifact_dir / "console-runner-invocation.json" console_stdout = artifact_dir / "console-runner-stdout.txt" console_stderr = artifact_dir / "console-runner-stderr.txt" group_classes = GROUP_CLASSES.get(args.check_group) if group_classes is None: return _result( "blocked", [f"Unsupported OpenCMIS TCK check group: {args.check_group}."], args, [], args.run_dir, artifact_dir, [], ) _write_session_parameters(session_path, args) _write_groups(groups_path, group_classes) command = _maven_command(args, session_path, groups_path) invocation = { "created_at": _now(), "command": command, "check_group": args.check_group, "group_classes": group_classes, "session_parameters": str(session_path), "groups_file": str(groups_path), } invocation_path.write_text(json.dumps(invocation, indent=2, sort_keys=True) + "\n", encoding="utf-8") artifact_refs = [ str(session_path.relative_to(artifact_dir)), str(groups_path.relative_to(artifact_dir)), str(invocation_path.relative_to(artifact_dir)), ] if not group_classes: return _result( "skipped", [f"No OpenCMIS TCK group classes are mapped for {args.check_group}; treated as a known-gap review group."], args, group_classes, args.run_dir, artifact_dir, artifact_refs, ) if args.dry_run: return _result( "skipped", ["Dry run completed; OpenCMIS TCK ConsoleRunner was not invoked."], args, group_classes, args.run_dir, artifact_dir, artifact_refs, ) completed = subprocess.run( command, capture_output=True, text=True, timeout=args.timeout_seconds, check=False, ) console_stdout.write_text(completed.stdout, encoding="utf-8") console_stderr.write_text(completed.stderr, encoding="utf-8") artifact_refs.extend( [ str(console_stdout.relative_to(artifact_dir)), str(console_stderr.relative_to(artifact_dir)), ] ) cases = _cases_from_console_output(completed.stdout) if completed.returncode == 0 and not any(case["status"] == "fail" for case in cases): status = "pass" else: status = "fail" return _result( status, [f"OpenCMIS TCK ConsoleRunner exited with {completed.returncode} for {args.check_group}."], args, group_classes, args.run_dir, artifact_dir, artifact_refs, cases=cases, returncode=completed.returncode, ) def _write_session_parameters(path: Path, args: argparse.Namespace) -> None: values = { "org.apache.chemistry.opencmis.binding.spi.type": "browser", "org.apache.chemistry.opencmis.binding.browser.url": args.browser_url, "org.apache.chemistry.opencmis.session.repository.id": args.repository_id, "org.apache.chemistry.opencmis.binding.browser.succinct": "true", "org.apache.chemistry.opencmis.binding.compression": "true", "org.apache.chemistry.opencmis.binding.cookies": "true", } if args.user is not None: values["org.apache.chemistry.opencmis.user"] = args.user if args.password is not None: values["org.apache.chemistry.opencmis.password"] = args.password path.write_text( "\n".join(f"{key}={value}" for key, value in values.items()) + "\n", encoding="utf-8", ) def _write_groups(path: Path, group_classes: list[str]) -> None: path.write_text("\n".join(group_classes) + ("\n" if group_classes else ""), encoding="utf-8") def _maven_command(args: argparse.Namespace, session_path: Path, groups_path: Path) -> list[str]: pom_path = args.extension_path / "runtime" / "opencmis-tck" / "pom.xml" return [ args.maven_executable, "-q", "-f", str(pom_path), "exec:java", "-Dexec.mainClass=org.apache.chemistry.opencmis.tck.runner.ConsoleRunner", f"-Dexec.args={session_path} {groups_path}", ] def _cases_from_console_output(output: str) -> list[dict[str, str]]: cases = [] for line in output.splitlines(): stripped = line.strip() upper = stripped.upper() if not stripped: continue if "UNEXPECTED_EXCEPTION" in upper: cases.append({"id": stripped[:120], "status": "infrastructure_error", "message": stripped}) elif "FAILURE" in upper: cases.append({"id": stripped[:120], "status": "fail", "message": stripped}) elif "WARNING" in upper: cases.append({"id": stripped[:120], "status": "warning", "message": stripped}) elif "SKIPPED" in upper: cases.append({"id": stripped[:120], "status": "skipped", "message": stripped}) elif "OK" in upper: cases.append({"id": stripped[:120], "status": "pass", "message": stripped}) return cases def _result( status: str, observations: list[str], args: argparse.Namespace, group_classes: list[str], run_dir: Path | None, artifact_dir: Path, artifact_refs: list[str], cases: list[dict[str, str]] | None = None, returncode: int | None = None, ) -> dict[str, Any]: cases = cases or [] counts: dict[str, int] = {} for case in cases: counts[case["status"]] = counts.get(case["status"], 0) + 1 if not counts: counts[status] = 1 return { "result": status, "observations": observations, "tests": cases, "facts": { "adapter": "opencmis-console-runner", "artifact_dir": str(artifact_dir), "check_group": args.check_group, "group_classes": group_classes, "returncode": returncode, "result_counts": counts, }, "artifact_refs": [ _artifact_ref(artifact_dir / ref, run_dir, artifact_dir) for ref in artifact_refs ], } def _artifact_ref(path: Path, run_dir: Path | None, artifact_dir: Path) -> str: resolved = path.resolve() if run_dir is not None: try: return str(resolved.relative_to(run_dir.resolve())) except ValueError: pass return str(resolved.relative_to(artifact_dir.resolve())) def _now() -> str: return datetime.now(timezone.utc).isoformat() if __name__ == "__main__": raise SystemExit(main())