# (c) cavaliba.com - tests / telemetry - counter buffer, request instrumentation, # revision / CRUD counts, object quantity gauges, per-id breakdown from datetime import timedelta from django.db import connection from django.test import TestCase, override_settings from django.test.utils import CaptureQueriesContext from django.urls import reverse import app_data.telemetry as T import app_home.cache as cache from app_data.data import Instance from app_data.models import DataTelemetryRaw from tests import helper AAA = {"username": "testuser", "perms": ["p_data_admin"]} def _raws(key): """Return (id, type, value) of the raw events of a key.""" rows = DataTelemetryRaw.objects.filter(key=key) return list(rows.values_list("distinct_id", "type", "value")) def _ids(key): """Return the set of distinct ids seen for a key.""" return set(DataTelemetryRaw.objects.filter(key=key).values_list("distinct_id", flat=True)) class TelemetryKpiTest(TestCase): def setUp(self): cache.clear() T.reset_cache() # ---- counter buffer ---------------------------------------------- def test_bump_then_flush_writes_one_row_per_key_and_id(self): T.telemetry_bump("x.c") T.telemetry_bump("x.c") T.telemetry_bump("x.c", id="a", n=5) self.assertFalse(DataTelemetryRaw.objects.exists()) self.assertEqual(T.telemetry_flush(), 2) self.assertEqual(sorted(_raws("x.c")), [("", "counter", 2), ("a", "counter", 5)]) def test_flush_is_idempotent(self): T.telemetry_bump("x.c") T.telemetry_flush() self.assertEqual(T.telemetry_flush(), 0) self.assertEqual(len(_raws("x.c")), 1) def test_flush_drops_invalid_key_or_id(self): T.telemetry_bump("Bad Key") T.telemetry_bump("x.c", id="bad id") self.assertEqual(T.telemetry_flush(), 0) self.assertFalse(DataTelemetryRaw.objects.exists()) def test_flush_stores_nothing_when_disabled(self): helper.set_telemetry_config("TELEMETRY_ENABLED", "no") T.telemetry_bump("x.c") self.assertEqual(T.telemetry_flush(), 0) self.assertFalse(DataTelemetryRaw.objects.exists()) def test_buffer_is_capped(self): for i in range(T.BUFFER_MAX_ENTRIES + 10): T.telemetry_bump("x.c", id=f"id{i}") self.assertEqual(T.telemetry_flush(), T.BUFFER_MAX_ENTRIES) # ---- key cache --------------------------------------------------- def _distinct_queries(self, call): """Return the SELECT DISTINCT key queries one call makes.""" with CaptureQueriesContext(connection) as ctx: call() return [q["sql"] for q in ctx.captured_queries if "DISTINCT" in q["sql"]] def test_known_key_reads_no_key_cache(self): T.telemetry("auth.login") # first event fills the key cache self.assertEqual(self._distinct_queries(lambda: T.telemetry("auth.login")), []) def test_unknown_key_reads_the_key_cache_when_stale(self): T.telemetry("auth.login") T._known_keys["ts"] = 0.0 # cache never read yet self.assertTrue(self._distinct_queries(lambda: T.telemetry("new.key"))) self.assertTrue(DataTelemetryRaw.objects.filter(key="new.key").exists()) def test_new_key_recorded_without_reading_the_key_cache(self): # a fresh cache must not make an ad-hoc key (timed., API ingest) wait T.telemetry("auth.login") self.assertEqual(self._distinct_queries(lambda: T.telemetry("other.key")), []) self.assertTrue(DataTelemetryRaw.objects.filter(key="other.key").exists()) def test_sample_does_nothing_when_disabled(self): helper.set_telemetry_config("TELEMETRY_ENABLED", "no") helper.add_schema(classname="kpitest", field_definition={}) with CaptureQueriesContext(connection) as ctx: self.assertEqual(T.telemetry_sample(), 0) self.assertFalse([q for q in ctx.captured_queries if "data_instance" in q["sql"]]) # ---- HTTP request instrumentation -------------------------------- @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_web_request_records_view_duration_cache_and_log(self): helper.add_admin_user(login="unittest") self.client.get(reverse("telemetry:index")) self.assertIn("data.telemetry_index", _ids("view.request")) duration = _raws("http.duration") self.assertEqual(len(duration), 1) self.assertEqual(duration[0][0], "data.telemetry_index") self.assertEqual(duration[0][1], "timer") self.assertGreater(duration[0][2], 0) self.assertTrue(_ids("log.event")) self.assertTrue(_ids("cache.get") & {"hit_local", "hit_shared", "miss"}) self.assertTrue(_ids("cache.session") & {"hit", "miss", "nocache"}) # ---- revisions and CRUD per schema ------------------------------- def test_instance_write_counts_per_schema_and_action(self): helper.add_schema( classname="kpitest", field_definition={"title": {"dataformat": "string", "displayname": "Title"}}, ) instance = Instance(classname="kpitest") instance.merge_import({"classname": "kpitest", "keyname": "obj01"}) instance.save(aaa=AAA) instance.merge_import({"classname": "kpitest", "keyname": "obj01", "displayname": "New"}) instance.save(aaa=AAA) instance.delete(aaa=AAA) writes = [row for row in _raws("data.write") if row[0] == "kpitest"] self.assertEqual(len(writes), 3) actions = DataTelemetryRaw.objects.filter(key="data.write.action") self.assertEqual( sorted(actions.values_list("svalue", flat=True)), ["create", "delete", "update"] ) # ---- object quantities ------------------------------------------- def test_sample_records_object_gauges(self): helper.add_schema(classname="kpitest", field_definition={}) helper.add_instance(classname="kpitest", keyname="obj01") helper.add_instance(classname="kpitest", keyname="obj02") T.telemetry_sample() objects = {row[0]: row[2] for row in _raws("data.objects")} self.assertEqual(objects["kpitest"], 2) self.assertEqual({row[1] for row in _raws("data.objects")}, {"gauge"}) total = _raws("data.objects.total") self.assertEqual(len(total), 1) self.assertGreaterEqual(total[0][2], 2) self.assertEqual(set(_ids("data.rows")), set(T.SAMPLED_TABLES)) # ---- widgets and dashboards -------------------------------------- def test_every_widget_spec_resolves(self): for name, spec in T.WIDGETS.items(): with self.subTest(widget=name): data = T.widget_data(dict(spec)) self.assertEqual(data["w"], spec.get("w", "value")) def test_every_dashboard_widget_exists(self): for name, board in T.DASHBOARDS.items(): for entry in board["widgets"]: with self.subTest(dashboard=name, widget=entry): self.assertIn(entry, T.WIDGETS) # ---- per-id breakdown -------------------------------------------- def test_id_breakdown_sums_per_id_sorted(self): now = helper.TELEMETRY_T0 for id, count in (("a", 1), ("b", 3)): for _ in range(count): helper.add_telemetry_raw(now, "x.c", id=id) T.telemetry_aggregate(now=now + timedelta(hours=2)) rows = T.get_id_breakdown("x.c", "sum", "hour", now, now + timedelta(hours=1)) self.assertEqual(rows, [{"id": "b", "value": 3}, {"id": "a", "value": 1}]) def test_id_breakdown_refused_for_breakdown_function(self): helper.add_telemetry_raw(helper.TELEMETRY_T0, "x.s", "set", svalue="ok") T.telemetry_aggregate(now=helper.TELEMETRY_T0 + timedelta(hours=2)) with self.assertRaises(T.TelemetryError): T.get_id_breakdown("x.s", "breakdown", "hour") def test_idtable_widget_rows(self): now = helper.TELEMETRY_T0 helper.add_telemetry_raw(now, "x.c", id="a") helper.add_telemetry_raw(now, "x.c", id="a") helper.add_telemetry_raw(now, "x.c", id="b") T.telemetry_aggregate(now=now + timedelta(hours=2)) data = T.widget_data({ "w": "idtable", "key": "x.c", "period": "hour", "start": now, "end": now, }) # fmt: skip self.assertEqual(data["w"], "idtable") self.assertEqual([row["label"] for row in data["table_rows"]], ["a", "b"]) self.assertEqual([row["display"] for row in data["table_rows"]], ["2", "1"])