Skip to content

test_helpers

ipw.tests.analysis.test_helpers

Tests for analysis helper functions.

TestLoadMetricsDataset

Test dataset loading.

Source code in intelligence-per-watt/src/ipw/tests/analysis/test_helpers.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class TestLoadMetricsDataset:
    """Test dataset loading."""

    def test_loads_single_dataset(self, tmp_path: Path) -> None:
        # Create a simple dataset
        dataset = Dataset.from_list([{"id": 1}, {"id": 2}])
        dataset.save_to_disk(str(tmp_path / "data"))

        loaded = load_metrics_dataset(tmp_path / "data")
        assert isinstance(loaded, Dataset)
        assert len(loaded) == 2

    def test_loads_first_split_from_dataset_dict(self, tmp_path: Path) -> None:
        # Create a DatasetDict
        dataset = Dataset.from_list([{"id": 1}])
        dataset_dict = DatasetDict({"train": dataset})
        dataset_dict.save_to_disk(str(tmp_path / "data"))

        loaded = load_metrics_dataset(tmp_path / "data")
        assert isinstance(loaded, Dataset)
        assert len(loaded) == 1

    def test_raises_on_empty_dataset_dict(self, tmp_path: Path) -> None:
        dataset_dict = DatasetDict({})
        dataset_dict.save_to_disk(str(tmp_path / "data"))

        with pytest.raises(RuntimeError, match="No splits found"):
            load_metrics_dataset(tmp_path / "data")

    def test_raises_on_empty_dataset(self, tmp_path: Path) -> None:
        from datasets import Features, Value

        # Create empty dataset with schema
        dataset = Dataset.from_dict(
            {"id": []},
            features=Features({"id": Value("int64")}),
        )
        try:
            dataset.save_to_disk(str(tmp_path / "data"))
        except (IndexError, Exception):
            pytest.skip("datasets library cannot save empty datasets in this version")

        with pytest.raises((RuntimeError, IndexError)):
            # Our code raises RuntimeError("Empty dataset"), but some versions
            # of the datasets library raise IndexError during load_from_disk
            # for 0-shard datasets before our check runs.
            load_metrics_dataset(tmp_path / "data")

TestResolveModelName

Test model name resolution.

Source code in intelligence-per-watt/src/ipw/tests/analysis/test_helpers.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
class TestResolveModelName:
    """Test model name resolution."""

    def test_returns_requested_model_when_available(self) -> None:
        dataset = Dataset.from_list(
            [
                {"model_metrics": {"llama": {"value": 1}}},
            ]
        )

        model = resolve_model_name(dataset, "llama", Path("/fake"))
        assert model == "llama"

    def test_raises_when_requested_model_not_found(self) -> None:
        dataset = Dataset.from_list(
            [
                {"model_metrics": {"llama": {"value": 1}}},
            ]
        )

        with pytest.raises(RuntimeError, match="Model 'gpt4' not found"):
            resolve_model_name(dataset, "gpt4", Path("/fake"))

    def test_infers_single_model_automatically(self) -> None:
        dataset = Dataset.from_list(
            [
                {"model_metrics": {"llama": {"value": 1}}},
                {"model_metrics": {"llama": {"value": 2}}},
            ]
        )

        model = resolve_model_name(dataset, None, Path("/fake"))
        assert model == "llama"

    def test_raises_when_multiple_models_and_none_requested(self) -> None:
        dataset = Dataset.from_list(
            [
                {"model_metrics": {"llama": {"value": 1}}},
                {"model_metrics": {"gpt4": {"value": 2}}},
            ]
        )

        with pytest.raises(RuntimeError, match="multiple models"):
            resolve_model_name(dataset, None, Path("/fake"))

    def test_raises_when_no_models_found(self) -> None:
        dataset = Dataset.from_list(
            [
                {"other_field": "value"},
            ]
        )

        with pytest.raises(RuntimeError, match="Could not infer model name"):
            resolve_model_name(dataset, None, Path("/fake"))

    def test_handles_empty_model_metrics(self) -> None:
        dataset = Dataset.from_list(
            [
                {"model_metrics": {}},
                {"model_metrics": {"llama": {"value": 1}}},
            ]
        )

        model = resolve_model_name(dataset, None, Path("/fake"))
        assert model == "llama"

    def test_preserves_model_order(self) -> None:
        # First non-empty model should be returned when single model
        dataset = Dataset.from_list(
            [
                {"model_metrics": {}},
                {"model_metrics": {"alpha": {"value": 1}}},
                {"model_metrics": {"alpha": {"value": 2}}},
            ]
        )

        model = resolve_model_name(dataset, None, Path("/fake"))
        assert model == "alpha"

TestIterModelEntries

Test model entry iteration.

Source code in intelligence-per-watt/src/ipw/tests/analysis/test_helpers.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
class TestIterModelEntries:
    """Test model entry iteration."""

    def test_yields_entries_for_matching_model(self) -> None:
        dataset = Dataset.from_list(
            [
                {"model_metrics": {"llama": {"tokens": 100}}},
                {"model_metrics": {"llama": {"tokens": 200}}},
            ]
        )

        entries = list(iter_model_entries(dataset, "llama"))
        assert len(entries) == 2
        assert entries[0]["tokens"] == 100
        assert entries[1]["tokens"] == 200

    def test_skips_entries_without_model(self) -> None:
        dataset = Dataset.from_list(
            [
                {"model_metrics": {"llama": {"tokens": 100}}},
                {"model_metrics": {"gpt4": {"tokens": 200}}},
                {"model_metrics": {"llama": {"tokens": 300}}},
            ]
        )

        entries = list(iter_model_entries(dataset, "llama"))
        assert len(entries) == 2
        assert entries[0]["tokens"] == 100
        assert entries[1]["tokens"] == 300

    def test_skips_records_without_model_metrics(self) -> None:
        dataset = Dataset.from_list(
            [
                {"model_metrics": None, "other_field": "value"},
                {"model_metrics": {"llama": {"tokens": 100}}, "other_field": None},
            ]
        )

        entries = list(iter_model_entries(dataset, "llama"))
        assert len(entries) == 1

    def test_skips_non_mapping_model_metrics(self) -> None:
        # Note: HuggingFace Dataset will normalize types, so we can't mix string and dict
        # Instead, test with None and dict
        dataset = Dataset.from_list(
            [
                {"model_metrics": None},
                {"model_metrics": {"llama": {"tokens": 100}}},
            ]
        )

        entries = list(iter_model_entries(dataset, "llama"))
        assert len(entries) == 1

    def test_returns_empty_when_no_matches(self) -> None:
        dataset = Dataset.from_list(
            [
                {"model_metrics": {"gpt4": {"tokens": 100}}},
            ]
        )

        entries = list(iter_model_entries(dataset, "llama"))
        assert len(entries) == 0

TestCollectRunMetadata

Test metadata collection.

Source code in intelligence-per-watt/src/ipw/tests/analysis/test_helpers.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
class TestCollectRunMetadata:
    """Test metadata collection."""

    def test_extracts_system_info_from_first_entry(self) -> None:
        entries = [
            {"system_info": {"cpu": "Intel"}, "gpu_info": {}},
            {"system_info": {"cpu": "AMD"}, "gpu_info": {}},
        ]

        system_info, _ = collect_run_metadata(entries)
        assert system_info == {"cpu": "Intel"}

    def test_extracts_gpu_info_from_first_entry(self) -> None:
        entries = [
            {"system_info": {}, "gpu_info": {"name": "RTX 3090"}},
            {"system_info": {}, "gpu_info": {"name": "RTX 4090"}},
        ]

        _, gpu_info = collect_run_metadata(entries)
        assert gpu_info == {"name": "RTX 3090"}

    def test_returns_empty_when_no_metadata(self) -> None:
        entries = [
            {"other_field": "value"},
        ]

        system_info, gpu_info = collect_run_metadata(entries)
        assert system_info == {}
        assert gpu_info == {}

    def test_skips_non_mapping_metadata(self) -> None:
        entries = [
            {"system_info": "not a dict", "gpu_info": "not a dict"},
            {"system_info": {"cpu": "Intel"}, "gpu_info": {"name": "RTX 3090"}},
        ]

        system_info, gpu_info = collect_run_metadata(entries)
        assert system_info == {"cpu": "Intel"}
        assert gpu_info == {"name": "RTX 3090"}

    def test_stops_early_when_both_found(self) -> None:
        # Create a mock iterator to verify early stopping
        def make_entries():
            yield {"system_info": {"cpu": "Intel"}, "gpu_info": {"name": "RTX"}}
            # This shouldn't be accessed
            raise RuntimeError("Should not iterate past first complete entry")

        system_info, gpu_info = collect_run_metadata(make_entries())
        assert system_info == {"cpu": "Intel"}
        assert gpu_info == {"name": "RTX"}

    def test_continues_until_both_found(self) -> None:
        entries = [
            {"system_info": {"cpu": "Intel"}},
            {"gpu_info": {"name": "RTX"}},
        ]

        system_info, gpu_info = collect_run_metadata(entries)
        assert system_info == {"cpu": "Intel"}
        assert gpu_info == {"name": "RTX"}