hub-core/tests/test_conformance.py

from __future__ import annotations

from fastapi.testclient import TestClient

from hub_core.conformance import ConformanceHarness, find_secret_violations
from hub_core.runtime.app import create_app
from hub_core.runtime.cli import build_parser
from hub_core.runtime.config import RuntimeSettings
from hub_core.runtime.store import InMemoryPortStore


def isolated_target() -> TestClient:
    settings = RuntimeSettings(environment="test", backend="memory", allow_ephemeral=True)
    return TestClient(create_app(settings=settings, port_store=InMemoryPortStore()))


def test_implemented_tier_2_and_3_profile_passes_reference_runtime() -> None:
    report = ConformanceHarness(isolated_target()).run()

    assert report.passed
    assert report.passed_count == 12
    assert {check.check_id for check in report.checks} == {
        "C1",
        "C2",
        "C3",
        "C4",
        "C5",
        "C6",
        "C7",
        "C8",
        "C9",
        "C10",
        "F2",
        "F3",
    }
    assert all(check.status == "pass" for check in report.checks)
    assert report.to_dict()["summary"] == {"passed": 12, "total": 12}


def test_projection_rebuild_scenario_leaves_separate_provenance_bearing_views() -> None:
    target = isolated_target()
    assert ConformanceHarness(target).run().passed

    progress = target.get("/ports/projections/progress_events").json()
    interaction = target.get("/ports/projections/interaction_events").json()

    assert progress["data"]["rebuild_from"] == ["progress_events"]
    assert interaction["data"]["rebuild_from"] == ["interaction_events"]
    assert {item["family"] for item in progress["data"]["items"]} == {"progress"}
    assert {item["family"] for item in interaction["data"]["items"]} == {"interaction"}
    assert progress["provenance"]["content_hash"]
    assert interaction["provenance"]["content_hash"]


def test_secret_heuristic_reports_paths_without_echoing_values() -> None:
    value = {
        "nested": {"api_token": "do-not-echo"},
        "database": "postgresql://runtime:do-not-echo@example.invalid/hub",
        "safe": "https://ops-hub.example.invalid/docs",
    }

    assert find_secret_violations(value) == ["$.nested.api_token", "$.database"]


def test_cli_exposes_remote_conformance_runner() -> None:
    args = build_parser(RuntimeSettings()).parse_args(
        ["conformance", "--base-url", "http://runtime.invalid", "--json"]
    )

    assert args.command == "conformance"
    assert args.base_url == "http://runtime.invalid"
    assert args.as_json is True