58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
from services.pchome_backfill_status import (
|
|
PchomeBackfillAlreadyRunning,
|
|
fail_pchome_backfill_run,
|
|
finish_pchome_backfill_run,
|
|
get_pchome_backfill_status,
|
|
start_pchome_backfill_run,
|
|
update_pchome_backfill_run,
|
|
)
|
|
|
|
|
|
def test_pchome_backfill_status_tracks_running_and_completion(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("PCHOME_BACKFILL_STATUS_PATH", str(tmp_path / "status.json"))
|
|
|
|
run = start_pchome_backfill_run(limit=12, operator="tester")
|
|
running = get_pchome_backfill_status()
|
|
|
|
assert running["running"] is True
|
|
assert running["status"] == "running"
|
|
assert running["current_run"]["run_id"] == run["run_id"]
|
|
assert running["current_run"]["limit"] == 12
|
|
|
|
try:
|
|
start_pchome_backfill_run(limit=5)
|
|
except PchomeBackfillAlreadyRunning as exc:
|
|
assert exc.status["running"] is True
|
|
else:
|
|
raise AssertionError("expected active PChome backfill guard")
|
|
|
|
update_pchome_backfill_run(
|
|
run["run_id"],
|
|
stage="matching",
|
|
result={"total_skus": 12, "matched": 3},
|
|
)
|
|
finish_pchome_backfill_run(
|
|
run["run_id"],
|
|
result={"total_skus": 12, "matched": 3},
|
|
pick_result={"written": 9},
|
|
)
|
|
completed = get_pchome_backfill_status()
|
|
|
|
assert completed["running"] is False
|
|
assert completed["status"] == "completed"
|
|
assert completed["last_result"]["matched"] == 3
|
|
assert completed["recent_runs"][0]["pick_result"]["written"] == 9
|
|
|
|
|
|
def test_pchome_backfill_status_records_failure(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("PCHOME_BACKFILL_STATUS_PATH", str(tmp_path / "status.json"))
|
|
|
|
run = start_pchome_backfill_run(limit=8)
|
|
fail_pchome_backfill_run(run["run_id"], "crawler timeout")
|
|
failed = get_pchome_backfill_status()
|
|
|
|
assert failed["running"] is False
|
|
assert failed["status"] == "failed"
|
|
assert failed["last_error"] == "crawler timeout"
|
|
assert failed["recent_runs"][0]["last_error"] == "crawler timeout"
|