documentlanguagemodel Public

Watch 0 Fork 0 Star 0
Python · 8303 bytes Raw Blame History
1 """Path resolution + layout creation."""
2
3 from __future__ import annotations
4
5 from pathlib import Path
6
7 import pytest
8
9 from dlm.store import errors
10 from dlm.store.layout import (
11 ADAPTER_DIR,
12 LOCK_FILENAME,
13 LOGS_DIR,
14 MANIFEST_FILENAME,
15 )
16 from dlm.store.paths import StorePath, _current_os_name, dlm_home, ensure_home, for_dlm
17
18 VALID_ID = "01HZ4X7TGZM3J1A2B3C4D5E6F7"
19
20
21 class TestDlmHome:
22 def test_current_os_name_passthrough(self) -> None:
23 import os
24
25 assert _current_os_name() == os.name
26
27 def test_override_takes_precedence(self, tmp_path: Path) -> None:
28 assert dlm_home(override=tmp_path / "custom") == (tmp_path / "custom").resolve()
29
30 def test_env_var_respected(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
31 monkeypatch.setenv("DLM_HOME", str(tmp_path / "env-home"))
32 assert dlm_home() == (tmp_path / "env-home").resolve()
33
34 def test_explicit_override_beats_env(
35 self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
36 ) -> None:
37 monkeypatch.setenv("DLM_HOME", str(tmp_path / "env"))
38 override = tmp_path / "cli"
39 assert dlm_home(override=override) == override.resolve()
40
41 def test_default_on_posix(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
42 monkeypatch.delenv("DLM_HOME", raising=False)
43 monkeypatch.setattr("dlm.store.paths._current_os_name", lambda: "posix")
44 monkeypatch.setattr(Path, "home", lambda: tmp_path / "u")
45 assert dlm_home() == tmp_path / "u" / ".dlm"
46
47 def test_default_on_nt_prefers_appdata(
48 self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
49 ) -> None:
50 monkeypatch.delenv("DLM_HOME", raising=False)
51 monkeypatch.setenv("APPDATA", str(tmp_path / "AppData" / "Roaming"))
52 monkeypatch.setattr("dlm.store.paths._current_os_name", lambda: "nt")
53 assert dlm_home() == (tmp_path / "AppData" / "Roaming").resolve() / "dlm"
54
55
56 class TestEnsureHome:
57 def test_creates_store_subdir(self, tmp_path: Path) -> None:
58 home = ensure_home(override=tmp_path / "h")
59 assert home.exists()
60 assert (home / "store").exists()
61
62
63 class TestForDlm:
64 def test_returns_store_path_under_home(self, tmp_path: Path) -> None:
65 sp = for_dlm(VALID_ID, home=tmp_path)
66 assert sp.root == tmp_path.resolve() / "store" / VALID_ID
67
68 def test_empty_id_rejected(self, tmp_path: Path) -> None:
69 with pytest.raises(ValueError, match="non-empty"):
70 for_dlm("", home=tmp_path)
71
72
73 class TestStorePathAccessors:
74 @pytest.fixture
75 def store(self, tmp_path: Path) -> StorePath:
76 return for_dlm(VALID_ID, home=tmp_path)
77
78 def test_manifest_path(self, store: StorePath) -> None:
79 assert store.manifest.name == MANIFEST_FILENAME
80 assert store.manifest.parent == store.root
81
82 def test_lock_path(self, store: StorePath) -> None:
83 assert store.lock.name == LOCK_FILENAME
84
85 def test_training_state_paths(self, store: StorePath) -> None:
86 assert store.training_state.name == "training_state.pt"
87 assert store.training_state_sha.name == "training_state.pt.sha256"
88
89 def test_adapter_subpaths(self, store: StorePath) -> None:
90 assert store.adapter.name == ADAPTER_DIR
91 assert store.adapter_versions.parent == store.adapter
92 assert store.adapter_version(1).name == "v0001"
93 assert store.adapter_version(1234).name == "v1234"
94
95 def test_logs_dir(self, store: StorePath) -> None:
96 assert store.logs.name == LOGS_DIR
97
98 def test_replay_paths(self, store: StorePath) -> None:
99 assert store.replay_corpus.name == "corpus.zst"
100 assert store.replay_index.name == "index.json"
101
102 def test_adapter_version_zero_rejected(self, store: StorePath) -> None:
103 with pytest.raises(ValueError, match="1-indexed"):
104 store.adapter_version(0)
105
106 def test_cache_dir_for_slug(self, store: StorePath) -> None:
107 assert store.cache_dir_for("qwen2.5-1.5b").name == "qwen2.5-1.5b"
108
109 def test_cache_dir_for_empty_rejected(self, store: StorePath) -> None:
110 with pytest.raises(ValueError):
111 store.cache_dir_for("")
112
113 def test_export_quant_dir(self, store: StorePath) -> None:
114 assert store.export_quant_dir("Q4_K_M").name == "Q4_K_M"
115
116 def test_export_quant_empty_rejected(self, store: StorePath) -> None:
117 with pytest.raises(ValueError):
118 store.export_quant_dir("")
119
120 def test_blob_dir(self, store: StorePath) -> None:
121 assert store.blob_dir.name == "blobs"
122 assert store.blob_dir.parent == store.root
123
124 def test_vl_cache_dir(self, store: StorePath) -> None:
125 assert store.vl_cache_dir.name == "vl-cache"
126 assert store.vl_cache_dir.parent == store.root
127
128 def test_other_lazy_dirs(self, store: StorePath) -> None:
129 assert store.tokenized_cache_dir.name == "tokenized-cache"
130 assert store.audio_cache_dir.name == "audio-cache"
131 assert store.audio_waveform_cache_dir.name == "audio-waveform-cache"
132 assert store.controls_dir.name == "controls"
133 assert store.control_file("demo").name == "demo.safetensors"
134 assert store.control_meta("demo").name == "demo.meta.json"
135
136 def test_blob_and_vl_cache_lazy(self, tmp_path: Path) -> None:
137 sp = for_dlm(VALID_ID, home=tmp_path)
138 sp.ensure_layout()
139 assert not sp.blob_dir.exists()
140 assert not sp.vl_cache_dir.exists()
141
142 def test_exists_reflects_store_root(self, tmp_path: Path) -> None:
143 sp = for_dlm(VALID_ID, home=tmp_path)
144 assert sp.exists() is False
145 sp.ensure_layout()
146 assert sp.exists() is True
147
148
149 class TestEnsureLayout:
150 @pytest.fixture
151 def store(self, tmp_path: Path) -> StorePath:
152 sp = for_dlm(VALID_ID, home=tmp_path)
153 sp.ensure_layout()
154 return sp
155
156 def test_root_created(self, store: StorePath) -> None:
157 assert store.root.is_dir()
158
159 def test_always_on_dirs_created(self, store: StorePath) -> None:
160 assert store.adapter.is_dir()
161 assert store.adapter_versions.is_dir()
162 assert store.logs.is_dir()
163
164 def test_lazy_dirs_not_created(self, store: StorePath) -> None:
165 # Sprint 08/11/06 own these; Sprint 04 must not materialize them.
166 assert not store.replay.exists()
167 assert not store.exports.exists()
168 assert not store.cache.exists()
169
170 def test_is_idempotent(self, store: StorePath) -> None:
171 store.ensure_layout()
172 store.ensure_layout()
173 assert store.adapter.is_dir()
174
175
176 class TestAdapterCurrentPointer:
177 @pytest.fixture
178 def store(self, tmp_path: Path) -> StorePath:
179 sp = for_dlm(VALID_ID, home=tmp_path)
180 sp.ensure_layout()
181 return sp
182
183 def test_resolve_returns_none_when_absent(self, store: StorePath) -> None:
184 assert store.resolve_current_adapter() is None
185
186 def test_set_and_resolve_roundtrip(self, store: StorePath) -> None:
187 version_dir = store.adapter_version(1)
188 version_dir.mkdir(parents=True, exist_ok=True)
189 store.set_current_adapter(version_dir)
190 resolved = store.resolve_current_adapter()
191 assert resolved == version_dir.resolve()
192
193 def test_set_rejects_outside_root(self, store: StorePath, tmp_path: Path) -> None:
194 rogue = tmp_path / "rogue"
195 rogue.mkdir()
196 with pytest.raises(ValueError, match="outside store root"):
197 store.set_current_adapter(rogue)
198
199 def test_empty_pointer_returns_none(self, store: StorePath) -> None:
200 store.adapter_current_pointer.write_text("", encoding="utf-8")
201 assert store.resolve_current_adapter() is None
202
203 def test_resolve_rejects_escape_via_parent_refs(self, store: StorePath) -> None:
204 # An adversarial pointer pointing outside the root (`..`) must fail.
205 store.adapter_current_pointer.write_text("../../escape", encoding="utf-8")
206 with pytest.raises(ValueError, match="escapes store root"):
207 store.resolve_current_adapter()
208
209
210 class TestErrorsSurface:
211 """Sanity-check the errors module exports."""
212
213 def test_unknown_store_error_message(self, tmp_path: Path) -> None:
214 err = errors.UnknownStoreError("deadbeef", tmp_path)
215 assert "deadbeef" in str(err)