97 lines
3.5 KiB
Python
97 lines
3.5 KiB
Python
"""End-to-end graph tests on the mock backend (fully offline).
|
|
|
|
Drives the LangGraph state machine through the human interrupt with an auto
|
|
resolver, and asserts the gate outcome + tracker write-back behaviour.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from langgraph.types import Command
|
|
|
|
from config.settings import get_settings
|
|
from pricing_agent.graph import build_graph
|
|
from pricing_agent.schemas import Decision, SkuInput
|
|
|
|
|
|
@pytest.fixture()
|
|
def settings(tmp_path):
|
|
s = get_settings()
|
|
# Redirect all writes into the temp dir; use the offline mock provider.
|
|
s.data_backend = "mock"
|
|
s.tracker_backend = "csv"
|
|
s.tracker_csv_out = str(tmp_path / "out.csv")
|
|
s.audit_log_path = str(tmp_path / "audit.csv")
|
|
fixtures = Path(__file__).parent / "fixtures" / "amazon_mock.json"
|
|
s.mock_fixtures_path = str(fixtures)
|
|
return s
|
|
|
|
|
|
def _run(graph, sku_input, auto):
|
|
cfg = {"configurable": {"thread_id": f"sku:{sku_input.sku}"}}
|
|
state = graph.invoke({"sku_input": sku_input}, config=cfg)
|
|
interrupts = state.get("__interrupt__")
|
|
assert interrupts, "graph should pause at human_review"
|
|
payload = interrupts[0].value
|
|
if auto == "smart":
|
|
action = "approve" if payload["decision"] == Decision.APPROVED.value else "reject"
|
|
else:
|
|
action = auto
|
|
resume = {"action": action, "price": None, "note": "test"}
|
|
return graph.invoke(Command(resume=resume), config=cfg)
|
|
|
|
|
|
def test_microfiber_sku_approved_and_written(settings, tmp_path):
|
|
graph = build_graph()
|
|
si = SkuInput(
|
|
sku="UBMICROFIBER4PCQUEENGREY",
|
|
asin="B0MICROFIBERQ",
|
|
product_name="Microfiber 4 Piece Queen Sheet Set - Grey",
|
|
landed_cost=8.00,
|
|
target_price=24.99,
|
|
referral_pct=0.12, # modeled; gate still uses worst-case 15%
|
|
)
|
|
state = _run(graph, si, "smart")
|
|
d = state["decision"]
|
|
assert d.decision == Decision.APPROVED
|
|
assert state["written"] is True
|
|
# Golden margin figures survive end to end.
|
|
assert round(d.fee_stack.break_even_price, 2) == 16.76
|
|
assert round(d.fee_stack.map_floor, 2) == 19.00
|
|
# Tracker row written with the right status.
|
|
out = Path(settings.tracker_csv_out).read_text(encoding="utf-8")
|
|
assert "Pricing Approved" in out
|
|
assert "UBMICROFIBER4PCQUEENGREY" in out
|
|
|
|
|
|
def test_suppressed_sku_blocked_and_not_written(settings, tmp_path):
|
|
graph = build_graph()
|
|
si = SkuInput(
|
|
sku="UBSUPPRESSEDTOWEL",
|
|
asin="B0SUPPRESSED",
|
|
landed_cost=9.00,
|
|
target_price=29.99,
|
|
)
|
|
state = _run(graph, si, "smart")
|
|
d = state["decision"]
|
|
assert d.decision == Decision.BLOCKED
|
|
# smart-mode rejects non-approved => nothing written to the pricing columns.
|
|
assert state["written"] is False
|
|
out_path = Path(settings.tracker_csv_out)
|
|
assert (not out_path.exists()) or ("Pricing Approved" not in out_path.read_text())
|
|
|
|
|
|
def test_human_edit_reprices_and_recomputes(settings, tmp_path):
|
|
graph = build_graph()
|
|
si = SkuInput(sku="UBEDIT", asin="B0MICROFIBERQ", landed_cost=8.00, target_price=24.99)
|
|
cfg = {"configurable": {"thread_id": "sku:UBEDIT"}}
|
|
state = graph.invoke({"sku_input": si}, config=cfg)
|
|
resume = {"action": "edit", "price": 26.99, "note": "bump"}
|
|
state = graph.invoke(Command(resume=resume), config=cfg)
|
|
d = state["decision"]
|
|
assert d.recommended_price == 26.99
|
|
# Margin recomputed at the edited price (higher price => higher CM).
|
|
assert d.fee_stack.selling_price == 26.99
|
|
assert state["written"] is True
|