Skip to content

conftest

ipw.tests.conftest

Shared test fixtures for the IPW test suite.

synthetic_telemetry_readings()

List of TelemetryReading with all fields (GPU + CPU + ANE).

Source code in intelligence-per-watt/src/ipw/tests/conftest.py 
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
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
@pytest.fixture()
def synthetic_telemetry_readings() -> List[TelemetryReading]:
    """List of TelemetryReading with all fields (GPU + CPU + ANE)."""
    return [
        TelemetryReading(
            power_watts=150.0,
            energy_joules=100.0,
            temperature_celsius=65.0,
            gpu_memory_usage_mb=4096.0,
            gpu_memory_total_mb=8192.0,
            cpu_memory_usage_mb=16384.0,
            cpu_power_watts=45.0,
            cpu_energy_joules=50.0,
            ane_power_watts=5.0,
            ane_energy_joules=10.0,
            gpu_compute_utilization_pct=85.0,
            gpu_memory_bandwidth_utilization_pct=60.0,
            gpu_tensor_core_utilization_pct=70.0,
            platform="linux",
            timestamp_nanos=1000000000,
            system_info=SystemInfo(
                os_name="Linux",
                os_version="6.8.0",
                kernel_version="6.8.0-60-generic",
                host_name="testhost",
                cpu_count=16,
                cpu_brand="AMD EPYC",
            ),
            gpu_info=GpuInfo(
                name="RTX 4090",
                vendor="NVIDIA",
                device_id=0,
                device_type="discrete",
                backend="cuda",
            ),
        ),
        TelemetryReading(
            power_watts=155.0,
            energy_joules=120.0,
            temperature_celsius=67.0,
            gpu_memory_usage_mb=4100.0,
            gpu_memory_total_mb=8192.0,
            cpu_memory_usage_mb=16400.0,
            cpu_power_watts=47.0,
            cpu_energy_joules=55.0,
            ane_power_watts=5.5,
            ane_energy_joules=12.0,
            gpu_compute_utilization_pct=88.0,
            gpu_memory_bandwidth_utilization_pct=62.0,
            gpu_tensor_core_utilization_pct=72.0,
            platform="linux",
            timestamp_nanos=2000000000,
            system_info=SystemInfo(
                os_name="Linux",
                os_version="6.8.0",
                kernel_version="6.8.0-60-generic",
                host_name="testhost",
                cpu_count=16,
                cpu_brand="AMD EPYC",
            ),
            gpu_info=GpuInfo(
                name="RTX 4090",
                vendor="NVIDIA",
                device_id=0,
                device_type="discrete",
                backend="cuda",
            ),
        ),
    ]

mock_agent()

Returns a mock agent producing predictable AgentRunResult.

Source code in intelligence-per-watt/src/ipw/tests/conftest.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
@pytest.fixture()
def mock_agent() -> MagicMock:
    """Returns a mock agent producing predictable AgentRunResult."""
    agent = MagicMock()
    agent.run.return_value = AgentRunResult(
        content="mock answer",
        tool_calls_attempted=2,
        tool_calls_succeeded=2,
        tool_names_used=["calculator", "web_search"],
        num_turns=3,
        input_tokens=100,
        output_tokens=50,
        cost_usd=0.005,
    )
    return agent

mock_dataset()

DatasetProvider mock with N known records.

Source code in intelligence-per-watt/src/ipw/tests/conftest.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
@pytest.fixture()
def mock_dataset() -> MagicMock:
    """DatasetProvider mock with N known records."""
    dataset = MagicMock()
    records = [
        DatasetRecord(
            problem=f"Question {i}",
            answer=f"Answer {i}",
            subject=f"subject_{i}",
            dataset_metadata={"index": i, "dataset_name": "test"},
        )
        for i in range(5)
    ]
    dataset.iter_records.return_value = iter(records)
    dataset.__iter__ = lambda self: iter(records)
    dataset.size.return_value = len(records)
    dataset.dataset_id = "test"
    dataset.dataset_name = "TestDataset"
    return dataset

sample_event_recorder()

EventRecorder with pre-populated events.

Source code in intelligence-per-watt/src/ipw/tests/conftest.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
@pytest.fixture()
def sample_event_recorder() -> EventRecorder:
    """EventRecorder with pre-populated events."""
    recorder = EventRecorder()
    recorder.record("lm_inference_start", model="test-model")
    recorder.record("tool_call_start", tool="calculator")
    recorder.record("tool_call_end", tool="calculator", result="42")
    recorder.record(
        "lm_inference_end",
        model="test-model",
        prompt_tokens=100,
        completion_tokens=50,
    )
    return recorder

mock_output_dir(tmp_path)

tmp_path-based output directory for test runs.

Source code in intelligence-per-watt/src/ipw/tests/conftest.py 
146
147
148
149
150
151
@pytest.fixture()
def mock_output_dir(tmp_path: Path) -> Path:
    """tmp_path-based output directory for test runs."""
    output = tmp_path / "test_output"
    output.mkdir(parents=True, exist_ok=True)
    return output