76 lines
1.9 KiB
Python
76 lines
1.9 KiB
Python
"""HTTP download result wiring in MachineSession."""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import requests
|
|
|
|
from spruce.download_result import PERMANENT_MISSING, TRANSIENT
|
|
from spruce.session import MachineSession
|
|
|
|
|
|
def _minimal_config() -> dict:
|
|
return {
|
|
"base_url": "http://127.0.0.1:8010/",
|
|
"image_base_url": "http://127.0.0.1:8011/",
|
|
"username": "u",
|
|
"password": "p",
|
|
"timeout": 10,
|
|
"request_delay": 0.0,
|
|
"workers": 2,
|
|
}
|
|
|
|
|
|
def _machine() -> dict:
|
|
return {
|
|
"label": "T [AMR-0]",
|
|
"option_value": "x",
|
|
"machine_id": "0",
|
|
}
|
|
|
|
|
|
def test_download_file_http_404_result(tmp_path: Path):
|
|
dest = tmp_path / "f.bin"
|
|
sess = MachineSession(_machine(), _minimal_config())
|
|
|
|
def fail_get(*_a, **_k):
|
|
r = requests.Response()
|
|
r.status_code = 404
|
|
r.url = "http://example/mosaic.jpg"
|
|
r.reason = "Not Found"
|
|
r.raise_for_status()
|
|
|
|
sess.http.get = fail_get # type: ignore[method-assign]
|
|
|
|
res = sess.download_file("http://example/mosaic.jpg", dest, retries=1)
|
|
assert res.size == 0
|
|
assert res.status_code == 404
|
|
assert res.error_class == PERMANENT_MISSING
|
|
assert "404" in (res.error or "")
|
|
|
|
|
|
def test_download_tile_row_on_failure(tmp_path: Path):
|
|
dest = tmp_path / "t.jpg"
|
|
tile = {
|
|
"scan_id": 1,
|
|
"row_index": 0,
|
|
"col_index": 0,
|
|
"x_mm": 0.0,
|
|
"y_mm": 0.0,
|
|
"url": "http://example/tile",
|
|
}
|
|
sess = MachineSession(_machine(), _minimal_config())
|
|
|
|
def fail_get(*_a, **_k):
|
|
r = requests.Response()
|
|
r.status_code = 500
|
|
r.url = "http://example"
|
|
r.reason = "Server Error"
|
|
r.raise_for_status()
|
|
|
|
sess.http.get = fail_get # type: ignore[method-assign]
|
|
row = sess.download_tile(tile, dest, dry_run=False)
|
|
assert row["status"] == "failed"
|
|
assert row["error_code"] == "500"
|
|
assert row["error_class"] == TRANSIENT
|