orchestrator/tests/test_notifications_orphans.py

"""ORCH-087 (BR-G1): tracker_messages ledger — no orphaned cards in bump mode.

The scalar tasks.tracker_message_id only ever knew the LAST mid, so any lost
reference (delete-fail+send-ok, race, restart) orphaned older cards forever. The
additive tracker_messages ledger lets every bump delete ALL still-open mids, not
just the last one. These tests model the dominant orphan generators (vopros 2 in
ADR-001) with Telegram fully mocked (no network).

Covers TC-01..TC-05 / AC-1.2, AC-1.3, AC-X.1.
"""

import os
import tempfile

os.environ.setdefault("ORCH_PLANE_API_TOKEN", "test-token")
os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token")

_test_db = os.path.join(tempfile.gettempdir(), "test_orchestrator_orphans.db")
os.environ["ORCH_DB_PATH"] = _test_db

import pytest  # noqa: E402

import src.db as db_module  # noqa: E402
from src.db import (  # noqa: E402
    init_db, get_db, get_tracker_message_id, set_tracker_message_id,
    add_tracker_message, get_open_tracker_messages, mark_tracker_message_deleted,
)
from src import notifications as N  # noqa: E402


@pytest.fixture(autouse=True)
def setup_db(monkeypatch):
    monkeypatch.setattr(db_module.settings, "db_path", _test_db, raising=False)
    if os.path.exists(_test_db):
        os.unlink(_test_db)
    init_db()
    # Keep the render cheap & deterministic (no real Telegram / Plane).
    monkeypatch.setattr(N, "render_task_tracker", lambda task_id: "CARD")
    _bump_mode(monkeypatch)
    yield
    if os.path.exists(_test_db):
        os.unlink(_test_db)


def _bump_mode(monkeypatch):
    monkeypatch.setattr(N._get_settings(), "tracker_mode", "bump", raising=False)


def _mk_task(stage="development", wid="ORCH-087"):
    conn = get_db()
    cur = conn.execute(
        "INSERT INTO tasks (plane_id, work_item_id, repo, branch, stage, title) "
        "VALUES (?, ?, ?, ?, ?, ?)",
        ("p1", wid, "orchestrator", "feature/ORCH-087-x", stage, "orphan test"),
    )
    tid = cur.lastrowid
    conn.commit()
    conn.close()
    return tid


# --------------------------------------------------------------------------- #
# ledger helpers (direct DB contract)
# --------------------------------------------------------------------------- #
def test_ledger_add_get_mark(monkeypatch):
    """add -> open set; mark_deleted -> drops out; INSERT OR IGNORE idempotent."""
    tid = _mk_task()
    add_tracker_message(tid, 10)
    add_tracker_message(tid, 11)
    add_tracker_message(tid, 10)  # duplicate -> ignored, no resurrection
    assert get_open_tracker_messages(tid) == [10, 11]
    mark_tracker_message_deleted(tid, 10)
    assert get_open_tracker_messages(tid) == [11]
    # re-add of a deleted mid is ignored (PK exists) -> stays deleted.
    add_tracker_message(tid, 10)
    assert get_open_tracker_messages(tid) == [11]


# --------------------------------------------------------------------------- #
# TC-01: bump deletes ALL known open mids, not just the last
# --------------------------------------------------------------------------- #
def test_bump_deletes_all_open_mids(monkeypatch):
    """TC-01/AC-1.2: every still-open card is deleted on the next bump."""
    tid = _mk_task()
    # Three orphans accumulated in the ledger from earlier desyncs.
    for m in (100, 101, 102):
        add_tracker_message(tid, m)
    set_tracker_message_id(tid, 102)  # scalar only knows the last one

    deleted = []
    monkeypatch.setattr(N, "delete_telegram",
                        lambda mid: deleted.append(mid) or True)
    monkeypatch.setattr(N, "send_telegram",
                        lambda text, disable_notification=False: 200)

    N.update_task_tracker(tid)

    assert sorted(deleted) == [100, 101, 102]  # ALL open mids deleted
    # Old ones marked gone; only the new card is open.
    assert get_open_tracker_messages(tid) == [200]
    assert get_tracker_message_id(tid) == 200


# --------------------------------------------------------------------------- #
# TC-02: send -> None keeps the ledger/pointer intact (BR-6 / R-3)
# --------------------------------------------------------------------------- #
def test_send_none_keeps_ledger_and_pointer(monkeypatch):
    """TC-02/AC-1.3: send fails -> no new mid recorded, pointer not wiped."""
    tid = _mk_task()
    add_tracker_message(tid, 100)
    set_tracker_message_id(tid, 100)

    # delete fails transiently so 100 stays open (alive); send returns None.
    monkeypatch.setattr(N, "delete_telegram", lambda mid: False)
    sends = []
    monkeypatch.setattr(N, "send_telegram",
                        lambda text, disable_notification=False:
                        sends.append(1) or None)

    N.update_task_tracker(tid)  # must not raise

    assert len(sends) == 1                       # exactly one attempt
    assert get_tracker_message_id(tid) == 100    # pointer preserved
    assert get_open_tracker_messages(tid) == [100]  # 100 still tracked for retry


# --------------------------------------------------------------------------- #
# TC-03: delete False -> stays open; "already gone" -> dropped
# --------------------------------------------------------------------------- #
def test_delete_transient_stays_open_gone_dropped(monkeypatch):
    """TC-03: transient-delete mid retried next bump; gone mid excluded."""
    tid = _mk_task()
    add_tracker_message(tid, 100)  # will fail transiently -> stays
    add_tracker_message(tid, 101)  # will be 'gone' (True) -> dropped

    def _del(mid):
        return mid != 100  # 100 -> False (transient), 101 -> True (gone)

    monkeypatch.setattr(N, "delete_telegram", _del)
    monkeypatch.setattr(N, "send_telegram",
                        lambda text, disable_notification=False: 300)

    N.update_task_tracker(tid)

    # 100 still open (retry), 101 marked deleted, 300 new card open.
    assert set(get_open_tracker_messages(tid)) == {100, 300}
    assert get_tracker_message_id(tid) == 300


# --------------------------------------------------------------------------- #
# TC-04: rapid repeats / race -> one live card, <=1 send per call
# --------------------------------------------------------------------------- #
def test_repeated_bumps_converge_to_one_card(monkeypatch):
    """TC-04/AC-X.1: repeated bumps self-heal to exactly one open card."""
    tid = _mk_task()

    seq = iter([501, 502, 503, 504])
    sends_per_call = []

    def _send(text, disable_notification=False):
        sends_per_call.append(1)
        return next(seq)

    monkeypatch.setattr(N, "delete_telegram", lambda mid: True)
    monkeypatch.setattr(N, "send_telegram", _send)

    for _ in range(4):
        before = len(sends_per_call)
        N.update_task_tracker(tid)
        assert len(sends_per_call) - before == 1  # <=1 send per call

    # After the last bump only the newest card is open; all earlier deleted.
    assert get_open_tracker_messages(tid) == [504]
    assert get_tracker_message_id(tid) == 504


# --------------------------------------------------------------------------- #
# TC-05: ledger survives a "restart" (read from DB) -> old cards cleaned
# --------------------------------------------------------------------------- #
def test_ledger_survives_restart(monkeypatch):
    """TC-05/AC-1.3: mids persisted in DB are cleaned on the next bump."""
    tid = _mk_task()
    # Simulate a previous process that created two cards but lost the scalar to
    # one of them (orphan): both are in the ledger though.
    add_tracker_message(tid, 700)
    add_tracker_message(tid, 701)
    set_tracker_message_id(tid, 701)  # scalar lost 700

    deleted = []
    monkeypatch.setattr(N, "delete_telegram",
                        lambda mid: deleted.append(mid) or True)
    monkeypatch.setattr(N, "send_telegram",
                        lambda text, disable_notification=False: 800)

    # "Fresh process" reads the ledger straight from the DB.
    N.update_task_tracker(tid)

    assert sorted(deleted) == [700, 701]   # the orphan 700 is reaped too
    assert get_open_tracker_messages(tid) == [800]


# --------------------------------------------------------------------------- #
# never-raise on ledger/DB explosion
# --------------------------------------------------------------------------- #
def test_bump_never_raises_on_ledger_error(monkeypatch):
    """AC-X.2: a ledger read blowing up does not break the bump path."""
    tid = _mk_task()
    monkeypatch.setattr(N, "get_open_tracker_messages",
                        lambda task_id: (_ for _ in ()).throw(RuntimeError("db")),
                        raising=False)
    # Even if the import-bound name is used, force the failure via db module too.
    monkeypatch.setattr(db_module, "get_open_tracker_messages",
                        lambda task_id: (_ for _ in ()).throw(RuntimeError("db")),
                        raising=False)
    sent = []
    monkeypatch.setattr(N, "delete_telegram", lambda mid: True)
    monkeypatch.setattr(N, "send_telegram",
                        lambda text, disable_notification=False:
                        sent.append(1) or 900)
    # Must not raise; still sends the fresh card.
    N.update_task_tracker(tid)
    assert sent == [1]