generated from coulomb/repo-seed
990 lines
40 KiB
Python
990 lines
40 KiB
Python
from __future__ import annotations
|
|
|
|
import http.client
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
|
|
from guide_board.discovery import discover_extensions
|
|
from guide_board.execution import run_assessment
|
|
from guide_board.planning import (
|
|
build_run_plan,
|
|
validate_assessment_profile,
|
|
validate_target_profile,
|
|
)
|
|
from guide_board.retention import build_trend_summary
|
|
from guide_board.service import ServiceHandle, start_service
|
|
from open_cmis_tck.bootstrap import TCK_COORDINATE, check_runtime
|
|
from open_cmis_tck.normalization import (
|
|
aggregate_case_result,
|
|
parse_text_report,
|
|
result_counts,
|
|
)
|
|
from open_cmis_tck.profile import validate_cmis_profile_config
|
|
from open_cmis_tck.scorecard import build_scorecard, write_scorecard
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
CORE_ROOT = ROOT.parent / "guide-board"
|
|
|
|
|
|
class OpenCmisTckExtensionTests(unittest.TestCase):
|
|
def test_extension_manifest_discovers_from_repo_root(self) -> None:
|
|
extensions = {
|
|
extension.id: extension
|
|
for extension in discover_extensions(CORE_ROOT, [ROOT])
|
|
}
|
|
|
|
self.assertIn("open-cmis-tck", extensions)
|
|
self.assertEqual(extensions["open-cmis-tck"].source, "external")
|
|
self.assertEqual(extensions["open-cmis-tck"].path, ROOT)
|
|
|
|
def test_builds_cmis_baseline_plan_from_external_extension(self) -> None:
|
|
assessment = validate_assessment_profile(
|
|
ROOT / "profiles" / "assessments" / "cmis-browser-baseline.json"
|
|
)
|
|
plan = build_run_plan(
|
|
CORE_ROOT,
|
|
ROOT / "profiles" / "targets" / "kontextual-cmis-compat.json",
|
|
ROOT / "profiles" / "assessments" / "cmis-browser-baseline.json",
|
|
[ROOT],
|
|
)
|
|
|
|
self.assertEqual(assessment["extension_refs"], ["open-cmis-tck"])
|
|
self.assertEqual(plan["extension_snapshots"][0]["id"], "open-cmis-tck")
|
|
self.assertEqual(plan["extension_snapshots"][0]["source"], "external")
|
|
self.assertEqual(plan["extension_snapshots"][0]["path"], str(ROOT))
|
|
self.assertEqual(len(plan["ordered_steps"]), 3)
|
|
|
|
def test_validates_cmis_profile_config_with_actionable_diagnostics(self) -> None:
|
|
target = json.loads(
|
|
(ROOT / "profiles" / "targets" / "kontextual-cmis-compat.json").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
assessment = json.loads(
|
|
(ROOT / "profiles" / "assessments" / "cmis-browser-baseline.json").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
|
|
diagnostics = validate_cmis_profile_config(target, assessment)
|
|
|
|
self.assertEqual(diagnostics["status"], "valid")
|
|
self.assertEqual(
|
|
diagnostics["cmis_config"]["browser_binding_url"],
|
|
"http://127.0.0.1:8000/cmis/compat-tck/browser",
|
|
)
|
|
self.assertEqual(diagnostics["cmis_config"]["repository_id"], "compat-tck")
|
|
self.assertEqual(diagnostics["cmis_config"]["auth_mode"], "anonymous")
|
|
|
|
authenticated = dict(target)
|
|
authenticated["credentials_ref"] = "env:CMIS_TCK_USER,CMIS_TCK_PASSWORD"
|
|
authenticated_diagnostics = validate_cmis_profile_config(authenticated, assessment)
|
|
self.assertEqual(authenticated_diagnostics["cmis_config"]["auth_mode"], "env")
|
|
|
|
broken = dict(target)
|
|
broken["endpoints"] = []
|
|
broken_diagnostics = validate_cmis_profile_config(broken, assessment)
|
|
self.assertEqual(broken_diagnostics["status"], "invalid")
|
|
self.assertIn(
|
|
"Add one endpoint with binding 'cmis-browser'",
|
|
broken_diagnostics["diagnostics"][0]["message"],
|
|
)
|
|
|
|
def test_target_profile_templates_validate(self) -> None:
|
|
template_dir = ROOT / "profiles" / "targets" / "templates"
|
|
for name in [
|
|
"cmis-browser-anonymous.json",
|
|
"cmis-browser-basic-auth-env.json",
|
|
"cmis-browser-basic-auth-file.json",
|
|
]:
|
|
with self.subTest(name=name):
|
|
profile = validate_target_profile(template_dir / name)
|
|
self.assertEqual(profile["subject_type"], "cmis-browser-binding-endpoint")
|
|
|
|
def test_bootstrap_reports_local_tck_runtime_posture(self) -> None:
|
|
with TemporaryDirectory() as temporary_directory:
|
|
output = Path(temporary_directory) / "runtime-summary.json"
|
|
|
|
summary = check_runtime(ROOT, output, resolve=False)
|
|
|
|
self.assertIn(summary["status"], {"ready", "blocked"})
|
|
self.assertEqual(summary["tck"]["coordinate"], TCK_COORDINATE)
|
|
self.assertEqual(
|
|
summary["tck"]["runner_class"],
|
|
"org.apache.chemistry.opencmis.tck.runner.ConsoleRunner",
|
|
)
|
|
self.assertTrue(output.exists())
|
|
|
|
def test_bootstrap_ready_path_with_fake_local_toolchain(self) -> None:
|
|
with TemporaryDirectory() as temporary_directory:
|
|
temp_root = Path(temporary_directory)
|
|
bin_dir = temp_root / "bin"
|
|
bin_dir.mkdir()
|
|
java = bin_dir / "java"
|
|
maven = bin_dir / "mvn"
|
|
java.write_text("#!/usr/bin/env sh\necho 'openjdk version \"17\"' >&2\n", encoding="utf-8")
|
|
maven.write_text("#!/usr/bin/env sh\necho 'Apache Maven 3.9.0'\n", encoding="utf-8")
|
|
java.chmod(0o755)
|
|
maven.chmod(0o755)
|
|
output = temp_root / "runtime-summary.json"
|
|
original_path = os.environ.get("PATH", "")
|
|
os.environ["PATH"] = f"{bin_dir}{os.pathsep}{original_path}"
|
|
try:
|
|
summary = check_runtime(ROOT, output, resolve=False)
|
|
finally:
|
|
os.environ["PATH"] = original_path
|
|
|
|
self.assertEqual(summary["status"], "ready")
|
|
self.assertTrue(summary["runtime"]["java"]["available"])
|
|
self.assertTrue(summary["runtime"]["maven"]["available"])
|
|
self.assertTrue(output.exists())
|
|
|
|
def test_parses_native_opencmis_text_report_fixture(self) -> None:
|
|
fixture = (ROOT / "tests" / "fixtures" / "opencmis-text-report-sanitized.txt").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
|
|
cases = parse_text_report(
|
|
fixture,
|
|
"repository-type",
|
|
["org.apache.chemistry.opencmis.tck.tests.basics.BasicsTestGroup"],
|
|
)
|
|
counts = result_counts(cases)
|
|
warning = next(case for case in cases if case["status"] == "warning")
|
|
failure = next(case for case in cases if case["status"] == "fail")
|
|
|
|
self.assertEqual(
|
|
counts,
|
|
{
|
|
"fail": 1,
|
|
"infrastructure_error": 1,
|
|
"pass": 2,
|
|
"skipped": 1,
|
|
"warning": 1,
|
|
},
|
|
)
|
|
self.assertEqual(aggregate_case_result(counts, 0), "infrastructure_error")
|
|
self.assertEqual(warning["status_native"], "WARNING")
|
|
self.assertEqual(warning["test_name"], "Repository Info Test")
|
|
self.assertEqual(warning["source_location"], {"file": "SecurityTest.java", "line": 52})
|
|
self.assertEqual(failure["message"], "Test folder could not be created.")
|
|
|
|
def test_console_adapter_dry_run_writes_session_and_group_files(self) -> None:
|
|
with TemporaryDirectory() as temporary_directory:
|
|
temp_root = Path(temporary_directory)
|
|
run_dir = temp_root / "run"
|
|
artifact_dir = run_dir / "artifacts" / "open-cmis-tck" / "tck" / "repository-type"
|
|
completed = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(ROOT / "adapters" / "opencmis_console_adapter.py"),
|
|
"--browser-url",
|
|
"http://127.0.0.1:8000/cmis/browser",
|
|
"--repository-id",
|
|
"local-test-repository",
|
|
"--check-group",
|
|
"repository-type",
|
|
"--artifact-dir",
|
|
str(artifact_dir),
|
|
"--run-dir",
|
|
str(run_dir),
|
|
"--extension-path",
|
|
str(ROOT),
|
|
"--dry-run",
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
result = json.loads(completed.stdout)
|
|
|
|
self.assertEqual(completed.returncode, 0)
|
|
self.assertEqual(result["result"], "skipped")
|
|
self.assertIn(
|
|
"artifacts/open-cmis-tck/tck/repository-type/session.properties.redacted",
|
|
result["artifact_refs"],
|
|
)
|
|
self.assertIn(
|
|
"org.apache.chemistry.opencmis.binding.spi.type=browser",
|
|
(artifact_dir / "session.properties.redacted").read_text(encoding="utf-8"),
|
|
)
|
|
self.assertIn(
|
|
"org.apache.chemistry.opencmis.tck.tests.basics.BasicsTestGroup",
|
|
(artifact_dir / "groups.txt").read_text(encoding="utf-8"),
|
|
)
|
|
self.assertFalse((artifact_dir / "session-private.properties").exists())
|
|
|
|
def test_console_adapter_uses_env_credentials_without_persisting_secret(self) -> None:
|
|
with TemporaryDirectory() as temporary_directory:
|
|
temp_root = Path(temporary_directory)
|
|
run_dir = temp_root / "run"
|
|
artifact_dir = run_dir / "artifacts" / "open-cmis-tck" / "tck" / "repository-type"
|
|
env = {
|
|
**os.environ,
|
|
"CMIS_TEST_USER": "alice",
|
|
"CMIS_TEST_PASSWORD": "secret-password",
|
|
}
|
|
completed = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(ROOT / "adapters" / "opencmis_console_adapter.py"),
|
|
"--browser-url",
|
|
"http://127.0.0.1:8000/cmis/browser",
|
|
"--repository-id",
|
|
"local-test-repository",
|
|
"--check-group",
|
|
"repository-type",
|
|
"--artifact-dir",
|
|
str(artifact_dir),
|
|
"--run-dir",
|
|
str(run_dir),
|
|
"--extension-path",
|
|
str(ROOT),
|
|
"--credentials-ref",
|
|
"env:CMIS_TEST_USER,CMIS_TEST_PASSWORD",
|
|
"--dry-run",
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
env=env,
|
|
)
|
|
result = json.loads(completed.stdout)
|
|
redacted = (artifact_dir / "session.properties.redacted").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
|
|
self.assertEqual(completed.returncode, 0)
|
|
self.assertEqual(result["result"], "skipped")
|
|
self.assertIn("org.apache.chemistry.opencmis.user=alice", redacted)
|
|
self.assertIn("org.apache.chemistry.opencmis.password=<redacted>", redacted)
|
|
self.assertNotIn("secret-password", redacted)
|
|
self.assertFalse((artifact_dir / "session-private.properties").exists())
|
|
|
|
def test_runs_cmis_preflight_against_local_endpoint(self) -> None:
|
|
server = HTTPServer(("127.0.0.1", 0), _CmisHandler)
|
|
thread = threading.Thread(target=server.serve_forever)
|
|
thread.daemon = True
|
|
thread.start()
|
|
try:
|
|
with TemporaryDirectory() as temporary_directory:
|
|
temp_root = Path(temporary_directory)
|
|
target_path = temp_root / "target.json"
|
|
assessment_path = temp_root / "assessment.json"
|
|
_write_target(target_path, server.server_port, "local-cmis-test")
|
|
_write_assessment(
|
|
assessment_path,
|
|
"local-cmis-preflight",
|
|
"local-cmis-test",
|
|
[],
|
|
None,
|
|
)
|
|
|
|
result = run_assessment(
|
|
CORE_ROOT,
|
|
target_path,
|
|
assessment_path,
|
|
temp_root / "run",
|
|
[ROOT],
|
|
)
|
|
run_dir = Path(result["run_dir"])
|
|
evidence = json.loads(
|
|
(run_dir / "normalized" / "evidence.json").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)["evidence"]
|
|
package = json.loads(
|
|
(run_dir / "reports" / "assessment-package.json").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
|
|
self.assertEqual(result["status"], "completed")
|
|
self.assertEqual(evidence[0]["result"], "pass")
|
|
self.assertEqual(
|
|
evidence[0]["facts"]["repository_ids"],
|
|
["local-test-repository"],
|
|
)
|
|
posture = {
|
|
item["requirement_ref"]: item["status"]
|
|
for item in evidence[0]["facts"]["capability_posture"]
|
|
}
|
|
self.assertEqual(posture["cmis.repository-info"], "supported")
|
|
self.assertEqual(len(package["artifact_manifest"]), 2)
|
|
self.assertTrue(
|
|
(
|
|
run_dir
|
|
/ "artifacts"
|
|
/ "open-cmis-tck"
|
|
/ "preflight"
|
|
/ "response-metadata.json"
|
|
).exists()
|
|
)
|
|
finally:
|
|
server.shutdown()
|
|
thread.join(timeout=5)
|
|
server.server_close()
|
|
|
|
def test_preflight_accepts_unsupported_optional_capability_as_known_gap(self) -> None:
|
|
server = HTTPServer(("127.0.0.1", 0), _CmisHandler)
|
|
thread = threading.Thread(target=server.serve_forever)
|
|
thread.daemon = True
|
|
thread.start()
|
|
try:
|
|
with TemporaryDirectory() as temporary_directory:
|
|
temp_root = Path(temporary_directory)
|
|
target_path = temp_root / "target.json"
|
|
assessment_path = temp_root / "assessment.json"
|
|
_write_target(target_path, server.server_port, "local-cmis-query-gap")
|
|
target = json.loads(target_path.read_text(encoding="utf-8"))
|
|
target["declared_capabilities"].append("cmis.query")
|
|
target["known_gaps"].append(
|
|
{
|
|
"id": "query-not-targeted",
|
|
"requirement_refs": ["cmis.query"],
|
|
"reason": "The local fixture deliberately reports no query support.",
|
|
"status": "unsupported_by_design",
|
|
}
|
|
)
|
|
target_path.write_text(json.dumps(target), encoding="utf-8")
|
|
_write_assessment(
|
|
assessment_path,
|
|
"local-cmis-known-gap",
|
|
"local-cmis-query-gap",
|
|
[],
|
|
None,
|
|
)
|
|
|
|
result = run_assessment(
|
|
CORE_ROOT,
|
|
target_path,
|
|
assessment_path,
|
|
temp_root / "run",
|
|
[ROOT],
|
|
)
|
|
evidence = json.loads(
|
|
(Path(result["run_dir"]) / "normalized" / "evidence.json").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)["evidence"]
|
|
posture = {
|
|
item["requirement_ref"]: item["status"]
|
|
for item in evidence[0]["facts"]["capability_posture"]
|
|
}
|
|
|
|
self.assertEqual(result["status"], "completed")
|
|
self.assertEqual(posture["cmis.query"], "expected_gap")
|
|
finally:
|
|
server.shutdown()
|
|
thread.join(timeout=5)
|
|
server.server_close()
|
|
|
|
def test_runs_cmis_tck_command_wrapper_boundary(self) -> None:
|
|
server = HTTPServer(("127.0.0.1", 0), _CmisHandler)
|
|
thread = threading.Thread(target=server.serve_forever)
|
|
thread.daemon = True
|
|
thread.start()
|
|
try:
|
|
with TemporaryDirectory() as temporary_directory:
|
|
temp_root = Path(temporary_directory)
|
|
target_path = temp_root / "target.json"
|
|
assessment_path = temp_root / "assessment.json"
|
|
waiver_path = temp_root / "waivers.json"
|
|
_write_target(target_path, server.server_port, "local-cmis-command-test")
|
|
_write_assessment(
|
|
assessment_path,
|
|
"local-cmis-command-boundary",
|
|
"local-cmis-command-test",
|
|
["repository-type"],
|
|
str(waiver_path),
|
|
)
|
|
_write_command_waiver(waiver_path, "local-cmis-command-test")
|
|
|
|
result = run_assessment(
|
|
CORE_ROOT,
|
|
target_path,
|
|
assessment_path,
|
|
temp_root / "run",
|
|
[ROOT],
|
|
)
|
|
run_dir = Path(result["run_dir"])
|
|
evidence = json.loads(
|
|
(run_dir / "normalized" / "evidence.json").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)["evidence"]
|
|
findings = json.loads(
|
|
(run_dir / "normalized" / "findings.json").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)["findings"]
|
|
mappings = json.loads(
|
|
(run_dir / "normalized" / "mappings.json").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)["mappings"]
|
|
|
|
self.assertEqual(result["status"], "blocked")
|
|
self.assertEqual(evidence[0]["result"], "pass")
|
|
self.assertEqual(evidence[1]["result"], "blocked")
|
|
self.assertEqual(evidence[1]["facts"]["runner_kind"], "command")
|
|
self.assertIn(
|
|
evidence[1]["facts"]["blocked_reason"],
|
|
{"missing_dependency", "tck_invocation_not_configured"},
|
|
)
|
|
self.assertEqual(findings[0]["waiver_ref"], "local-command-wrapper-bootstrap")
|
|
self.assertEqual({mapping["target_id"] for mapping in mappings}, {"repository-type"})
|
|
finally:
|
|
server.shutdown()
|
|
thread.join(timeout=5)
|
|
server.server_close()
|
|
|
|
def test_runs_configured_tck_command_and_normalizes_json_results(self) -> None:
|
|
server = HTTPServer(("127.0.0.1", 0), _CmisHandler)
|
|
thread = threading.Thread(target=server.serve_forever)
|
|
thread.daemon = True
|
|
thread.start()
|
|
try:
|
|
with TemporaryDirectory() as temporary_directory:
|
|
temp_root = Path(temporary_directory)
|
|
target_path = temp_root / "target.json"
|
|
assessment_path = temp_root / "assessment.json"
|
|
fake_tck = temp_root / "fake_tck.py"
|
|
fake_tck.write_text(
|
|
"\n".join(
|
|
[
|
|
"import json",
|
|
"print(json.dumps({",
|
|
" 'tests': [",
|
|
" {'id': 'repository-info', 'status': 'pass'},",
|
|
" {'id': 'type-definitions', 'status': 'pass'}",
|
|
" ]",
|
|
"}))",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
_write_target(target_path, server.server_port, "local-cmis-configured-tck")
|
|
_write_assessment(
|
|
assessment_path,
|
|
"local-cmis-configured-tck",
|
|
"local-cmis-configured-tck",
|
|
["repository-type"],
|
|
None,
|
|
{
|
|
"requires_java_maven": False,
|
|
"repository_id": "local-test-repository",
|
|
"command": [
|
|
sys.executable,
|
|
str(fake_tck),
|
|
"--url",
|
|
"{browser_url}",
|
|
"--repository",
|
|
"{repository_id}",
|
|
"--group",
|
|
"{check_group}",
|
|
],
|
|
},
|
|
)
|
|
|
|
result = run_assessment(
|
|
CORE_ROOT,
|
|
target_path,
|
|
assessment_path,
|
|
temp_root / "run",
|
|
[ROOT],
|
|
)
|
|
run_dir = Path(result["run_dir"])
|
|
evidence = json.loads(
|
|
(run_dir / "normalized" / "evidence.json").read_text(encoding="utf-8")
|
|
)["evidence"]
|
|
retention = json.loads(
|
|
(run_dir / "retention-summary.json").read_text(encoding="utf-8")
|
|
)
|
|
trend = build_trend_summary(temp_root)
|
|
|
|
self.assertEqual(result["status"], "completed")
|
|
self.assertEqual(evidence[1]["result"], "pass")
|
|
self.assertEqual(evidence[1]["facts"]["normalizer"], "json-cases")
|
|
self.assertEqual(evidence[1]["facts"]["result_counts"], {"pass": 2})
|
|
self.assertTrue(
|
|
(
|
|
run_dir
|
|
/ "artifacts"
|
|
/ "open-cmis-tck"
|
|
/ "tck"
|
|
/ "repository-type"
|
|
/ "stdout.log"
|
|
).exists()
|
|
)
|
|
self.assertEqual(retention["summary"]["status"], "completed")
|
|
self.assertGreaterEqual(retention["summary"]["artifact_count"], 4)
|
|
self.assertEqual(trend["run_count"], 1)
|
|
|
|
scorecard = build_scorecard(run_dir)
|
|
self.assertEqual(scorecard["run_id"], result["run_id"])
|
|
groups = {group["id"]: group for group in scorecard["groups"]}
|
|
self.assertEqual(groups["repository-type"]["status"], "demonstrated")
|
|
self.assertEqual(groups["repository-type"]["score"], 4)
|
|
self.assertEqual(groups["object-content"]["status"], "not_assessed")
|
|
written = write_scorecard(run_dir)
|
|
self.assertTrue(Path(written["json"]).exists())
|
|
self.assertTrue(Path(written["markdown"]).exists())
|
|
finally:
|
|
server.shutdown()
|
|
thread.join(timeout=5)
|
|
server.server_close()
|
|
|
|
def test_runs_configured_tck_command_and_normalizes_text_report_results(self) -> None:
|
|
server = HTTPServer(("127.0.0.1", 0), _CmisHandler)
|
|
thread = threading.Thread(target=server.serve_forever)
|
|
thread.daemon = True
|
|
thread.start()
|
|
try:
|
|
with TemporaryDirectory() as temporary_directory:
|
|
temp_root = Path(temporary_directory)
|
|
target_path = temp_root / "target.json"
|
|
assessment_path = temp_root / "assessment.json"
|
|
fake_tck = temp_root / "fake_tck_text.py"
|
|
fixture = ROOT / "tests" / "fixtures" / "opencmis-text-report-sanitized.txt"
|
|
fake_tck.write_text(
|
|
"\n".join(
|
|
[
|
|
"from pathlib import Path",
|
|
f"print(Path({str(fixture)!r}).read_text(encoding='utf-8'))",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
_write_target(target_path, server.server_port, "local-cmis-text-tck")
|
|
_write_assessment(
|
|
assessment_path,
|
|
"local-cmis-text-tck",
|
|
"local-cmis-text-tck",
|
|
["repository-type"],
|
|
None,
|
|
{
|
|
"requires_java_maven": False,
|
|
"repository_id": "local-test-repository",
|
|
"command": [sys.executable, str(fake_tck)],
|
|
},
|
|
)
|
|
|
|
result = run_assessment(
|
|
CORE_ROOT,
|
|
target_path,
|
|
assessment_path,
|
|
temp_root / "run",
|
|
[ROOT],
|
|
)
|
|
run_dir = Path(result["run_dir"])
|
|
evidence = json.loads(
|
|
(run_dir / "normalized" / "evidence.json").read_text(encoding="utf-8")
|
|
)["evidence"]
|
|
cases = evidence[1]["facts"]["cases"]
|
|
|
|
self.assertEqual(result["status"], "infrastructure_error")
|
|
self.assertEqual(evidence[1]["result"], "infrastructure_error")
|
|
self.assertEqual(evidence[1]["facts"]["normalizer"], "opencmis-text-report")
|
|
self.assertEqual(
|
|
evidence[1]["facts"]["result_counts"],
|
|
{
|
|
"fail": 1,
|
|
"infrastructure_error": 1,
|
|
"pass": 2,
|
|
"skipped": 1,
|
|
"warning": 1,
|
|
},
|
|
)
|
|
self.assertEqual(cases[0]["status_native"], "OK")
|
|
self.assertEqual(cases[0]["group_name"], "Basics Test Group")
|
|
self.assertEqual(cases[0]["test_name"], "Repository Info Test")
|
|
self.assertIn(
|
|
"artifacts/open-cmis-tck/tck/repository-type/stdout.log",
|
|
evidence[1]["artifact_refs"],
|
|
)
|
|
finally:
|
|
server.shutdown()
|
|
thread.join(timeout=5)
|
|
server.server_close()
|
|
|
|
def test_guide_board_dry_run_invokes_console_adapter_and_captures_artifacts(self) -> None:
|
|
server = HTTPServer(("127.0.0.1", 0), _CmisHandler)
|
|
thread = threading.Thread(target=server.serve_forever)
|
|
thread.daemon = True
|
|
thread.start()
|
|
try:
|
|
with TemporaryDirectory() as temporary_directory:
|
|
temp_root = Path(temporary_directory)
|
|
target_path = temp_root / "target.json"
|
|
assessment_path = temp_root / "assessment.json"
|
|
_write_target(target_path, server.server_port, "local-cmis-guide-dry-run")
|
|
_write_assessment(
|
|
assessment_path,
|
|
"local-cmis-guide-dry-run",
|
|
"local-cmis-guide-dry-run",
|
|
["repository-type"],
|
|
None,
|
|
{
|
|
"requires_java_maven": False,
|
|
"repository_id": "local-test-repository",
|
|
"command": [
|
|
sys.executable,
|
|
str(ROOT / "adapters" / "opencmis_console_adapter.py"),
|
|
"--browser-url",
|
|
"{browser_url}",
|
|
"--repository-id",
|
|
"{repository_id}",
|
|
"--check-group",
|
|
"{check_group}",
|
|
"--artifact-dir",
|
|
"{artifact_dir}",
|
|
"--run-dir",
|
|
"{run_dir}",
|
|
"--extension-path",
|
|
"{extension_path}",
|
|
"--credentials-ref",
|
|
"{credentials_ref}",
|
|
"--target-profile-dir",
|
|
"{target_profile_dir}",
|
|
"--timeout-seconds",
|
|
"{timeout_seconds}",
|
|
"--dry-run",
|
|
],
|
|
},
|
|
)
|
|
|
|
result = run_assessment(
|
|
CORE_ROOT,
|
|
target_path,
|
|
assessment_path,
|
|
temp_root / "run",
|
|
[ROOT],
|
|
)
|
|
run_dir = Path(result["run_dir"])
|
|
evidence = json.loads(
|
|
(run_dir / "normalized" / "evidence.json").read_text(encoding="utf-8")
|
|
)["evidence"]
|
|
package = json.loads(
|
|
(run_dir / "reports" / "assessment-package.json").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
artifact_paths = {
|
|
item["path"] for item in package["artifact_manifest"]
|
|
}
|
|
|
|
self.assertEqual(result["status"], "completed")
|
|
self.assertEqual(evidence[1]["result"], "skipped")
|
|
self.assertEqual(
|
|
evidence[1]["facts"]["adapter"],
|
|
"opencmis-console-runner",
|
|
)
|
|
self.assertIn(
|
|
"artifacts/open-cmis-tck/tck/repository-type/session.properties.redacted",
|
|
artifact_paths,
|
|
)
|
|
self.assertIn(
|
|
"artifacts/open-cmis-tck/tck/repository-type/groups.txt",
|
|
artifact_paths,
|
|
)
|
|
self.assertFalse(
|
|
(
|
|
run_dir
|
|
/ "artifacts"
|
|
/ "open-cmis-tck"
|
|
/ "tck"
|
|
/ "repository-type"
|
|
/ "session-private.properties"
|
|
).exists()
|
|
)
|
|
finally:
|
|
server.shutdown()
|
|
thread.join(timeout=5)
|
|
server.server_close()
|
|
|
|
def test_guide_board_service_runs_cmis_extension(self) -> None:
|
|
server = HTTPServer(("127.0.0.1", 0), _CmisHandler)
|
|
thread = threading.Thread(target=server.serve_forever)
|
|
thread.daemon = True
|
|
thread.start()
|
|
service = start_service(CORE_ROOT, [ROOT], host="127.0.0.1", port=0)
|
|
try:
|
|
with TemporaryDirectory() as temporary_directory:
|
|
temp_root = Path(temporary_directory)
|
|
target_path = temp_root / "target.json"
|
|
assessment_path = temp_root / "assessment.json"
|
|
_write_target(target_path, server.server_port, "local-cmis-service")
|
|
_write_assessment(
|
|
assessment_path,
|
|
"local-cmis-service",
|
|
"local-cmis-service",
|
|
[],
|
|
None,
|
|
)
|
|
|
|
extensions = _request_json(service, "GET", "/extensions")
|
|
self.assertIn(
|
|
"open-cmis-tck",
|
|
[extension["id"] for extension in extensions["extensions"]],
|
|
)
|
|
|
|
job = _request_json(
|
|
service,
|
|
"POST",
|
|
"/runs",
|
|
{
|
|
"target": str(target_path),
|
|
"assessment": str(assessment_path),
|
|
"output_dir": str(temp_root / "service-run"),
|
|
},
|
|
expected_status=202,
|
|
)
|
|
status = _wait_for_job(service, job["job_id"])
|
|
reports = _request_json(service, "GET", f"/runs/{job['job_id']}/reports")
|
|
|
|
self.assertEqual(status["status"], "succeeded")
|
|
self.assertEqual(status["result"]["status"], "completed")
|
|
self.assertEqual(
|
|
reports["assessment_package"]["json"]["extensions"][0]["id"],
|
|
"open-cmis-tck",
|
|
)
|
|
finally:
|
|
service.stop()
|
|
server.shutdown()
|
|
thread.join(timeout=5)
|
|
server.server_close()
|
|
|
|
def test_preflight_failure_blocks_downstream_checks(self) -> None:
|
|
with TemporaryDirectory() as temporary_directory:
|
|
temp_root = Path(temporary_directory)
|
|
target_path = temp_root / "target.json"
|
|
assessment_path = temp_root / "assessment.json"
|
|
_write_failing_target(target_path)
|
|
_write_assessment(
|
|
assessment_path,
|
|
"local-cmis-preflight-gate",
|
|
"local-cmis-preflight-failure",
|
|
["repository-type"],
|
|
None,
|
|
)
|
|
|
|
result = run_assessment(
|
|
CORE_ROOT,
|
|
target_path,
|
|
assessment_path,
|
|
temp_root / "run",
|
|
[ROOT],
|
|
)
|
|
run_dir = Path(result["run_dir"])
|
|
evidence = json.loads(
|
|
(run_dir / "normalized" / "evidence.json").read_text(encoding="utf-8")
|
|
)["evidence"]
|
|
findings = json.loads(
|
|
(run_dir / "normalized" / "findings.json").read_text(encoding="utf-8")
|
|
)["findings"]
|
|
|
|
self.assertEqual(result["status"], "infrastructure_error")
|
|
self.assertEqual(evidence[0]["result"], "infrastructure_error")
|
|
self.assertEqual(evidence[1]["result"], "blocked")
|
|
self.assertEqual(evidence[1]["facts"]["blocked_reason"], "preflight_failed")
|
|
self.assertFalse((run_dir / "artifacts" / "runner-contexts").exists())
|
|
self.assertEqual(findings[1]["classification"], "preflight_failed")
|
|
self.assertTrue(findings[1]["expected"])
|
|
|
|
|
|
def _write_target(path: Path, port: int, target_id: str) -> None:
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"id": target_id,
|
|
"subject_type": "cmis-browser-binding-endpoint",
|
|
"subject_name": "Local CMIS Test",
|
|
"environment": "test",
|
|
"scope": ["preflight", "tck-wrapper"],
|
|
"endpoints": [
|
|
{
|
|
"id": "browser-binding",
|
|
"url": f"http://127.0.0.1:{port}/cmis/browser",
|
|
"binding": "cmis-browser",
|
|
}
|
|
],
|
|
"artifacts": [],
|
|
"credentials_ref": None,
|
|
"declared_capabilities": ["cmis.repository-info"],
|
|
"known_gaps": [],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def _write_failing_target(path: Path) -> None:
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"id": "local-cmis-preflight-failure",
|
|
"subject_type": "cmis-browser-binding-endpoint",
|
|
"subject_name": "Local CMIS Preflight Failure",
|
|
"environment": "test",
|
|
"scope": ["preflight", "tck-wrapper"],
|
|
"endpoints": [
|
|
{
|
|
"id": "browser-binding",
|
|
"url": "http://127.0.0.1:9/cmis/browser",
|
|
"binding": "cmis-browser",
|
|
}
|
|
],
|
|
"artifacts": [],
|
|
"credentials_ref": None,
|
|
"declared_capabilities": ["cmis.repository-info"],
|
|
"known_gaps": [],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def _write_assessment(
|
|
path: Path,
|
|
assessment_id: str,
|
|
target_id: str,
|
|
check_groups: list[str],
|
|
waiver_ref: str | None,
|
|
opencmis_policy: dict[str, object] | None = None,
|
|
) -> None:
|
|
runtime_policy: dict[str, object] = {
|
|
"offline": False,
|
|
"timeout_seconds": 15,
|
|
}
|
|
if opencmis_policy is not None:
|
|
runtime_policy["opencmis_tck"] = opencmis_policy
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"id": assessment_id,
|
|
"framework_refs": ["cmis.browser-binding.compatibility.v1"],
|
|
"extension_refs": ["open-cmis-tck"],
|
|
"target_profile_ref": target_id,
|
|
"selected_check_groups": {"open-cmis-tck": check_groups},
|
|
"expectations_ref": None,
|
|
"waivers_ref": waiver_ref,
|
|
"output_policy": {
|
|
"report_formats": ["json", "markdown"],
|
|
"artifact_retention": "summary-only",
|
|
},
|
|
"retention_policy": {
|
|
"summary_days": 365,
|
|
"raw_artifact_days": 0,
|
|
},
|
|
"runtime_policy": runtime_policy,
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def _write_command_waiver(path: Path, target_id: str) -> None:
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"id": "local-cmis-command-waivers",
|
|
"target_profile_ref": target_id,
|
|
"waivers": [
|
|
{
|
|
"id": "local-command-wrapper-bootstrap",
|
|
"scope": "test",
|
|
"requirement_refs": [],
|
|
"check_refs": ["check-group:open-cmis-tck:repository-type"],
|
|
"result_refs": ["blocked"],
|
|
"classification_refs": [],
|
|
"reason": "The test stops before invoking the Java/Maven TCK.",
|
|
"owner": "open-cmis-tck-tests",
|
|
"approved_by": "open-cmis-tck-tests",
|
|
"created_at": "2026-05-07",
|
|
"expires_at": "2099-12-31",
|
|
"review_status": "approved",
|
|
}
|
|
],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def _request_json(
|
|
service: ServiceHandle,
|
|
method: str,
|
|
path: str,
|
|
payload: dict[str, object] | None = None,
|
|
expected_status: int = 200,
|
|
) -> dict[str, object]:
|
|
connection = http.client.HTTPConnection(service.host, service.port, timeout=5)
|
|
body = None
|
|
headers = {}
|
|
if payload is not None:
|
|
body = json.dumps(payload).encode("utf-8")
|
|
headers["Content-Type"] = "application/json"
|
|
try:
|
|
connection.request(method, path, body=body, headers=headers)
|
|
response = connection.getresponse()
|
|
data = response.read().decode("utf-8")
|
|
finally:
|
|
connection.close()
|
|
if response.status != expected_status:
|
|
raise AssertionError(f"expected HTTP {expected_status}, got {response.status}: {data}")
|
|
value = json.loads(data)
|
|
if not isinstance(value, dict):
|
|
raise AssertionError(f"expected JSON object response, got {type(value).__name__}")
|
|
return value
|
|
|
|
|
|
def _wait_for_job(service: ServiceHandle, job_id: str) -> dict[str, object]:
|
|
for _ in range(50):
|
|
status = _request_json(service, "GET", f"/runs/{job_id}")
|
|
if status["status"] in {"succeeded", "failed"}:
|
|
return status
|
|
time.sleep(0.05)
|
|
raise AssertionError(f"job did not finish: {job_id}")
|
|
|
|
|
|
class _CmisHandler(BaseHTTPRequestHandler):
|
|
def do_GET(self) -> None:
|
|
body = json.dumps(
|
|
{
|
|
"local-test-repository": {
|
|
"repositoryId": "local-test-repository",
|
|
"repositoryName": "Local Test Repository",
|
|
"cmisVersionSupported": "1.1",
|
|
"capabilities": {
|
|
"capabilityACL": "discover",
|
|
"capabilityChanges": "none",
|
|
"capabilityGetDescendants": True,
|
|
"capabilityGetFolderTree": True,
|
|
"capabilityQuery": "none",
|
|
},
|
|
}
|
|
}
|
|
).encode("utf-8")
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, format: str, *args: object) -> None:
|
|
return
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|