/ proof-of-concept / cymsim
Sign in
proof-of-concept / cymsim
a3f35ac
Branches
main
cymsim / tests / test_cache_run.py
268 lines · 11.8 KBCodeBlameHistory
2import copy
3import hashlib
6import threading
7import urllib.error
11import pytest
12from test_resume import _assert_same_files
16from cymsim.cymsim_file import load_cymsim
17from cymsim.simulate import Simulator
19PREFIX = "/api/apps/cymsim/entries/"
28 def __init__(self):
32 class Handler(BaseHTTPRequestHandler):
34 key = self.path[len(PREFIX):]
35 data = self.rfile.read(int(self.headers["Content-Length"]))
36 if self.headers.get("X-Passcode") != PASSCODE:
37 status, body = 401, b'{"error":"Invalid passcode"}'
38 else:
39 cache.entries[key] = data
40 cache.metas[key] = json.loads(urllib.parse.unquote(self.headers.get("X-Artifact-Meta") or "null"))
41 status, body = 201, json.dumps({"etag": hashlib.md5(data).hexdigest()}).encode()
42 self.send_response(status)
43 self.send_header("Content-Length", str(len(body)))
44 self.end_headers()
45 self.wfile.write(body)
46 cache.requests.append(("PUT", key, status))
50 if self.path.startswith(meta_prefix):
51 key = self.path[len(meta_prefix):]
52 body = json.dumps({"key": key, "meta": cache.metas.get(key)}).encode()
53 self.send_response(200 if key in cache.entries else 404)
54 self.send_header("Content-Length", str(len(body)))
55 self.end_headers()
56 self.wfile.write(body)
57 return
59 data = cache.entries.get(key)
60 if data is None:
61 status = 404
62 self.send_response(404)
63 self.end_headers()
64 else:
65 etag = f'"{hashlib.md5(data).hexdigest()}"'
66 status = 304 if self.headers.get("If-None-Match") == etag else 200
67 self.send_response(status)
68 self.send_header("ETag", etag)
69 if status == 200:
70 self.send_header("Content-Length", str(len(data)))
71 self.end_headers()
72 if status == 200:
73 self.wfile.write(data)
76 def log_message(self, *args):
77 pass
79 self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
80 self.url = f"http://127.0.0.1:{self.server.server_address[1]}{PREFIX[:-len('entries/')]}"
81 threading.Thread(target=self.server.serve_forever, daemon=True).start()
88@pytest.fixture
90 monkeypatch.delenv("CYMSIM_CACHE_KEY", raising=False)
92 yield c
93 c.server.shutdown()
96def _config(base, duration):
97 cfg = copy.deepcopy(base)
98 cfg.numerics.duration = duration
99 cfg.output.displacement_rate = 200.0
100 cfg.output.probes = [(0.1, 0.05), (0.05, -0.12)]
101 return cfg
104@pytest.fixture(scope="module")
105def runs(coarse_config, tmp_path_factory):
106 """Uninterrupted runs of 0.02 and 0.05 s, saved."""
107 d = tmp_path_factory.mktemp("runs")
108 out = {}
109 for T in (0.02, 0.05):
110 out[T] = d / f"{T}.cymsim"
111 Simulator(_config(coarse_config, T)).run(save_path=str(out[T]))
112 return out
115def test_without_an_entry_the_run_starts_from_rest(coarse_config, runs, cache, tmp_path):
116 out = tmp_path / "out.cymsim"
117 cached_run(_config(coarse_config, 0.05), cache.url, save_path=str(out), directory=tmp_path / "c")
118 _assert_same_files(out, runs[0.05])
122def test_a_covering_entry_is_cut_without_building_the_model(coarse_config, runs, cache, tmp_path, monkeypatch):
123 cfg = _config(coarse_config, 0.03)
124 ref = tmp_path / "ref.cymsim"
125 Simulator(cfg).run(save_path=str(ref), resume=load_cymsim(runs[0.05]))
126 cache.put(cfg, runs[0.05])
128 def no_model(*args, **kwargs):
129 raise AssertionError("the model was built")
131 monkeypatch.setattr(Simulator, "__init__", no_model)
132 out = tmp_path / "out.cymsim"
133 res = cached_run(cfg, cache.url, save_path=str(out), directory=tmp_path / "c")
134 assert res.complete and len(res.pressure_rayleigh) == round(44100 * 0.03)
135 _assert_same_files(out, ref) # including the state, whose overhang has grown
136 # the energy log stops where an uninterrupted 0.03 s run stops, as on the resume path
137 cached_log, ref_log = load_cymsim(runs[0.05]).energy_log, load_cymsim(ref).energy_log
138 assert res.energy_log == ref_log and len(ref_log) < len(cached_log)
139 assert res.energy_log[-1].time <= 0.03 + 0.02 and cached_log[-1].time > 0.05
142def test_a_shorter_entry_is_continued(coarse_config, runs, cache, tmp_path):
143 cfg = _config(coarse_config, 0.05)
144 cache.put(cfg, runs[0.02])
145 out = tmp_path / "out.cymsim"
146 cached_run(cfg, cache.url, save_path=str(out), directory=tmp_path / "c")
147 _assert_same_files(out, runs[0.05])
150def test_an_entry_of_another_configuration_is_ignored(coarse_config, runs, cache, tmp_path, caplog):
151 cfg = _config(coarse_config, 0.02)
152 other = copy.deepcopy(cfg)
153 other.hits[0].velocity *= 2
154 other_path = tmp_path / "other.cymsim"
155 Simulator(other).run(save_path=str(other_path))
156 cache.entries[cache_key(cfg)] = other_path.read_bytes()
157 out = tmp_path / "out.cymsim"
158 with caplog.at_level(logging.WARNING, logger="cymsim"):
159 cached_run(cfg, cache.url, save_path=str(out), directory=tmp_path / "c")
160 assert "different configuration" in caplog.text
161 _assert_same_files(out, runs[0.02])
164def test_local_copies(coarse_config, runs, cache, tmp_path):
165 cfg, d = _config(coarse_config, 0.02), tmp_path / "c"
166 key = cache_key(cfg)
167 cache.put(cfg, runs[0.02])
168 p = fetch(cache.url, key, d)
169 assert p == d / f"{key}.cymsim" and p.read_bytes() == runs[0.02].read_bytes()
170 assert fetch(cache.url, key, d) == p
173 assert fetch(cache.url, key, d).read_bytes() == runs[0.05].read_bytes()
174 dead = "http://127.0.0.1:9/api/apps/cymsim" # unreachable: the local copy serves
175 assert fetch(dead, key, d) == p
176 with pytest.raises(urllib.error.URLError):
177 fetch(dead, "v1/missing", d)
178 del cache.entries[key] # deleted remotely: the local copy goes too
179 assert fetch(cache.url, key, d) is None and not p.exists()
182def test_the_acceleration_record_bypasses_the_cache(coarse_config, cache, tmp_path):
183 cfg = _config(coarse_config, 0.02)
184 cfg.output.record_acceleration = True
185 res = cached_run(cfg, cache.url, directory=tmp_path / "c")
186 assert res.acceleration is not None and cache.requests == []
189def test_cli_cache(coarse_config, runs, cache, tmp_path, monkeypatch):
190 cfg = _config(coarse_config, 0.03)
191 cfg.output.wav = cfg.output.npz = cfg.output.cascade_plot = False
192 cfg.output.prefix = str(tmp_path / "x")
193 ypath = tmp_path / "c.yaml"
194 cfg.to_yaml(ypath)
195 cache.put(cfg, runs[0.05])
196 monkeypatch.setenv("CYMSIM_CACHE_DIR", str(tmp_path / "c"))
197 out = tmp_path / "out.cymsim"
198 assert main(["run", str(ypath), "--cache", cache.url, "-o", str(out), "--quiet"]) == 0
199 assert len(load_cymsim(out).pressure_rayleigh) == round(44100 * 0.03)
200 monkeypatch.setenv("CYMSIM_CACHE_URL", cache.url)
201 assert main(["run", str(ypath), "-o", str(out), "--quiet"]) == 0
202 assert main(["run", str(ypath), "--no-cache", "-o", str(out), "--quiet"]) == 0
207def _load_entry(cache, key, path):
208 path.write_bytes(cache.entries[key])
209 return path
212def test_a_computed_run_is_uploaded(coarse_config, runs, cache, tmp_path):
213 cfg, d = _config(coarse_config, 0.02), tmp_path / "c"
214 key = cache_key(cfg)
215 cached_run(cfg, cache.url, directory=d, passcode=PASSCODE)
216 _assert_same_files(_load_entry(cache, key, tmp_path / "e.cymsim"), runs[0.02])
217 meta = entry_meta(cache.url, key)
218 assert meta["n_steps_done"] == load_cymsim(runs[0.02]).n_steps_done and meta["complete"]
219 (seg,) = meta["segments"]
220 assert seg["from"] == 0 and seg["to"] == meta["n_steps_done"] and seg["device"].startswith("cpu")
221 assert meta["config"] == json.loads(json.dumps(key_fields(cfg)))
222 # the upload is the local copy: the next lookup is a 304 and uploads nothing
223 res = cached_run(cfg, cache.url, directory=d, passcode=PASSCODE)
224 assert res.complete and cache.requests[-1] == ("GET", key, 304)
225 assert [r for r in cache.requests if r[0] == "PUT"] == [("PUT", key, 201)]
228def test_a_continued_run_replaces_the_shorter_entry(coarse_config, runs, cache, tmp_path, caplog):
229 cfg = _config(coarse_config, 0.05)
230 key = cache_key(cfg)
231 n = load_cymsim(runs[0.02]).n_steps_done
232 cache.put(cfg, runs[0.02], {"n_steps_done": n, "segments": [{"from": 0, "to": n, "device": "gpu:Fake"}]})
233 with caplog.at_level(logging.WARNING, logger="cymsim"):
234 cached_run(cfg, cache.url, directory=tmp_path / "c", passcode=PASSCODE)
235 assert "integrated on gpu:Fake" in caplog.text
236 _assert_same_files(_load_entry(cache, key, tmp_path / "e.cymsim"), runs[0.05])
237 segs = cache.metas[key]["segments"]
238 assert [(g["from"], g["to"]) for g in segs] == [(0, n), (n, load_cymsim(runs[0.05]).n_steps_done)]
241def test_reading_without_a_passcode_uploads_nothing(coarse_config, cache, tmp_path):
242 cached_run(_config(coarse_config, 0.02), cache.url, directory=tmp_path / "c")
243 assert not cache.entries and all(r[0] == "GET" for r in cache.requests)
246def test_a_cut_run_is_not_uploaded(coarse_config, runs, cache, tmp_path):
247 cfg = _config(coarse_config, 0.02)
248 cache.put(cfg, runs[0.05], {"n_steps_done": load_cymsim(runs[0.05]).n_steps_done})
249 cached_run(cfg, cache.url, directory=tmp_path / "c", passcode=PASSCODE)
250 assert all(r[0] == "GET" for r in cache.requests)
253def test_a_longer_entry_is_never_replaced(coarse_config, runs, cache, tmp_path):
254 cfg = _config(coarse_config, 0.02)
255 long_steps = load_cymsim(runs[0.05]).n_steps_done
256 cache.put(cfg, runs[0.05], {"n_steps_done": long_steps})
257 assert not publish(load_cymsim(runs[0.02]), cache.url, cache_key(cfg), PASSCODE, [], directory=tmp_path / "c")
258 assert all(r[0] == "GET" for r in cache.requests)
259 assert cache.entries[cache_key(cfg)] == runs[0.05].read_bytes()
262def test_a_failed_upload_keeps_the_run(coarse_config, runs, cache, tmp_path, caplog):
263 cfg, out = _config(coarse_config, 0.02), tmp_path / "out.cymsim"
264 with caplog.at_level(logging.ERROR, logger="cymsim"):
265 res = cached_run(cfg, cache.url, save_path=str(out), directory=tmp_path / "c", passcode="wrong")
266 assert res.complete and "upload of" in caplog.text and "Invalid passcode" in caplog.text
267 assert ("PUT", cache_key(cfg), 401) in cache.requests and not cache.entries
268 _assert_same_files(out, runs[0.02])
↑↓ move↵openescclose