"""Reading runs from a cache: a local HTTP server stands in for the artifact-cache app.""" import copy import hashlib import json import logging import threading import urllib.error import urllib.parse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import pytest from test_resume import _assert_same_files from cymsim.cache import cache_key, cached_run, entry_meta, fetch, key_fields, publish from cymsim.cli import main from cymsim.cymsim_file import load_cymsim from cymsim.simulate import Simulator PREFIX = "/api/apps/cymsim/entries/" PASSCODE = "secret" class FakeCache: """The artifact-cache worker's GET (with ETag revalidation), GET meta and PUT (with passcode).""" def __init__(self): self.entries, self.metas, self.requests = {}, {}, [] cache = self class Handler(BaseHTTPRequestHandler): def do_PUT(self): key = self.path[len(PREFIX):] data = self.rfile.read(int(self.headers["Content-Length"])) if self.headers.get("X-Passcode") != PASSCODE: status, body = 401, b'{"error":"Invalid passcode"}' else: cache.entries[key] = data cache.metas[key] = json.loads(urllib.parse.unquote(self.headers.get("X-Artifact-Meta") or "null")) status, body = 201, json.dumps({"etag": hashlib.md5(data).hexdigest()}).encode() self.send_response(status) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) cache.requests.append(("PUT", key, status)) def do_GET(self): meta_prefix = PREFIX.replace("/entries/", "/meta/") if self.path.startswith(meta_prefix): key = self.path[len(meta_prefix):] body = json.dumps({"key": key, "meta": cache.metas.get(key)}).encode() self.send_response(200 if key in cache.entries else 404) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) return key = self.path[len(PREFIX):] if self.path.startswith(PREFIX) else None data = cache.entries.get(key) if data is None: status = 404 self.send_response(404) self.end_headers() else: etag = f'"{hashlib.md5(data).hexdigest()}"' status = 304 if self.headers.get("If-None-Match") == etag else 200 self.send_response(status) self.send_header("ETag", etag) if status == 200: self.send_header("Content-Length", str(len(data))) self.end_headers() if status == 200: self.wfile.write(data) cache.requests.append(("GET", key, status)) def log_message(self, *args): pass self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) self.url = f"http://127.0.0.1:{self.server.server_address[1]}{PREFIX[:-len('entries/')]}" threading.Thread(target=self.server.serve_forever, daemon=True).start() def put(self, cfg, path, meta=None): self.entries[cache_key(cfg)] = path.read_bytes() self.metas[cache_key(cfg)] = meta @pytest.fixture def cache(monkeypatch): monkeypatch.delenv("CYMSIM_CACHE_KEY", raising=False) c = FakeCache() yield c c.server.shutdown() def _config(base, duration): cfg = copy.deepcopy(base) cfg.numerics.duration = duration cfg.output.displacement_rate = 200.0 cfg.output.probes = [(0.1, 0.05), (0.05, -0.12)] return cfg @pytest.fixture(scope="module") def runs(coarse_config, tmp_path_factory): """Uninterrupted runs of 0.02 and 0.05 s, saved.""" d = tmp_path_factory.mktemp("runs") out = {} for T in (0.02, 0.05): out[T] = d / f"{T}.cymsim" Simulator(_config(coarse_config, T)).run(save_path=str(out[T])) return out def test_without_an_entry_the_run_starts_from_rest(coarse_config, runs, cache, tmp_path): out = tmp_path / "out.cymsim" cached_run(_config(coarse_config, 0.05), cache.url, save_path=str(out), directory=tmp_path / "c") _assert_same_files(out, runs[0.05]) assert cache.requests == [("GET", cache_key(_config(coarse_config, 0.05)), 404)] def test_a_covering_entry_is_cut_without_building_the_model(coarse_config, runs, cache, tmp_path, monkeypatch): cfg = _config(coarse_config, 0.03) ref = tmp_path / "ref.cymsim" Simulator(cfg).run(save_path=str(ref), resume=load_cymsim(runs[0.05])) cache.put(cfg, runs[0.05]) def no_model(*args, **kwargs): raise AssertionError("the model was built") monkeypatch.setattr(Simulator, "__init__", no_model) out = tmp_path / "out.cymsim" res = cached_run(cfg, cache.url, save_path=str(out), directory=tmp_path / "c") assert res.complete and len(res.pressure_rayleigh) == round(44100 * 0.03) _assert_same_files(out, ref) # including the state, whose overhang has grown # the energy log stops where an uninterrupted 0.03 s run stops, as on the resume path cached_log, ref_log = load_cymsim(runs[0.05]).energy_log, load_cymsim(ref).energy_log assert res.energy_log == ref_log and len(ref_log) < len(cached_log) assert res.energy_log[-1].time <= 0.03 + 0.02 and cached_log[-1].time > 0.05 def test_a_shorter_entry_is_continued(coarse_config, runs, cache, tmp_path): cfg = _config(coarse_config, 0.05) cache.put(cfg, runs[0.02]) out = tmp_path / "out.cymsim" cached_run(cfg, cache.url, save_path=str(out), directory=tmp_path / "c") _assert_same_files(out, runs[0.05]) def test_an_entry_of_another_configuration_is_ignored(coarse_config, runs, cache, tmp_path, caplog): cfg = _config(coarse_config, 0.02) other = copy.deepcopy(cfg) other.hits[0].velocity *= 2 other_path = tmp_path / "other.cymsim" Simulator(other).run(save_path=str(other_path)) cache.entries[cache_key(cfg)] = other_path.read_bytes() out = tmp_path / "out.cymsim" with caplog.at_level(logging.WARNING, logger="cymsim"): cached_run(cfg, cache.url, save_path=str(out), directory=tmp_path / "c") assert "different configuration" in caplog.text _assert_same_files(out, runs[0.02]) def test_local_copies(coarse_config, runs, cache, tmp_path): cfg, d = _config(coarse_config, 0.02), tmp_path / "c" key = cache_key(cfg) cache.put(cfg, runs[0.02]) p = fetch(cache.url, key, d) assert p == d / f"{key}.cymsim" and p.read_bytes() == runs[0.02].read_bytes() assert fetch(cache.url, key, d) == p assert [s for _, _, s in cache.requests] == [200, 304] cache.entries[key] = runs[0.05].read_bytes() # replaced remotely: downloaded again assert fetch(cache.url, key, d).read_bytes() == runs[0.05].read_bytes() dead = "http://127.0.0.1:9/api/apps/cymsim" # unreachable: the local copy serves assert fetch(dead, key, d) == p with pytest.raises(urllib.error.URLError): fetch(dead, "v1/missing", d) del cache.entries[key] # deleted remotely: the local copy goes too assert fetch(cache.url, key, d) is None and not p.exists() def test_the_acceleration_record_bypasses_the_cache(coarse_config, cache, tmp_path): cfg = _config(coarse_config, 0.02) cfg.output.record_acceleration = True res = cached_run(cfg, cache.url, directory=tmp_path / "c") assert res.acceleration is not None and cache.requests == [] def test_cli_cache(coarse_config, runs, cache, tmp_path, monkeypatch): cfg = _config(coarse_config, 0.03) cfg.output.wav = cfg.output.npz = cfg.output.cascade_plot = False cfg.output.prefix = str(tmp_path / "x") ypath = tmp_path / "c.yaml" cfg.to_yaml(ypath) cache.put(cfg, runs[0.05]) monkeypatch.setenv("CYMSIM_CACHE_DIR", str(tmp_path / "c")) out = tmp_path / "out.cymsim" assert main(["run", str(ypath), "--cache", cache.url, "-o", str(out), "--quiet"]) == 0 assert len(load_cymsim(out).pressure_rayleigh) == round(44100 * 0.03) monkeypatch.setenv("CYMSIM_CACHE_URL", cache.url) assert main(["run", str(ypath), "-o", str(out), "--quiet"]) == 0 assert main(["run", str(ypath), "--no-cache", "-o", str(out), "--quiet"]) == 0 assert [s for _, _, s in cache.requests] == [200, 304] assert main(["run", str(ypath), "--cache", cache.url, "--resume", str(out), "--quiet"]) == 2 def _load_entry(cache, key, path): path.write_bytes(cache.entries[key]) return path def test_a_computed_run_is_uploaded(coarse_config, runs, cache, tmp_path): cfg, d = _config(coarse_config, 0.02), tmp_path / "c" key = cache_key(cfg) cached_run(cfg, cache.url, directory=d, passcode=PASSCODE) _assert_same_files(_load_entry(cache, key, tmp_path / "e.cymsim"), runs[0.02]) meta = entry_meta(cache.url, key) assert meta["n_steps_done"] == load_cymsim(runs[0.02]).n_steps_done and meta["complete"] (seg,) = meta["segments"] assert seg["from"] == 0 and seg["to"] == meta["n_steps_done"] and seg["device"].startswith("cpu") assert meta["config"] == json.loads(json.dumps(key_fields(cfg))) # the upload is the local copy: the next lookup is a 304 and uploads nothing res = cached_run(cfg, cache.url, directory=d, passcode=PASSCODE) assert res.complete and cache.requests[-1] == ("GET", key, 304) assert [r for r in cache.requests if r[0] == "PUT"] == [("PUT", key, 201)] def test_a_continued_run_replaces_the_shorter_entry(coarse_config, runs, cache, tmp_path, caplog): cfg = _config(coarse_config, 0.05) key = cache_key(cfg) n = load_cymsim(runs[0.02]).n_steps_done cache.put(cfg, runs[0.02], {"n_steps_done": n, "segments": [{"from": 0, "to": n, "device": "gpu:Fake"}]}) with caplog.at_level(logging.WARNING, logger="cymsim"): cached_run(cfg, cache.url, directory=tmp_path / "c", passcode=PASSCODE) assert "integrated on gpu:Fake" in caplog.text _assert_same_files(_load_entry(cache, key, tmp_path / "e.cymsim"), runs[0.05]) segs = cache.metas[key]["segments"] assert [(g["from"], g["to"]) for g in segs] == [(0, n), (n, load_cymsim(runs[0.05]).n_steps_done)] def test_reading_without_a_passcode_uploads_nothing(coarse_config, cache, tmp_path): cached_run(_config(coarse_config, 0.02), cache.url, directory=tmp_path / "c") assert not cache.entries and all(r[0] == "GET" for r in cache.requests) def test_a_cut_run_is_not_uploaded(coarse_config, runs, cache, tmp_path): cfg = _config(coarse_config, 0.02) cache.put(cfg, runs[0.05], {"n_steps_done": load_cymsim(runs[0.05]).n_steps_done}) cached_run(cfg, cache.url, directory=tmp_path / "c", passcode=PASSCODE) assert all(r[0] == "GET" for r in cache.requests) def test_a_longer_entry_is_never_replaced(coarse_config, runs, cache, tmp_path): cfg = _config(coarse_config, 0.02) long_steps = load_cymsim(runs[0.05]).n_steps_done cache.put(cfg, runs[0.05], {"n_steps_done": long_steps}) assert not publish(load_cymsim(runs[0.02]), cache.url, cache_key(cfg), PASSCODE, [], directory=tmp_path / "c") assert all(r[0] == "GET" for r in cache.requests) assert cache.entries[cache_key(cfg)] == runs[0.05].read_bytes() def test_a_failed_upload_keeps_the_run(coarse_config, runs, cache, tmp_path, caplog): cfg, out = _config(coarse_config, 0.02), tmp_path / "out.cymsim" with caplog.at_level(logging.ERROR, logger="cymsim"): res = cached_run(cfg, cache.url, save_path=str(out), directory=tmp_path / "c", passcode="wrong") assert res.complete and "upload of" in caplog.text and "Invalid passcode" in caplog.text assert ("PUT", cache_key(cfg), 401) in cache.requests and not cache.entries _assert_same_files(out, runs[0.02])