content-find-agent-new/tests/test_walk_engine_pagination.py

from content_agent.business_modules.walk_engine import run_bounded_walk
from tests.p6_walk_helpers import FakeWalkPlatformClient, build_initial_walk_context


def test_walk_engine_pagination_uses_explicit_cursor(tmp_path):
    context = build_initial_walk_context(tmp_path)
    # M3: mock judgment scores 60 (review/pending); pagination requires a pooled
    # success query, so promote the seed decision to pool to exercise paging.
    _override_decisions(context, "ADD_TO_CONTENT_POOL", "success")
    client = FakeWalkPlatformClient()

    result = run_bounded_walk(platform_client=client, **context)

    page_calls = [
        call for call in client.search_calls
        if call.get("search_query_generation_method") == "query_next_page"
    ]
    assert page_calls[0]["page_cursor"] == "10"
    assert any(row["walk_action"] == "fetch_next_page" for row in result["walk_actions"])


def test_walk_engine_pagination_skips_without_cursor(tmp_path):
    context = build_initial_walk_context(tmp_path)
    context["discovered_content_items"][0]["next_cursor"] = ""
    client = FakeWalkPlatformClient()

    run_bounded_walk(platform_client=client, **context)

    assert not [
        call for call in client.search_calls
        if call.get("search_query_generation_method") == "query_next_page"
    ]


def _override_decisions(context, action, effect_status):
    for decision in context["rule_decisions"]:
        decision["decision_action"] = action
        decision["search_query_effect_status"] = effect_status


def test_pagination_requires_success_effect_status(tmp_path):
    context = build_initial_walk_context(tmp_path)
    _override_decisions(context, "KEEP_CONTENT_FOR_REVIEW", "pending")
    client = FakeWalkPlatformClient()

    result = run_bounded_walk(platform_client=client, **context)

    assert not [
        call for call in client.search_calls
        if call.get("search_query_generation_method") == "query_next_page"
    ]
    assert not [row for row in result["walk_actions"] if row["edge_id"] == "query_next_page"]


def test_rule_blocked_query_does_not_page_even_with_cursor(tmp_path):
    context = build_initial_walk_context(tmp_path)
    _override_decisions(context, "REJECT_CONTENT", "rule_blocked")
    assert context["discovered_content_items"][0]["has_more"]
    client = FakeWalkPlatformClient()

    result = run_bounded_walk(platform_client=client, **context)

    assert not [
        call for call in client.search_calls
        if call.get("search_query_generation_method") == "query_next_page"
    ]
    assert not [row for row in result["walk_actions"] if row["edge_id"] == "query_next_page"]


def test_missing_decision_does_not_page(tmp_path):
    context = build_initial_walk_context(tmp_path)
    context["rule_decisions"] = []
    client = FakeWalkPlatformClient()

    result = run_bounded_walk(platform_client=client, **context)

    assert not [
        call for call in client.search_calls
        if call.get("search_query_generation_method") == "query_next_page"
    ]
    assert not [row for row in result["walk_actions"] if row["edge_id"] == "query_next_page"]


def test_success_query_with_cursor_pages_once(tmp_path):
    context = build_initial_walk_context(tmp_path)
    # M3: mock judgment now lands in review/pending; pagination requires a pooled
    # success query, so promote the seed decision to pool.
    _override_decisions(context, "ADD_TO_CONTENT_POOL", "success")
    client = FakeWalkPlatformClient()

    result = run_bounded_walk(platform_client=client, **context)

    page_actions = [
        row for row in result["walk_actions"]
        if row["edge_id"] == "query_next_page" and row["walk_status"] == "success"
    ]
    assert len(page_actions) == 1
    assert page_actions[0]["raw_payload"]["rule_pack_execution"] == {
        "executed": True,
        "executed_rule_pack_id": "douyin_content_discovery_rule_pack_v1",
        "reason": "content_decision_reused_for_walk_gate",
    }