# (c) cavaliba.com - tests / data / exporter view - form, options and result file import json import yaml from django.test import TestCase, override_settings from django.urls import reverse import app_home.cache as cache from app_data.models import DataTask from tests import helper @override_settings(CELERY_TASK_ALWAYS_EAGER=True) class ViewExporterDownloadTest(TestCase): def setUp(self): helper.isolate_filestore(self) cache.clear() helper.add_admin_user(login="unittest") helper.add_schema( classname="vexp", field_definition={ "mystring": {"dataformat": "string", "displayname": "MyString"}, }, ) helper.add_instance(classname="vexp", keyname="vexp-01", fields={"mystring": "one"}) helper.add_instance(classname="vexp", keyname="vexp-02", fields={"mystring": "two"}) helper.add_instance( classname="_dataview", keyname="vexp_small", fields={ "target_class": "vexp", "content": "columns:\n - keyname\n - MyString:\n from: mystring\n", }, ) def _post(self, **extra): data = {"schema": "vexp", "format": "csv", "columns": "*"} data.update(extra) return self.client.post(reverse("app_data:data_exporter"), data) def _text(self, **extra): data = {"schema": "vexp", "format": "csv", "columns": "*"} data.update(extra) return helper.export_text(self, **data) # --------------------------------------------------------------- # GET - form + prepopulation # --------------------------------------------------------------- @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_blank_tool_renders(self): response = self.client.get(reverse("app_data:data_exporter")) self.assertEqual(response.status_code, 200) self.assertEqual(response.context["schema_name"], "") self.assertEqual(response.context["catalog"], []) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_get_offers_no_task(self): response = self.client.get(reverse("app_data:data_exporter")) self.assertIsNone(response.context["task_handle"]) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_schema_param_prepopulates(self): response = self.client.get(reverse("app_data:data_exporter"), {"schema": "vexp"}) self.assertEqual(response.status_code, 200) self.assertEqual(response.context["schema_name"], "vexp") self.assertIn("mystring", [e["name"] for e in response.context["catalog"]]) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_query_param_prepopulates(self): response = self.client.get( reverse("app_data:data_exporter"), {"schema": "vexp", "query": "mystring:one"} ) self.assertEqual(response.context["values"]["query"], "mystring:one") @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_get_does_not_touch_the_session_dataview(self): self.client.get(reverse("app_data:data_exporter"), {"schema": "vexp", "dv": "vexp_small"}) self.assertNotIn("dataviews", self.client.session) # --------------------------------------------------------------- # POST - every export is a task, the file is its attachment # --------------------------------------------------------------- @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_post_stays_on_the_page_with_a_task_handle(self): response = self._post() self.assertEqual(response.status_code, 200) self.assertIn("text/html", response["Content-Type"]) self.assertEqual(response.context["task_handle"], str(DataTask.objects.first().handle)) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_two_row_export_is_a_task_too(self): # no threshold any more: the smallest export goes through the worker task = helper.export_submit(self, schema="vexp", format="csv", columns="*") self.assertEqual(task.state, "DONE") self.assertEqual(task.output["total"], 2) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_csv_result_file(self): lines = self._text().splitlines() self.assertTrue(lines[0].startswith("_schema,keyname")) self.assertEqual(len(lines), 3) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_result_filename_carries_the_schema(self): task = helper.export_submit(self, schema="vexp", format="csv", columns="*") self.assertTrue(task.attachment["filename"].startswith("vexp-")) self.assertTrue(task.attachment["filename"].endswith(".csv")) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_yaml_result_file(self): data = yaml.safe_load(self._text(format="yaml")) self.assertEqual(len(data), 2) self.assertEqual(data[0]["classname"], "vexp") @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_json_result_file(self): data = json.loads(self._text(format="json")) self.assertEqual(len(data), 2) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_query_narrows_the_export(self): lines = self._text(query="mystring:one").splitlines() self.assertEqual(len(lines), 2) self.assertIn("vexp-01", lines[1]) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_explicit_columns_respected(self): lines = self._text(columns="keyname,mystring").splitlines() self.assertEqual(lines[0], "_schema,keyname,mystring") @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_schema_column_is_always_emitted(self): lines = self._text(columns="keyname").splitlines() self.assertEqual(lines[0], "_schema,keyname") @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_action_column_added_first(self): lines = self._text(action="noop").splitlines() self.assertTrue(lines[0].startswith("_action,_schema,keyname")) self.assertTrue(lines[1].startswith("noop,vexp,")) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_custom_delimiter(self): lines = self._text(columns="keyname", delimiter=";").splitlines() self.assertEqual(lines[0], "_schema;keyname") # --------------------------------------------------------------- # POST - rejected input stays on the form, keeps the values, no task # --------------------------------------------------------------- @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_bad_format_rerenders_the_form(self): response = self._post(format="xlsx") self.assertEqual(response.status_code, 200) self.assertIn("text/html", response["Content-Type"]) self.assertIsNone(response.context["task_handle"]) self.assertEqual(DataTask.objects.count(), 0) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_rejected_input_keeps_the_checked_columns(self): response = self._post(format="xlsx", columns="keyname,mystring") self.assertIn("mystring", response.context["checked"]) self.assertNotIn("displayname", response.context["checked"]) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_unknown_schema_rerenders_the_form(self): response = self._post(schema="nosuchschema") self.assertEqual(response.status_code, 200) self.assertIsNone(response.context["task_handle"]) self.assertEqual(DataTask.objects.count(), 0) class ViewExporterPermissionTest(TestCase): def setUp(self): cache.clear() helper.add_schema( classname="vexpperm", field_definition={ "mystring": {"dataformat": "string", "displayname": "MyString"}, }, ) helper.add_instance(classname="vexpperm", keyname="vexpperm-01") @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_without_p_data_export_redirected(self): helper.add_user_with_perm(login="unittest", perms=["p_data_access"]) response = self.client.get(reverse("app_data:data_exporter")) self.assertEqual(response.status_code, 302) @override_settings(CAVALIBA_AUTH_MODE="unittest") def test_with_p_data_export_allowed(self): helper.add_user_with_perm(login="unittest", perms=["p_data_export"]) response = self.client.get(reverse("app_data:data_exporter")) self.assertEqual(response.status_code, 200)