| """nexa telemetry: the span contract, and the two things it must never do. |
| |
| Telemetry is the one subsystem here that is allowed to fail silently, and that |
| licence is exactly why it needs pinning. Two invariants carry the whole design: |
| |
| it must not break the tool a file it cannot write is a line lost, |
| never an exception reaching the caller |
| |
| it must not leak the work prompts and answers stay out unless |
| somebody explicitly asked for them |
| |
| The module reads its configuration once, at import, into module-level flags. |
| Tests that need a different configuration set the flag rather than the |
| environment variable: setting the variable after import changes nothing, which |
| is the trap this docstring exists to stop the next person falling into. |
| """ |
| import io |
| import json |
| import os |
|
|
| import pytest |
|
|
| import telemetry |
|
|
|
|
| @pytest.fixture() |
| def journal(tmp_path, monkeypatch): |
| """A telemetry file of this test's own, enabled, read back as records.""" |
| path = tmp_path / "telemetry.jsonl" |
| monkeypatch.setattr(telemetry, "_FILE", str(path)) |
| monkeypatch.setattr(telemetry, "_ENABLED", True) |
|
|
| def records(): |
| if not path.exists(): |
| return [] |
| with io.open(str(path), encoding="utf-8") as handle: |
| return [json.loads(line) for line in handle if line.strip()] |
|
|
| records.path = path |
| return records |
|
|
|
|
| |
|
|
| def test_an_event_carries_its_schema_version(journal): |
| telemetry.emit("localai.test", {"rules": 3}) |
| row = journal()[0] |
| assert row["event"] == "localai.test" |
| assert row["schema_version"] == telemetry.SCHEMA_VERSION |
| assert row["rules"] == 3 |
| assert row["timestamp_utc"].endswith("+00:00"), "UTC, and says so" |
|
|
|
|
| def test_disabled_telemetry_writes_nothing(tmp_path, monkeypatch): |
| path = tmp_path / "telemetry.jsonl" |
| monkeypatch.setattr(telemetry, "_FILE", str(path)) |
| monkeypatch.setattr(telemetry, "_ENABLED", False) |
| telemetry.emit("localai.test", {"rules": 3}) |
| assert not path.exists() |
|
|
|
|
| def test_a_file_that_cannot_be_written_does_not_raise(monkeypatch, tmp_path): |
| """The load-bearing promise. A directory that does not exist is the |
| ordinary way this happens, and an exception here would end a run over a |
| logging problem.""" |
| monkeypatch.setattr(telemetry, "_ENABLED", True) |
| monkeypatch.setattr( |
| telemetry, "_FILE", str(tmp_path / "no" / "such" / "dir" / "t.jsonl")) |
| telemetry.emit("localai.test", {"rules": 3}) |
|
|
|
|
| |
|
|
| def test_a_span_emits_one_event_with_its_duration(journal): |
| with telemetry.span("localai.test", {"stack": "odoo"}) as attrs: |
| attrs["verdict"] = "PASS" |
| rows = journal() |
| assert len(rows) == 1 |
| assert rows[0]["stack"] == "odoo" |
| assert rows[0]["verdict"] == "PASS", "the yielded dict is the event" |
| assert rows[0]["duration_ms"] >= 0 |
|
|
|
|
| def test_a_bad_outcome_is_an_attribute_and_not_an_error(journal): |
| """A violation is a successful check, exactly as a REFUSE is a successful |
| decision. Recording either as an error would make the error count useless |
| for finding the thing errors are for.""" |
| with telemetry.span("localai.verify") as attrs: |
| attrs["verdict"] = "FAIL" |
| attrs["violations"] = 2 |
| row = journal()[0] |
| assert row["verdict"] == "FAIL" |
| assert "error" not in row |
|
|
|
|
| def test_an_exception_is_recorded_and_re_raised(journal): |
| """Recorded, because a run that died is the most interesting row in the |
| file. Re-raised, because swallowing it would make telemetry decide what |
| the caller does about a failure.""" |
| with pytest.raises(ValueError): |
| with telemetry.span("localai.test") as attrs: |
| attrs["stack"] = "odoo" |
| raise ValueError("boom") |
| row = journal()[0] |
| assert row["error"] == "ValueError" |
| assert row["stack"] == "odoo", "attributes set before the raise survive" |
| assert row["duration_ms"] >= 0 |
|
|
|
|
| def test_a_correlation_id_travels_with_the_event(journal): |
| with telemetry.span("localai.run", correlation_id="abc-123"): |
| pass |
| assert journal()[0]["correlation_id"] == "abc-123" |
|
|
|
|
| |
|
|
| def test_content_is_withheld_unless_it_was_asked_for(monkeypatch): |
| """The other load-bearing promise. Prompts and answers are client work, |
| and a telemetry file is the last place they should accumulate.""" |
| monkeypatch.setattr(telemetry, "_CONTENT", False) |
| assert telemetry.content("a customer's question") is None |
| monkeypatch.setattr(telemetry, "_CONTENT", True) |
| assert telemetry.content("a customer's question") == "a customer's question" |
|
|
|
|
| def test_status_reports_the_configuration_it_actually_has(journal): |
| state = telemetry.status() |
| assert state["enabled"] is True |
| assert state["file"] == str(journal.path) |
| assert state["content_capture"] in (True, False) |
| assert state["otel_bridge"] in (True, False) |
|
|
|
|
| def test_a_disabled_journal_does_not_name_its_file(monkeypatch): |
| """Nothing is being written there, and printing a path invites somebody to |
| go looking for events that were never recorded.""" |
| monkeypatch.setattr(telemetry, "_ENABLED", False) |
| assert telemetry.status()["file"] is None |
|
|
|
|
| |
|
|
| def test_the_lock_is_taken_and_released_on_a_real_descriptor(tmp_path): |
| """Whichever primitive this platform has. The append path depends on the |
| pair balancing: a lock taken and not released is a file no later writer |
| can append to.""" |
| path = tmp_path / "locked.bin" |
| path.write_bytes(b"x") |
| fd = os.open(str(path), os.O_RDWR | getattr(os, "O_BINARY", 0)) |
| try: |
| assert telemetry._lock(fd) is True |
| telemetry._unlock(fd) |
| assert telemetry._lock(fd) is True, "released, so it can be taken again" |
| telemetry._unlock(fd) |
| finally: |
| os.close(fd) |
|
|
|
|
| def test_a_platform_with_no_locking_primitive_still_appends(journal, monkeypatch): |
| """Better an unlocked append than no telemetry at all -- some network |
| shares support neither call.""" |
| monkeypatch.setattr(telemetry, "_MSVCRT", None) |
| monkeypatch.setattr(telemetry, "_FCNTL", None) |
| telemetry.emit("localai.test", {"rules": 1}) |
| assert journal()[0]["rules"] == 1 |
|
|
|
|
| def test_appends_accumulate_rather_than_overwrite(journal): |
| for n in range(5): |
| telemetry.emit("localai.test", {"n": n}) |
| assert [row["n"] for row in journal()] == [0, 1, 2, 3, 4] |
|
|
|
|
| |
|
|
| def test_an_unencodable_attribute_does_not_reach_the_caller(journal): |
| """json.dumps sat outside _append's try, so one attribute nothing could |
| encode raised straight out of emit -- through the docstring promising it |
| never raises, and into whatever was being measured.""" |
| telemetry.emit("localai.test", {"rules": {"a", "b"}}) |
| assert journal() == [], "an event that cannot be encoded is dropped" |
| telemetry.emit("localai.test", {"rules": ["a", "b"]}) |
| assert len(journal()) == 1, "and the next one still writes" |
|
|
|
|
| def test_an_unencodable_attribute_does_not_replace_the_real_error(journal): |
| """Worse than a lost event. Inside span() that TypeError REPLACED the |
| exception the caller was already handling, so the error you were shown was |
| about JSON and the error you actually had was somewhere else entirely.""" |
| with pytest.raises(ValueError, match="the real error"): |
| with telemetry.span("localai.test") as attrs: |
| attrs["unencodable"] = object() |
| raise ValueError("the real error") |
|
|
