generated from coulomb/repo-seed
1342 lines
54 KiB
Python
1342 lines
54 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
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.archive import archive_run
|
|
from open_cmis_tck.bootstrap import TCK_COORDINATE, check_runtime
|
|
from open_cmis_tck.log_review import build_log_review, write_log_review
|
|
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_inmemory_pilot_profiles_validate(self) -> None:
|
|
target = validate_target_profile(ROOT / "profiles" / "targets" / "opencmis-inmemory-local.json")
|
|
assessment = validate_assessment_profile(
|
|
ROOT / "profiles" / "assessments" / "cmis-browser-inmemory-pilot.json"
|
|
)
|
|
plan = build_run_plan(
|
|
CORE_ROOT,
|
|
ROOT / "profiles" / "targets" / "opencmis-inmemory-local.json",
|
|
ROOT / "profiles" / "assessments" / "cmis-browser-inmemory-pilot.json",
|
|
[ROOT],
|
|
)
|
|
|
|
self.assertEqual(target["credentials_ref"], "env:OPENCMIS_INMEMORY_USER,OPENCMIS_INMEMORY_PASSWORD")
|
|
self.assertEqual(assessment["target_profile_ref"], "opencmis-inmemory-local")
|
|
self.assertEqual(
|
|
[step["id"] for step in plan["ordered_steps"]],
|
|
["preflight:open-cmis-tck", "check-group:open-cmis-tck:repository-type"],
|
|
)
|
|
|
|
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_log_review_classifies_loopback_warning_and_closed_warning(self) -> None:
|
|
with TemporaryDirectory() as temporary_directory:
|
|
temp_root = Path(temporary_directory)
|
|
previous_run = temp_root / "previous"
|
|
current_run = temp_root / "current"
|
|
_write_review_run(
|
|
previous_run,
|
|
"run-previous",
|
|
"http://127.0.0.1:8010/cmis/browser",
|
|
["cmis.repository-info", "cmis.object-services"],
|
|
[
|
|
_opencmis_case(
|
|
"repository-type",
|
|
"warning",
|
|
"WARNING",
|
|
"Security Test (BROWSER)",
|
|
"HTTPS is not used. Credentials might be transferred as plain text!",
|
|
"SecurityTest.java",
|
|
67,
|
|
),
|
|
_opencmis_case(
|
|
"object-content",
|
|
"warning",
|
|
"WARNING",
|
|
"Set, Append, and Delete Content Test (BROWSER)",
|
|
"appendContentStream() is not supported!",
|
|
"SetAndDeleteContentTest.java",
|
|
200,
|
|
),
|
|
],
|
|
)
|
|
_write_review_run(
|
|
current_run,
|
|
"run-current",
|
|
"http://127.0.0.1:8010/cmis/browser",
|
|
["cmis.repository-info", "cmis.object-services"],
|
|
[
|
|
_opencmis_case(
|
|
"repository-type",
|
|
"warning",
|
|
"WARNING",
|
|
"Security Test (BROWSER)",
|
|
"HTTPS is not used. Credentials might be transferred as plain text!",
|
|
"SecurityTest.java",
|
|
67,
|
|
),
|
|
_opencmis_case(
|
|
"object-content",
|
|
"skipped",
|
|
"SKIPPED",
|
|
"Create and Delete Relationship Test (BROWSER)",
|
|
"Relationship type 'cmis:relationship' is not creatable!",
|
|
"AbstractSessionTest.java",
|
|
634,
|
|
),
|
|
],
|
|
)
|
|
|
|
review = build_log_review(current_run, previous_run_dir=previous_run)
|
|
written = write_log_review(current_run, previous_run_dir=previous_run)
|
|
|
|
self.assertEqual(review["summary"]["status"], "pass_with_review_notes")
|
|
self.assertEqual(review["summary"]["accepted_warnings"], 1)
|
|
self.assertEqual(review["summary"]["unaccepted_warnings"], 0)
|
|
self.assertEqual(review["summary"]["closed_warnings"], 1)
|
|
self.assertEqual(review["summary"]["expected_skips"], 1)
|
|
self.assertEqual(
|
|
review["warnings"][0]["classification"],
|
|
"accepted_local_loopback_transport",
|
|
)
|
|
self.assertEqual(
|
|
review["closed_warnings"][0]["message"],
|
|
"appendContentStream() is not supported!",
|
|
)
|
|
self.assertTrue(Path(written["json"]).exists())
|
|
self.assertIn(
|
|
"OpenCMIS Log Review",
|
|
Path(written["markdown"]).read_text(encoding="utf-8"),
|
|
)
|
|
|
|
def test_log_review_flags_non_loopback_http_warning_as_deployment_blocker(self) -> None:
|
|
with TemporaryDirectory() as temporary_directory:
|
|
run_dir = Path(temporary_directory) / "run"
|
|
_write_review_run(
|
|
run_dir,
|
|
"run-production-http",
|
|
"http://cmis.example.test/browser",
|
|
["cmis.repository-info"],
|
|
[
|
|
_opencmis_case(
|
|
"repository-type",
|
|
"warning",
|
|
"WARNING",
|
|
"Security Test (BROWSER)",
|
|
"HTTPS is not used. Credentials might be transferred as plain text!",
|
|
"SecurityTest.java",
|
|
67,
|
|
)
|
|
],
|
|
environment="production",
|
|
)
|
|
|
|
review = build_log_review(run_dir)
|
|
|
|
self.assertEqual(review["summary"]["status"], "review_required")
|
|
self.assertEqual(review["summary"]["unaccepted_warnings"], 1)
|
|
self.assertEqual(review["warnings"][0]["classification"], "deployment_transport_blocker")
|
|
self.assertEqual(review["warnings"][0]["severity"], "blocker")
|
|
|
|
def test_log_review_marks_advertised_capability_skip_for_review(self) -> None:
|
|
with TemporaryDirectory() as temporary_directory:
|
|
run_dir = Path(temporary_directory) / "run"
|
|
_write_review_run(
|
|
run_dir,
|
|
"run-relationship-skip",
|
|
"http://127.0.0.1:8010/cmis/browser",
|
|
["cmis.repository-info", "cmis.relationships"],
|
|
[
|
|
_opencmis_case(
|
|
"object-content",
|
|
"skipped",
|
|
"SKIPPED",
|
|
"Create and Delete Relationship Test (BROWSER)",
|
|
"Relationship type 'cmis:relationship' is not creatable!",
|
|
"AbstractSessionTest.java",
|
|
634,
|
|
)
|
|
],
|
|
)
|
|
|
|
review = build_log_review(run_dir)
|
|
|
|
self.assertEqual(review["summary"]["status"], "review_required")
|
|
self.assertEqual(review["summary"]["unexpected_skips"], 1)
|
|
self.assertFalse(review["skips"][0]["expected"])
|
|
self.assertEqual(review["skips"][0]["classification"], "advertised_capability_not_exercised")
|
|
|
|
def test_archive_run_copies_evidence_and_writes_hash_manifest(self) -> None:
|
|
with TemporaryDirectory() as temporary_directory:
|
|
temp_root = Path(temporary_directory)
|
|
run_dir = temp_root / "run"
|
|
_write_review_run(
|
|
run_dir,
|
|
"run-archive",
|
|
"http://127.0.0.1:8010/cmis/browser",
|
|
["cmis.repository-info"],
|
|
[],
|
|
)
|
|
(run_dir / "reports" / "report.md").write_text("# Report\n", encoding="utf-8")
|
|
|
|
manifest = archive_run(run_dir, temp_root / "archive")
|
|
archive_dir = Path(manifest["archive_dir"])
|
|
manifest_path = archive_dir / "archive-manifest.json"
|
|
|
|
self.assertTrue((archive_dir / "reports" / "report.md").exists())
|
|
self.assertTrue(manifest_path.exists())
|
|
self.assertEqual(manifest["run_id"], "run-archive")
|
|
self.assertIn(
|
|
"normalized/evidence.json",
|
|
{item["path"] for item in manifest["files"]},
|
|
)
|
|
self.assertTrue(all(len(item["sha256"]) == 64 for item in manifest["files"]))
|
|
|
|
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_preflight_uses_env_credentials_for_basic_auth(self) -> None:
|
|
server = HTTPServer(("127.0.0.1", 0), _BasicAuthCmisHandler)
|
|
thread = threading.Thread(target=server.serve_forever)
|
|
thread.daemon = True
|
|
thread.start()
|
|
original_user = os.environ.get("CMIS_AUTH_USER")
|
|
original_password = os.environ.get("CMIS_AUTH_PASSWORD")
|
|
os.environ["CMIS_AUTH_USER"] = "alice"
|
|
os.environ["CMIS_AUTH_PASSWORD"] = "secret"
|
|
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-auth-preflight")
|
|
target = json.loads(target_path.read_text(encoding="utf-8"))
|
|
target["credentials_ref"] = "env:CMIS_AUTH_USER,CMIS_AUTH_PASSWORD"
|
|
target_path.write_text(json.dumps(target), encoding="utf-8")
|
|
_write_assessment(
|
|
assessment_path,
|
|
"local-cmis-auth-preflight",
|
|
"local-cmis-auth-preflight",
|
|
[],
|
|
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"]
|
|
|
|
self.assertEqual(result["status"], "completed")
|
|
self.assertEqual(evidence[0]["result"], "pass")
|
|
self.assertEqual(evidence[0]["facts"]["auth_mode"], "env")
|
|
finally:
|
|
if original_user is None:
|
|
os.environ.pop("CMIS_AUTH_USER", None)
|
|
else:
|
|
os.environ["CMIS_AUTH_USER"] = original_user
|
|
if original_password is None:
|
|
os.environ.pop("CMIS_AUTH_PASSWORD", None)
|
|
else:
|
|
os.environ["CMIS_AUTH_PASSWORD"] = original_password
|
|
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_review_run(
|
|
path: Path,
|
|
run_id: str,
|
|
browser_url: str,
|
|
declared_capabilities: list[str],
|
|
cases: list[dict[str, object]],
|
|
*,
|
|
target_id: str = "kontextual-cmis-compat",
|
|
environment: str = "local",
|
|
) -> None:
|
|
groups = sorted({str(case["selected_check_group"]) for case in cases}) or ["repository-type"]
|
|
(path / "normalized").mkdir(parents=True)
|
|
(path / "reports").mkdir()
|
|
for group in groups:
|
|
artifact_dir = path / "artifacts" / "open-cmis-tck" / "tck" / group
|
|
artifact_dir.mkdir(parents=True, exist_ok=True)
|
|
(artifact_dir / "console-runner-stderr.txt").write_text("", encoding="utf-8")
|
|
(artifact_dir / "stderr.log").write_text("", encoding="utf-8")
|
|
path.joinpath("run.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"id": run_id,
|
|
"target_profile_ref": target_id,
|
|
"assessment_profile_ref": "cmis-browser-baseline",
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
path.joinpath("target-profile.snapshot.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"id": target_id,
|
|
"environment": environment,
|
|
"endpoints": [
|
|
{
|
|
"id": "browser-binding",
|
|
"url": browser_url,
|
|
"binding": "cmis-browser",
|
|
}
|
|
],
|
|
"declared_capabilities": declared_capabilities,
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
path.joinpath("assessment-profile.snapshot.json").write_text(
|
|
json.dumps({"id": "cmis-browser-baseline"}),
|
|
encoding="utf-8",
|
|
)
|
|
evidence = []
|
|
for group in groups:
|
|
group_cases = [case for case in cases if case["selected_check_group"] == group]
|
|
evidence.append(
|
|
{
|
|
"id": f"evidence:check-group:open-cmis-tck:{group}",
|
|
"check_id": f"check-group:open-cmis-tck:{group}",
|
|
"result": "warning" if any(case["status"] == "warning" for case in group_cases) else "pass",
|
|
"facts": {
|
|
"selected_check_group": group,
|
|
"browser_binding_url": browser_url,
|
|
"cases": group_cases,
|
|
},
|
|
}
|
|
)
|
|
path.joinpath("normalized", "evidence.json").write_text(
|
|
json.dumps({"evidence": evidence}),
|
|
encoding="utf-8",
|
|
)
|
|
path.joinpath("normalized", "findings.json").write_text(
|
|
json.dumps({"findings": []}),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def _opencmis_case(
|
|
selected_check_group: str,
|
|
status: str,
|
|
status_native: str,
|
|
test_name: str,
|
|
message: str,
|
|
source_file: str,
|
|
source_line: int,
|
|
) -> dict[str, object]:
|
|
return {
|
|
"id": f"opencmis-tck:{selected_check_group}:{test_name.lower().replace(' ', '-')}",
|
|
"status": status,
|
|
"status_native": status_native,
|
|
"selected_check_group": selected_check_group,
|
|
"group_name": "OpenCMIS Test Group",
|
|
"test_name": test_name,
|
|
"message": message,
|
|
"source_location": {
|
|
"file": source_file,
|
|
"line": source_line,
|
|
},
|
|
}
|
|
|
|
|
|
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
|
|
|
|
|
|
class _BasicAuthCmisHandler(_CmisHandler):
|
|
def do_GET(self) -> None:
|
|
expected = "Basic " + base64.b64encode(b"alice:secret").decode("ascii")
|
|
if self.headers.get("Authorization") != expected:
|
|
body = b"unauthorized"
|
|
self.send_response(401)
|
|
self.send_header("Content-Type", "text/plain")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
return
|
|
super().do_GET()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|