# (c) cavaliba.com - tests / api / load import json from django.test import TestCase, override_settings from django.urls import reverse import app_home.cache as cache from app_data.data import Instance from app_data.loader import load_broker from app_data.models import DataInstance from tests import helper class ApiLoadAuthTest(TestCase): def setUp(self): cache.clear() helper.add_schema( classname="loadauth", field_definition={"mystring": {"dataformat": "string", "displayname": "MyString"}}, ) def _post(self, body, headers=None, params=""): url = reverse("api:api_load") + params return self.client.post( url, data=body, content_type="application/json", headers=headers or {} ) def test_no_auth_header(self): response = self._post('{"classname": "loadauth", "keyname": "a"}') self.assertEqual(response.status_code, 401) def test_readonly_key_denied(self): headers = helper.add_apikey(keyname="load_ro", acl="p_api_load\np_data_admin") # flip is_readonly directly on the just-created apikey instance instance = Instance.from_keyname(classname="_apikey", keyname="load_ro") instance.merge_import({"is_readonly": True}) instance.save(action="update") response = self._post('{"classname": "loadauth", "keyname": "a"}', headers=headers) self.assertEqual(response.status_code, 401) def test_get_not_allowed(self): headers = helper.add_apikey_admin() url = reverse("api:api_load") response = self.client.get(url, headers=headers) self.assertEqual(response.status_code, 405) def test_missing_p_api_load_permission_denied(self): # has a real per-object create permission, but not the endpoint-level p_api_load headers = helper.add_apikey(keyname="load_nogate", acl="p_data_create") response = self._post('{"classname": "loadauth", "keyname": "a"}', headers=headers) self.assertEqual(response.status_code, 401) class ApiLoadInputTest(TestCase): def setUp(self): cache.clear() self.headers = helper.add_apikey_admin() def _post(self, body, params=""): url = reverse("api:api_load") + params return self.client.post( url, data=body, content_type="application/json", headers=self.headers ) def test_invalid_json_syntax(self): response = self._post("{not json") self.assertEqual(response.status_code, 400) def test_body_is_a_number(self): response = self._post("42") self.assertEqual(response.status_code, 400) def test_body_is_a_string(self): response = self._post('"just a string"') self.assertEqual(response.status_code, 400) def test_empty_body(self): response = self._post("") self.assertEqual(response.status_code, 400) def test_invalid_pipeline(self): response = self._post( '{"classname": "loadauth", "keyname": "a"}', params="?pipeline=doesnotexist" ) self.assertEqual(response.status_code, 400) def test_invalid_action(self): response = self._post( '{"classname": "loadauth", "keyname": "a"}', params="?action=obliterate" ) self.assertEqual(response.status_code, 400) class ApiLoadLogicTest(TestCase): def setUp(self): cache.clear() self.admin_headers = helper.add_apikey_admin() helper.add_schema( classname="loadtest", field_definition={ "mystring": {"dataformat": "string", "displayname": "MyString"}, "tags": {"dataformat": "string", "displayname": "Tags", "cardinal_max": 0}, }, ) def _post(self, body, headers=None, params=""): url = reverse("api:api_load") + params return self.client.post( url, data=body if isinstance(body, str) else json.dumps(body), content_type="application/json", headers=headers or self.admin_headers, ) def test_single_object_not_wrapped_in_list(self): body = {"classname": "loadtest", "keyname": "single01", "mystring": "hello"} response = self._post(body) self.assertEqual(response.status_code, 200) data = response.json() self.assertEqual(data["count"], 1) self.assertEqual(data["count_ok"], 1) self.assertIsNotNone( Instance.from_keyname(classname="loadtest", keyname="single01", expand=False) ) def test_batch_of_objects_all_created(self): body = [ {"classname": "loadtest", "keyname": "row1", "mystring": "a"}, {"classname": "loadtest", "keyname": "row2", "mystring": "b"}, {"classname": "loadtest", "keyname": "row3", "mystring": "c"}, ] response = self._post(body) self.assertEqual(response.status_code, 200) data = response.json() self.assertEqual(data["count"], 3) self.assertEqual(data["count_ok"], 3) self.assertEqual(data["count_ko"], 0) for keyname in ("row1", "row2", "row3"): self.assertIsNotNone( Instance.from_keyname(classname="loadtest", keyname=keyname, expand=False) ) def test_empty_list(self): response = self._post([]) self.assertEqual(response.status_code, 200) data = response.json() self.assertEqual(data["count"], 0) self.assertEqual(data["count_ok"], 0) def test_missing_classname_reported_as_ko_with_error(self): body = [{"keyname": "noschema"}] response = self._post(body) self.assertEqual(response.status_code, 200) data = response.json() self.assertEqual(data["count_ko"], 1) self.assertEqual(len(data["results"]), 1) self.assertEqual(data["results"][0]["status"], "ko") self.assertIn("error", data["results"][0]) def test_unknown_schema_reported_as_ko(self): body = [{"classname": "doesnotexist_schema", "keyname": "x"}] response = self._post(body) data = response.json() self.assertEqual(data["count_ko"], 1) self.assertEqual(data["results"][0]["schema"], "doesnotexist_schema") self.assertEqual(data["results"][0]["keyname"], "x") def test_non_dict_entry_reported_as_ko(self): body = ["not-an-object", {"classname": "loadtest", "keyname": "row9", "mystring": "z"}] response = self._post(body) data = response.json() self.assertEqual(data["count"], 2) self.assertEqual(data["count_ko"], 1) self.assertEqual(data["count_ok"], 1) def test_permission_denied_create_not_persisted(self): headers = helper.add_apikey(keyname="load_noperm", acl="p_api_load") body = [{"classname": "loadtest", "keyname": "denied01", "mystring": "x"}] response = self._post(body, headers=headers) data = response.json() self.assertEqual(data["count_ko"], 1) self.assertIsNone( Instance.from_keyname(classname="loadtest", keyname="denied01", expand=False) ) def test_action_query_param_overrides_object_action(self): # target does not exist yet: action=update (no create) must fail body = [{"classname": "loadtest", "keyname": "willnotexist", "mystring": "x"}] response = self._post(body, params="?action=update") data = response.json() self.assertEqual(data["count_ko"], 1) self.assertIsNone( Instance.from_keyname(classname="loadtest", keyname="willnotexist", expand=False) ) def test_action_query_param_create_ignores_object_action_delete(self): body = [ { "classname": "loadtest", "keyname": "forcedcreate", "mystring": "x", "_action": "delete", } ] response = self._post(body, params="?action=create") data = response.json() self.assertEqual(data["count_ok"], 1) self.assertIsNotNone( Instance.from_keyname(classname="loadtest", keyname="forcedcreate", expand=False) ) def test_pipeline_discard_not_counted_ok_or_ko(self): load_broker( datalist=[ { "classname": "_pipeline", "_action": "create", "keyname": "load_discard_all", "displayname": "load_discard_all", "is_enabled": True, "content": "tasks:\n- discard: ['']\n", } ], aaa={"perms": ["p_pipeline_create"]}, ) body = [{"classname": "loadtest", "keyname": "discardme", "mystring": "x"}] response = self._post(body, params="?pipeline=load_discard_all") data = response.json() self.assertEqual(data["count_ok"], 0) self.assertEqual(data["count_ko"], 0) self.assertEqual(data["count_discarded"], 1) self.assertIsNone( Instance.from_keyname(classname="loadtest", keyname="discardme", expand=False) ) def test_dryrun_true_does_not_persist_but_reports_ok(self): body = [{"classname": "loadtest", "keyname": "dryrun01", "mystring": "x"}] response = self._post(body, params="?dryrun=true") data = response.json() self.assertTrue(data["dryrun"]) self.assertEqual(data["count_ok"], 1) self.assertIsNone( Instance.from_keyname(classname="loadtest", keyname="dryrun01", expand=False) ) def test_dryrun_true_is_shallow_does_not_check_permission(self): # dryrun never calls load_broker(), so it can't catch permission # denial - it only reports pipeline discard/keep, not "would this # actually succeed". This is a known, documented limitation. headers = helper.add_apikey(keyname="load_dryrun_noperm", acl="p_api_load") body = [{"classname": "loadtest", "keyname": "dryrun02", "mystring": "x"}] response = self._post(body, headers=headers, params="?dryrun=true") data = response.json() self.assertEqual(data["count_ok"], 1) self.assertIsNone( Instance.from_keyname(classname="loadtest", keyname="dryrun02", expand=False) ) def test_verbose_false_omits_ok_entries(self): body = [{"classname": "loadtest", "keyname": "verbosefalse", "mystring": "x"}] response = self._post(body) data = response.json() self.assertEqual(data["results"], []) def test_verbose_true_includes_ok_entries(self): body = [{"classname": "loadtest", "keyname": "verbosetrue", "mystring": "x"}] response = self._post(body, params="?verbose=true") data = response.json() self.assertEqual(len(data["results"]), 1) self.assertEqual(data["results"][0]["status"], "ok") self.assertEqual(data["results"][0]["keyname"], "verbosetrue") @override_settings(CAVALIBA_LOAD_BATCH_MAX_ERROR=2) def test_abort_past_max_error(self): body = [{"classname": "doesnotexist_schema", "keyname": f"x{i}"} for i in range(1, 10)] response = self._post(body) data = response.json() self.assertTrue(data["aborted"]) processed = data["count_ok"] + data["count_ko"] + data["count_discarded"] self.assertLess(processed, data["count"]) def test_sync_param_sets_last_sync(self): body = [{"classname": "loadtest", "keyname": "syncedobj", "mystring": "x"}] self._post(body, params="?sync=true") iobj = DataInstance.objects.get(classname="loadtest", keyname="syncedobj") self.assertIsNotNone(iobj.last_sync) def test_no_sync_param_last_sync_null(self): body = [{"classname": "loadtest", "keyname": "unsyncedobj", "mystring": "x"}] self._post(body) iobj = DataInstance.objects.get(classname="loadtest", keyname="unsyncedobj") self.assertIsNone(iobj.last_sync) def test_action_query_param_append_unions_values(self): load_broker( [{"classname": "loadtest", "keyname": "appendme", "tags": ["a"]}], aaa={"perms": ["p_data_admin"]}, ) body = [{"classname": "loadtest", "keyname": "appendme", "tags": ["b"]}] response = self._post(body, params="?action=append") data = response.json() self.assertEqual(response.status_code, 200) self.assertEqual(data["count_ok"], 1) instance = Instance.from_keyname(classname="loadtest", keyname="appendme", expand=False) self.assertEqual(instance.fields["tags"].value, ["a", "b"]) def test_action_query_param_unappend_removes_values(self): load_broker( [{"classname": "loadtest", "keyname": "unappendme", "tags": ["a", "b"]}], aaa={"perms": ["p_data_admin"]}, ) body = [{"classname": "loadtest", "keyname": "unappendme", "tags": ["a"]}] response = self._post(body, params="?action=unappend") data = response.json() self.assertEqual(response.status_code, 200) self.assertEqual(data["count_ok"], 1) instance = Instance.from_keyname(classname="loadtest", keyname="unappendme", expand=False) self.assertEqual(instance.fields["tags"].value, ["b"]) def test_response_has_duration_and_counts(self): body = [{"classname": "loadtest", "keyname": "durationcheck", "mystring": "x"}] response = self._post(body) data = response.json() self.assertIn("duration", data) self.assertIsInstance(data["duration"], float) self.assertIn("count", data) self.assertIn("count_ok", data) self.assertIn("count_ko", data) self.assertIn("count_discarded", data)