Gitea-Tools/tests/test_issue_708_not_connected_classification.py

"""Regression tests for Issue #708 B2: not-connected is not Connected-but-unattached.

The first #708 slice counted a namespace as ``missing`` only when it appeared in
``connected_servers``. A *required* namespace absent from that inventory was therefore
never counted at all, so the session reported ``attachment_healthy: true`` with
``error_type: None`` while holding no attachment proof for it — and the gate text told the
operator "the host reports Connected" about a service nothing had reported Connected.

These tests pin the corrected distinction:

* required and not connected never yields a healthy verdict,
* it is typed ``mcp_required_namespaces_not_connected``, never the #708 condition,
* ``mcp_connected_namespaces_missing`` stays reserved for genuinely Connected namespaces,
* both categories still fail review and merge closed, with their own reason,
* per-namespace ``connected``/``attached`` evidence is reported accurately,
* one session's evidence cannot clear another session's block.
"""

import mcp_namespace_health


ROLES = ["gitea-author", "gitea-reviewer", "gitea-merger", "gitea-tools"]

NOT_CONNECTED = mcp_namespace_health.ERROR_REQUIRED_NAMESPACES_NOT_CONNECTED
CONNECTED_MISSING = mcp_namespace_health.ERROR_CONNECTED_NAMESPACES_MISSING


def _assess(connected, attached, required):
    return mcp_namespace_health.assess_connected_namespace_attachment(
        connected_servers=connected,
        attached_session_namespaces=attached,
        required_namespaces=required,
    )


def _store(assessment):
    """The recorder contract: per-namespace verdicts feed the gate."""
    return dict(assessment["namespace_conditions"])


# --- the four evidence combinations ----------------------------------------


def test_connected_and_attached_is_healthy():
    res = _assess(ROLES, ROLES, ROLES)
    assert res["attachment_healthy"] is True
    assert res["error_type"] is None
    assert res["missing_namespaces"] == []
    assert res["not_connected_namespaces"] == []
    assert res["discovery_status"] == "namespaces_attached"


def test_connected_but_unattached_keeps_the_708_condition():
    res = _assess(ROLES, ["gitea-author"], ROLES)
    assert res["attachment_healthy"] is False
    assert res["error_type"] == CONNECTED_MISSING
    assert res["not_connected_namespaces"] == []
    assert sorted(res["missing_namespaces"]) == [
        "gitea-merger",
        "gitea-reviewer",
        "gitea-tools",
    ]
    assert res["discovery_status"] == "connected_but_namespaces_missing"


def test_required_but_not_connected_is_never_healthy():
    """The reviewer's exact reproduction from review 637."""
    res = _assess(
        ["gitea-reviewer"], ["gitea-reviewer"], ["gitea-reviewer", "gitea-merger"]
    )
    assert res["attachment_healthy"] is False
    assert res["success"] is False
    assert res["error_type"] is not None
    assert res["telemetry"]["not_connected_count"] == 1


def test_required_but_not_connected_is_typed_distinctly():
    res = _assess(
        ["gitea-reviewer"], ["gitea-reviewer"], ["gitea-reviewer", "gitea-merger"]
    )
    assert res["error_type"] == NOT_CONNECTED
    assert res["discovery_status"] == "required_namespaces_not_connected"
    assert res["not_connected_namespaces"] == ["gitea-merger"]
    # Not collapsed into the #708 condition, nor into config drift (#672) or
    # transport-closed (#584).
    assert CONNECTED_MISSING not in res["error_types"]
    assert res["missing_namespaces"] == []
    assert res["error_type"] not in {"mcp_config_drift", "transport_closed"}


def test_not_connected_reason_makes_no_connected_claim():
    res = _assess(
        ["gitea-reviewer"], ["gitea-reviewer"], ["gitea-reviewer", "gitea-merger"]
    )
    about_merger = [r for r in res["reasons"] if "gitea-merger" in r]
    assert about_merger, "the not-connected namespace must be named in the reasons"
    for reason in about_merger:
        assert "report Connected at host/CLI layer" not in reason
    assert "gitea-merger" in res["exact_next_action"]


def test_neither_connected_nor_attached_reports_disconnected():
    res = _assess([], [], ROLES)
    assert res["attachment_healthy"] is False
    assert res["discovery_status"] == "disconnected"
    assert res["error_type"] == NOT_CONNECTED
    assert sorted(res["not_connected_namespaces"]) == sorted(ROLES)
    assert res["missing_namespaces"] == []


# --- mixed required namespaces ---------------------------------------------


def test_mixed_connected_attached_and_not_connected():
    res = _assess(
        ["gitea-author"], ["gitea-author"], ["gitea-author", "gitea-merger"]
    )
    assert res["attachment_healthy"] is False
    assert res["not_connected_namespaces"] == ["gitea-merger"]
    assert res["missing_namespaces"] == []
    conditions = res["namespace_conditions"]
    assert conditions["gitea-author"]["attachment_healthy"] is True
    assert conditions["gitea-author"]["condition"] is None
    assert conditions["gitea-merger"]["attachment_healthy"] is False
    assert conditions["gitea-merger"]["condition"] == NOT_CONNECTED


def test_both_conditions_present_are_both_reported():
    """One namespace Connected-but-unattached, another never connected."""
    res = _assess(
        ["gitea-author", "gitea-reviewer"],
        ["gitea-author"],
        ["gitea-author", "gitea-reviewer", "gitea-merger"],
    )
    assert res["missing_namespaces"] == ["gitea-reviewer"]
    assert res["not_connected_namespaces"] == ["gitea-merger"]
    # Neither condition is hidden by the other; error_type names the primary one.
    assert sorted(res["error_types"]) == sorted([NOT_CONNECTED, CONNECTED_MISSING])
    assert res["error_type"] == NOT_CONNECTED


# --- unknown, partial, malformed, contradictory evidence --------------------


def test_unknown_attachment_evidence_is_not_healthy():
    """No session tool surface reported at all is unproven, not proven good."""
    res = _assess(ROLES, None, ROLES)
    assert res["attachment_healthy"] is False
    assert res["telemetry"]["attached_count"] == 0
    assert res["error_type"] == CONNECTED_MISSING


def test_partial_evidence_gates_only_the_unproven_roles():
    res = _assess(ROLES, ["gitea-author", "gitea-tools"], ROLES)
    store = _store(res)
    assert mcp_namespace_health.attachment_gate_from_session("work_issue", store) == []
    assert mcp_namespace_health.attachment_gate_from_session("review_pr", store)
    assert mcp_namespace_health.attachment_gate_from_session("merge_pr", store)


def test_malformed_namespace_entries_are_discarded_not_trusted():
    """Blank and whitespace-only names must not become namespaces or proof."""
    res = mcp_namespace_health.assess_connected_namespace_attachment(
        connected_servers=["gitea-author", "", "   "],
        attached_session_namespaces=["gitea-author", ""],
        required_namespaces=["gitea-author", "  "],
    )
    assert "" not in res["proof_of_connected_vs_attached"]
    assert "   " not in res["proof_of_connected_vs_attached"]
    assert res["attachment_healthy"] is True
    assert res["telemetry"]["required_count"] == 1


def test_contradictory_attached_without_connected_fails_closed():
    """Attached in the session yet absent from the connected inventory."""
    res = _assess(["gitea-author"], ["gitea-author", "gitea-merger"], ROLES)
    assert res["attachment_healthy"] is False
    assert "gitea-merger" in res["contradictory_namespaces"]
    assert "gitea-merger" in res["not_connected_namespaces"]
    assert res["namespace_conditions"]["gitea-merger"]["contradictory_evidence"] is True
    assert res["namespace_conditions"]["gitea-merger"]["attachment_healthy"] is False
    assert any("contradictory evidence" in r for r in res["reasons"])
    assert res["telemetry"]["contradictory_count"] >= 1


def test_duplicate_required_entries_are_counted_once():
    res = _assess(["gitea-author"], ["gitea-author"], ["gitea-author", "gitea-author"])
    assert res["telemetry"]["required_count"] == 1
    assert res["attachment_healthy"] is True


# --- review and merge behaviour for both failure categories -----------------


def test_merge_fails_closed_when_required_namespace_never_connected():
    store = _store(_assess(["gitea-author"], ["gitea-author"], ROLES))
    reasons = mcp_namespace_health.attachment_gate_from_session("merge_pr", store)
    assert reasons
    assert NOT_CONNECTED in reasons[0]
    assert "fail closed, #708" in reasons[0]


def test_review_fails_closed_when_required_namespace_never_connected():
    store = _store(_assess(["gitea-author"], ["gitea-author"], ROLES))
    reasons = mcp_namespace_health.attachment_gate_from_session("review_pr", store)
    assert reasons
    assert NOT_CONNECTED in reasons[0]
    assert "fail closed, #708" in reasons[0]


def test_not_connected_block_never_claims_the_host_reports_connected():
    store = _store(_assess(["gitea-author"], ["gitea-author"], ROLES))
    reasons = mcp_namespace_health.attachment_gate_from_session("merge_pr", store)
    assert "the host reports Connected" not in reasons[0]
    assert "absent from the connected-service inventory" in reasons[0]


def test_connected_but_unattached_block_still_says_connected():
    store = _store(_assess(ROLES, ["gitea-author"], ROLES))
    reasons = mcp_namespace_health.attachment_gate_from_session("merge_pr", store)
    assert CONNECTED_MISSING in reasons[0]
    assert "the host reports Connected" in reasons[0]


def test_both_categories_offer_only_the_sanctioned_recovery():
    for store in (
        _store(_assess(ROLES, [], ROLES)),
        _store(_assess(["gitea-author"], ["gitea-author"], ROLES)),
    ):
        reasons = mcp_namespace_health.attachment_gate_from_session("merge_pr", store)
        joined = " ".join(reasons)
        assert "reconnect" in joined.lower()
        assert "Workflow Safety Hard Stop (#708)" in joined
        for forbidden in ("pkill", "chmod", "curl ", ".env", "sys.path"):
            assert forbidden not in reasons[0]


def test_entry_without_connected_evidence_blocks_without_asserting_either():
    """A legacy/partial store entry must fail closed and claim nothing it cannot prove."""
    store = {"gitea-merger": {"namespace": "gitea-merger", "attached": False}}
    reasons = mcp_namespace_health.attachment_gate_from_session("merge_pr", store)
    assert reasons
    assert "no connected-status evidence" in reasons[0]
    assert "the host reports Connected" not in reasons[0]
    assert "fail closed, #708" in reasons[0]


# --- session isolation ------------------------------------------------------


def test_one_session_evidence_does_not_clear_another_session_block():
    """Attachment evidence is per-session state; it must not travel between sessions."""
    blocked_session = _store(_assess(["gitea-author"], ["gitea-author"], ROLES))
    healthy_session = _store(_assess(ROLES, ROLES, ROLES))

    assert (
        mcp_namespace_health.attachment_gate_from_session("merge_pr", healthy_session)
        == []
    )
    # The healthy session's verdict is not consulted for the blocked session.
    assert mcp_namespace_health.attachment_gate_from_session("merge_pr", blocked_session)
    # And the blocked session's store is unchanged by the healthy one existing.
    assert blocked_session["gitea-merger"]["attachment_healthy"] is False


def test_gate_reads_only_the_store_it_is_given():
    healthy_session = _store(_assess(ROLES, ROLES, ROLES))
    assert mcp_namespace_health.attachment_gate_from_session("merge_pr", {}) == []
    assert mcp_namespace_health.attachment_gate_from_session("merge_pr", None) == []
    assert (
        mcp_namespace_health.attachment_gate_from_session("merge_pr", healthy_session)
        == []
    )


# --- telemetry stays secret-free -------------------------------------------


def test_not_connected_telemetry_leaks_no_secrets():
    res = _assess(["gitea-author"], ["gitea-author"], ROLES)
    blob = repr(res["telemetry"]).lower()
    for leak in ("token", "authorization", "password", "secret", "/users/", "http"):
        assert leak not in blob