# (c) cavaliba.com - tests / api / rawfile (was api import) import json import yaml from django.test import TestCase 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 tests import helper class ApiRawfileAuthTest(TestCase): def setUp(self): cache.clear() helper.add_schema( classname="rawfileauth", field_definition={"mystring": {"dataformat": "string", "displayname": "MyString"}}, ) def _post(self, body, headers=None, params=""): url = reverse("api:api_rawfile") + 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": "rawfileauth", "keyname": "a"}]') self.assertEqual(response.status_code, 401) def test_only_p_data_import_is_not_enough(self): # p_data_import gates the UI import tool, not the API - the API needs # its own p_api_rawfile grant headers = helper.add_apikey(keyname="rawfile_uionly", acl="p_data_import\np_data_create") response = self._post('[{"classname": "rawfileauth", "keyname": "a"}]', headers=headers) self.assertEqual(response.status_code, 401) def test_readonly_key_denied(self): headers = helper.add_apikey(keyname="rawfile_ro", acl="p_api_rawfile\np_data_admin") instance = Instance.from_keyname(classname="_apikey", keyname="rawfile_ro") instance.merge_import({"is_readonly": True}) instance.save(action="update") response = self._post('[{"classname": "rawfileauth", "keyname": "a"}]', headers=headers) self.assertEqual(response.status_code, 401) def test_get_not_allowed(self): headers = helper.add_apikey_admin() response = self.client.get(reverse("api:api_rawfile"), headers=headers) self.assertEqual(response.status_code, 405) class ApiRawfileInputTest(TestCase): def setUp(self): cache.clear() self.headers = helper.add_apikey_admin() def test_invalid_content_type(self): response = self.client.post( reverse("api:api_rawfile"), data="whatever", content_type="text/plain", headers=self.headers, ) self.assertEqual(response.status_code, 400) def test_invalid_action(self): response = self.client.post( reverse("api:api_rawfile") + "?action=obliterate", data="[]", content_type="application/json", headers=self.headers, ) self.assertEqual(response.status_code, 400) def test_invalid_encoding(self): response = self.client.post( reverse("api:api_rawfile") + "?encoding=shift-jis", data=b"classname,keyname\n", content_type="text/csv", headers=self.headers, ) self.assertEqual(response.status_code, 400) def test_invalid_pipeline(self): response = self.client.post( reverse("api:api_rawfile") + "?pipeline=doesnotexist", data="[]", content_type="application/json", headers=self.headers, ) self.assertEqual(response.status_code, 405) class ApiRawfileMultiTest(TestCase): """Regression coverage for the /api/rawfile/ (formerly /api/import/) N^2 bug: the view used to call load_broker(datalist=datalist, ...) *inside* the per-item loop (reprocessing the whole batch on every iteration instead of just the current item), which also made "discard" a no-op (a discarded item still got created via a sibling iteration's full-list reload) and crashed with UnboundLocalError whenever every item was discarded (or the batch was empty).""" def setUp(self): cache.clear() helper.add_schema( classname="apimultitest", field_definition={"mystring": {"dataformat": "string", "displayname": "MyString"}}, ) self.headers = helper.add_apikey_admin() def _post_rawfile(self, body, content_type="application/json", params=""): url = f"/api/rawfile/{params}" return self.client.post(url, data=body, content_type=content_type, headers=self.headers) def test_multi_item_import_creates_all_items_once(self): body = json.dumps( [ {"classname": "apimultitest", "keyname": "row1", "mystring": "a"}, {"classname": "apimultitest", "keyname": "row2", "mystring": "b"}, {"classname": "apimultitest", "keyname": "row3", "mystring": "c"}, ] ) response = self._post_rawfile(body) self.assertEqual(response.status_code, 200) data = response.json() self.assertEqual(data["count"], 3) self.assertEqual(data["count_ok"], 3) for keyname in ("row1", "row2", "row3"): self.assertIsNotNone( Instance.from_keyname(classname="apimultitest", keyname=keyname, expand=False) ) def test_empty_list_does_not_crash(self): response = self._post_rawfile("[]") self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["count"], 0) def test_pipeline_discard_all_does_not_crash(self): datalist = yaml.safe_load(""" - classname: _pipeline keyname: pipeline_discard_all displayname: pipeline_discard_all is_enabled: True content: | tasks: - discard: [''] """) load_broker(datalist=datalist, aaa={"perms": ["p_pipeline_create"]}) body = json.dumps( [ {"classname": "apimultitest", "keyname": "rowA", "mystring": "a"}, {"classname": "apimultitest", "keyname": "rowB", "mystring": "b"}, ] ) response = self._post_rawfile(body, params="?pipeline=pipeline_discard_all") self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["count"], 0) self.assertIsNone( Instance.from_keyname(classname="apimultitest", keyname="rowA", expand=False) ) self.assertIsNone( Instance.from_keyname(classname="apimultitest", keyname="rowB", expand=False) ) def test_pipeline_discard_one_item_excludes_only_that_item(self): """A discarded item must not be re-created via a sibling iteration's full-list reload - the core N^2 bug.""" datalist = yaml.safe_load(""" - classname: _pipeline keyname: pipeline_discard_one displayname: pipeline_discard_one is_enabled: True content: | tasks: - set_condition: [SKIP, field_match, keyname, '^skipme$'] - discard: [SKIP] """) load_broker(datalist=datalist, aaa={"perms": ["p_pipeline_create"]}) body = json.dumps( [ {"classname": "apimultitest", "keyname": "skipme", "mystring": "a"}, {"classname": "apimultitest", "keyname": "keepme", "mystring": "b"}, ] ) response = self._post_rawfile(body, params="?pipeline=pipeline_discard_one") self.assertEqual(response.status_code, 200) data = response.json() self.assertEqual(data["count"], 1) self.assertEqual(data["count_ok"], 1) self.assertIsNone( Instance.from_keyname(classname="apimultitest", keyname="skipme", expand=False) ) self.assertIsNotNone( Instance.from_keyname(classname="apimultitest", keyname="keepme", expand=False) ) class ApiRawfileOptionsTest(TestCase): """New in v4.4.0: ?action=, ?encoding=, ?dryrun= (rename from /api/import/).""" def setUp(self): cache.clear() helper.add_schema( classname="rawfileopt", field_definition={ "mystring": {"dataformat": "string", "displayname": "MyString"}, "tags": {"dataformat": "string", "displayname": "Tags", "cardinal_max": 0}, }, ) self.headers = helper.add_apikey_admin() def test_action_query_param_overrides_object_action(self): # target does not exist yet: action=update (no create) must fail body = json.dumps([{"classname": "rawfileopt", "keyname": "willnotexist"}]) response = self.client.post( "/api/rawfile/?action=update", data=body, content_type="application/json", headers=self.headers, ) data = response.json() self.assertEqual(data["count_ko"], 1) self.assertIsNone( Instance.from_keyname(classname="rawfileopt", keyname="willnotexist", expand=False) ) def test_action_query_param_create_ignores_object_action_delete(self): body = json.dumps( [ { "classname": "rawfileopt", "keyname": "forcedcreate", "_action": "delete", } ] ) response = self.client.post( "/api/rawfile/?action=create", data=body, content_type="application/json", headers=self.headers, ) data = response.json() self.assertEqual(data["count_ok"], 1) self.assertIsNotNone( Instance.from_keyname(classname="rawfileopt", keyname="forcedcreate", expand=False) ) def test_action_query_param_append_unions_values(self): load_broker( [{"classname": "rawfileopt", "keyname": "appendme", "tags": ["a"]}], aaa={"perms": ["p_data_admin"]}, ) body = json.dumps([{"classname": "rawfileopt", "keyname": "appendme", "tags": ["b"]}]) response = self.client.post( "/api/rawfile/?action=append", data=body, content_type="application/json", headers=self.headers, ) self.assertEqual(response.status_code, 200) data = response.json() self.assertEqual(data["count_ok"], 1) instance = Instance.from_keyname(classname="rawfileopt", keyname="appendme", expand=False) self.assertEqual(instance.fields["tags"].value, ["a", "b"]) def test_action_query_param_unappend_removes_values(self): load_broker( [{"classname": "rawfileopt", "keyname": "unappendme", "tags": ["a", "b"]}], aaa={"perms": ["p_data_admin"]}, ) body = json.dumps([{"classname": "rawfileopt", "keyname": "unappendme", "tags": ["a"]}]) response = self.client.post( "/api/rawfile/?action=unappend", data=body, content_type="application/json", headers=self.headers, ) self.assertEqual(response.status_code, 200) data = response.json() self.assertEqual(data["count_ok"], 1) instance = Instance.from_keyname(classname="rawfileopt", keyname="unappendme", expand=False) self.assertEqual(instance.fields["tags"].value, ["b"]) def test_encoding_cp1252_csv_decodes_correctly(self): # 0xE9 in cp1252 is 'é' csv_body = "classname,keyname,mystring\r\nrawfileopt,cp1252row,caf\xe9\r\n".encode("cp1252") response = self.client.post( "/api/rawfile/?encoding=cp1252", data=csv_body, content_type="text/csv", headers=self.headers, ) self.assertEqual(response.status_code, 200) data = response.json() self.assertEqual(data["count_ok"], 1) instance = Instance.from_keyname(classname="rawfileopt", keyname="cp1252row") self.assertEqual(instance.fields["mystring"].get_first_value(), "café") def test_default_encoding_is_utf8(self): csv_body = "classname,keyname,mystring\r\nrawfileopt,utf8row,café\r\n".encode() response = self.client.post( "/api/rawfile/", data=csv_body, content_type="text/csv", headers=self.headers, ) self.assertEqual(response.status_code, 200) data = response.json() self.assertEqual(data["count_ok"], 1) def test_delimiter_and_schema_query_params(self): csv_body = b"keyname;mystring\r\nrow1;hello\r\n" response = self.client.post( "/api/rawfile/?delimiter=;&schema=rawfileopt", data=csv_body, content_type="text/csv", headers=self.headers, ) self.assertEqual(response.status_code, 200) data = response.json() self.assertEqual(data["count_ok"], 1) self.assertIsNotNone( Instance.from_keyname(classname="rawfileopt", keyname="row1", expand=False) ) def test_csv_classname_column_works_without_schema_param(self): # a literal "classname" CSV column works on its own - csv.DictReader # populates it like any other column, no special-casing needed csv_body = b"classname,keyname\r\nrawfileopt,row2\r\n" response = self.client.post( "/api/rawfile/", data=csv_body, content_type="text/csv", headers=self.headers, ) data = response.json() self.assertEqual(data["count_ok"], 1) def test_csv_schema_column_is_not_auto_detected(self): # unlike load_file_csv() (used by the UI / "cavaliba load"), csv_body() # here does NOT rename a "schema" column to "classname" - only the # ?schema= query param (or pipeline.classname) injects it csv_body = b"schema,keyname\r\nrawfileopt,row3\r\n" response = self.client.post( "/api/rawfile/", data=csv_body, content_type="text/csv", headers=self.headers, ) data = response.json() self.assertEqual(data["count_ko"], 1) self.assertIn("missing classname", data["errors"][0]) def test_schema_param_overrides_existing_classname_column(self): csv_body = b"classname,keyname\r\nwrongschema,row4\r\n" response = self.client.post( "/api/rawfile/?schema=rawfileopt", data=csv_body, content_type="text/csv", headers=self.headers, ) data = response.json() self.assertEqual(data["count_ok"], 1) self.assertIsNotNone( Instance.from_keyname(classname="rawfileopt", keyname="row4", expand=False) ) def test_dryrun_true_does_not_persist_but_reports_ok(self): body = json.dumps([{"classname": "rawfileopt", "keyname": "dryrunrow"}]) response = self.client.post( "/api/rawfile/?dryrun=true", data=body, content_type="application/json", headers=self.headers, ) self.assertEqual(response.status_code, 200) data = response.json() self.assertTrue(data["dryrun"]) self.assertEqual(data["count_ok"], 1) self.assertIsNone( Instance.from_keyname(classname="rawfileopt", keyname="dryrunrow", expand=False) ) def test_dryrun_false_by_default(self): body = json.dumps([{"classname": "rawfileopt", "keyname": "notdryrun"}]) response = self.client.post( "/api/rawfile/", data=body, content_type="application/json", headers=self.headers, ) data = response.json() self.assertFalse(data["dryrun"]) self.assertIsNotNone( Instance.from_keyname(classname="rawfileopt", keyname="notdryrun", expand=False) ) def test_dryrun_true_pipeline_discard_still_discarded(self): load_broker( datalist=[ { "classname": "_pipeline", "_action": "create", "keyname": "rawfile_dryrun_discard", "displayname": "rawfile_dryrun_discard", "is_enabled": True, "content": "tasks:\n- discard: ['']\n", } ], aaa={"perms": ["p_pipeline_create"]}, ) body = json.dumps([{"classname": "rawfileopt", "keyname": "dryrundiscard"}]) response = self.client.post( "/api/rawfile/?dryrun=true&pipeline=rawfile_dryrun_discard", data=body, content_type="application/json", headers=self.headers, ) data = response.json() self.assertEqual(data["count"], 0) self.assertEqual(data["count_ok"], 0)