355 lines
13 KiB
Python
355 lines
13 KiB
Python
"""Tests for the two one-time migrations.
|
|
|
|
These are the only steps in the project that mutate the archive or destroy state, so
|
|
the properties tested here are the ones that make them safe to run: dry runs change
|
|
nothing, re-runs are no-ops, collisions are skipped rather than forced, and nothing
|
|
is renamed until the labels it would destroy have been exported and recorded.
|
|
"""
|
|
|
|
import csv
|
|
import gzip
|
|
import json
|
|
import os
|
|
|
|
import pytest
|
|
|
|
import migrate_unrename
|
|
import migrate_urlcache
|
|
from conftest import solar_disc, write_fits
|
|
from suvi import db, paths
|
|
|
|
T0 = 1715299200
|
|
|
|
|
|
def make_frame_file(root, satellite=16, wavelength=171, index=0, label=None):
|
|
when = T0 + index * paths.CADENCE
|
|
name = paths.FrameName(satellite, wavelength, when, when + paths.CADENCE, "1-0-2")
|
|
path = os.path.join(str(root), *name.relpath(label).split("/"))
|
|
write_fits(path, solar_disc(size=32, radius=10, peak=3.0))
|
|
return name, path
|
|
|
|
|
|
# ------------------------------------------------------------------- url cache
|
|
|
|
|
|
@pytest.fixture
|
|
def url_json(tmp_path):
|
|
payload = {f"https://example.test/file{i}.fits": 1700000000.0 + i for i in range(50)}
|
|
path = tmp_path / "file_database.json"
|
|
path.write_text(json.dumps(payload))
|
|
return path, payload
|
|
|
|
|
|
def test_urlcache_dry_run_writes_nothing(url_json, db_path):
|
|
path, payload = url_json
|
|
assert migrate_urlcache.main(["--json", str(path), "--db", db_path]) == 0
|
|
assert not os.path.exists(db_path)
|
|
assert path.exists()
|
|
|
|
|
|
def test_urlcache_migrates_every_record(url_json, db_path):
|
|
path, payload = url_json
|
|
assert migrate_urlcache.main(["--json", str(path), "--db", db_path, "--apply"]) == 0
|
|
conn = db.connect(db_path, readonly=True)
|
|
assert conn.execute("SELECT count(*) c FROM remote_file").fetchone()["c"] == len(payload)
|
|
for url, mtime in list(payload.items())[:5]:
|
|
assert db.get_remote_mtime(conn, url) == mtime
|
|
conn.close()
|
|
|
|
|
|
def test_urlcache_leaves_the_json_alone_unless_asked(url_json, db_path):
|
|
path, _ = url_json
|
|
migrate_urlcache.main(["--json", str(path), "--db", db_path, "--apply"])
|
|
assert path.exists()
|
|
assert not os.path.exists(str(path) + ".bak")
|
|
|
|
|
|
def test_urlcache_backs_up_only_after_verifying(url_json, db_path):
|
|
path, _ = url_json
|
|
migrate_urlcache.main(
|
|
["--json", str(path), "--db", db_path, "--apply", "--backup-json"]
|
|
)
|
|
assert not path.exists()
|
|
assert os.path.exists(str(path) + ".bak")
|
|
|
|
|
|
def test_urlcache_is_idempotent(url_json, db_path):
|
|
path, payload = url_json
|
|
migrate_urlcache.main(["--json", str(path), "--db", db_path, "--apply"])
|
|
migrate_urlcache.main(["--json", str(path), "--db", db_path, "--apply"])
|
|
conn = db.connect(db_path, readonly=True)
|
|
assert conn.execute("SELECT count(*) c FROM remote_file").fetchone()["c"] == len(payload)
|
|
conn.close()
|
|
|
|
|
|
def test_urlcache_refuses_malformed_records(tmp_path, db_path):
|
|
path = tmp_path / "bad.json"
|
|
path.write_text(json.dumps({"https://example.test/a": "not a timestamp"}))
|
|
assert migrate_urlcache.main(["--json", str(path), "--db", db_path, "--apply"]) == 1
|
|
assert not os.path.exists(db_path)
|
|
|
|
|
|
def test_urlcache_refuses_a_non_object(tmp_path, db_path):
|
|
path = tmp_path / "list.json"
|
|
path.write_text(json.dumps(["a", "b"]))
|
|
assert migrate_urlcache.main(["--json", str(path), "--db", db_path, "--apply"]) == 1
|
|
|
|
|
|
def test_urlcache_reports_a_missing_source(tmp_path, db_path):
|
|
assert migrate_urlcache.main(["--json", str(tmp_path / "nope.json"), "--db", db_path]) == 1
|
|
|
|
|
|
# -------------------------------------------------------------------- un-rename
|
|
|
|
|
|
@pytest.fixture
|
|
def labelled_archive(archive):
|
|
"""An archive in the pre-migration state: suffixed files plus error plots."""
|
|
made = []
|
|
for index in range(4):
|
|
made.append(make_frame_file(archive, index=index, label="f"))
|
|
for index in range(4, 6):
|
|
name, path = make_frame_file(archive, index=index, label="e")
|
|
made.append((name, path))
|
|
plot = os.path.join(os.path.dirname(path), name.error_plot_name())
|
|
open(plot, "w").write("diagnostic")
|
|
make_frame_file(archive, index=6, label=None) # never processed
|
|
return archive, made
|
|
|
|
|
|
def test_unrename_dry_run_changes_nothing(labelled_archive, db_path):
|
|
archive, _ = labelled_archive
|
|
before = sorted(os.listdir(_day_dir(archive)))
|
|
assert migrate_unrename.main(["--root", str(archive), "--db", db_path]) == 0
|
|
assert sorted(os.listdir(_day_dir(archive))) == before
|
|
assert not os.path.exists(db_path)
|
|
|
|
|
|
def _day_dir(archive):
|
|
return os.path.join(str(archive), "goes16/l2/data/suvi-l2-ci171/2024/05/10")
|
|
|
|
|
|
def test_unrename_strips_both_suffixes(labelled_archive, db_path):
|
|
archive, _ = labelled_archive
|
|
assert migrate_unrename.main(["--root", str(archive), "--db", db_path, "--apply"]) == 0
|
|
names = os.listdir(_day_dir(archive))
|
|
assert not [n for n in names if n.endswith("_f.fits") or n.endswith("_e.fits")]
|
|
assert len([n for n in names if n.endswith(".fits")]) == 7
|
|
|
|
|
|
def test_unrename_deletes_diagnostic_plots(labelled_archive, db_path):
|
|
archive, _ = labelled_archive
|
|
migrate_unrename.main(["--root", str(archive), "--db", db_path, "--apply"])
|
|
assert not [n for n in os.listdir(_day_dir(archive)) if n.endswith(".jpg")]
|
|
|
|
|
|
def test_unrename_keeps_plots_when_asked(labelled_archive, db_path):
|
|
archive, _ = labelled_archive
|
|
migrate_unrename.main(
|
|
["--root", str(archive), "--db", db_path, "--apply", "--keep-plots"]
|
|
)
|
|
assert len([n for n in os.listdir(_day_dir(archive)) if n.endswith(".jpg")]) == 2
|
|
|
|
|
|
def test_unrename_exports_labels_before_renaming(labelled_archive, db_path, tmp_path):
|
|
archive, _ = labelled_archive
|
|
export = tmp_path / "labels.csv.gz"
|
|
migrate_unrename.main(
|
|
["--root", str(archive), "--db", db_path, "--apply", "--export", str(export)]
|
|
)
|
|
assert export.exists()
|
|
with gzip.open(export, "rt") as handle:
|
|
rows = list(csv.DictReader(handle))
|
|
assert len(rows) == 6
|
|
assert sum(1 for r in rows if r["legacy_label"] == "f") == 4
|
|
assert sum(1 for r in rows if r["legacy_label"] == "e") == 2
|
|
# The export must name the *restored* path, so it can be rejoined after renaming.
|
|
for row in rows:
|
|
assert not row["relpath"].endswith("_f.fits")
|
|
assert os.path.exists(paths.abspath(row["relpath"], str(archive)))
|
|
|
|
|
|
def test_unrename_records_verdicts_in_the_index(labelled_archive, db_path):
|
|
archive, _ = labelled_archive
|
|
migrate_unrename.main(["--root", str(archive), "--db", db_path, "--apply"])
|
|
conn = db.connect(db_path, readonly=True)
|
|
run_id = db.latest_run_id(conn, migrate_unrename.LEGACY_RUN_NAME)
|
|
assert run_id is not None
|
|
rows = conn.execute(
|
|
"SELECT verdict, count(*) c FROM detection WHERE run_id = ? GROUP BY verdict",
|
|
(run_id,),
|
|
).fetchall()
|
|
counts = {row["verdict"]: row["c"] for row in rows}
|
|
assert counts == {"good": 4, "bad": 2}
|
|
conn.close()
|
|
|
|
|
|
def test_unrename_is_idempotent(labelled_archive, db_path):
|
|
archive, _ = labelled_archive
|
|
migrate_unrename.main(["--root", str(archive), "--db", db_path, "--apply"])
|
|
names_after_first = sorted(os.listdir(_day_dir(archive)))
|
|
assert migrate_unrename.main(["--root", str(archive), "--db", db_path, "--apply"]) == 0
|
|
assert sorted(os.listdir(_day_dir(archive))) == names_after_first
|
|
|
|
|
|
def test_unrename_skips_collisions_rather_than_overwriting(archive, db_path):
|
|
"""If both the suffixed and unsuffixed names exist, neither may be destroyed."""
|
|
name, suffixed = make_frame_file(archive, index=0, label="f")
|
|
plain = os.path.join(os.path.dirname(suffixed), name.filename())
|
|
write_fits(plain, solar_disc(size=32, radius=10, peak=1.0))
|
|
plain_bytes = open(plain, "rb").read()
|
|
|
|
migrate_unrename.main(["--root", str(archive), "--db", db_path, "--apply"])
|
|
assert os.path.exists(suffixed), "the suffixed file was destroyed"
|
|
assert open(plain, "rb").read() == plain_bytes, "the existing file was overwritten"
|
|
|
|
|
|
def test_unrename_can_be_scoped_to_one_band(archive, db_path):
|
|
make_frame_file(archive, wavelength=171, index=0, label="f")
|
|
make_frame_file(archive, wavelength=304, index=0, label="f")
|
|
migrate_unrename.main(
|
|
["--root", str(archive), "--db", db_path, "--apply", "--wavelength", "171"]
|
|
)
|
|
band_171 = os.listdir(_day_dir(archive))
|
|
band_304 = os.listdir(
|
|
os.path.join(str(archive), "goes16/l2/data/suvi-l2-ci304/2024/05/10")
|
|
)
|
|
assert not [n for n in band_171 if n.endswith("_f.fits")]
|
|
assert [n for n in band_304 if n.endswith("_f.fits")]
|
|
|
|
|
|
def test_unrename_on_an_already_clean_archive(archive, db_path):
|
|
make_frame_file(archive, index=0, label=None)
|
|
assert migrate_unrename.main(["--root", str(archive), "--db", db_path, "--apply"]) == 0
|
|
|
|
|
|
def test_unrename_reports_a_missing_root(tmp_path, db_path):
|
|
assert migrate_unrename.main(
|
|
["--root", str(tmp_path / "nope"), "--db", db_path, "--apply"]
|
|
) == 1
|
|
|
|
|
|
def test_walk_archive_finds_only_frames(labelled_archive):
|
|
archive, _ = labelled_archive
|
|
found = list(migrate_unrename.walk_archive(str(archive), (16,), (171,)))
|
|
assert len(found) == 7
|
|
assert all(name is not None for _, _, name in found)
|
|
|
|
|
|
def test_walk_surfaces_unreadable_directories(archive, monkeypatch):
|
|
"""A filesystem error must not look like an empty archive.
|
|
|
|
os.walk ignores errors by default, so a transient ENFILE on the shared mount
|
|
made the walk yield nothing and the migration report "already migrated" --
|
|
silently skipping hundreds of thousands of files it should have renamed.
|
|
"""
|
|
make_frame_file(archive, index=0, label="f")
|
|
|
|
def explode(path):
|
|
raise OSError(23, "Too many open files in system")
|
|
|
|
monkeypatch.setattr(os, "scandir", explode)
|
|
with pytest.raises(OSError):
|
|
list(migrate_unrename.walk_archive(str(archive), (16,), (171,)))
|
|
|
|
|
|
def test_unrename_records_and_skips_completed_chunks(labelled_archive, db_path):
|
|
"""A resume must skip finished band-years without touching the filesystem.
|
|
|
|
Re-walking completed subtrees on every resume is what made the migration
|
|
traverse the archive several times over, and traversal is what exhausts the
|
|
mount's file handles.
|
|
"""
|
|
archive, _ = labelled_archive
|
|
migrate_unrename.main(["--root", str(archive), "--db", db_path, "--apply"])
|
|
|
|
conn = db.connect(db_path, readonly=True)
|
|
assert db.get_meta(conn, migrate_unrename.chunk_key(171, 2024)) == "done"
|
|
conn.close()
|
|
|
|
def explode(*args, **kwargs):
|
|
raise AssertionError("resume re-read the filesystem for a completed chunk")
|
|
|
|
import unittest.mock as mock
|
|
|
|
with mock.patch.object(migrate_unrename, "walk_archive", explode):
|
|
assert migrate_unrename.main(
|
|
["--root", str(archive), "--db", db_path, "--apply"]
|
|
) == 0
|
|
|
|
|
|
def test_recheck_forces_a_completed_chunk_to_be_re_examined(labelled_archive, db_path):
|
|
archive, _ = labelled_archive
|
|
migrate_unrename.main(["--root", str(archive), "--db", db_path, "--apply"])
|
|
calls = []
|
|
real = migrate_unrename.walk_archive
|
|
|
|
def spy(*args, **kwargs):
|
|
calls.append(args)
|
|
return real(*args, **kwargs)
|
|
|
|
import unittest.mock as mock
|
|
|
|
with mock.patch.object(migrate_unrename, "walk_archive", spy):
|
|
migrate_unrename.main(
|
|
["--root", str(archive), "--db", db_path, "--apply", "--recheck"]
|
|
)
|
|
assert calls, "--recheck should re-examine the archive"
|
|
|
|
|
|
def test_a_chunk_with_failed_renames_is_not_marked_done(labelled_archive, db_path, monkeypatch):
|
|
"""Marking a partially failed chunk complete would make a resume skip real work."""
|
|
archive, _ = labelled_archive
|
|
real_rename = os.rename
|
|
state = {"n": 0}
|
|
|
|
def flaky(src, dst):
|
|
state["n"] += 1
|
|
if state["n"] == 2:
|
|
raise OSError(23, "Too many open files in system")
|
|
return real_rename(src, dst)
|
|
|
|
monkeypatch.setattr(os, "rename", flaky)
|
|
migrate_unrename.main(["--root", str(archive), "--db", db_path, "--apply"])
|
|
|
|
conn = db.connect(db_path, readonly=True)
|
|
assert db.get_meta(conn, migrate_unrename.chunk_key(171, 2024)) is None
|
|
conn.close()
|
|
|
|
|
|
# --------------------------------------------------------------- review sampler
|
|
|
|
|
|
def test_judge_timestamp_classifies_every_band(archive):
|
|
"""The sampler's unit is a frame, but a composite needs all six bands."""
|
|
import filter_FITS
|
|
|
|
band_paths = {}
|
|
for band in (94, 131, 171, 195, 284, 304):
|
|
name, path = make_frame_file(archive, wavelength=band, index=0)
|
|
band_paths[band] = path
|
|
|
|
satellite, timestamp, verdicts = filter_FITS._judge_timestamp((16, name.t_start, band_paths))
|
|
assert satellite == 16
|
|
assert set(verdicts) == {94, 131, 171, 195, 284, 304}
|
|
for verdict, reason, scores in verdicts.values():
|
|
assert verdict in ("good", "bad")
|
|
assert isinstance(scores, dict)
|
|
|
|
|
|
def test_judge_timestamp_flags_a_damaged_band(archive):
|
|
"""A blank band must be flagged, and the others left alone."""
|
|
import numpy as np
|
|
|
|
import filter_FITS
|
|
|
|
band_paths = {}
|
|
for band in (94, 131, 171, 195, 284, 304):
|
|
name, path = make_frame_file(archive, wavelength=band, index=0)
|
|
band_paths[band] = path
|
|
# Replace one band with a frame carrying no signal at all.
|
|
write_fits(band_paths[171], np.zeros((32, 32), dtype="float32"))
|
|
|
|
_, _, verdicts = filter_FITS._judge_timestamp((16, name.t_start, band_paths))
|
|
assert verdicts[171][0] == "bad"
|
|
assert verdicts[171][1], "a flagged band must carry a reason"
|