| """Bounded idle-gated retries never qualify contamination, drift, or partial runs.""" |
| from __future__ import annotations |
|
|
| import json |
| from pathlib import Path |
|
|
| import pytest |
|
|
| from d2 import greedy_reference as G |
| from d2 import smoke_prepare as P |
| from d2.decode_smoke_dry import DrySmoke |
|
|
| REPO = Path(__file__).resolve().parents[4] |
|
|
|
|
| def fixture(tmp_path, monkeypatch, failure='none'): |
| monkeypatch.setattr(G, 'PRIVATE', tmp_path) |
| monkeypatch.setattr(P, 'PRIVATE', tmp_path) |
| io = DrySmoke(REPO, tmp_path) |
| spec = io.inputs() |
| capture = Path(spec['greedy_reference']['path']) |
| monkeypatch.setattr(G, 'K1_CAPTURE', capture) |
| proof_path = Path(spec['greedy_reference']['request_receipt']) |
| proof = json.loads(proof_path.read_text()) |
| proof['request']['priority'] = 0 |
| proof_path.write_text(json.dumps(proof)) |
| spec['requests'][2]['body']['priority'] = 0 |
| spec['greedy_reference']['request_receipt_sha256'] = G.digest(proof_path) |
| monkeypatch.setattr(G, 'historical_request', lambda *args: proof['request']) |
| inputs = tmp_path/'input.json' |
| inputs.write_text(json.dumps(spec)) |
|
|
| class Fake: |
| count = 0 |
| metric_calls = 0 |
| calls = [] |
| pauses = 0 |
| def metrics(self): |
| self.metric_calls += 1 |
| gauges = 'vllm:num_requests_running{} 0\nvllm:num_requests_waiting{} 0\n' |
| if failure == 'busy' or (failure == 'busy_once' and self.metric_calls == 1): |
| gauges = 'vllm:num_requests_running{} 1\nvllm:num_requests_waiting{} 0\n' |
| return io.latency_metrics(self.count)+gauges |
| def pause(self): self.pauses += 1 |
| def post(self, body, attempt): |
| self.calls.append(dict(body)) |
| self.count += 1 |
| if failure in ('contaminated', 'contaminated_once') and (failure == 'contaminated' or len(self.calls) == 1): |
| self.count += 1 |
| if failure == 'foreign_live' and len(self.calls) == 1: |
| raise G.Contamination('foreign work; owned child cancelled') |
| response = {'choices':[{'finish_reason':'stop','message':{'content': |
| 'drift' if failure == 'drift' else 'synthetic-greedy'}}], |
| 'usage':{'prompt_tokens_details':{'cached_tokens':0}}} |
| (attempt/'response.json').write_text(json.dumps(response)) |
| return response |
| return Fake(), inputs |
|
|
|
|
| @pytest.mark.parametrize('failure', ['none','busy_once','contaminated_once','foreign_live']) |
| def test_reference_collects_five_warm_repeats_with_bounded_retries(tmp_path, monkeypatch, failure): |
| io, inputs = fixture(tmp_path, monkeypatch, failure) |
| result = G.capture(io, inputs, {}, tmp_path/'r2', lambda: {}) |
| assert result['qualified'] is True and result['repeats'] == 5 and result['warmups'] == 1 |
| assert result['requests_sent'] == (7 if failure in ('contaminated_once','foreign_live') else 6) |
| assert all(body['priority'] == 0 for body in io.calls) |
| assert result['attempts'] == (6 if failure == 'none' else 7) |
| assert io.pauses == (0 if failure == 'none' else 1) |
| qualified = json.loads((tmp_path/'smoke-input-r2-qualified.json').read_text()) |
| G.checked_spec(qualified) |
| count = len(io.calls) |
| with pytest.raises(ValueError, match='overwrite'): |
| G.capture(io, inputs, {}, tmp_path/'other', lambda: {}) |
| assert len(io.calls) == count |
|
|
|
|
| @pytest.mark.parametrize('failure', ['busy','contaminated','drift','runtime']) |
| def test_reference_never_qualifies_bad_or_exhausted_collection(tmp_path, monkeypatch, failure): |
| io, inputs = fixture(tmp_path, monkeypatch, failure) |
| with pytest.raises(ValueError): |
| G.capture(io, inputs, {}, tmp_path/'r2', lambda: {'changed':True} if failure == 'runtime' else {}) |
| assert not (tmp_path/'smoke-input-r2-qualified.json').exists() |
| assert not (tmp_path/'r2/k1-request-receipt.json').exists() |
| assert len(io.calls) == (0 if failure == 'busy' else 1 if failure == 'drift' else 6 if failure == 'runtime' else 10) |
| if failure in ('busy','contaminated'): |
| assert (tmp_path/'r2/checkpoint-10.json').exists() |
|
|
|
|
| def test_offline_qualification_rechecks_saved_bytes_without_sending(tmp_path, monkeypatch): |
| io, inputs = fixture(tmp_path, monkeypatch) |
| G.capture(io, inputs, {}, tmp_path/'r2', lambda: {}) |
| (tmp_path/'smoke-input-r2-qualified.json').unlink() |
| (tmp_path/'r2/k1-request-receipt.json').unlink() |
| (tmp_path/'r2/summary.json').unlink() |
| def forbidden(*args, **kwargs): raise AssertionError('offline recovery sent network/process') |
| monkeypatch.setattr(G.subprocess, 'Popen', forbidden) |
| assert G.qualify_saved(inputs, tmp_path/'r2', {}, 'synthetic recovery')['qualified'] |
| assert len(io.calls) == 6 |
| (tmp_path/'r2/attempt-02/response.json').write_text('{}') |
| with pytest.raises(ValueError, match='response changed'): |
| G.qualify_saved(inputs, tmp_path/'r2', {}) |
|
|
|
|
| def test_docker_mount_order_does_not_change_runtime_identity(tmp_path, monkeypatch): |
| profiles = REPO/'offload/operations/profiles.v1.json' |
| profile = next(p for p in json.loads(profiles.read_text())['profiles'] if p['name'] == 'K1-327680') |
| env = [k+'='+v for k,v in profile['match']['env'].items() if v is not None] |
| mounts = [{'Destination':'/z'},{'Destination':'/a'}] |
| def output(argv, **kwargs): |
| if argv[-1] == 'docker exec dsv41-serve cat /serve.sh': |
| return '/srv/dsv41/legal-v2a --max-model-len 327680 --port 8000' |
| mounts.reverse() |
| return json.dumps([{'Image':G.IMAGE,'Id':'a'*64,'State':{'Running':True}, |
| 'Config':{'Env':env},'Mounts':mounts}]) |
| monkeypatch.setattr(G.subprocess,'check_output',output) |
| assert G.runtime_identity(REPO) == G.runtime_identity(REPO) |
|
|
