# (c) cavaliba.com - tests / views_importer - progress modal + abort """ The import progress modal, served by app_data:data_import_progress and shaped by task_ui.progress_context(). Footer and header live inside the swapped element so they follow the task's state; figures fall back to the task's `output` when there is no `progress` at all, which is the usual case (task_import batches every 20 rows, so nothing smaller ever reports progress). """ 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 app_data.task_manager import create_datatask, finish_task, update_progress from tests import helper OUTPUT_DONE = {"count_ok": 7, "count_ko": 1, "errors": ["row3: nope"], "aborted": False, "total": 8} @override_settings(CAVALIBA_AUTH_MODE="unittest") class ViewImporterProgressTest(TestCase): def setUp(self): cache.clear() helper.add_user_with_perm(login="unittest", perms=["p_data_import"]) self.task = create_datatask( name="Import csv - sites.csv", params={"aaa": {"username": "unittest"}}, owner_type="user", owner_id="unittest", ) def _get(self, handle=None): return self.client.get( reverse("app_data:data_import_progress", args=[handle or self.task.handle]) ) def _running(self): DataTask.objects.filter(handle=self.task.handle).update(state="RUNNING") # --------------------------------------------------------------- # running : abort + hide, self-polling # --------------------------------------------------------------- def test_running_offers_abort_and_hide(self): self._running() response = self._get() self.assertContains(response, "Abort") self.assertContains(response, "Hide") self.assertNotContains(response, ">Close<") def test_running_keeps_polling(self): self._running() response = self._get() self.assertContains(response, "every 2s") def test_running_warns_that_abort_is_not_a_rollback(self): self._running() response = self._get() self.assertContains(response, "rows already written stay written") def test_live_counts_come_from_progress(self): self._running() update_progress( self.task.handle, percent=50, count=10, total=20, message="9 ok, 1 ko so far", extra={"count_ok": 9, "count_ko": 1}, ) response = self._get() self.assertEqual(response.context["count_ok"], 9) self.assertEqual(response.context["count_ko"], 1) self.assertEqual(response.context["total"], 20) # --------------------------------------------------------------- # terminal : close only, figures from output # --------------------------------------------------------------- def test_done_offers_close_only(self): finish_task(self.task.handle, state="DONE", output=OUTPUT_DONE) response = self._get() self.assertContains(response, "Close") self.assertNotContains(response, "Abort") self.assertNotContains(response, ">Hide<") def test_done_stops_polling(self): finish_task(self.task.handle, state="DONE", output=OUTPUT_DONE) response = self._get() self.assertNotContains(response, "every 2s") def test_figures_fall_back_to_output_without_any_progress(self): """A task under task_import's batch size never calls on_batch, so it has no `progress` at all - the common case for an import.""" finish_task(self.task.handle, state="DONE", output=OUTPUT_DONE) response = self._get() # finish_task() only ever stamps percent; not one counter is in there progress = DataTask.objects.get(handle=self.task.handle).progress self.assertEqual(progress, {"percent": 100}) self.assertEqual(response.context["total"], 8) self.assertEqual(response.context["count_ok"], 7) self.assertEqual(response.context["count_ko"], 1) self.assertEqual(response.context["percent"], 100) def test_errors_are_shown(self): finish_task(self.task.handle, state="DONE", output=OUTPUT_DONE) response = self._get() self.assertContains(response, "row3: nope") def test_errors_offer_a_download_once_finished(self): finish_task(self.task.handle, state="DONE", output=OUTPUT_DONE) response = self._get() self.assertContains( response, reverse("app_data:task_errors", kwargs={"handle": self.task.handle}) ) def test_errors_download_is_not_offered_while_running(self): task = DataTask.objects.get(handle=self.task.handle) task.state = "RUNNING" task.progress = {"percent": 40, "errors_sample": ["row3: nope"]} task.save(update_fields=["state", "progress"]) response = self._get() self.assertContains(response, "row3: nope") self.assertNotContains( response, reverse("app_data:task_errors", kwargs={"handle": self.task.handle}) ) def test_failed_task_shows_its_reason(self): finish_task(self.task.handle, state="FAILED", output={"error": "task not queued: boom"}) response = self._get() self.assertContains(response, "Import failed") self.assertContains(response, "task not queued: boom") def test_early_abort_is_called_out(self): finish_task(self.task.handle, state="DONE", output={**OUTPUT_DONE, "aborted": True}) response = self._get() self.assertContains(response, "too many failed rows") def test_links_to_the_full_task_report_once_finished(self): finish_task(self.task.handle, state="DONE", output=OUTPUT_DONE) response = self._get() self.assertContains( response, reverse("app_data:task_detail", kwargs={"handle": self.task.handle}) ) # --------------------------------------------------------------- # access # --------------------------------------------------------------- def test_someone_elses_task_is_not_found(self): other = create_datatask(name="Import csv - x", owner_type="user", owner_id="someuser") response = self._get(handle=other.handle) self.assertEqual(response.status_code, 404) @override_settings(CAVALIBA_AUTH_MODE="unittest") class ViewImporterProgressTaskAllTest(TestCase): """p_task_all widens the modal to anyone's task, same rule as views_task.""" def setUp(self): cache.clear() helper.add_user_with_perm(login="unittest", perms=["p_data_import", "p_task_all"]) def test_sees_someone_elses_task(self): other = create_datatask(name="Import csv - x", owner_type="user", owner_id="someuser") response = self.client.get(reverse("app_data:data_import_progress", args=[other.handle])) self.assertEqual(response.status_code, 200) @override_settings(CAVALIBA_AUTH_MODE="unittest") class ViewImporterProgressNoPermTest(TestCase): """The modal is gated on the tool's own permission, p_data_import.""" def setUp(self): cache.clear() helper.add_user_with_perm(login="unittest", perms=[]) def test_refused_without_p_data_import(self): task = create_datatask(name="Import csv - x", owner_type="user", owner_id="unittest") response = self.client.get(reverse("app_data:data_import_progress", args=[task.handle])) self.assertEqual(response.status_code, 302) @override_settings(CAVALIBA_AUTH_MODE="unittest") class ViewImporterAbortTest(TestCase): def setUp(self): cache.clear() helper.add_user_with_perm(login="unittest", perms=["p_data_import"]) self.task = create_datatask( name="Import csv - sites.csv", params={"aaa": {"username": "unittest"}}, owner_type="user", owner_id="unittest", ) def _abort(self, handle=None, method="post"): url = reverse("app_data:data_import_abort", args=[handle or self.task.handle]) return getattr(self.client, method)(url) def test_abort_marks_the_task_aborted_and_re_renders(self): DataTask.objects.filter(handle=self.task.handle).update(state="RUNNING") response = self._abort() self.assertEqual(response.status_code, 200) self.assertEqual(DataTask.objects.get(handle=self.task.handle).state, "ABORTED") self.assertContains(response, "Close") self.assertNotContains(response, "Abort") def test_abort_of_a_finished_task_changes_nothing(self): finish_task(self.task.handle, state="DONE", output=OUTPUT_DONE) response = self._abort() self.assertEqual(response.status_code, 200) self.assertEqual(DataTask.objects.get(handle=self.task.handle).state, "DONE") def test_get_is_not_allowed(self): response = self._abort(method="get") self.assertEqual(response.status_code, 405) def test_someone_elses_task_is_not_found(self): other = create_datatask(name="Import csv - x", owner_type="user", owner_id="someuser") response = self._abort(handle=other.handle) self.assertEqual(response.status_code, 404) self.assertEqual(DataTask.objects.get(handle=other.handle).state, "QUEUED")