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 localguid-stability rule (release
must NOT pass deletedfolder) and the lazy guid-refind are both invisible to a call-count
check but critical to correctness.
"""

import pytest

from conftest import FakePPS
from pipeline import (
    ALLOWED_STEPS,
    describe_pipeline,
    pipeline_destination,
    run_pipeline,
)
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,
        "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 test_release_only_has_no_deletedfolder():
    pps = FakePPS()
    run_pipeline(pps, _cfg(["release"]), "Quarantine", _chunk())
    acts = pps.acts()
    assert len(acts) == 1
    assert acts[0]["action"] == "release"
    assert acts[0]["deletedfolder"] is None    # THE localguid-stability rule
    assert acts[0]["folder"] == "Quarantine"


def test_release_move_is_two_calls_zero_searches():
    pps = FakePPS()
    run_pipeline(pps, _cfg(["release", "move"]), "Quarantine", _chunk())
    assert [a["action"] for a in pps.acts()] == ["release", "move"]
    assert pps.acts()[0]["deletedfolder"] is None
    assert pps.acts()[1]["targetfolder"] == "Rep"
    assert pps.acts()[1]["folder"] == "Quarantine"
    assert pps.searches() == []                 # lazy refind: no search needed


def test_move_fallback_refinds_in_deleted_folder():
    class MovingPPS(FakePPS):
        def act(self, action, folder, localguids, *, targetfolder=None, deletedfolder=None, scan=False):
            if action == "move" and folder == "Quarantine":
                self.calls.append({"m": "act", "action": action, "folder": folder,
                                   "localguids": list(localguids), "targetfolder": targetfolder,
                                   "deletedfolder": deletedfolder, "scan": scan})
                raise PPSError("release relocated it")
            return super().act(action, folder, localguids, targetfolder=targetfolder,
                               deletedfolder=deletedfolder, scan=scan)

    pps = MovingPPS(records=[{"guid": "g1", "localguid": "D:1"}, {"guid": "g2", "localguid": "D:2"}])
    run_pipeline(pps, _cfg(["release", "move"]), "Quarantine", _chunk())
    # release, failed move on Quarantine, search Deleted, move from Deleted
    assert [s["folder"] for s in pps.searches()] == ["Deleted"]
    moves = [a for a in pps.acts() if a["action"] == "move"]
    assert moves[-1]["folder"] == "Deleted"
    assert sorted(moves[-1]["localguids"]) == ["D:1", "D:2"]


def test_release_move_delete_refinds_between_move_and_delete():
    pps = FakePPS(records=[{"guid": "g1", "localguid": "R:1"}, {"guid": "g2", "localguid": "R:2"}])
    run_pipeline(pps, _cfg(["release", "move", "delete"]), "Quarantine", _chunk())
    actions = [a["action"] for a in pps.acts()]
    assert actions == ["release", "move", "delete"]
    # delete happens in the move_target after a refind there
    assert len(pps.searches()) == 1 and pps.searches()[0]["folder"] == "Rep"
    delete = [a for a in pps.acts() if a["action"] == "delete"][0]
    assert delete["folder"] == "Rep"
    assert delete["deletedfolder"] == "Deleted"


def test_refind_no_guids_raises():
    pps = FakePPS()
    chunk = [{"localguid": "6:6:1"}]   # no guid
    with pytest.raises(PPSError):
        # force a refind by making move fail
        class P(FakePPS):
            def act(self, *a, **k):
                if a[0] == "move":
                    raise PPSError("relocated")
                return super().act(*a, **k)
        run_pipeline(P(), _cfg(["release", "move"]), "Quarantine", chunk)


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


def test_describe_pipeline():
    text = describe_pipeline(_cfg(["release", "move", "delete"]))
    assert "released in place" in text and "Rep" in text and "Deleted" in text
    assert "→" in text


def test_allowed_steps_is_canonical_order():
    assert ALLOWED_STEPS == ("release", "move", "delete")


def test_step_delay_waits_between_steps():
    pps = FakePPS()
    waits = []
    run_pipeline(pps, _cfg(["release", "move"], step_delay=60), "Quarantine", _chunk(),
                 sleep=waits.append)
    # One wait, of the configured length, between the two steps.
    assert waits == [60]


def test_no_delay_before_first_or_when_zero():
    pps = FakePPS()
    waits = []
    run_pipeline(pps, _cfg(["release"], step_delay=60), "Quarantine", _chunk(), sleep=waits.append)
    assert waits == []                      # single step -> no wait
    waits.clear()
    run_pipeline(pps, _cfg(["release", "move"], step_delay=0), "Quarantine", _chunk(),
                 sleep=waits.append)
    assert waits == []                      # delay 0 -> disabled


def test_delay_scales_with_step_count():
    pps = FakePPS(records=[{"guid": "g1", "localguid": "R:1"}, {"guid": "g2", "localguid": "R:2"}])
    waits = []
    run_pipeline(pps, _cfg(["release", "move", "delete"], step_delay=30), "Quarantine",
                 _chunk(), sleep=waits.append)
    assert waits == [30, 30]                # before move, before delete