test_views.py

            excs.filter(
                kind=constants.flow_rule_kind.grading).count(), 1)
        description = excs[0].rule.get("description")
        self.assertIsNotNone(description)
        self.assertIn("89.1%", description)

        self.assertAddMessageCallCount(1)
        self.assertAddMessageCalledWith("Exception granted to ", reset=True)

    def test_grading_rule_generates_grade(self):
        # not generates_grade
        self.post_grant_exception_stage_3_view(
            data=self.get_default_post_data(
                # untick generates_grade
                create_grading_exception=True))

        self.assertEqual(self.mock_validate_session_grading_rule.call_count, 1)
        self.mock_validate_session_grading_rule.reset_mock()

        excs = models.FlowRuleException.objects.all()
        self.assertEqual(excs.count(), 1)
        self.assertEqual(
            excs.filter(
                kind=constants.flow_rule_kind.grading).count(), 1)
        generates_grade = excs[0].rule.get("generates_grade")
        self.assertFalse(generates_grade)

        self.assertAddMessageCallCount(1)
        self.assertAddMessageCalledWith("Exception granted to ", reset=True)

        # second, generates_grade
        self.post_grant_exception_stage_3_view(
            data=self.get_default_post_data(
                create_grading_exception=True, generates_grade=True))

        self.assertEqual(self.mock_validate_session_grading_rule.call_count, 1)

        excs = models.FlowRuleException.objects.all()
        self.assertEqual(excs.count(), 2)
        self.assertEqual(
            excs.filter(
                kind=constants.flow_rule_kind.grading).count(), 2)
        generates_grade = excs[1].rule.get("generates_grade")
        self.assertTrue(generates_grade)

        self.assertAddMessageCallCount(1)
        self.assertAddMessageCalledWith("Exception granted to ", reset=True)

    @unittest.skipIf(six.PY2, "PY2 doesn't support subTest")
    def test_grading_rule_item_set(self):
        from random import random
        items = ["credit_percent", "bonus_points", "max_points",
                 "max_points_enforced_cap"]
        for item in items:
            value = random() * 100
            with self.subTest(item=item):
                data = self.get_default_post_data(
                    create_grading_exception=True)
                data[item] = value

                resp = self.post_grant_exception_stage_3_view(data=data)
                self.assertRedirects(resp, self.get_grant_exception_url(),
                                     fetch_redirect_response=False)

                self.assertEqual(
                    self.mock_validate_session_grading_rule.call_count, 1)
                self.mock_validate_session_grading_rule.reset_mock()

                rule = models.FlowRuleException.objects.last().rule
                self.assertEqual(rule[item], value)
                for other_item in items:
                    if other_item != item:
                        self.assertEqual(rule.get(other_item), None)

    def test_create_grading_exception_restrict_to_same_tag(self):
        # restrict_to_same_tag, while session access_rules_tag is none
        flow_desc_access_rule_tags = ["fdesc_tag1", "fdesc_tag2"]
        hacked_flow_desc = self.get_hacked_flow_desc_with_access_rule_tags(
            flow_desc_access_rule_tags)

        with mock.patch("course.content.get_flow_desc") as mock_get_flow_desc:
            mock_get_flow_desc.return_value = hacked_flow_desc

            resp = self.post_grant_exception_stage_3_view(
                data=self.get_default_post_data(
                    create_grading_exception=True,
                    set_access_rules_tag=[flow_desc_access_rule_tags[1]],
                    restrict_to_same_tag=True
                ))

            self.assertFormErrorLoose(resp, None)
            self.assertRedirects(
                resp, self.get_grant_exception_url(),
                fetch_redirect_response=False)

            self.assertEqual(models.FlowRuleException.objects.count(), 1)
            self.assertEqual(
                models.FlowRuleException.objects.filter(
                    kind=constants.flow_rule_kind.grading).count(), 1)

            # the session had the same tag as in the grading rule
            self.fs.refresh_from_db()
            self.assertEqual(
                self.fs.access_rules_tag, flow_desc_access_rule_tags[1])

            exc_rule = models.FlowRuleException.objects.last().rule
            self.assertEqual(
                exc_rule.get("if_has_tag"), flow_desc_access_rule_tags[1])

    # }}}

    def test_form_invalid(self):
        with mock.patch(
                "course.views.ExceptionStage3Form.is_valid") as mock_is_valid:
            mock_is_valid.return_value = False
            resp = self.post_grant_exception_stage_3_view(
                data=self.get_default_post_data())
            self.assertEqual(resp.status_code, 200)

            self.assertEqual(models.FlowRuleException.objects.count(), 0)
            self.assertEqual(self.mock_validate_session_access_rule.call_count, 0)
            self.assertEqual(self.mock_validate_session_grading_rule.call_count, 0)


# {{{ test views.monitor_task

PYTRACEBACK = """\
Traceback (most recent call last):
  File "foo.py", line 2, in foofunc
    don't matter
  File "bar.py", line 3, in barfunc
    don't matter
Doesn't matter: really!\
"""


class MonitorTaskTest(SingleCourseTestMixin, TestCase):
    # test views.monitor_task
    def mock_task(self, name, state, result, traceback=None):
        return {
            'id': uuid(), 'name': name, 'state': state,
            'result': result, 'traceback': traceback,
        }

    def save_result(self, app, task):
        traceback = task.get('traceback') or 'Some traceback'
        state = task['state']
        if state == states.SUCCESS:
            app.backend.mark_as_done(task['id'], task['result'])
        elif state == states.RETRY:
            app.backend.mark_as_retry(
                task['id'], task['result'], traceback=traceback,
            )
        elif state == states.FAILURE:
            app.backend.mark_as_failure(
                task['id'], task['result'], traceback=traceback)
        elif state == states.REVOKED:
            app.backend.mark_as_revoked(
                task_id=task['id'], reason="blabla", state=state)
        elif state == states.STARTED:
            app.backend.mark_as_started(task['id'], **task['result'])
        else:
            app.backend.store_result(
                task['id'], task['result'], state,
            )

    def get_monitor_url(self, task_id):
        from django.urls import reverse
        return reverse("relate-monitor_task", kwargs={"task_id": task_id})

    def get_monitor_view(self, task_id):
        return self.c.get(self.get_monitor_url(task_id))

    def test_user_not_authenticated(self):
        message = "This is good!"
        task = self.mock_task('task', states.SUCCESS, {"message": message})
        self.save_result(app, task)
        with self.temporarily_switch_to_user(None):
            resp = self.get_monitor_view(task['id'])
        self.assertEqual(resp.status_code, 302)

    def test_state_success(self):
        message = "This is good!"
        task = self.mock_task('task', states.SUCCESS, {"message": message})
        self.save_result(app, task)
        resp = self.get_monitor_view(task['id'])
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(resp, "state", task['state'])
        self.assertResponseContextEqual(resp, "progress_statement", message)
        self.assertResponseContextIsNone(resp, "progress_percent")
        self.assertResponseContextIsNone(resp, "traceback")

    def test_state_success_result_not_dict(self):
        task = self.mock_task('task', states.SUCCESS, "This is good!")
        self.save_result(app, task)
        resp = self.get_monitor_view(task['id'])
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(resp, "state", task['state'])
        self.assertResponseContextIsNone(resp, "progress_percent")
        self.assertResponseContextIsNone(resp, "progress_statement")
        self.assertResponseContextIsNone(resp, "traceback")

    def test_state_success_result_contains_no_message(self):
        task = self.mock_task('task', states.SUCCESS, {"log": "This is good!"})
        self.save_result(app, task)
        resp = self.get_monitor_view(task['id'])
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(resp, "state", task['state'])
        self.assertResponseContextIsNone(resp, "progress_percent")
        self.assertResponseContextIsNone(resp, "progress_statement")
        self.assertResponseContextIsNone(resp, "traceback")

    def test_state_progress(self):
        task = self.mock_task("progressing", "PROGRESS",
                         {'current': 20, 'total': 40})
        self.save_result(app, task)
        resp = self.get_monitor_view(task['id'])
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(resp, "state", task['state'])
        self.assertResponseContextEqual(resp, "progress_percent", 50)
        self.assertResponseContextEqual(
            resp, "progress_statement", "20 out of 40 items processed.")
        self.assertResponseContextIsNone(resp, "traceback")

    def test_state_progress_total_zero(self):
        task = self.mock_task("progressing", "PROGRESS",
                         {'current': 0, 'total': 0})
        self.save_result(app, task)
        resp = self.get_monitor_view(task['id'])
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(resp, "state", task['state'])
        self.assertResponseContextIsNone(resp, "progress_percent")
        self.assertResponseContextEqual(
            resp, "progress_statement", "0 out of 0 items processed.")
        self.assertResponseContextIsNone(resp, "traceback")

    def test_state_failure(self):
        self.instructor_participation.user.is_staff = True
        self.instructor_participation.user.save()
        task = self.mock_task("failure", states.FAILURE,
                              KeyError("foo"),
                              PYTRACEBACK)
        self.save_result(app, task)
        with self.temporarily_switch_to_user(self.superuser):
            resp = self.get_monitor_view(task['id'])
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(resp, "state", task['state'])
        self.assertResponseContextIsNone(resp, "progress_percent")
        self.assertResponseContextIsNone(resp, "progress_statement")
        self.assertResponseContextEqual(resp, "traceback", PYTRACEBACK)

    def test_state_failure_request_user_not_staff(self):
        task = self.mock_task("failure", states.FAILURE,
                              KeyError("foo"),
                              PYTRACEBACK)
        self.save_result(app, task)
        resp = self.get_monitor_view(task['id'])

        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(resp, "state", task['state'])
        self.assertResponseContextIsNone(resp, "progress_percent")
        self.assertResponseContextIsNone(resp, "progress_statement")
        self.assertResponseContextIsNone(resp, "traceback")

    def test_pending(self):
        state = states.PENDING
        task = self.mock_task("task", state,
                              {'current': 20, 'total': 40})
        self.save_result(app, task)
        resp = self.get_monitor_view(task['id'])
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(resp, "state", state)
        self.assertResponseContextIsNone(resp, "progress_percent")
        self.assertResponseContextIsNone(resp, "progress_statement")
        self.assertResponseContextIsNone(resp, "traceback")

    def test_received(self):
        state = states.RECEIVED
        task = self.mock_task("task", state,
                              {'current': 20, 'total': 40})
        self.save_result(app, task)
        resp = self.get_monitor_view(task['id'])
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(resp, "state", state)
        self.assertResponseContextIsNone(resp, "progress_percent")
        self.assertResponseContextIsNone(resp, "progress_statement")
        self.assertResponseContextIsNone(resp, "traceback")

    def test_started(self):
        state = states.STARTED
        task = self.mock_task("task", state,
                              {'foo': "foo", 'bar': "bar"})
        self.save_result(app, task)
        resp = self.get_monitor_view(task['id'])
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(resp, "state", state)
        self.assertResponseContextIsNone(resp, "progress_percent")
        self.assertResponseContextIsNone(resp, "progress_statement")
        self.assertResponseContextIsNone(resp, "traceback")

    def test_retry(self):
        state = states.RETRY
        task = self.mock_task("task", state,
                              KeyError())
        self.save_result(app, task)
        resp = self.get_monitor_view(task['id'])
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(resp, "state", state)
        self.assertResponseContextIsNone(resp, "progress_percent")
        self.assertResponseContextIsNone(resp, "progress_statement")
        self.assertResponseContextIsNone(resp, "traceback")

    def test_revoked(self):
        state = states.REVOKED
        task = self.mock_task("task", state, {})
        self.save_result(app, task)
        resp = self.get_monitor_view(task['id'])
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(resp, "state", state)
        self.assertResponseContextIsNone(resp, "progress_percent")
        self.assertResponseContextIsNone(resp, "progress_statement")
        self.assertResponseContextIsNone(resp, "traceback")

# }}}

# vim: fdm=marker