poprhythm
08a29d124a
- spruce.exif: tag_mosaic_jpeg_for_scan_dir, resolve_machine_label_for_scan_dir; ProcessingSoftware for tile-stitched mosaics - spruce.settings: load_config(require_credentials=False) for config without login - scripts/tag_mosaic_exif.py and tests; stitch script --write-exif path
127 lines
3.6 KiB
Python
127 lines
3.6 KiB
Python
"""Tests for spruce.settings — config loading and worker clamping."""
|
|
|
|
import logging
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from spruce.settings import (
|
|
MAX_SAFE_WORKERS,
|
|
SCANS_CSV_FIELDS,
|
|
TILES_CSV_FIELDS,
|
|
_clamp_workers,
|
|
load_config,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _clamp_workers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_clamp_workers_below_limit():
|
|
assert _clamp_workers(2) == 2
|
|
|
|
|
|
def test_clamp_workers_at_limit():
|
|
assert _clamp_workers(MAX_SAFE_WORKERS) == MAX_SAFE_WORKERS
|
|
|
|
|
|
def test_clamp_workers_above_limit_caps(caplog):
|
|
with caplog.at_level(logging.WARNING):
|
|
result = _clamp_workers(MAX_SAFE_WORKERS + 1)
|
|
assert result == MAX_SAFE_WORKERS
|
|
assert "exceeds the safe limit" in caplog.text
|
|
|
|
|
|
def test_clamp_workers_zero():
|
|
assert _clamp_workers(0) == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# load_config
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _write_config(tmp_path, **overrides):
|
|
base = {
|
|
"username": "testuser",
|
|
"password": "testpass",
|
|
}
|
|
base.update(overrides)
|
|
path = tmp_path / "config.yaml"
|
|
path.write_text(yaml.dump(base))
|
|
return str(path)
|
|
|
|
|
|
def test_load_config_defaults(tmp_path):
|
|
path = _write_config(tmp_path)
|
|
cfg = load_config(path)
|
|
assert cfg["base_url"] == "http://205.149.147.131:8010/"
|
|
assert cfg["workers"] == 2
|
|
assert cfg["timeout"] == 60
|
|
assert cfg["request_delay"] == 0.5
|
|
assert cfg["output_dir"] == "archives"
|
|
assert cfg["write_exif"] is True
|
|
assert cfg["machine_metadata"] == {}
|
|
|
|
|
|
def test_load_config_overrides(tmp_path):
|
|
path = _write_config(tmp_path, workers=3, output_dir="my_archives")
|
|
cfg = load_config(path)
|
|
assert cfg["workers"] == 3
|
|
assert cfg["output_dir"] == "my_archives"
|
|
|
|
|
|
def test_load_config_caps_workers(tmp_path, caplog):
|
|
path = _write_config(tmp_path, workers=MAX_SAFE_WORKERS + 2)
|
|
with caplog.at_level(logging.WARNING):
|
|
cfg = load_config(path)
|
|
assert cfg["workers"] == MAX_SAFE_WORKERS
|
|
assert "exceeds the safe limit" in caplog.text
|
|
|
|
|
|
def test_load_config_missing_username_exits(tmp_path):
|
|
path = tmp_path / "config.yaml"
|
|
path.write_text(yaml.dump({"password": "x"}))
|
|
with pytest.raises(SystemExit):
|
|
load_config(str(path))
|
|
|
|
|
|
def test_load_config_missing_password_exits(tmp_path):
|
|
path = tmp_path / "config.yaml"
|
|
path.write_text(yaml.dump({"username": "x"}))
|
|
with pytest.raises(SystemExit):
|
|
load_config(str(path))
|
|
|
|
|
|
def test_load_config_optional_credentials(tmp_path):
|
|
path = tmp_path / "config.yaml"
|
|
path.write_text(
|
|
yaml.dump({"machine_metadata": {"A [B]": {"plot_number": 2}}})
|
|
)
|
|
cfg = load_config(str(path), require_credentials=False)
|
|
assert cfg["machine_metadata"]["A [B]"]["plot_number"] == 2
|
|
assert cfg["write_exif"] is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CSV schemas (failure columns)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_scans_csv_fields_includes_mosaic_failure_columns():
|
|
for name in (
|
|
"mosaic_download_status",
|
|
"mosaic_error",
|
|
"mosaic_error_code",
|
|
"mosaic_error_class",
|
|
):
|
|
assert name in SCANS_CSV_FIELDS
|
|
|
|
|
|
def test_tiles_csv_fields_includes_status_and_error_columns():
|
|
for name in ("status", "error", "error_code", "error_class"):
|
|
assert name in TILES_CSV_FIELDS
|
|
assert TILES_CSV_FIELDS.index("status") < TILES_CSV_FIELDS.index("downloaded_at")
|