# (c) cavaliba.com - tests data export import hashlib import json import os from django.conf import settings 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.dataview import DataView from app_data.revision import revision_add_raw from tests import helper class DataExportTest(TestCase): def setUp(self): cache.clear() helper.add_admin_user(login="unittest") helper.add_schema( classname="test1", field_definition={ "mystring": {"dataformat": "string", "displayname": "MyString"}, }, ) helper.add_instance( classname="test1", keyname="test1-01", fields={ "mystring": "data for test1-01", }, ) helper.add_instance( classname="_dataview", keyname="test1_enumerate_inject", fields={ "target_class": "test1", "content": "columns:\n - keyname\n - Widget:\n from: my_enumerate_subfields__widget\n", }, ) helper.add_instance( classname="_dataview", keyname="test1_welcome", fields={ "target_class": "test1", "content": "columns:\n - keyname\n - invalid:\n from: non_existent_column\n", }, ) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_data_export_csv(self): response = self.client.post( reverse("app_data:data_export"), { "classname": "test1", "page": "allpages", "dv": "___ALL___", "query": "", "format": "csv", }, ) self.assertEqual(response.status_code, 200) self.assertContains(response, "test1-01") @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_data_export_yaml(self): response = self.client.post( reverse("app_data:data_export"), { "classname": "test1", "page": "allpages", "dv": "___ALL___", "query": "", "format": "yaml", }, ) self.assertEqual(response.status_code, 200) self.assertContains(response, "test1-01") @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_data_export_json(self): response = self.client.post( reverse("app_data:data_export"), { "classname": "test1", "page": "allpages", "dv": "___ALL___", "query": "", "format": "json", }, ) self.assertEqual(response.status_code, 200) self.assertContains(response, "test1-01") @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_data_export_yaml_dataview1(self): response = self.client.post( reverse("app_data:data_export"), { "classname": "test1", "page": "allpages", "dv": "test1_enumerate_inject", "query": "", "format": "yaml", }, ) self.assertEqual(response.status_code, 200) self.assertContains(response, "test1-01") # header/key is the real source field name, not the view's "Widget" label, # so the export stays importable self.assertContains(response, "my_enumerate_subfields__widget") self.assertNotContains(response, "Widget") @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_data_export_yaml_dataview2(self): response = self.client.post( reverse("app_data:data_export"), { "classname": "test1", "page": "allpages", "dv": "test1_welcome", "query": "", "format": "yaml", }, ) self.assertEqual(response.status_code, 200) self.assertContains(response, "test1-01") # header/key is the real source field name, not the view's "invalid" label self.assertContains(response, "non_existent_column") self.assertNotContains(response, "invalid") @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_data_export_json_dataview(self): response = self.client.post( reverse("app_data:data_export"), { "classname": "test1", "page": "allpages", "dv": "test1_welcome", "query": "", "format": "json", }, ) self.assertEqual(response.status_code, 200) self.assertContains(response, "test1-01") @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_export_viewer_get(self): response = self.client.get(reverse("app_data:data_export")) self.assertEqual(response.status_code, 200) self.assertContains(response, "Export Viewer") @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_export_viewer_file_list_and_view(self): export_folder = str(settings.CAVALIBA_EXPORT_FOLDER) os.makedirs(export_folder, exist_ok=True) test_file = os.path.join(export_folder, "unittest.txt") test_content = "This is a test export file\n" with open(test_file, "w") as f: f.write(test_content) try: response = self.client.get(reverse("app_data:data_export")) self.assertEqual(response.status_code, 200) self.assertContains(response, "unittest.txt") filename_md5 = hashlib.md5(b"unittest.txt").hexdigest() download_response = self.client.get( reverse("app_data:data_export") + f"?dl={filename_md5}" ) self.assertEqual(download_response.status_code, 200) self.assertIn("Content-Disposition", download_response) self.assertIn("attachment", download_response["Content-Disposition"]) view_response = self.client.get( reverse("app_data:data_export") + f"?view={filename_md5}" ) self.assertEqual(view_response.status_code, 200) self.assertContains(view_response, test_content.strip()) self.assertContains(view_response, "unittest.txt") finally: if os.path.exists(test_file): os.remove(test_file) class DataExportRefsRevTest(TestCase): def setUp(self): cache.clear() helper.add_admin_user(login="unittest") helper.add_user(login="testuser01") helper.add_schema( classname="test0", field_definition={ "mystring": {"dataformat": "string", "displayname": "MyString"}, }, ) helper.add_instance( classname="test0", keyname="test0-01", fields={ "mystring": "data for test0-01", }, ) helper.add_schema( classname="test1", field_definition={ "mystring": {"dataformat": "string", "displayname": "MyString"}, "my_user": {"dataformat": "user", "displayname": "My User"}, "my_schema_test0": { "dataformat": "schema", "dataformat_ext": "test0", "displayname": "Schema Test0", }, }, ) helper.add_instance( classname="test1", keyname="test1-01", fields={ "mystring": "data for test1-01", "my_user": "testuser01", "my_schema_test0": "test0-01", }, ) def _post(self, extra=None): data = { "classname": "test1", "page": "allpages", "dv": "___ALL___", "query": "", "format": "yaml", } if extra: data.update(extra) return self.client.post(reverse("app_data:data_export"), data) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_export_yaml_refs_user(self): response = self._post({"refs": "user"}) self.assertEqual(response.status_code, 200) content = response.content.decode() self.assertIn("_refs", content) self.assertIn("testuser01", content) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_export_json_refs_user(self): response = self._post({"format": "json", "refs": "user"}) self.assertEqual(response.status_code, 200) data = json.loads(response.content) refs_found = any("_refs" in item for item in data) self.assertTrue(refs_found) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_export_yaml_refs_schema(self): response = self._post({"refs": "test0"}) self.assertEqual(response.status_code, 200) content = response.content.decode() self.assertIn("_refs", content) self.assertIn("test0", content) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_export_yaml_refs_empty_no_refs_key(self): response = self._post({"refs": ""}) self.assertEqual(response.status_code, 200) content = response.content.decode() self.assertNotIn("_refs", content) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_export_yaml_refs_unknown_no_refs_key(self): response = self._post({"refs": "nonexistent_schema"}) self.assertEqual(response.status_code, 200) content = response.content.decode() self.assertNotIn("_refs", content) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_export_csv_refs_ignored(self): response = self._post({"format": "csv", "refs": "user"}) self.assertEqual(response.status_code, 200) self.assertContains(response, "test1-01") @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_export_yaml_rev_with_revisions(self): revision_add_raw(aaa=None, classname="test1", keyname="test1-01", action="edit") response = self._post({"rev": "3"}) self.assertEqual(response.status_code, 200) content = response.content.decode() self.assertIn("_revision", content) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_export_json_rev_with_revisions(self): revision_add_raw(aaa=None, classname="test1", keyname="test1-01", action="edit") response = self._post({"format": "json", "rev": "1"}) self.assertEqual(response.status_code, 200) data = json.loads(response.content) rev_found = any("_revision" in item for item in data) self.assertTrue(rev_found) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_export_yaml_rev_no_revisions(self): response = self._post({"rev": "5"}) self.assertEqual(response.status_code, 200) content = response.content.decode() self.assertNotIn("_revision", content) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_export_yaml_rev_invalid_value(self): response = self._post({"rev": "notanumber"}) self.assertEqual(response.status_code, 200) self.assertContains(response, "test1-01") @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_export_csv_rev_ignored(self): revision_add_raw(aaa=None, classname="test1", keyname="test1-01", action="edit") response = self._post({"format": "csv", "rev": "1"}) self.assertEqual(response.status_code, 200) self.assertContains(response, "test1-01") class TestGetDictForExportRevision(TestCase): def setUp(self): cache.clear() helper.add_schema(classname="revtest") helper.add_instance(classname="revtest", keyname="obj01") for i in range(15): revision_add_raw(classname="revtest", keyname="obj01", action="edit") def test_rev_none_no_revision_key(self): instance = Instance.from_keyname("revtest", "obj01") data = instance.get_dict_for_export(rev=None) self.assertNotIn("_revision", data) def test_rev_zero_no_revision_key(self): instance = Instance.from_keyname("revtest", "obj01") data = instance.get_dict_for_export(rev=0) self.assertNotIn("_revision", data) def test_rev_positive_returns_revision_key(self): instance = Instance.from_keyname("revtest", "obj01") data = instance.get_dict_for_export(rev=5) self.assertIn("_revision", data) def test_rev_limits_count(self): instance = Instance.from_keyname("revtest", "obj01") data = instance.get_dict_for_export(rev=5) self.assertEqual(len(data["_revision"]), 5) def test_rev_capped_at_available(self): instance = Instance.from_keyname("revtest", "obj01") data = instance.get_dict_for_export(rev=100) self.assertEqual(len(data["_revision"]), 15) def test_rev_entry_fields(self): instance = Instance.from_keyname("revtest", "obj01") data = instance.get_dict_for_export(rev=1) entry = data["_revision"][0] self.assertIn("login", entry) self.assertIn("date", entry) self.assertIn("action", entry) def test_rev_no_revisions_in_db(self): helper.add_instance(classname="revtest", keyname="obj02") instance = Instance.from_keyname("revtest", "obj02") data = instance.get_dict_for_export(rev=10) self.assertNotIn("_revision", data) class TestGetDictForExportRefs(TestCase): def setUp(self): cache.clear() # schema with a user field and a schema field helper.add_schema( classname="refstest", field_definition={ "owner": { "dataformat": "user", "displayname": "Owner", "cardinal_min": 0, "cardinal_max": 1, }, "linked": { "dataformat": "schema", "dataformat_ext": "refsother", "displayname": "Linked", "cardinal_min": 0, "cardinal_max": 1, }, }, ) helper.add_schema(classname="refsother") helper.add_instance( classname="refsother", keyname="other01", fields={"displayname": "Other 01"} ) helper.add_instance( classname="refstest", keyname="obj01", fields={ "owner": "testuser01", "linked": "other01", }, ) # user referenced by obj01, with extra fields for refs expansion user = helper.add_user(login="testuser01", displayname="Test User 01") user.set_field_value_single("email", "testuser01@example.com") user.set_field_value_single("mobile", "+33600000001") user.set_field_value_single("external_id", "EXT001") user.save(aaa={"username": "unittest"}, action="update") def _export(self, refs=None): instance = Instance.from_keyname("refstest", "obj01") return instance.get_dict_for_export(refs=refs) def test_refs_none_no_refs_key(self): data = self._export(refs=None) self.assertNotIn("_refs", data) def test_refs_empty_list_no_refs_key(self): data = self._export(refs=[]) self.assertNotIn("_refs", data) def test_refs_unknown_type_no_refs_key(self): data = self._export(refs=["nonexistent"]) self.assertNotIn("_refs", data) def test_refs_user_present(self): data = self._export(refs=["user"]) self.assertIn("_refs", data) self.assertIn("user", data["_refs"]) def test_refs_user_contains_expected_login(self): data = self._export(refs=["user"]) logins = [u["keyname"] for u in data["_refs"]["user"]] self.assertIn("testuser01", logins) def test_refs_user_entry_fields(self): data = self._export(refs=["user"]) entry = data["_refs"]["user"][0] self.assertIn("keyname", entry) self.assertIn("displayname", entry) def test_refs_user_extra_fields(self): data = self._export(refs=["user"]) entry = next(u for u in data["_refs"]["user"] if u["keyname"] == "testuser01") self.assertEqual(entry.get("email"), "testuser01@example.com") self.assertEqual(entry.get("mobile"), "+33600000001") self.assertEqual(entry.get("external_id"), "EXT001") def test_refs_schema_present(self): data = self._export(refs=["refsother"]) self.assertIn("_refs", data) self.assertIn("refsother", data["_refs"]) def test_refs_schema_contains_expected_keyname(self): data = self._export(refs=["refsother"]) keynames = [i["keyname"] for i in data["_refs"]["refsother"]] self.assertIn("other01", keynames) def test_refs_schema_entry_fields(self): data = self._export(refs=["refsother"]) entry = data["_refs"]["refsother"][0] self.assertIn("keyname", entry) self.assertIn("classname", entry) self.assertIn("displayname", entry) def test_refs_multiple_types(self): data = self._export(refs=["user", "refsother"]) self.assertIn("_refs", data) self.assertIn("user", data["_refs"]) self.assertIn("refsother", data["_refs"]) def test_refs_star_expands_to_schemas(self): data = self._export(refs=["*"]) self.assertIn("_refs", data) self.assertIn("refsother", data["_refs"]) def test_refs_star_with_user(self): data = self._export(refs=["*", "user"]) self.assertIn("_refs", data) self.assertIn("user", data["_refs"]) self.assertIn("refsother", data["_refs"]) class TestGetDictForExportFields(TestCase): def setUp(self): cache.clear() helper.add_schema( classname="fieldstest", field_definition={ "alpha": {"dataformat": "string", "displayname": "Alpha"}, "beta": {"dataformat": "string", "displayname": "Beta"}, "charlie1": {"dataformat": "string", "displayname": "Charlie1"}, "charlie2": {"dataformat": "string", "displayname": "Charlie2"}, }, ) helper.add_instance( classname="fieldstest", keyname="obj01", fields={ "alpha": "A", "beta": "B", "charlie1": "C1", "charlie2": "C2", }, ) def _export(self, fields=None): instance = Instance.from_keyname("fieldstest", "obj01") return instance.get_dict_for_export(fields=fields) def test_fields_none_all_present(self): data = self._export(fields=None) for f in ("alpha", "beta", "charlie1", "charlie2"): self.assertIn(f, data) def test_fields_star_all_present(self): data = self._export(fields="*") for f in ("alpha", "beta", "charlie1", "charlie2"): self.assertIn(f, data) def test_fields_empty_string_all_present(self): data = self._export(fields="") for f in ("alpha", "beta", "charlie1", "charlie2"): self.assertIn(f, data) def test_fields_whitelist_exact(self): data = self._export(fields="alpha,beta") self.assertIn("alpha", data) self.assertIn("beta", data) self.assertNotIn("charlie1", data) self.assertNotIn("charlie2", data) def test_fields_whitelist_wildcard(self): data = self._export(fields="charlie*") self.assertNotIn("alpha", data) self.assertNotIn("beta", data) self.assertIn("charlie1", data) self.assertIn("charlie2", data) def test_fields_blacklist_exact(self): data = self._export(fields="!alpha,beta") self.assertNotIn("alpha", data) self.assertNotIn("beta", data) self.assertIn("charlie1", data) self.assertIn("charlie2", data) def test_fields_blacklist_wildcard(self): data = self._export(fields="!charlie*") self.assertIn("alpha", data) self.assertIn("beta", data) self.assertNotIn("charlie1", data) self.assertNotIn("charlie2", data) def test_fields_exclude_all_keeps_builtin_attributes(self): data = self._export(fields="!*") self.assertNotIn("alpha", data) self.assertNotIn("beta", data) self.assertNotIn("charlie1", data) self.assertNotIn("charlie2", data) self.assertEqual(data["keyname"], "obj01") self.assertEqual(data["classname"], "fieldstest") self.assertNotIn("id", data) def test_fields_whitespace_tolerant(self): data = self._export(fields=" alpha , beta ") self.assertIn("alpha", data) self.assertIn("beta", data) self.assertNotIn("charlie1", data) class TestGetCsvColumnsFields(TestCase): def setUp(self): cache.clear() helper.add_schema( classname="csvfieldstest", field_definition={ "alpha": {"dataformat": "string", "displayname": "Alpha"}, "charlie1": {"dataformat": "string", "displayname": "Charlie1"}, }, ) helper.add_instance( classname="csvfieldstest", keyname="obj01", fields={"alpha": "A", "charlie1": "C1"}, ) def test_fields_none_all_columns(self): instance = Instance.from_keyname("csvfieldstest", "obj01") columns = instance.get_csv_columns(fields=None) self.assertIn("alpha", columns) self.assertIn("charlie1", columns) def test_fields_whitelist(self): instance = Instance.from_keyname("csvfieldstest", "obj01") columns = instance.get_csv_columns(fields="alpha") self.assertIn("alpha", columns) self.assertNotIn("charlie1", columns) def test_fields_blacklist_keeps_only_keyname(self): # keyname is the only fixed column - schema/is_enabled/displayname are # all opt-out via the fields filter, same as regular fields instance = Instance.from_keyname("csvfieldstest", "obj01") columns = instance.get_csv_columns(fields="!*") self.assertNotIn("alpha", columns) self.assertNotIn("charlie1", columns) self.assertNotIn("classname", columns) self.assertNotIn("schema", columns) self.assertIn("keyname", columns) self.assertNotIn("is_enabled", columns) self.assertNotIn("displayname", columns) def test_fields_none_includes_displayname(self): # no filter provided means "*" (all fields, including displayname) instance = Instance.from_keyname("csvfieldstest", "obj01") columns = instance.get_csv_columns(fields=None) self.assertIn("displayname", columns) self.assertIn("schema", columns) self.assertNotIn("classname", columns) self.assertIn("keyname", columns) self.assertIn("is_enabled", columns) def test_fields_star_includes_displayname(self): instance = Instance.from_keyname("csvfieldstest", "obj01") columns = instance.get_csv_columns(fields="*") self.assertIn("displayname", columns) def test_fields_explicit_whitelist_includes_displayname(self): instance = Instance.from_keyname("csvfieldstest", "obj01") columns = instance.get_csv_columns(fields="displayname") self.assertIn("displayname", columns) self.assertNotIn("alpha", columns) class DataExportFieldsViewTest(TestCase): def setUp(self): cache.clear() helper.add_admin_user(login="unittest") helper.add_schema( classname="test2", field_definition={ "keep_me": {"dataformat": "string", "displayname": "Keep Me"}, "drop_me": {"dataformat": "string", "displayname": "Drop Me"}, }, ) helper.add_instance( classname="test2", keyname="test2-01", fields={"keep_me": "keepvalue", "drop_me": "dropvalue"}, ) helper.add_instance( classname="_dataview", keyname="test2_view", fields={ "target_class": "test2", "content": "columns:\n - keyname\n - drop_me\n", }, ) def _post(self, extra=None): data = { "classname": "test2", "page": "allpages", "dv": "___ALL___", "query": "", "format": "yaml", } if extra: data.update(extra) return self.client.post(reverse("app_data:data_export"), data) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_fields_whitelist_excludes_other_field(self): response = self._post({"fields": "keep_me"}) self.assertEqual(response.status_code, 200) content = response.content.decode() self.assertIn("keepvalue", content) self.assertNotIn("dropvalue", content) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_fields_blacklist_excludes_field(self): response = self._post({"fields": "!drop_me"}) self.assertEqual(response.status_code, 200) content = response.content.decode() self.assertIn("keepvalue", content) self.assertNotIn("dropvalue", content) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_fields_default_includes_all(self): response = self._post() self.assertEqual(response.status_code, 200) content = response.content.decode() self.assertIn("keepvalue", content) self.assertIn("dropvalue", content) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_fields_csv_whitelist(self): response = self._post({"format": "csv", "fields": "keep_me"}) self.assertEqual(response.status_code, 200) content = response.content.decode() self.assertIn("keepvalue", content) self.assertNotIn("dropvalue", content) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_fields_blacklist_applies_with_named_dataview(self): # even though the dataview explicitly lists "drop_me" as a column, # the fields filter still drops it response = self._post({"dv": "test2_view", "fields": "!drop_me"}) self.assertEqual(response.status_code, 200) content = response.content.decode() self.assertNotIn("dropvalue", content) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_fields_whitelist_applies_with_named_dataview(self): response = self._post({"dv": "test2_view", "fields": "keyname"}) self.assertEqual(response.status_code, 200) content = response.content.decode() self.assertNotIn("dropvalue", content) self.assertIn("test2-01", content) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_fields_default_keeps_dataview_columns(self): response = self._post({"dv": "test2_view"}) self.assertEqual(response.status_code, 200) content = response.content.decode() self.assertIn("dropvalue", content) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_fields_blacklist_applies_to_renamed_dataview_column(self): # dataview column is a dict-form rename ("MyDrop" label, "from: drop_me" # source) - the filter must match the source field, not the label helper.add_instance( classname="_dataview", keyname="test2_view_renamed", fields={ "target_class": "test2", "content": "columns:\n - keyname\n - MyDrop:\n from: drop_me\n", }, ) response = self._post({"dv": "test2_view_renamed", "fields": "!drop_me"}) self.assertEqual(response.status_code, 200) content = response.content.decode() self.assertNotIn("dropvalue", content) self.assertNotIn("MyDrop", content) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_fields_keyname_always_included_in_dataview(self): response = self._post({"dv": "test2_view", "fields": "!*"}) self.assertEqual(response.status_code, 200) content = response.content.decode() self.assertNotIn("dropvalue", content) self.assertIn("test2-01", content) class TestDataViewFieldsFilter(TestCase): def setUp(self): cache.clear() helper.add_schema( classname="dvfieldstest", field_definition={ "keep_me": {"dataformat": "string", "displayname": "Keep Me"}, "drop_me": {"dataformat": "string", "displayname": "Drop Me"}, }, ) helper.add_instance( classname="dvfieldstest", keyname="obj01", fields={"keep_me": "keepvalue", "drop_me": "dropvalue"}, ) helper.add_instance( classname="_dataview", keyname="dvfieldstest_plain", fields={ "target_class": "dvfieldstest", "content": "columns:\n - keyname\n - keep_me\n - drop_me\n", }, ) helper.add_instance( classname="_dataview", keyname="dvfieldstest_renamed", fields={ "target_class": "dvfieldstest", "content": "columns:\n - keyname\n - MyDrop:\n from: drop_me\n", }, ) helper.add_instance( classname="_dataview", keyname="dvfieldstest_classname", fields={ "target_class": "dvfieldstest", "content": ( "columns:\n - keyname\n - classname\n - schema:\n from: classname\n" ), }, ) helper.add_instance( classname="_dataview", keyname="dvfieldstest_no_identity", fields={ "target_class": "dvfieldstest", "content": "columns:\n - keep_me\n", }, ) helper.add_instance( classname="_dataview", keyname="dvfieldstest_empty", fields={ "target_class": "dvfieldstest", "content": "columns: []\n", }, ) def test_get_columns_none_all_columns(self): # classname isn't in this view's own columns, so it's guaranteed/prepended dataview = DataView(keyname="dvfieldstest_plain") columns = dataview.get_columns(fields=None) self.assertEqual(columns, ["classname", "keyname", "keep_me", "drop_me"]) def test_get_columns_whitelist(self): dataview = DataView(keyname="dvfieldstest_plain") columns = dataview.get_columns(fields="keep_me") self.assertEqual(columns, ["classname", "keyname", "keep_me"]) def test_get_columns_blacklist(self): dataview = DataView(keyname="dvfieldstest_plain") columns = dataview.get_columns(fields="!drop_me") self.assertEqual(columns, ["classname", "keyname", "keep_me"]) def test_get_columns_blacklist_matches_renamed_column_source(self): dataview = DataView(keyname="dvfieldstest_renamed") columns = dataview.get_columns(fields="!drop_me") self.assertEqual(columns, ["classname", "keyname"]) def test_get_columns_whitelist_keeps_keyname(self): dataview = DataView(keyname="dvfieldstest_plain") columns = dataview.get_columns(fields="!*") self.assertEqual(columns, ["classname", "keyname"]) def test_get_columns_guarantees_classname_and_keyname_by_source_not_label(self): # dvfieldstest_classname explicitly defines both (keyname plain, classname # via plain + a "schema" rename, deduped down to one) - already covered, # so no forced prepend dataview = DataView(keyname="dvfieldstest_classname") columns = dataview.get_columns() self.assertEqual(columns, ["keyname", "classname"]) def test_get_columns_renamed_column_uses_source_name(self): # the real-world case: a view renames "keyname" to a human label (here # drop_me -> "MyDrop") - the export column must be the real field name dataview = DataView(keyname="dvfieldstest_renamed") columns = dataview.get_columns() self.assertEqual(columns, ["classname", "keyname", "drop_me"]) self.assertNotIn("MyDrop", columns) def test_filter_renamed_column_value_keyed_by_source_name(self): instance = Instance.from_keyname("dvfieldstest", "obj01") dataview = DataView(keyname="dvfieldstest_renamed") columns = dataview.get_columns() values = dataview.filter(instance=instance, mode="csv") result = dict(zip(columns, values)) self.assertEqual(result["drop_me"], "dropvalue") self.assertNotIn("MyDrop", result) def test_get_columns_view_with_neither_identity_column(self): # a view listing only a regular field still gets classname/keyname # prepended, so the export stays re-importable via cavaliba load dataview = DataView(keyname="dvfieldstest_no_identity") columns = dataview.get_columns() self.assertEqual(columns, ["classname", "keyname", "keep_me"]) def test_get_columns_empty_view_still_guarantees_identity(self): dataview = DataView(keyname="dvfieldstest_empty") columns = dataview.get_columns() self.assertEqual(columns, ["classname", "keyname"]) def test_filter_view_with_neither_identity_column(self): instance = Instance.from_keyname("dvfieldstest", "obj01") dataview = DataView(keyname="dvfieldstest_no_identity") values = dataview.filter(instance=instance, mode="csv") columns = dataview.get_columns() result = dict(zip(columns, values)) self.assertEqual(result["classname"], "dvfieldstest") self.assertEqual(result["keyname"], "obj01") self.assertEqual(result["keep_me"], "keepvalue") def test_filter_empty_view_still_guarantees_identity(self): instance = Instance.from_keyname("dvfieldstest", "obj01") dataview = DataView(keyname="dvfieldstest_empty") values = dataview.filter(instance=instance, mode="csv") self.assertEqual(values, ["dvfieldstest", "obj01"]) def test_filter_and_get_columns_stay_aligned_with_action_and_no_identity(self): instance = Instance.from_keyname("dvfieldstest", "obj01") dataview = DataView(keyname="dvfieldstest_no_identity") columns = dataview.get_columns(action="noop") values = dataview.filter(instance=instance, mode="csv", action="noop") self.assertEqual(columns, ["_action", "classname", "keyname", "keep_me"]) self.assertEqual(values, ["noop", "dvfieldstest", "obj01", "keepvalue"]) def test_filter_blacklist_drops_value(self): instance = Instance.from_keyname("dvfieldstest", "obj01") dataview = DataView(keyname="dvfieldstest_plain") values = dataview.filter(instance=instance, mode="value", fields="!drop_me") # classname, keyname, keep_me (drop_me blacklisted) self.assertEqual(len(values), 3) def test_filter_stays_aligned_with_get_columns(self): instance = Instance.from_keyname("dvfieldstest", "obj01") dataview = DataView(keyname="dvfieldstest_plain") columns = dataview.get_columns(fields="!drop_me") values = dataview.filter(instance=instance, mode="csv", fields="!drop_me") self.assertEqual(len(columns), len(values)) self.assertEqual(dict(zip(columns, values))["keep_me"], "keepvalue") def test_classname_plain_column_resolves_value(self): instance = Instance.from_keyname("dvfieldstest", "obj01") dataview = DataView(keyname="dvfieldstest_classname") columns = dataview.get_columns() values = dataview.filter(instance=instance, mode="csv") self.assertEqual(dict(zip(columns, values))["classname"], "dvfieldstest") def test_classname_duplicate_source_deduplicated(self): # view has both a plain "classname" column and a "schema" rename of # "classname" - the second (same source) is dropped, not double-exported instance = Instance.from_keyname("dvfieldstest", "obj01") dataview = DataView(keyname="dvfieldstest_classname") columns = dataview.get_columns() values = dataview.filter(instance=instance, mode="csv") self.assertEqual(columns, ["keyname", "classname"]) self.assertEqual(len(columns), len(values)) self.assertNotIn("schema", columns) def test_classname_survives_blacklist(self): dataview = DataView(keyname="dvfieldstest_classname") columns = dataview.get_columns(fields="!*") self.assertEqual(columns, ["keyname", "classname"]) class DataExportActionNoopTest(TestCase): def setUp(self): cache.clear() helper.add_admin_user(login="unittest") helper.add_schema( classname="noopmount", field_definition={ "mystring": {"dataformat": "string", "displayname": "MyString"}, }, ) helper.add_instance( classname="noopmount", keyname="noop-01", fields={"mystring": "value1"}, ) def _post(self, extra=None): data = { "classname": "noopmount", "page": "allpages", "dv": "___ALL___", "query": "", "format": "yaml", } if extra: data.update(extra) return self.client.post(reverse("app_data:data_export"), data) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_yaml_includes_action_noop(self): response = self._post() content = response.content.decode() self.assertIn("_action: noop", content) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_json_includes_action_noop(self): response = self._post({"format": "json"}) data = json.loads(response.content) self.assertTrue(all(item.get("_action") == "noop" for item in data)) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_csv_has_no_action_column(self): # unlike YAML/JSON, CSV never gets a safe-by-default "_action" column response = self._post({"format": "csv"}) content = response.content.decode() header = content.strip().splitlines()[0].split(",") self.assertNotIn("_action", header) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_csv_has_no_action_column_with_named_dataview(self): helper.add_instance( classname="_dataview", keyname="noopmount_view", fields={ "target_class": "noopmount", "content": "columns:\n - keyname\n - mystring\n", }, ) response = self._post({"format": "csv", "dv": "noopmount_view"}) content = response.content.decode() header = content.strip().splitlines()[0].split(",") self.assertNotIn("_action", header) class DataExportCsvOptionsTest(TestCase): def setUp(self): cache.clear() helper.add_admin_user(login="unittest") helper.add_schema( classname="csvopttest", field_definition={ "mystring": {"dataformat": "string", "displayname": "MyString"}, }, ) helper.add_instance( classname="csvopttest", keyname="csvopt-01", fields={"mystring": "café"}, ) def _post(self, extra=None, follow=False): data = { "classname": "csvopttest", "page": "allpages", "dv": "___ALL___", "query": "", "format": "csv", } if extra: data.update(extra) return self.client.post(reverse("app_data:data_export"), data, follow=follow) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_default_delimiter_is_comma_family(self): response = self._post() self.assertEqual(response.status_code, 200) content = response.content.decode() self.assertIn("csvopt-01", content) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_custom_delimiter_semicolon(self): response = self._post({"delimiter": ";"}) self.assertEqual(response.status_code, 200) header = response.content.decode().splitlines()[0] self.assertIn(";", header) self.assertNotIn(",", header) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_custom_delimiter_pipe(self): response = self._post({"delimiter": "|"}) self.assertEqual(response.status_code, 200) header = response.content.decode().splitlines()[0] self.assertIn("|", header) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_invalid_delimiter_rejected(self): response = self._post({"delimiter": "::"}, follow=True) self.assertContains(response, "single character") @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_default_encoding_is_utf8(self): response = self._post() self.assertEqual(response.status_code, 200) self.assertIn("utf-8", response["Content-Type"]) self.assertIn("café", response.content.decode("utf-8")) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_custom_encoding_latin1(self): response = self._post({"encoding": "iso-8859-1"}) self.assertEqual(response.status_code, 200) self.assertIn("iso-8859-1", response["Content-Type"]) self.assertIn("café", response.content.decode("iso-8859-1")) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_unknown_encoding_rejected(self): response = self._post({"encoding": "not-a-real-encoding"}, follow=True) self.assertContains(response, "Unknown CSV encoding") @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_encoding_cannot_represent_data_rejected(self): response = self._post({"encoding": "ascii"}, follow=True) self.assertContains(response, "cannot be encoded")