# (c) cavaliba.com - tests / api / telemetry import json from datetime import timedelta from django.test import TestCase from django.urls import reverse from django.utils import timezone import app_data.telemetry as T import app_home.cache as cache from app_data.api.telemetry import INGEST_MAX_EVENTS from app_data.data import Instance from app_data.models import DataTelemetryRaw from tests import helper def _readonly_key(keyname, acl): headers = helper.add_apikey(keyname=keyname, acl=acl) instance = Instance.from_keyname(classname="_apikey", keyname=keyname) instance.merge_import({"is_readonly": True}) instance.save(action="update") return headers class ApiTelemetryAuthTest(TestCase): def setUp(self): cache.clear() T.reset_cache() def test_no_auth_header(self): self.assertEqual(self.client.get(reverse("api:api_telemetry")).status_code, 401) def test_missing_read_permission(self): headers = helper.add_apikey(keyname="tlm_w", acl="p_telemetry_write") self.assertEqual( self.client.get(reverse("api:api_telemetry"), headers=headers).status_code, 401 ) def test_post_needs_write_permission(self): headers = helper.add_apikey(keyname="tlm_r", acl="p_telemetry_read") response = self.client.post( reverse("api:api_telemetry"), data='{"key": "x.c"}', content_type="application/json", headers=headers, ) self.assertEqual(response.status_code, 401) def test_post_readonly_key(self): headers = _readonly_key("tlm_ro", "p_telemetry_write") response = self.client.post( reverse("api:api_telemetry"), data='{"key": "x.c"}', content_type="application/json", headers=headers, ) self.assertEqual(response.status_code, 401) def test_wrong_method(self): headers = helper.add_apikey(keyname="tlm_rw", acl="p_telemetry_read\np_telemetry_write") self.assertEqual( self.client.put(reverse("api:api_telemetry"), headers=headers).status_code, 405 ) url = reverse("api:api_telemetry_detail", args=["auth.login"]) self.assertEqual(self.client.post(url, headers=headers).status_code, 405) class ApiTelemetryIngestTest(TestCase): def setUp(self): cache.clear() T.reset_cache() self.headers = helper.add_apikey(keyname="tlm_w", acl="p_telemetry_write") def _post(self, body): data = body if isinstance(body, str) else json.dumps(body) return self.client.post( reverse("api:api_telemetry"), data=data, content_type="application/json", headers=self.headers, ) def test_post_one_event(self): response = self._post({"key": "x.t", "type": "timer", "id": "a", "value": 12.5}) self.assertEqual(response.status_code, 201) self.assertEqual(response.json(), {"accepted": 1, "rejected": []}) raw = DataTelemetryRaw.objects.get(key="x.t") self.assertEqual( (raw.key, raw.type, raw.distinct_id, raw.value), ("x.t", "timer", "a", 12.5) ) def test_post_list_with_rejected_events(self): response = self._post( [{"key": "auth.login"}, {"key": "Bad Key"}, {"key": "auth.login", "type": "timer"}] ) self.assertEqual(response.status_code, 201) self.assertEqual(response.json(), {"accepted": 1, "rejected": [1, 2]}) def test_post_invalid_body(self): for body in [ "notjson", [], [{"key": "x.c"}] * (INGEST_MAX_EVENTS + 1), [{"key": "x.c", "extra": 1}], [{"value": 1}], [{"key": "x.c", "value": True}], [{"key": "x.c", "id": 3}], ]: self.assertEqual(self._post(body).status_code, 400, body) # none of the rejected events was stored (other keys come from the instrumentation # of the request itself: api.request, http.duration, log.event, cache.get) self.assertFalse(DataTelemetryRaw.objects.filter(key="x.c").exists()) class ApiTelemetryReadTest(TestCase): def setUp(self): cache.clear() T.reset_cache() self.headers = helper.add_apikey(keyname="tlm_r", acl="p_telemetry_read") now = timezone.now() helper.add_telemetry_raw(now - timedelta(minutes=10), "auth.login", id="oidc") T.telemetry_aggregate(now=now) def _get(self, url, query=""): return self.client.get(url + query, headers=self.headers) def test_list(self): response = self._get(reverse("api:api_telemetry")) self.assertEqual(response.status_code, 200) self.assertIn("auth.login", [k["key"] for k in response.json()]) def test_detail(self): response = self._get(reverse("api:api_telemetry_detail", args=["auth.login"])) self.assertEqual(response.status_code, 200) body = response.json() self.assertEqual((body["type"], body["ids"]), ("counter", ["oidc"])) self.assertIsInstance(body["ranges"]["minute"]["first"], str) def test_detail_unknown_key(self): self.assertEqual( self._get(reverse("api:api_telemetry_detail", args=["no.key"])).status_code, 404 ) def test_value(self): url = reverse("api:api_telemetry_data", args=["auth.login", "minute", "sum"]) body = self._get(url, "?at=now-10m").json() self.assertEqual((body["value"], body["function"], body["id"]), (1, "sum", "")) self.assertIsInstance(body["start"], str) def test_series(self): url = reverse("api:api_telemetry_data", args=["auth.login", "minute", "sum"]) body = self._get(url, "?start=now-12m&end=now-8m&id=oidc").json() self.assertEqual(len(body["values"]), 5) self.assertEqual(sum(p["value"] for p in body["values"]), 1) def test_invalid_queries(self): for args, query, status in [ (["auth.login", "minute", "sum"], "?at=now&start=now-1h", 400), (["auth.login", "minute", "p95"], "", 400), (["auth.login", "decade", "sum"], "", 400), (["auth.login", "minute", "sum"], "?at=yesterday", 400), (["auth.login", "minute", "sum"], "?start=now-1w", 400), (["no.key", "minute", "sum"], "", 404), ]: response = self._get(reverse("api:api_telemetry_data", args=args), query) self.assertEqual(response.status_code, status, (args, query))