# (c) cavaliba.com - tests / api / log import json from django.core.cache import cache as cache_django from django.test import TestCase from django.urls import reverse import app_home.cache as cache from app_data import crypto from app_data.data import Instance from app_home.log import log2 from app_home.models import CavalibaEvent def add_apikey_log(keyname, acl, readonly=False): secret = f"{keyname}_secret" instance = Instance(classname="_apikey", keyname=keyname) instance.merge_import({ "secret": crypto.hash_create(secret), "ip_filter": "*", "is_readonly": readonly, "is_enabled": True, "acl": acl, }) # fmt: skip instance.save(action="create") return {"HTTP_X_CAVALIBA_KEY": f"{keyname} {secret}"} def seed_events(): CavalibaEvent.objects.all().delete() log2.reset(channel="web", actor="alice", request_id="rid-seed") log2.access("seed.read", target=("server", "web01")) log2.ok("seed.update", target=("server", "web01")) log2.fail("seed.fail", reason="not_found", target=("site", "paris"), extra={"k": "v"}) class LogAPITestCase(TestCase): def setUp(self): cache.clear() cache_django.clear() self.view = add_apikey_log("logview", "p_log_view") self.admin = add_apikey_log("logadmin", "p_log_view\np_log_admin") self.readonly = add_apikey_log("logro", "p_log_view\np_log_admin", readonly=True) self.noperm = add_apikey_log("lognoperm", "") self.writer = add_apikey_log("logwriter", "p_log_write") self.writer_ro = add_apikey_log("logwriterro", "p_log_write", readonly=True) seed_events() def _json(self, response): if response.streaming: return json.loads(b"".join(response.streaming_content)) return json.loads(response.content) class APILogAuthTest(LogAPITestCase): def test_no_auth(self): for name in ("api:api_log", "api:api_log_count", "api:api_log_export"): self.assertEqual(self.client.get(reverse(name)).status_code, 401) self.assertEqual( self.client.post(reverse("api:api_log_purge") + "?mode=all").status_code, 401 ) def test_no_permission(self): response = self.client.get(reverse("api:api_log"), **self.noperm) self.assertEqual(response.status_code, 401) def test_view_key_cannot_purge(self): response = self.client.post(reverse("api:api_log_purge") + "?mode=all", **self.view) self.assertEqual(response.status_code, 401) self.assertTrue(CavalibaEvent.objects.filter(action="seed.fail").exists()) def test_readonly_key_cannot_purge(self): response = self.client.post(reverse("api:api_log_purge") + "?mode=all", **self.readonly) self.assertEqual(response.status_code, 401) self.assertTrue(CavalibaEvent.objects.filter(action="seed.fail").exists()) denied = CavalibaEvent.objects.get(action="log.purge", outcome="denied") self.assertEqual(denied.reason, "readonly_key") class APILogInputTest(LogAPITestCase): def test_wrong_methods(self): self.assertEqual(self.client.put(reverse("api:api_log"), **self.view).status_code, 405) self.assertEqual( self.client.post(reverse("api:api_log_count"), **self.view).status_code, 405 ) response = self.client.get(reverse("api:api_log_purge") + "?mode=all", **self.admin) self.assertEqual(response.status_code, 405) def test_invalid_filter_values_rejected(self): for query in ("level=loud", "outcome=maybe", "channel=x", "since=2026-13-45", "until=bad"): response = self.client.get(reverse("api:api_log") + "?" + query, **self.view) self.assertEqual(response.status_code, 400, query) self.assertIn("error", self._json(response)) def test_invalid_paging_rejected(self): for query in ("page=0", "size=0", "size=1001"): response = self.client.get(reverse("api:api_log") + "?" + query, **self.view) self.assertEqual(response.status_code, 400, query) def test_export_format_and_purge_mode_required(self): response = self.client.get(reverse("api:api_log_export") + "?format=xml", **self.view) self.assertEqual(response.status_code, 400) response = self.client.post(reverse("api:api_log_purge"), **self.admin) self.assertEqual(response.status_code, 400) def test_unknown_id_is_404(self): response = self.client.get(reverse("api:api_log_detail", args=[999999]), **self.view) self.assertEqual(response.status_code, 404) class APILogLogicTest(LogAPITestCase): def test_search_filters(self): url = reverse("api:api_log") + "?target=server:web01&size=50" actions = {e["action"] for e in self._json(self.client.get(url, **self.view))["events"]} self.assertEqual(actions, {"seed.read", "seed.update"}) def test_search_has_no_default_level(self): url = reverse("api:api_log") + "?request_id=rid-seed&size=50" events = self._json(self.client.get(url, **self.view))["events"] self.assertIn("seed.read", {e["action"] for e in events}) def test_search_pagination(self): url = reverse("api:api_log") + "?request_id=rid-seed&size=2" first = self._json(self.client.get(url, **self.view)) self.assertTrue(first["next"]) self.assertEqual([e["action"] for e in first["events"]], ["seed.fail", "seed.update"]) second = self._json(self.client.get(url + "&page=2", **self.view)) self.assertFalse(second["next"]) self.assertEqual([e["action"] for e in second["events"]], ["seed.read"]) def test_count(self): url = reverse("api:api_log_count") + "?request_id=rid-seed&level=INFO" reply = self._json(self.client.get(url, **self.view)) self.assertEqual(reply["count"], 2) self.assertEqual(reply["filters"], {"request_id": "rid-seed", "level": "INFO"}) def test_purge_retention_and_all(self): reply = self._json( self.client.post(reverse("api:api_log_purge") + "?mode=retention", **self.admin) ) self.assertEqual(reply, {"mode": "retention", "deleted": 0}) reply = self._json( self.client.post(reverse("api:api_log_purge") + "?mode=all", **self.admin) ) self.assertEqual(reply["mode"], "all") self.assertGreater(reply["deleted"], 0) self.assertFalse(CavalibaEvent.objects.filter(action__startswith="seed.").exists()) def test_reads_are_logged_with_the_key_as_actor(self): self.client.get(reverse("api:api_log"), **self.view) event = CavalibaEvent.objects.get(action="log.search") self.assertEqual( (event.actor, event.actor_kind, event.channel), ("API:logview", "apikey", "api") ) class APILogResponseTest(LogAPITestCase): def test_search_shape(self): reply = self._json( self.client.get(reverse("api:api_log") + "?request_id=rid-seed", **self.view) ) self.assertEqual(set(reply), {"page", "size", "next", "events"}) self.assertEqual((reply["page"], reply["size"]), (1, 10)) event = reply["events"][0] for key in ("id", "created", "level", "action", "outcome", "actor", "target_type", "extra"): self.assertIn(key, event) def test_detail_shape(self): seeded = CavalibaEvent.objects.get(action="seed.fail") reply = self._json( self.client.get(reverse("api:api_log_detail", args=[seeded.id]), **self.view) ) self.assertEqual(reply["id"], seeded.id) self.assertEqual((reply["reason"], reply["extra"]), ("not_found", {"k": "v"})) def test_yaml_output(self): response = self.client.get(reverse("api:api_log_count") + "?o=yaml", **self.view) self.assertIn("count:", response.content.decode()) def test_export_csv_and_json(self): url = reverse("api:api_log_export") + "?request_id=rid-seed" response = self.client.get(url, **self.view) self.assertEqual(response["Content-Type"], "text/csv; charset=utf-8") lines = b"".join(response.streaming_content).decode().strip().splitlines() self.assertTrue(lines[0].startswith("id,created,level")) self.assertEqual(len(lines), 4) rows = self._json(self.client.get(url + "&format=json", **self.view)) self.assertEqual(len(rows), 3) class APILogIngestTest(LogAPITestCase): def _post(self, body, key=None): data = body if isinstance(body, str) else json.dumps(body) return self.client.post(reverse("api:api_log"), data=data, content_type="application/json", **(key or self.writer)) # fmt: skip def test_requires_p_log_write(self): self.assertEqual(self._post({"action": "x.y"}, key=self.admin).status_code, 401) self.assertEqual(self._post({"action": "x.y"}, key=self.writer_ro).status_code, 401) self.assertFalse(CavalibaEvent.objects.filter(action="x.y").exists()) def test_single_event_with_forced_identity(self): response = self._post({"action": "backup.run", "outcome": "failure", "reason": "disk_full", "target_type": "server", "target_id": "web01", "count": 3, "duration_ms": 1200, "extra": {"job": "nightly"}}) # fmt: skip self.assertEqual(response.status_code, 200) reply = self._json(response) self.assertEqual(reply["count"], 1) event = CavalibaEvent.objects.get(id=reply["ids"][0]) self.assertEqual( (event.action, event.outcome, event.level), ("backup.run", "failure", "ERROR") ) self.assertEqual((event.target_type, event.target_id), ("server", "web01")) self.assertEqual( (event.count, event.duration_ms, event.extra), (3, 1200, {"job": "nightly"}) ) self.assertEqual((event.actor, event.actor_kind), ("API:logwriter", "apikey")) self.assertEqual((event.channel, event.origin), ("api", "/api/log/")) def test_batch_and_level_override(self): reply = self._json(self._post([ {"action": "a.one"}, {"action": "a.two", "outcome": "denied", "level": "error"}, ])) # fmt: skip self.assertEqual(reply["count"], 2) events = {e.action: e for e in CavalibaEvent.objects.filter(id__in=reply["ids"])} self.assertEqual((events["a.one"].outcome, events["a.one"].level), ("success", "INFO")) self.assertEqual((events["a.two"].outcome, events["a.two"].level), ("denied", "ERROR")) def test_identity_fields_refused(self): for field in ("actor", "actor_effective", "client_ip", "channel", "request_id", "created"): response = self._post({"action": "x.y", field: "forged"}) self.assertEqual(response.status_code, 400, field) self.assertIn(field, self._json(response)["error"]) def test_invalid_events_rejected(self): cases = [ "not json", [], {}, {"action": ""}, {"action": 5}, ["x"], {"action": "x.y", "outcome": "maybe"}, {"action": "x.y", "level": "loud"}, {"action": "x.y", "count": "3"}, {"action": "x.y", "count": True}, {"action": "x.y", "extra": "text"}, {"action": "x.y", "reason": 1}, ] # fmt: skip for body in cases: self.assertEqual(self._post(body).status_code, 400, body) def test_batch_is_all_or_nothing(self): before = CavalibaEvent.objects.filter(action="ok.first").count() response = self._post( [{"action": "ok.first"}, {"action": "bad.second", "outcome": "maybe"}] ) self.assertEqual(response.status_code, 400) self.assertIn("event #1", self._json(response)["error"]) self.assertEqual(CavalibaEvent.objects.filter(action="ok.first").count(), before)