Add 1_write_events/tests/test_core.py

1_write_events/tests/test_core.py

@@ -0,0 +1,89 @@
+import json
+import os
+from pathlib import Path
+import pytest
+
+import src.write_events.core as core
+
+OUTPUT_PATH = Path('output/write_events.json')
+
+@pytest.fixture(autouse=True)
+def cleanup_output():
+    if OUTPUT_PATH.exists():
+        OUTPUT_PATH.unlink()
+    yield
+    if OUTPUT_PATH.exists():
+        OUTPUT_PATH.unlink()
+
+
+def _load_events():
+    with open(OUTPUT_PATH, 'r', encoding='utf-8') as f:
+        data = json.load(f)
+    assert isinstance(data, list)
+    return data
+
+
+def test_record_write_event_creates_file(tmp_path, monkeypatch):
+    target = tmp_path / 'write_events.json'
+    monkeypatch.setattr(core, 'OUTPUT_PATH', target)
+    core.record_write_event('base_raw_write', '52a9f', 'C03_01')
+    assert target.exists(), 'Output JSON file should be created.'
+    data = json.loads(target.read_text())
+    assert isinstance(data, list)
+    assert len(data) == 1
+    ev = data[0]
+    # Ensure keys and types
+    expected_keys = {'timestamp', 'field_tag', 'new_value_hash', 'corr_id', 'cpu', 'ktime_ns'}
+    assert expected_keys.issubset(ev.keys())
+    assert ev['field_tag'] == 'base_raw_write'
+    assert ev['new_value_hash'] == '52a9f'
+    assert ev['corr_id'] == 'C03_01'
+    assert isinstance(ev['cpu'], int)
+    assert isinstance(ev['ktime_ns'], int)
+
+
+def test_record_write_event_appends(monkeypatch, tmp_path):
+    target = tmp_path / 'write_events.json'
+    monkeypatch.setattr(core, 'OUTPUT_PATH', target)
+    core.record_write_event('base_raw_write', '111aa', 'C03_01')
+    core.record_write_event('nsec_base_write', '999bb', 'C03_01')
+    data = json.loads(target.read_text())
+    assert len(data) == 2
+    tags = [ev['field_tag'] for ev in data]
+    assert tags == ['base_raw_write', 'nsec_base_write']
+
+
+def test_input_validation(monkeypatch, tmp_path):
+    target = tmp_path / 'write_events.json'
+    monkeypatch.setattr(core, 'OUTPUT_PATH', target)
+    # Invalid types should raise AssertionError or TypeError
+    with pytest.raises((AssertionError, TypeError)):
+        core.record_write_event(123, 'abc', 'cid')
+    with pytest.raises((AssertionError, TypeError)):
+        core.record_write_event('tag', 999, 'cid')
+    with pytest.raises((AssertionError, TypeError)):
+        core.record_write_event('tag', 'hash', None)
+
+
+def test_multiple_calls_persist(monkeypatch, tmp_path):
+    target = tmp_path / 'write_events.json'
+    monkeypatch.setattr(core, 'OUTPUT_PATH', target)
+    for i in range(3):
+        core.record_write_event('f', f'h{i}', 'C03_01')
+    data = json.loads(target.read_text())
+    assert len(data) == 3
+    hashes = [ev['new_value_hash'] for ev in data]
+    assert hashes == ['h0', 'h1', 'h2']
+
+
+def test_timestamp_and_cpu_values(monkeypatch, tmp_path):
+    target = tmp_path / 'write_events.json'
+    monkeypatch.setattr(core, 'OUTPUT_PATH', target)
+    core.record_write_event('field', 'valhash', 'C03_01')
+    data = json.loads(target.read_text())
+    ev = data[0]
+    assert isinstance(ev['timestamp'], str)
+    # CPU should be non-negative integer
+    assert ev['cpu'] >= 0
+    # ktime_ns should be plausible nanoseconds (non-negative)
+    assert ev['ktime_ns'] >= 0