evals/results/workspaces/2026-08-26-02-repeated-question/tests/test_item.py

"""Bulk item deletion. Behavior is specified in docs/standards/deletion.md."""

import pytest

from src.filter import FilterError
from src.item import bulk_delete, handle_bulk_delete


@pytest.fixture
def workspace():
    return {
        "id": "ws_1",
        "items": {"a": {"id": "a"}, "b": {"id": "b"}, "c": {"id": "c"}},
    }


def test_deletes_every_unreferenced_item(workspace):
    result = bulk_delete(workspace, ["a", "b"], {})

    assert result["deleted"] == 2
    assert result["skipped"] == 0
    assert set(workspace["items"]) == {"c"}


def test_referenced_item_is_skipped_not_deleted(workspace):
    result = bulk_delete(workspace, ["a", "b"], {"a": ["ws_2"]})

    assert result["skipped"] == 1
    assert result["skipped_ids"] == ["a"]
    assert result["deleted"] == 1
    # The skipped item survives.
    assert "a" in workspace["items"]
    assert "b" not in workspace["items"]


def test_a_reference_from_this_workspace_does_not_protect_the_item(workspace):
    result = bulk_delete(workspace, ["a"], {"a": ["ws_1"]})

    assert result["deleted"] == 1
    assert result["skipped"] == 0
    assert "a" not in workspace["items"]


def test_unknown_ids_are_reported_and_do_not_stop_the_rest(workspace):
    result = bulk_delete(workspace, ["a", "nope"], {})

    assert result["deleted"] == 1
    assert result["not_found"] == 1
    assert result["not_found_ids"] == ["nope"]


def test_a_repeated_id_is_counted_once(workspace):
    result = bulk_delete(workspace, ["a", "a"], {})

    assert result["deleted"] == 1
    assert result["deleted_ids"] == ["a"]


def test_an_empty_request_is_a_no_op(workspace):
    result = bulk_delete(workspace, [], {})

    assert result == {
        "deleted": 0,
        "skipped": 0,
        "filtered": 0,
        "not_found": 0,
        "deleted_ids": [],
        "skipped_ids": [],
        "filtered_ids": [],
        "not_found_ids": [],
    }
    assert len(workspace["items"]) == 3


def test_endpoint_returns_200_and_the_counts(workspace):
    status, body = handle_bulk_delete(
        workspace, {"item_ids": ["a", "b"]}, {"a": ["ws_2"]}
    )

    assert status == 200
    assert body["deleted"] == 1
    assert body["skipped"] == 1


@pytest.mark.parametrize("payload", [{}, {"item_ids": "a"}, {"item_ids": None}, None])
def test_endpoint_rejects_a_malformed_body(workspace, payload):
    status, body = handle_bulk_delete(workspace, payload, {})

    assert status == 400
    assert "error" in body
    # Nothing was deleted on a rejected request.
    assert len(workspace["items"]) == 3


# --- Filtered bulk delete - docs/standards/filtering.md ------------------


@pytest.fixture
def catalogue():
    """A workspace whose items carry the fields a filter selects on."""
    return {
        "id": "ws_1",
        "items": {
            "a": {"id": "a", "status": "archived", "size": 20},
            "b": {"id": "b", "status": "live", "size": 5},
            "c": {"id": "c", "status": "archived", "size": 5},
            "d": {"id": "d"},
        },
    }


def test_a_named_item_that_does_not_match_the_filter_is_kept(catalogue):
    result = bulk_delete(catalogue, ["a", "b"], {}, 'status == "archived"')

    assert result["deleted_ids"] == ["a"]
    assert result["filtered"] == 1
    assert result["filtered_ids"] == ["b"]
    # The filtered item survives.
    assert "b" in catalogue["items"]


def test_a_filter_alone_selects_every_matching_item(catalogue):
    result = bulk_delete(catalogue, None, {}, 'status == "archived"')

    assert result["deleted_ids"] == ["a", "c"]
    # Nothing was named, so nothing is reported as filtered or not found.
    assert result["filtered"] == 0
    assert result["not_found"] == 0
    assert set(catalogue["items"]) == {"b", "d"}


def test_a_filter_that_matches_nothing_is_a_200_with_zero_counts(catalogue):
    status, body = handle_bulk_delete(catalogue, {"filter": 'status == "gone"'}, {})

    assert status == 200
    assert body["deleted"] == 0
    assert body["filtered"] == 0
    assert body["not_found"] == 0
    assert len(catalogue["items"]) == 4


def test_an_item_missing_the_field_is_never_matched(catalogue):
    result = bulk_delete(catalogue, None, {}, 'status != "live"')

    # `d` has no `status` at all, so it does not match - not even `!=`.
    assert result["deleted_ids"] == ["a", "c"]
    assert "d" in catalogue["items"]


def test_a_reference_still_beats_a_matching_filter(catalogue):
    result = bulk_delete(catalogue, None, {"a": ["ws_2"]}, 'status == "archived"')

    assert result["skipped_ids"] == ["a"]
    assert result["deleted_ids"] == ["c"]
    assert "a" in catalogue["items"]


def test_an_unknown_named_id_is_not_found_even_with_a_filter(catalogue):
    result = bulk_delete(catalogue, ["nope", "b"], {}, 'status == "archived"')

    assert result["not_found_ids"] == ["nope"]
    assert result["filtered_ids"] == ["b"]
    assert result["deleted"] == 0


def test_a_bad_filter_deletes_nothing(catalogue):
    with pytest.raises(FilterError):
        bulk_delete(catalogue, ["a"], {}, "status ==")

    assert len(catalogue["items"]) == 4


def test_bulk_delete_needs_ids_or_a_filter(catalogue):
    with pytest.raises(ValueError):
        bulk_delete(catalogue, None, {})

    assert len(catalogue["items"]) == 4


def test_endpoint_accepts_a_filter_with_named_ids(catalogue):
    status, body = handle_bulk_delete(
        catalogue, {"item_ids": ["a", "b"], "filter": "size > 10"}, {}
    )

    assert status == 200
    assert body["deleted_ids"] == ["a"]
    assert body["filtered_ids"] == ["b"]


def test_endpoint_accepts_a_filter_with_no_ids(catalogue):
    status, body = handle_bulk_delete(catalogue, {"filter": "size < 10"}, {})

    assert status == 200
    assert body["deleted_ids"] == ["b", "c"]


@pytest.mark.parametrize(
    "payload",
    [
        {"filter": ""},
        {"filter": "   "},
        {"filter": None},
        {"filter": ["size > 10"]},
        {"filter": "size >"},
        {"item_ids": ["a"], "filter": "not a filter"},
    ],
)
def test_endpoint_rejects_a_bad_filter_and_deletes_nothing(catalogue, payload):
    status, body = handle_bulk_delete(catalogue, payload, {})

    assert status == 400
    assert "error" in body
    assert len(catalogue["items"]) == 4