134 lines
3.7 KiB
Python
134 lines
3.7 KiB
Python
"""Retry queue loading from scans.csv (mosaic_download_status=failed)."""
|
|
|
|
import csv
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from spruce.orchestrator import load_failed_scans_from_csv
|
|
from spruce.settings import SCANS_CSV_FIELDS
|
|
|
|
|
|
def _blank_row(**kwargs: str) -> dict[str, str]:
|
|
row = {k: "" for k in SCANS_CSV_FIELDS}
|
|
row.update(kwargs)
|
|
return row
|
|
|
|
|
|
def _write_scans_csv(path: Path, rows: list[dict[str, str]]) -> None:
|
|
with open(path, "w", newline="", encoding="utf-8") as fh:
|
|
w = csv.DictWriter(fh, fieldnames=SCANS_CSV_FIELDS)
|
|
w.writeheader()
|
|
for r in rows:
|
|
w.writerow({k: r.get(k, "") for k in SCANS_CSV_FIELDS})
|
|
|
|
|
|
def test_load_failed_scans_dedup_keeps_last_row(tmp_path: Path) -> None:
|
|
path = tmp_path / "scans.csv"
|
|
common = {
|
|
"machine": "BW1 [X]",
|
|
"machine_id": "1",
|
|
"scan_id": "100",
|
|
"mosaic_url": "http://x/m.jpg",
|
|
"mosaic_local_path": "",
|
|
"mosaic_on_disk": "False",
|
|
}
|
|
_write_scans_csv(
|
|
path,
|
|
[
|
|
_blank_row(
|
|
**common,
|
|
mosaic_download_status="failed",
|
|
mosaic_error_code="404",
|
|
scan_time="2020-01-01",
|
|
),
|
|
_blank_row(
|
|
**common,
|
|
mosaic_download_status="failed",
|
|
mosaic_error_code="404",
|
|
scan_time="2020-06-01",
|
|
),
|
|
],
|
|
)
|
|
out = load_failed_scans_from_csv(path, "BW1 [X]")
|
|
assert len(out) == 1
|
|
assert out[0]["scan_id"] == 100
|
|
assert out[0]["scan_time"] == "2020-06-01"
|
|
|
|
|
|
def test_load_failed_scans_since_year(tmp_path: Path) -> None:
|
|
path = tmp_path / "scans.csv"
|
|
base = {
|
|
"machine": "M",
|
|
"machine_id": "1",
|
|
"mosaic_url": "",
|
|
"mosaic_local_path": "",
|
|
"mosaic_on_disk": "",
|
|
"mosaic_download_status": "failed",
|
|
"mosaic_error_code": "404",
|
|
}
|
|
_write_scans_csv(
|
|
path,
|
|
[
|
|
_blank_row(**base, scan_id="1", scan_time="2022-12-31"),
|
|
_blank_row(**base, scan_id="2", scan_time="2023-01-01"),
|
|
_blank_row(**base, scan_id="3", scan_time=""),
|
|
],
|
|
)
|
|
out = load_failed_scans_from_csv(path, "M", since_year="2023")
|
|
ids = {s["scan_id"] for s in out}
|
|
assert ids == {2}
|
|
|
|
|
|
def test_load_failed_scans_error_code(tmp_path: Path) -> None:
|
|
path = tmp_path / "scans.csv"
|
|
base = {
|
|
"machine": "M",
|
|
"machine_id": "1",
|
|
"scan_time": "2024-01-01",
|
|
"mosaic_url": "",
|
|
"mosaic_local_path": "",
|
|
"mosaic_on_disk": "",
|
|
"mosaic_download_status": "failed",
|
|
}
|
|
_write_scans_csv(
|
|
path,
|
|
[
|
|
_blank_row(**base, scan_id="10", mosaic_error_code="404"),
|
|
_blank_row(**base, scan_id="11", mosaic_error_code="200"),
|
|
],
|
|
)
|
|
out = load_failed_scans_from_csv(path, "M", error_code="200")
|
|
assert [s["scan_id"] for s in out] == [11]
|
|
|
|
|
|
def test_load_failed_scans_excludes_downloaded(tmp_path: Path) -> None:
|
|
path = tmp_path / "scans.csv"
|
|
base = {
|
|
"machine": "M",
|
|
"machine_id": "1",
|
|
"scan_time": "2024-01-01",
|
|
"mosaic_url": "",
|
|
"mosaic_local_path": "",
|
|
"mosaic_on_disk": "True",
|
|
}
|
|
_write_scans_csv(
|
|
path,
|
|
[
|
|
_blank_row(
|
|
**base,
|
|
scan_id="5",
|
|
mosaic_download_status="downloaded",
|
|
mosaic_error_code="",
|
|
),
|
|
_blank_row(
|
|
**base,
|
|
scan_id="6",
|
|
mosaic_download_status="failed",
|
|
mosaic_error_code="404",
|
|
),
|
|
],
|
|
)
|
|
out = load_failed_scans_from_csv(path, "M")
|
|
assert [s["scan_id"] for s in out] == [6]
|