Gitea-Tools/tests/test_webui_notifications.py

"""Unit tests for Phase 3 Notifications and Human-Attention Console (#648)."""

from __future__ import annotations

import pytest
from starlette.testclient import TestClient

from webui.app import create_app
from webui.notifications import (
    ATTENTION_HUMAN_REQUIRED,
    ATTENTION_OPERATOR,
    ATTENTION_ROUTINE,
    CATEGORY_AUTH,
    CATEGORY_BLOCKER,
    CATEGORY_LEASE,
    CATEGORY_SYSTEM,
    CATEGORY_VALIDATION,
    CATEGORY_WORKFLOW,
    NotificationItem,
    NotificationSnapshot,
    classify_attention_event,
    load_notifications_snapshot,
    snapshot_to_dict,
)
from webui.notification_views import render_notifications_page
from webui.project_registry import load_registry
from webui.queue_loader import QueueItem, QueueSnapshot
from webui.lease_loader import CollisionWarning, LeaseSnapshot
from webui.system_health import DependencyProbe, SystemHealthSnapshot, VersionInfo, StaleRuntime


def test_classify_attention_event_rules():
    # 1. Critical escalation boundaries -> human-required
    att_cls, req_human = classify_attention_event(
        CATEGORY_AUTH, "Auth error", "Unauthorized access attempt", is_auth_failure=True
    )
    assert att_cls == ATTENTION_HUMAN_REQUIRED
    assert req_human is True

    att_cls, req_human = classify_attention_event(
        CATEGORY_SYSTEM, "Hard stop", "Hard stop triggered", is_hard_stop=True
    )
    assert att_cls == ATTENTION_HUMAN_REQUIRED
    assert req_human is True

    att_cls, req_human = classify_attention_event(
        CATEGORY_VALIDATION, "Validation Error", "Report validation failed", is_validation_failure=True
    )
    assert att_cls == ATTENTION_HUMAN_REQUIRED
    assert req_human is True

    # 2. Operational issues -> operator
    att_cls, req_human = classify_attention_event(
        CATEGORY_BLOCKER, "PR Blocked", "Merge conflict detected", is_blocker=True
    )
    assert att_cls == ATTENTION_OPERATOR
    assert req_human is False

    att_cls, req_human = classify_attention_event(
        CATEGORY_LEASE, "Lease Expired", "Session lease expired", is_stale=True
    )
    assert att_cls == ATTENTION_OPERATOR
    assert req_human is False

    # 3. Routine workflow transitions -> routine
    att_cls, req_human = classify_attention_event(
        CATEGORY_WORKFLOW, "PR Active", "PR in review"
    )
    assert att_cls == ATTENTION_ROUTINE
    assert req_human is False


def test_notification_snapshot_aggregation():
    reg = load_registry()
    proj_id = reg.projects[0].id if reg.projects else "gitea-tools"

    mock_queue = QueueSnapshot(
        project_id=proj_id,
        repo_label="org/repo",
        prs=(
            QueueItem(
                number=101,
                title="Blocked PR",
                badges=("blocked",),
                extra={},
            ),
            QueueItem(
                number=102,
                title="Normal PR",
                badges=("in-review",),
                extra={},
            ),
        ),
        issues=(),
        pr_pagination=None,
        issue_pagination=None,
    )

    mock_leases = LeaseSnapshot(
        project_id=proj_id,
        repo_label="org/repo",
        issue_lock=None,
        claim_inventory={},
        reviewer_leases=(
            {
                "pr_number": 101,
                "status": "expired",
                "is_expired": True,
            },
        ),
        duplicate_prs=(
            CollisionWarning(
                kind="duplicate_pr",
                message="Multiple open PRs for issue #101",
                issue_number=101,
                pr_numbers=(101, 103),
            ),
        ),
        duplicate_branches=(),
        collision_history=(),
        fetch_error=None,
    )

    mock_version = VersionInfo(
        git_sha="abc1234",
        git_describe="v1.0.0",
        control_plane_schema_version=1,
        python_version="3.11",
        known=True,
    )

    mock_stale = StaleRuntime(
        daemon_head="abc1234",
        checkout_head="abc1234",
        remote_head="abc1234",
        stale=False,
        determinable=True,
        mutation_safe=True,
        reasons=(),
    )

    mock_health = SystemHealthSnapshot(
        status="degraded",
        ready=False,
        readiness_complete=True,
        readiness_reasons=("Auth failure",),
        service="webui",
        mode="test",
        version=mock_version,
        started_at="2026-07-25T00:00:00Z",
        uptime_seconds=100.0,
        timestamp="2026-07-25T00:00:00Z",
        deep_probes_requested=True,
        dependencies=(
            DependencyProbe(
                name="auth_service",
                kind="auth",
                status="unauthorized",
                detail="Token expired",
                required=True,
            ),
        ),
        mcp_namespaces=(),
        stale_runtime=mock_stale,
        probe_errors=(),
    )

    snapshot = load_notifications_snapshot(
        proj_id,
        load_queue=lambda _id: mock_queue,
        load_leases=lambda **_kwargs: mock_leases,
        load_health=lambda **_kwargs: mock_health,
    )

    assert snapshot.project_id == proj_id
    assert snapshot.total_count == 5
    assert snapshot.human_required_count >= 1  # auth probe failure
    assert snapshot.operator_count >= 3  # blocked PR + expired lease + duplicate PR collision
    assert snapshot.routine_count >= 1  # normal PR

    # Inbox items should include operator and human-required items only
    inbox_classes = {item.attention_class for item in snapshot.inbox_items}
    assert ATTENTION_ROUTINE not in inbox_classes
    assert ATTENTION_OPERATOR in inbox_classes
    assert ATTENTION_HUMAN_REQUIRED in inbox_classes


def test_snapshot_to_dict_and_redaction():
    item = NotificationItem(
        id="notif-1",
        attention_class=ATTENTION_HUMAN_REQUIRED,
        category=CATEGORY_AUTH,
        title="Auth Error",
        summary="Failed auth header: Bearer secret_token_12345",
        work_kind="system",
        work_number=None,
        project_id="test-proj",
        repo_label="org/repo",
        created_at="2026-07-25T16:00:00Z",
        requires_human=True,
    )
    snap = NotificationSnapshot(
        project_id="test-proj",
        repo_label="org/repo",
        items=(item,),
        human_required_count=1,
        operator_count=0,
        routine_count=0,
        total_count=1,
    )

    data = snapshot_to_dict(snap)
    assert data["project_id"] == "test-proj"
    assert data["human_required_count"] == 1
    assert len(data["inbox_items"]) == 1

    # Redaction test
    summary = data["inbox_items"][0]["summary"]
    assert "secret_token_12345" not in summary
    assert "<redacted>" in summary or "Bearer" in summary


def test_notifications_html_views():
    item = NotificationItem(
        id="notif-1",
        attention_class=ATTENTION_HUMAN_REQUIRED,
        category=CATEGORY_AUTH,
        title="Critical Auth Failure",
        summary="Auth failure details",
        work_kind="issue",
        work_number=42,
        project_id="test-proj",
        repo_label="org/repo",
        created_at="2026-07-25T16:00:00Z",
        requires_human=True,
    )
    snap = NotificationSnapshot(
        project_id="test-proj",
        repo_label="org/repo",
        items=(item,),
        human_required_count=1,
        operator_count=0,
        routine_count=0,
        total_count=1,
    )

    html = render_notifications_page(snap, filter_class="inbox")
    assert "Notifications &amp; Attention Inbox" in html or "Notifications & Attention Inbox" in html
    assert "Critical Auth Failure" in html
    assert "HUMAN REQUIRED" in html
    assert "Human Required" in html


def test_notifications_app_routes():
    app = create_app()
    client = TestClient(app)

    # 1. HTML Route
    res = client.get("/notifications")
    assert res.status_code == 200
    assert "Notifications" in res.text
    assert "Attention Inbox" in res.text

    # 2. API Route /api/v1/notifications
    res_api = client.get("/api/v1/notifications")
    assert res_api.status_code == 200
    json_data = res_api.json()
    assert "human_required_count" in json_data
    assert "operator_count" in json_data
    assert "routine_count" in json_data
    assert "inbox_items" in json_data

    # 3. Compatibility Alias /api/notifications
    res_alias = client.get("/api/notifications")
    assert res_alias.status_code == 200
    assert res_alias.json()["project_id"] == json_data["project_id"]


def test_classify_ignores_human_authored_title_and_summary_keywords():
    """B1: keywords in human-authored titles must not escalate routine work (#905)."""
    # Routine transition whose title/summary mention critical-boundary words
    att_cls, req_human = classify_attention_event(
        CATEGORY_WORKFLOW,
        "record irrecoverable decision lock provenance",
        "PR #999 'record irrecoverable decision lock provenance' is in routine state in-review.",
    )
    assert att_cls == ATTENTION_ROUTINE
    assert req_human is False

    att_cls, req_human = classify_attention_event(
        CATEGORY_WORKFLOW,
        "fix unauthorized token path",
        "Issue #1 'fix unauthorized token path' state: claimed. hard stop docs only.",
    )
    assert att_cls == ATTENTION_ROUTINE
    assert req_human is False

    # Structured flags still escalate (machine-driven)
    att_cls, req_human = classify_attention_event(
        CATEGORY_SYSTEM,
        "anything",
        "anything with hard stop in text",
        is_hard_stop=True,
    )
    assert att_cls == ATTENTION_HUMAN_REQUIRED
    assert req_human is True


def test_notification_ids_are_unique_across_probe_errors_and_collisions():
    """B2: published notification ids must be unique within a snapshot (#905)."""
    reg = load_registry()
    proj_id = reg.projects[0].id if reg.projects else "gitea-tools"

    mock_queue = QueueSnapshot(
        project_id=proj_id,
        repo_label="org/repo",
        prs=(),
        issues=(),
        pr_pagination=None,
        issue_pagination=None,
    )
    mock_leases = LeaseSnapshot(
        project_id=proj_id,
        repo_label="org/repo",
        issue_lock=None,
        claim_inventory={},
        reviewer_leases=(),
        duplicate_prs=(
            CollisionWarning(
                kind="duplicate_pr",
                message="Multiple open PRs for issue #10",
                issue_number=10,
                pr_numbers=(10, 11),
            ),
            CollisionWarning(
                kind="duplicate_branch",
                message="Another collision without issue",
                issue_number=None,
                pr_numbers=(12, 13),
            ),
            CollisionWarning(
                kind="duplicate_pr",
                message="Second issue collision",
                issue_number=10,
                pr_numbers=(14, 15),
            ),
        ),
        duplicate_branches=(),
        collision_history=(),
        fetch_error=None,
    )
    mock_version = VersionInfo(
        git_sha="abc1234",
        git_describe="v1.0.0",
        control_plane_schema_version=1,
        python_version="3.11",
        known=True,
    )
    mock_stale = StaleRuntime(
        daemon_head="abc1234",
        checkout_head="abc1234",
        remote_head="abc1234",
        stale=False,
        determinable=True,
        mutation_safe=True,
        reasons=(),
    )
    mock_health = SystemHealthSnapshot(
        status="degraded",
        ready=False,
        readiness_complete=True,
        readiness_reasons=(),
        service="webui",
        mode="test",
        version=mock_version,
        started_at="2026-07-25T00:00:00Z",
        uptime_seconds=100.0,
        timestamp="2026-07-25T00:00:00Z",
        deep_probes_requested=True,
        dependencies=(),
        mcp_namespaces=(),
        stale_runtime=mock_stale,
        probe_errors=("error alpha", "error beta"),
    )

    snapshot = load_notifications_snapshot(
        proj_id,
        load_queue=lambda _id: mock_queue,
        load_leases=lambda **_kwargs: mock_leases,
        load_health=lambda **_kwargs: mock_health,
    )
    ids = [item.id for item in snapshot.items]
    assert len(ids) == len(set(ids)), f"duplicate notification ids: {ids}"
    assert any(i.startswith(f"notif-sys-err-{proj_id}-") for i in ids)
    assert any(i.startswith("notif-collision-") for i in ids)


def test_probe_errors_do_not_set_fetch_error():
    """B3: probe_errors must not be reported as fetch_error (#905)."""
    reg = load_registry()
    proj_id = reg.projects[0].id if reg.projects else "gitea-tools"

    mock_queue = QueueSnapshot(
        project_id=proj_id,
        repo_label="org/repo",
        prs=(),
        issues=(),
        pr_pagination=None,
        issue_pagination=None,
        fetch_error=None,
    )
    mock_leases = LeaseSnapshot(
        project_id=proj_id,
        repo_label="org/repo",
        issue_lock=None,
        claim_inventory={},
        reviewer_leases=(),
        duplicate_prs=(),
        duplicate_branches=(),
        collision_history=(),
        fetch_error=None,
    )
    mock_version = VersionInfo(
        git_sha="abc1234",
        git_describe="v1.0.0",
        control_plane_schema_version=1,
        python_version="3.11",
        known=True,
    )
    mock_stale = StaleRuntime(
        daemon_head="abc1234",
        checkout_head="abc1234",
        remote_head="abc1234",
        stale=False,
        determinable=True,
        mutation_safe=True,
        reasons=(),
    )
    mock_health = SystemHealthSnapshot(
        status="degraded",
        ready=False,
        readiness_complete=True,
        readiness_reasons=(),
        service="webui",
        mode="test",
        version=mock_version,
        started_at="2026-07-25T00:00:00Z",
        uptime_seconds=100.0,
        timestamp="2026-07-25T00:00:00Z",
        deep_probes_requested=True,
        dependencies=(),
        mcp_namespaces=(),
        stale_runtime=mock_stale,
        probe_errors=("probe blew up",),
    )

    snapshot = load_notifications_snapshot(
        proj_id,
        load_queue=lambda _id: mock_queue,
        load_leases=lambda **_kwargs: mock_leases,
        load_health=lambda **_kwargs: mock_health,
    )
    assert snapshot.fetch_error is None
    # probe errors still appear as items
    assert any("probe blew up" in item.summary for item in snapshot.items)