proofpoint-quarantine-portal/tests/test_report_release.py

"""The Report & Release pipeline — the highest-risk behaviour.

Asserts on exact PPS call kwargs, not just counts: the move-first order, the re-read of
localguids before every later step (never the cached ones), and the localguid-stability
rule (release must NOT pass deletedfolder) are all invisible to a call-count check but
critical to correctness. The last group covers the other half of the design — waiting
between steps must not block the job queue.
"""

import pytest

from conftest import FakePPS
from pipeline import (
    ALLOWED_STEPS,
    describe_pipeline,
    pipeline_destination,
    run_step,
)
from pps_client import PPSError


def _cfg(steps, move_target="Rep", deleted="Deleted", step_delay=0):
    return {
        "deleted_folder": deleted,
        "list_query": "from=*",
        "default_days_back": 7,
        "chunk_size": 25,
        "report_release": {
            "steps": steps,
            "move_target": move_target,
            "step_delay_seconds": step_delay,
        },
    }


def _chunk():
    return [
        {"localguid": "6:6:1", "guid": "g1"},
        {"localguid": "6:6:2", "guid": "g2"},
    ]


def _moved_records():
    """What a search of the move target returns once the messages have landed there."""
    return [{"guid": "g1", "localguid": "R:1"}, {"guid": "g2", "localguid": "R:2"}]


# ---------------------------------------------------------------- single steps

def test_first_step_uses_the_job_localguids_without_searching():
    pps = FakePPS()
    folder = run_step(pps, _cfg(["move", "release"]), "move", "Quarantine", _chunk(),
                      refresh=False)
    assert folder == "Rep"
    assert pps.searches() == []
    move = pps.acts()[0]
    assert move["action"] == "move" and move["folder"] == "Quarantine"
    assert move["localguids"] == ["6:6:1", "6:6:2"]
    assert move["targetfolder"] == "Rep"


def test_later_step_rereads_and_never_uses_cached_localguids():
    pps = FakePPS(records=_moved_records())
    folder = run_step(pps, _cfg(["move", "release"]), "release", "Rep", _chunk(),
                      refresh=True)
    assert folder == "Rep"
    assert [s["folder"] for s in pps.searches()] == ["Rep"]   # re-read in the NEW folder
    release = pps.acts()[0]
    assert release["action"] == "release"
    assert release["folder"] == "Rep"
    assert release["localguids"] == ["R:1", "R:2"]            # fresh, not "6:6:*"
    assert release["deletedfolder"] is None                   # THE stability rule


def test_release_carries_no_deletedfolder_even_as_first_step():
    pps = FakePPS()
    run_step(pps, _cfg(["release"]), "release", "Quarantine", _chunk(), refresh=False)
    assert pps.acts()[0]["deletedfolder"] is None


def test_reread_falls_back_to_the_deleted_folder():
    """Some deployments relocate messages to the deleted folder behind our back."""
    class OnlyDeleted(FakePPS):
        def search(self, folder, query, limit=200, days_back=7, enddate=None):
            self.calls.append({"m": "search", "folder": folder, "enddate": enddate})
            return _moved_records() if folder == "Deleted" else []

    pps = OnlyDeleted()
    run_step(pps, _cfg(["move", "release"]), "release", "Rep", _chunk(), refresh=True)
    assert [s["folder"] for s in pps.searches()] == ["Rep", "Deleted"]
    assert pps.acts()[0]["folder"] == "Deleted"
    assert pps.acts()[0]["localguids"] == ["R:1", "R:2"]


def test_move_retries_once_after_rereading_when_the_stored_handles_are_stale():
    class StaleHandles(FakePPS):
        def act(self, action, folder, localguids, **kw):
            result = super().act(action, folder, localguids, **kw)   # record the attempt
            if action == "move" and "6:6:1" in localguids:
                raise PPSError("no such localguid")
            return result

    pps = StaleHandles(records=_moved_records())
    run_step(pps, _cfg(["move"]), "move", "Quarantine", _chunk(), refresh=False)
    assert [s["folder"] for s in pps.searches()] == ["Quarantine"]
    assert [a["localguids"] for a in pps.acts()] == [["6:6:1", "6:6:2"], ["R:1", "R:2"]]
    assert pps.acts()[-1]["targetfolder"] == "Rep"


def test_reread_without_guids_raises():
    pps = FakePPS()
    with pytest.raises(PPSError):
        run_step(pps, _cfg(["move", "release"]), "release", "Rep",
                 [{"localguid": "6:6:1"}], refresh=True)   # no guid stored


# ---------------------------------------------------------------- descriptions

def test_pipeline_destination():
    assert pipeline_destination(_cfg(["release"])) is None
    assert pipeline_destination(_cfg(["move"])) == "Rep"
    assert pipeline_destination(_cfg(["move", "release"])) == "Rep"


def test_describe_pipeline_reads_in_execution_order():
    text = describe_pipeline(_cfg(["move", "release"]))
    assert text.index("Rep") < text.index("released")


def test_allowed_steps_is_canonical_order_and_never_deletes():
    assert ALLOWED_STEPS == ("move", "release")


# ---------------------------------------------------------------- worker driving

def _enqueue(queue):
    return queue.enqueue("report_release", "Quarantine", _chunk(), {}, user="dev@x")


def _run_next(queue):
    """Claim and process one job; returns the claimed job row (None if none is due)."""
    job = queue._claim_next()
    if job is not None:
        queue._process(job)
    return job


def _job(queue, job_id):
    return next(j for j in queue.recent_jobs(limit=50) if j["id"] == job_id)


def test_zero_delay_runs_every_step_in_one_pass(make_queue):
    pps = FakePPS(records=_moved_records())
    q = make_queue(pps, _cfg(["move", "release"], step_delay=0))
    job_id = _enqueue(q)

    _run_next(q)
    assert [a["action"] for a in pps.acts()] == ["move", "release"]
    assert _job(q, job_id)["status"] == "done"
    # Even with no wait, the release step re-reads rather than reusing the move's handles.
    assert pps.acts()[1]["localguids"] == ["R:1", "R:2"]


def test_delay_parks_the_job_instead_of_blocking_the_queue(make_queue):
    pps = FakePPS(records=_moved_records())
    q = make_queue(pps, _cfg(["move", "release"], step_delay=60))
    rr_id = _enqueue(q)
    other_id = q.enqueue("delete", "Quarantine", [{"localguid": "x"}], {}, user="dev@x")

    # First pass: the move, then the job parks itself.
    assert _run_next(q)["id"] == rr_id
    assert [a["action"] for a in pps.acts()] == ["move"]
    parked = _job(q, rr_id)
    assert parked["status"] == "pending" and parked["run_after"]

    # The job queued behind it is NOT blocked by the wait — it runs now.
    assert _run_next(q)["id"] == other_id
    assert [a["action"] for a in pps.acts()] == ["move", "delete"]
    assert _job(q, other_id)["status"] == "done"

    # Nothing else is due while the wait is on.
    assert q._claim_next() is None

    # Once the delay has passed the job is re-claimed and finishes on fresh handles.
    q._conn.execute("UPDATE jobs SET run_after='2000-01-01 00:00:00' WHERE id=?", (rr_id,))
    q._conn.commit()
    assert _run_next(q)["id"] == rr_id
    release = [a for a in pps.acts() if a["action"] == "release"][0]
    assert [s["folder"] for s in pps.searches()] == ["Rep"]
    assert release["folder"] == "Rep" and release["localguids"] == ["R:1", "R:2"]
    assert release["deletedfolder"] is None
    assert _job(q, rr_id)["status"] == "done"


def test_resume_keeps_the_steps_it_started_with(make_queue):
    """An admin edit during the wait must not change a job already in flight."""
    cfg = _cfg(["move", "release"], step_delay=60)
    pps = FakePPS(records=_moved_records())
    q = make_queue(pps, cfg)
    rr_id = _enqueue(q)
    _run_next(q)

    cfg["report_release"]["steps"] = ["move"]        # live config change during the wait
    q._conn.execute("UPDATE jobs SET run_after=NULL WHERE id=?", (rr_id,))
    q._conn.commit()
    _run_next(q)

    # Still released: the plan came from the job's state, not the edited config.
    assert [a["action"] for a in pps.acts()] == ["move", "release"]
    assert _job(q, rr_id)["status"] == "done"


def test_a_chunk_that_failed_a_step_is_skipped_by_later_steps(make_queue):
    class OneBadMove(FakePPS):
        def act(self, action, folder, localguids, **kw):
            result = super().act(action, folder, localguids, **kw)   # record the attempt
            if action == "move" and localguids == ["6:6:2"]:
                raise PPSError("boom")
            return result

    cfg = _cfg(["move", "release"], step_delay=0)
    cfg["chunk_size"] = 1                                   # one message per chunk
    pps = OneBadMove(records=[{"guid": "g1", "localguid": "R:1"}])
    q = make_queue(pps, cfg)
    rr_id = _enqueue(q)
    _run_next(q)

    # The message whose move failed is never released; the other one is.
    assert [(a["action"], a["localguids"]) for a in pps.acts()] == [
        ("move", ["6:6:1"]), ("move", ["6:6:2"]), ("release", ["R:1"])]
    row = _job(q, rr_id)
    # The failed move re-read first (see the retry above) and couldn't locate it either.
    assert row["status"] == "failed" and "not found" in row["error"]
    assert row["processed"] == 1


def test_retry_clears_the_parked_state(make_queue):
    pps = FakePPS(records=_moved_records())
    q = make_queue(pps, _cfg(["move", "release"], step_delay=60))
    rr_id = _enqueue(q)
    _run_next(q)
    q._conn.execute("UPDATE jobs SET status='failed' WHERE id=?", (rr_id,))
    q._conn.commit()

    assert q.retry(rr_id) is True
    row = q._conn.execute("SELECT run_after, state FROM jobs WHERE id=?", (rr_id,)).fetchone()
    assert row["run_after"] is None and row["state"] == "{}"
    # ...so it starts over at the first step, on the job's own localguids.
    _run_next(q)
    assert [a["action"] for a in pps.acts()] == ["move", "move"]