cc-ci/tests/unit/test_results.py

"""Unit tests for Phase-3 results assembly (harness.results), plan-phase3-results-ux.md §4.2 / R1/R3.

Covers JUnit parsing, stage roll-up, the tier→rung derivation (the documented mapping the level
depends on), and full results.json assembly incl. the U0 gate cases. Pure / tmp-file only. Run cold:
  cc-ci-run -m pytest tests/unit/test_results.py -q
"""

from __future__ import annotations

import json
import os
import sys

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "runner"))
from harness import results as R  # noqa: E402

JUNIT_PASS = """<?xml version="1.0"?>
<testsuites><testsuite name="pytest" tests="2">
<testcase classname="tests.x" name="test_a" time="0.012"/>
<testcase classname="tests.x" name="test_b" time="1.5"/>
</testsuite></testsuites>"""

JUNIT_MIXED = """<?xml version="1.0"?>
<testsuites><testsuite name="pytest" tests="3">
<testcase classname="tests.y" name="test_ok" time="0.1"/>
<testcase classname="tests.y" name="test_bad" time="0.2"><failure message="boom">trace</failure></testcase>
<testcase classname="tests.y" name="test_skipped" time="0"><skipped message="no deps"/></testcase>
</testsuite></testsuites>"""


def _write(tmp_path, name, content):
    p = tmp_path / name
    p.write_text(content)
    return str(p)


def test_parse_junit_pass(tmp_path):
    rows = R.parse_junit(_write(tmp_path, "p.xml", JUNIT_PASS))
    assert len(rows) == 2
    assert {r["status"] for r in rows} == {"pass"}
    assert rows[1]["ms"] == 1500


def test_parse_junit_mixed(tmp_path):
    rows = R.parse_junit(_write(tmp_path, "m.xml", JUNIT_MIXED))
    by = {r["name"]: r["status"] for r in rows}
    assert by == {"test_ok": "pass", "test_bad": "fail", "test_skipped": "skip"}


def test_parse_junit_missing_file_is_empty():
    assert R.parse_junit("/nonexistent/x.xml") == []


def test_collect_stages_orders_and_rolls_up(tmp_path):
    recs = [
        {
            "tier": "install",
            "source": "generic",
            "file": "g/test_install.py",
            "rc": 0,
            "junit": _write(tmp_path, "i.xml", JUNIT_PASS),
        },
        {
            "tier": "custom",
            "source": "cc-ci",
            "file": "c/test_x.py",
            "rc": 1,
            "junit": _write(tmp_path, "c.xml", JUNIT_MIXED),
        },
    ]
    stages = R.collect_stages(recs)
    assert [s["name"] for s in stages] == ["install", "custom"]  # install before custom
    assert stages[0]["status"] == "pass"
    assert stages[1]["status"] == "fail"  # the failure in JUNIT_MIXED
    assert len(stages[1]["tests"]) == 3


def test_collect_stages_synthesizes_when_no_junit():
    recs = [
        {
            "tier": "install",
            "source": "generic",
            "file": "g/test_install.py",
            "rc": 1,
            "junit": None,
        }
    ]
    stages = R.collect_stages(recs)
    assert stages[0]["status"] == "fail"
    assert len(stages[0]["tests"]) == 1


# ---- derive_rungs: the documented mapping ----


def _results(**kw):
    base = {
        "install": "pass",
        "upgrade": "pass",
        "backup": "pass",
        "restore": "pass",
        "custom": "pass",
    }
    base.update(kw)
    return base


def test_derive_rungs_full_stateful_sso():
    rungs = R.derive_rungs(
        _results(),
        backup_capable=True,
        declared=["keycloak"],
        deps_ready=True,
        sso_unverified=False,
        has_custom=True,
        has_repo_local=False,
        repo_local_passed=False,
    )
    assert rungs == {
        "install": "pass",
        "upgrade": "pass",
        "backup_restore": "pass",
        "functional": "pass",
        "integration": "pass",
        "recipe_local": "na",
    }


def test_derive_rungs_no_sso_surface_is_integration_na():
    rungs = R.derive_rungs(
        _results(),
        backup_capable=True,
        declared=[],
        deps_ready=True,
        sso_unverified=False,
        has_custom=True,
        has_repo_local=False,
        repo_local_passed=False,
    )
    assert rungs["integration"] == "na"
    assert rungs["functional"] == "pass"


def test_derive_rungs_stateless_backup_na():
    rungs = R.derive_rungs(
        _results(backup="skip", restore="skip", custom="skip"),
        backup_capable=False,
        declared=[],
        deps_ready=True,
        sso_unverified=False,
        has_custom=False,
        has_repo_local=False,
        repo_local_passed=False,
    )
    assert rungs["backup_restore"] == "na"
    assert rungs["functional"] == "na"


def test_derive_rungs_sso_unverified_is_integration_fail():
    rungs = R.derive_rungs(
        _results(),
        backup_capable=True,
        declared=["keycloak"],
        deps_ready=False,
        sso_unverified=True,
        has_custom=True,
        has_repo_local=False,
        repo_local_passed=False,
    )
    assert rungs["integration"] == "fail"


def test_derive_rungs_repo_local_pass():
    rungs = R.derive_rungs(
        _results(),
        backup_capable=True,
        declared=[],
        deps_ready=True,
        sso_unverified=False,
        has_custom=True,
        has_repo_local=True,
        repo_local_passed=True,
    )
    assert rungs["recipe_local"] == "pass"


# ---- build_results: end-to-end incl level + flags ----


def test_build_results_level_and_flags(tmp_path):
    recs = [
        {
            "tier": "install",
            "source": "generic",
            "file": "g/test_install.py",
            "rc": 0,
            "junit": _write(tmp_path, "i.xml", JUNIT_PASS),
        },
        {
            "tier": "custom",
            "source": "cc-ci",
            "file": "c/test_func.py",
            "rc": 0,
            "junit": _write(tmp_path, "c.xml", JUNIT_PASS),
        },
    ]
    data = R.build_results(
        recipe="hedgedoc",
        version="1.2.3",
        pr="7",
        ref="deadbeefcafe0000",
        records=recs,
        results=_results(),
        backup_capable=True,
        declared=[],
        deps_ready=True,
        sso_unverified=False,
        clean_teardown=True,
        no_secret_leak=True,
        finished_ts=1234.0,
    )
    # stateful, functional pass, no SSO surface, no repo-local → caps at L4
    assert data["level"] == 4
    assert "L5" in data["level_cap_reason"]
    assert data["recipe"] == "hedgedoc"
    assert data["ref"] == "deadbeefcafe"
    assert data["flags"] == {"clean_teardown": True, "no_secret_leak": True}
    assert [s["name"] for s in data["stages"]] == ["install", "custom"]


def test_build_results_capped_at_L1_on_upgrade_fail(tmp_path):
    recs = [
        {
            "tier": "install",
            "source": "generic",
            "file": "g/test_install.py",
            "rc": 0,
            "junit": _write(tmp_path, "i.xml", JUNIT_PASS),
        }
    ]
    data = R.build_results(
        recipe="x",
        version=None,
        pr="0",
        ref=None,
        records=recs,
        results=_results(upgrade="fail"),
        backup_capable=True,
        declared=[],
        deps_ready=True,
        sso_unverified=False,
        clean_teardown=True,
        no_secret_leak=True,
        finished_ts=0.0,
    )
    assert data["level"] == 1
    assert "L2" in data["level_cap_reason"]


# ---- classify_na / cap_intent: intentional-vs-accidental N/A (operator request) ----


def _rungs(**kw):
    base = {
        "install": "pass",
        "upgrade": "pass",
        "backup_restore": "pass",
        "functional": "pass",
        "integration": "na",
        "recipe_local": "na",
    }
    base.update(kw)
    return base


def test_classify_na_declared_vs_undeclared():
    rungs = _rungs(backup_restore="na", functional="na")
    info = R.classify_na(rungs, {"backup_restore": "stateless static server"})
    # backup_restore is declared intentional; functional is an undeclared gap-sensitive N/A.
    assert info["rungs"]["backup_restore"] == {
        "intent": "declared",
        "reason": "stateless static server",
    }
    assert info["rungs"]["functional"]["intent"] == "undeclared"
    assert info["gaps"] == ["functional"]  # backup_restore declared → not a gap
    assert info["stale_declared"] == []
    # structurally-optional N/A (integration, recipe_local) are recorded but never flagged as gaps.
    assert info["rungs"]["integration"]["intent"] == "undeclared"
    assert "integration" not in info["gaps"]


def test_classify_na_stale_declaration():
    # backup_restore actually ran (pass) but is declared N/A → stale opt-out, surfaced.
    rungs = _rungs(backup_restore="pass")
    info = R.classify_na(rungs, {"backup_restore": "stale reason"})
    assert info["stale_declared"] == ["backup_restore"]
    assert "backup_restore" not in info["rungs"]  # not N/A, so not in the per-rung N/A map


def test_cap_intent_declared_explains_cap():
    # install+upgrade pass, backup_restore declared-N/A → caps at L2 with an intentional clause.
    rungs = _rungs(backup_restore="na")
    info = R.classify_na(rungs, {"backup_restore": "no persistent data"})
    intent = R.cap_intent(rungs, 2, "L3 backup/restore (data integrity) N/A", info)
    assert intent == "intentional · no persistent data"


def test_cap_intent_undeclared_gap():
    rungs = _rungs(backup_restore="na")
    info = R.classify_na(rungs, None)
    intent = R.cap_intent(rungs, 2, "L3 backup/restore (data integrity) N/A", info)
    assert "possible coverage gap" in intent


def test_cap_intent_blank_when_not_capped_on_na():
    rungs = _rungs()  # full clean climb, capped only at integration (na, structurally optional)
    info = R.classify_na(rungs, None)
    # capping rung is integration (level 4) — structurally optional, so no intent clause.
    assert R.cap_intent(rungs, 4, "L5 integration N/A", info) == ""
    # and no cap at all → blank.
    assert R.cap_intent(rungs, 6, "", info) == ""


def test_build_results_threads_expected_na(tmp_path):
    recs = [
        {
            "tier": "install",
            "source": "generic",
            "file": "g/test_install.py",
            "rc": 0,
            "junit": _write(tmp_path, "i.xml", JUNIT_PASS),
        }
    ]
    data = R.build_results(
        recipe="custom-html-tiny",
        version="1.1.0",
        pr="0",
        ref=None,
        records=recs,
        results=_results(backup="skip", restore="skip", custom="skip"),
        backup_capable=False,  # no backupbot label → backup_restore N/A
        declared=[],
        deps_ready=True,
        sso_unverified=False,
        clean_teardown=True,
        no_secret_leak=True,
        finished_ts=0.0,
        expected_na={"backup_restore": "stateless static file server"},
    )
    # N/A still caps at L2 (never inflates), but now annotated intentional rather than flagged.
    assert data["level"] == 2
    assert "L3" in data["level_cap_reason"]
    assert data["level_cap_intent"] == "intentional · stateless static file server"
    assert data["na"]["rungs"]["backup_restore"]["intent"] == "declared"
    assert data["na"]["gaps"] == []


def test_write_results_roundtrip(tmp_path):
    data = {"run_id": "42", "level": 3, "stages": []}
    path = R.write_results(data, runs_dir_override=str(tmp_path))
    assert path.endswith("/42/results.json")
    with open(path) as f:
        assert json.load(f)["level"] == 3