401 lines
16 KiB
Python
401 lines
16 KiB
Python
"""Tests for keeping the index in step with the archive without traversing it.
|
|
|
|
The property under test throughout is that a *re-run reads nothing it does not have
|
|
to*. That is not an optimisation here: a full traversal of this archive exhausts the
|
|
file handles of the virtiofs mount it lives on, and the mount then refuses every open
|
|
until the guest drops its dentry cache. So "how many directories did we open" is a
|
|
correctness-adjacent measurement, and several tests assert on it directly.
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
|
|
import pytest
|
|
|
|
from conftest import solar_disc, write_fits
|
|
from suvi import db, index, paths
|
|
|
|
T0 = 1715299200
|
|
|
|
|
|
def add_frame(root, satellite=16, wavelength=171, index_=0, day=None, label=None):
|
|
when = T0 + index_ * paths.CADENCE
|
|
name = paths.FrameName(satellite, wavelength, when, when + paths.CADENCE, "1-0-2")
|
|
relpath = name.relpath(label)
|
|
if day:
|
|
relpath = relpath.replace("/2024/05/10/", f"/{day}/")
|
|
path = os.path.join(str(root), *relpath.split("/"))
|
|
write_fits(path, solar_disc(size=32, radius=10, peak=3.0))
|
|
return name, path, relpath
|
|
|
|
|
|
@pytest.fixture
|
|
def small_archive(archive):
|
|
for i in range(5):
|
|
add_frame(archive, index_=i)
|
|
for i in range(3):
|
|
add_frame(archive, wavelength=304, index_=i)
|
|
return archive
|
|
|
|
|
|
class CountingScandir:
|
|
"""Wraps os.scandir so a test can assert how much of the tree was opened."""
|
|
|
|
def __init__(self, monkeypatch, only_under=None):
|
|
self.paths = []
|
|
self._real = os.scandir
|
|
self._only = str(only_under) if only_under else None
|
|
monkeypatch.setattr(os, "scandir", self)
|
|
|
|
def __call__(self, path=".", *args, **kwargs):
|
|
text = str(path)
|
|
if self._only is None or text.startswith(self._only):
|
|
self.paths.append(text)
|
|
return self._real(path, *args, **kwargs)
|
|
|
|
#: .../YYYY/MM/DD -- the directories holding ~360 frames each. Month
|
|
#: directories must still be listed to discover day directories at all; it is
|
|
#: only the day directories, and the per-file lookups inside them, that a
|
|
#: no-change reconcile has to avoid.
|
|
_DAY_DIR = re.compile(r"/\d{4}/\d{2}/\d{2}/?$")
|
|
|
|
def opened_day_dirs(self):
|
|
return [p for p in self.paths if self._DAY_DIR.search(p)]
|
|
|
|
|
|
# ------------------------------------------------------------- day_directories
|
|
|
|
|
|
def test_day_directories_finds_every_day(small_archive):
|
|
found = list(index.day_directories(str(small_archive), (16,), (171, 304)))
|
|
assert len(found) == 2
|
|
for relpath, abspath in found:
|
|
assert relpath.endswith("2024/05/10")
|
|
assert os.path.isdir(abspath)
|
|
|
|
|
|
def test_day_directories_filters_by_year(small_archive):
|
|
add_frame(small_archive, day="2023/07/04")
|
|
assert len(list(index.day_directories(str(small_archive), (16,), (171,)))) == 2
|
|
only = list(index.day_directories(str(small_archive), (16,), (171,), years=[2023]))
|
|
assert len(only) == 1 and only[0][0].endswith("2023/07/04")
|
|
|
|
|
|
def test_day_directories_tolerates_a_missing_archive(tmp_path):
|
|
assert list(index.day_directories(str(tmp_path), (16,), (171,))) == []
|
|
|
|
|
|
# ------------------------------------------------------------------- reconcile
|
|
|
|
|
|
def test_first_reconcile_indexes_everything(small_archive, db_path):
|
|
conn = db.connect(db_path)
|
|
summary = index.reconcile(conn, str(small_archive), (16,), (171, 304))
|
|
assert summary["frames_added"] == 8
|
|
assert summary["directories_changed"] == 2
|
|
assert conn.execute("SELECT count(*) c FROM frame").fetchone()["c"] == 8
|
|
conn.close()
|
|
|
|
|
|
def test_second_reconcile_opens_no_day_directories(small_archive, db_path, monkeypatch):
|
|
"""The heart of it: an unchanged archive must cost directory stats, nothing more.
|
|
|
|
Re-reading directories that have not changed is what made a resume traverse the
|
|
whole archive, and traversal is what breaks the mount.
|
|
"""
|
|
conn = db.connect(db_path)
|
|
index.reconcile(conn, str(small_archive), (16,), (171, 304))
|
|
|
|
spy = CountingScandir(monkeypatch, only_under=small_archive)
|
|
summary = index.reconcile(conn, str(small_archive), (16,), (171, 304))
|
|
assert summary["directories_changed"] == 0
|
|
assert summary["frames_added"] == 0
|
|
assert spy.opened_day_dirs() == [], "re-read a directory that had not changed"
|
|
conn.close()
|
|
|
|
|
|
def test_reconcile_picks_up_a_new_frame(small_archive, db_path):
|
|
conn = db.connect(db_path)
|
|
index.reconcile(conn, str(small_archive), (16,), (171,))
|
|
add_frame(small_archive, index_=99)
|
|
summary = index.reconcile(conn, str(small_archive), (16,), (171,))
|
|
assert summary["frames_added"] == 1
|
|
assert summary["directories_changed"] == 1
|
|
conn.close()
|
|
|
|
|
|
def test_reconcile_notices_a_deleted_frame(small_archive, db_path):
|
|
conn = db.connect(db_path)
|
|
index.reconcile(conn, str(small_archive), (16,), (171,))
|
|
before = conn.execute("SELECT count(*) c FROM frame").fetchone()["c"]
|
|
|
|
_, path, _ = add_frame(small_archive, index_=0) # existing file
|
|
os.remove(path)
|
|
summary = index.reconcile(conn, str(small_archive), (16,), (171,))
|
|
assert summary["frames_removed"] == 1
|
|
assert conn.execute("SELECT count(*) c FROM frame").fetchone()["c"] == before - 1
|
|
conn.close()
|
|
|
|
|
|
def test_reconcile_handles_a_vanished_directory(small_archive, db_path):
|
|
import shutil
|
|
|
|
conn = db.connect(db_path)
|
|
index.reconcile(conn, str(small_archive), (16,), (171, 304))
|
|
shutil.rmtree(os.path.join(str(small_archive), "goes16/l2/data/suvi-l2-ci304"))
|
|
summary = index.reconcile(conn, str(small_archive), (16,), (171, 304))
|
|
assert summary["directories_vanished"] == 1
|
|
assert summary["frames_removed"] == 3
|
|
assert conn.execute("SELECT count(*) c FROM frame").fetchone()["c"] == 5
|
|
conn.close()
|
|
|
|
|
|
def test_force_rereads_everything(small_archive, db_path, monkeypatch):
|
|
conn = db.connect(db_path)
|
|
index.reconcile(conn, str(small_archive), (16,), (171,))
|
|
spy = CountingScandir(monkeypatch, only_under=small_archive)
|
|
summary = index.reconcile(conn, str(small_archive), (16,), (171,), force=True)
|
|
assert summary["directories_changed"] == 1
|
|
assert spy.opened_day_dirs(), "--full should re-read directories"
|
|
conn.close()
|
|
|
|
|
|
def test_reconcile_is_idempotent(small_archive, db_path):
|
|
conn = db.connect(db_path)
|
|
first = index.reconcile(conn, str(small_archive), (16,), (171,))
|
|
second = index.reconcile(conn, str(small_archive), (16,), (171,))
|
|
third = index.reconcile(conn, str(small_archive), (16,), (171,))
|
|
assert first["frames_added"] == 5
|
|
assert second["frames_added"] == third["frames_added"] == 0
|
|
assert conn.execute("SELECT count(*) c FROM frame").fetchone()["c"] == 5
|
|
conn.close()
|
|
|
|
|
|
def test_reconcile_ignores_non_frame_files(small_archive, db_path):
|
|
conn = db.connect(db_path)
|
|
day = os.path.join(str(small_archive), "goes16/l2/data/suvi-l2-ci171/2024/05/10")
|
|
open(os.path.join(day, "notes.txt"), "w").write("not a frame")
|
|
summary = index.reconcile(conn, str(small_archive), (16,), (171,))
|
|
assert summary["frames_added"] == 5
|
|
conn.close()
|
|
|
|
|
|
def test_reconcile_records_the_observed_mtime(small_archive, db_path):
|
|
conn = db.connect(db_path)
|
|
index.reconcile(conn, str(small_archive), (16,), (171,))
|
|
relpath = "goes16/l2/data/suvi-l2-ci171/2024/05/10"
|
|
abspath = os.path.join(str(small_archive), *relpath.split("/"))
|
|
assert db.get_dir_mtime(conn, relpath) == pytest.approx(os.stat(abspath).st_mtime)
|
|
conn.close()
|
|
|
|
|
|
# ------------------------------------------------------------ record_downloaded
|
|
|
|
|
|
def test_record_downloaded_indexes_without_any_traversal(archive, db_path, monkeypatch):
|
|
"""The cheapest path: the downloader already knows the file is there."""
|
|
conn = db.connect(db_path)
|
|
_, path, relpath = add_frame(archive)
|
|
|
|
spy = CountingScandir(monkeypatch, only_under=archive)
|
|
frame_id = index.record_downloaded(conn, path, str(archive))
|
|
assert frame_id is not None
|
|
assert spy.paths == [], "indexing a download must not scan the archive"
|
|
row = conn.execute("SELECT path FROM frame WHERE id = ?", (frame_id,)).fetchone()
|
|
assert row["path"] == relpath
|
|
conn.close()
|
|
|
|
|
|
def test_record_downloaded_ignores_other_files(archive, db_path):
|
|
conn = db.connect(db_path)
|
|
other = os.path.join(str(archive), "readme.txt")
|
|
open(other, "w").write("x")
|
|
assert index.record_downloaded(conn, other, str(archive)) is None
|
|
conn.close()
|
|
|
|
|
|
def test_record_downloaded_survives_a_missing_file(archive, db_path):
|
|
conn = db.connect(db_path)
|
|
_, path, _ = add_frame(archive)
|
|
os.remove(path)
|
|
assert index.record_downloaded(conn, path, str(archive)) is not None
|
|
conn.close()
|
|
|
|
|
|
# ------------------------------------------------------------------ db helpers
|
|
|
|
|
|
def test_frames_in_dir_scopes_to_one_directory(small_archive, db_path):
|
|
conn = db.connect(db_path)
|
|
index.reconcile(conn, str(small_archive), (16,), (171, 304))
|
|
in_171 = db.frames_in_dir(conn, "goes16/l2/data/suvi-l2-ci171/2024/05/10")
|
|
in_304 = db.frames_in_dir(conn, "goes16/l2/data/suvi-l2-ci304/2024/05/10")
|
|
assert len(in_171) == 5 and len(in_304) == 3
|
|
assert all("ci171" in name for name in in_171)
|
|
conn.close()
|
|
|
|
|
|
def test_schema_upgrades_from_version_one(tmp_path):
|
|
"""An index created before dir_scan existed must gain it, not be rejected."""
|
|
path = str(tmp_path / "old.sqlite")
|
|
conn = db.connect(path)
|
|
conn.execute("DROP TABLE dir_scan")
|
|
conn.execute("UPDATE meta SET value = '1' WHERE key = 'schema_version'")
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
conn = db.connect(path) # re-open triggers the upgrade
|
|
assert db.get_meta(conn, "schema_version") == str(db.SCHEMA_VERSION)
|
|
conn.execute("SELECT count(*) FROM dir_scan") # table now exists
|
|
conn.close()
|
|
|
|
|
|
def test_meta_round_trip(db_path):
|
|
conn = db.connect(db_path)
|
|
assert db.get_meta(conn, "absent") is None
|
|
assert db.get_meta(conn, "absent", "fallback") == "fallback"
|
|
db.set_meta(conn, "unrename_done:ci171:2024", "done")
|
|
db.set_meta(conn, "unrename_done:ci171:2024", "done") # upsert, not duplicate
|
|
assert db.get_meta(conn, "unrename_done:ci171:2024") == "done"
|
|
conn.close()
|
|
|
|
|
|
# ------------------------------------------------------------ slot collisions
|
|
|
|
|
|
def test_two_files_claiming_one_slot_do_not_break_indexing(archive, db_path):
|
|
"""The archive holds 1,742 such pairs; they must not block the whole index.
|
|
|
|
An older filter labelled some frames repeatedly, and the puller later
|
|
re-downloaded a clean copy it could no longer find under the published name --
|
|
leaving `X_v1-0-1.fits` beside a byte-identical `X_v1-0-1_f_f_f.fits`.
|
|
"""
|
|
_, path, _ = add_frame(archive) # canonical
|
|
add_frame(archive, label="f") # same slot, labelled
|
|
conn = db.connect(db_path)
|
|
summary = index.reconcile(conn, str(archive), (16,), (171,))
|
|
assert summary["frames_added"] == 1
|
|
assert len(summary["duplicate_slots"]) == 1
|
|
assert summary["duplicate_slots"][0].endswith("_f.fits")
|
|
row = conn.execute("SELECT path FROM frame").fetchone()
|
|
assert row["path"].endswith("v1-0-2.fits"), "kept the labelled copy over the published one"
|
|
conn.close()
|
|
|
|
|
|
def test_the_published_name_wins_over_a_multiply_labelled_one(archive, db_path):
|
|
_, path, _ = add_frame(archive)
|
|
name = paths.parse_frame_filename(os.path.basename(path))
|
|
triple = os.path.join(os.path.dirname(path), name.filename().replace(".fits", "_f_f_f.fits"))
|
|
write_fits(triple, solar_disc(size=32, radius=10, peak=3.0))
|
|
|
|
conn = db.connect(db_path)
|
|
summary = index.reconcile(conn, str(archive), (16,), (171,))
|
|
assert summary["frames_added"] == 1
|
|
assert conn.execute("SELECT path FROM frame").fetchone()["path"].endswith("v1-0-2.fits")
|
|
assert summary["duplicate_slots"][0].endswith("_f_f_f.fits")
|
|
conn.close()
|
|
|
|
|
|
def test_a_renamed_file_does_not_collide_with_its_old_row(archive, db_path):
|
|
"""Un-renaming moves a file within its directory; reconcile must cope.
|
|
|
|
The removal has to be applied before the insertion, or both names briefly claim
|
|
the same observation slot and the unique constraint fires.
|
|
"""
|
|
_, old_path, _ = add_frame(archive, label="f")
|
|
conn = db.connect(db_path)
|
|
index.reconcile(conn, str(archive), (16,), (171,))
|
|
assert conn.execute("SELECT count(*) c FROM frame").fetchone()["c"] == 1
|
|
|
|
name = paths.parse_frame_filename(os.path.basename(old_path))
|
|
os.rename(old_path, os.path.join(os.path.dirname(old_path), name.filename()))
|
|
|
|
summary = index.reconcile(conn, str(archive), (16,), (171,))
|
|
assert summary["frames_added"] == 1 and summary["frames_removed"] == 1
|
|
assert conn.execute("SELECT count(*) c FROM frame").fetchone()["c"] == 1
|
|
assert conn.execute("SELECT path FROM frame").fetchone()["path"].endswith("v1-0-2.fits")
|
|
conn.close()
|
|
|
|
|
|
# ------------------------------------------------- unreadable != empty
|
|
|
|
|
|
def test_unreadable_directory_raises_rather_than_looking_empty(small_archive, monkeypatch):
|
|
"""Regression: swallowing OSError made five of six bands look absent.
|
|
|
|
reconcile then treated their contents as deleted and dropped 205,618 index rows
|
|
for files that were still on disk. An unreadable directory has to stop the run.
|
|
"""
|
|
real = os.scandir
|
|
|
|
def refuse(path=".", *args, **kwargs):
|
|
if "suvi-l2-ci304" in str(path):
|
|
raise OSError(23, "Too many open files in system")
|
|
return real(path, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(os, "scandir", refuse)
|
|
with pytest.raises(OSError):
|
|
list(index.day_directories(str(small_archive), (16,), (171, 304)))
|
|
|
|
|
|
def test_a_missing_directory_is_still_treated_as_empty(tmp_path):
|
|
"""Only genuine absence may be silent."""
|
|
assert index._subdirs(str(tmp_path / "does-not-exist")) == []
|
|
|
|
|
|
def test_frames_are_not_purged_when_enumeration_comes_up_short(small_archive, db_path, monkeypatch):
|
|
"""Defence in depth: absence from the scan is not proof a directory is gone.
|
|
|
|
Even if enumeration misses a directory, its rows must survive as long as the
|
|
directory is still on disk.
|
|
"""
|
|
conn = db.connect(db_path)
|
|
index.reconcile(conn, str(small_archive), (16,), (171, 304))
|
|
before = conn.execute("SELECT count(*) c FROM frame").fetchone()["c"]
|
|
assert before == 8
|
|
|
|
# Enumerate only one band, as though the other had been missed entirely.
|
|
real = index.day_directories
|
|
partial = list(real(str(small_archive), (16,), (171,)))
|
|
monkeypatch.setattr(index, "day_directories", lambda *a, **k: iter(partial))
|
|
summary = index.reconcile(conn, str(small_archive), (16,), (171, 304))
|
|
assert summary["directories_vanished"] == 0
|
|
assert conn.execute("SELECT count(*) c FROM frame").fetchone()["c"] == before
|
|
conn.close()
|
|
|
|
|
|
def test_a_cold_build_reclaims_periodically(small_archive, db_path, monkeypatch):
|
|
"""A cold build reads every file, which is what exhausts the mount's handles.
|
|
|
|
It must hand them back as it goes, or the build cannot finish.
|
|
"""
|
|
calls = []
|
|
monkeypatch.setattr(index.vfs, "release_handles", lambda *a, **k: calls.append(1))
|
|
monkeypatch.setattr(index.vfs, "RELIEF_INTERVAL", 3)
|
|
conn = db.connect(db_path)
|
|
index.reconcile(conn, str(small_archive), (16,), (171, 304))
|
|
assert calls, "cold build never reclaimed"
|
|
conn.close()
|
|
|
|
|
|
def test_reclaim_can_be_disabled(small_archive, db_path, monkeypatch):
|
|
def explode(*a, **k):
|
|
raise AssertionError("reclaimed despite relief=False")
|
|
|
|
monkeypatch.setattr(index.vfs, "release_handles", explode)
|
|
monkeypatch.setattr(index.vfs, "RELIEF_INTERVAL", 1)
|
|
conn = db.connect(db_path)
|
|
index.reconcile(conn, str(small_archive), (16,), (171,), relief=False)
|
|
conn.close()
|
|
|
|
|
|
def test_a_steady_state_run_does_not_reclaim(small_archive, db_path, monkeypatch):
|
|
"""Nothing changed means nothing was read, so there is nothing to hand back."""
|
|
conn = db.connect(db_path)
|
|
index.reconcile(conn, str(small_archive), (16,), (171,))
|
|
calls = []
|
|
monkeypatch.setattr(index.vfs, "release_handles", lambda *a, **k: calls.append(1))
|
|
monkeypatch.setattr(index.vfs, "RELIEF_INTERVAL", 1)
|
|
index.reconcile(conn, str(small_archive), (16,), (171,))
|
|
assert calls == []
|
|
conn.close()
|