test_enrollment.py

class EnrollmentPreapprovalTestMixin(LocmemBackendTestsMixin,
                                     EnrollmentTestBaseMixin):

    @classmethod
    def setUpTestData(cls):  # noqa
        super().setUpTestData()
        cls.non_ptcp_active_user1.institutional_id_verified = True
        cls.non_ptcp_active_user1.save()
        cls.non_ptcp_active_user2.institutional_id_verified = False
        cls.non_ptcp_active_user2.save()

    @property
    def preapprove_data_emails(self):
        preapproved_user = [self.non_ptcp_active_user1,
                            self.non_ptcp_active_user2]
        preapproved_data = [u.email for u in preapproved_user]
        preapproved_data.insert(1, "  ")  # empty line
        preapproved_data.insert(0, "  ")  # empty line
        return preapproved_data

    @property
    def preapprove_data_institutional_ids(self):
        preapproved_user = [self.non_ptcp_active_user1,
                            self.non_ptcp_active_user2,
                            self.non_ptcp_unconfirmed_user1]
        preapproved_data = [u.institutional_id for u in preapproved_user]
        preapproved_data.insert(1, "  ")  # empty line
        preapproved_data.insert(0, "  ")  # empty line
        return preapproved_data

    @property
    def preapproval_url(self):
        return reverse("relate-create_preapprovals",
                            args=[self.course.identifier])

    @property
    def default_preapprove_role(self):
        role, _ = (ParticipationRole.objects.get_or_create(
            course=self.course, identifier="student"))
        return [str(role.pk)]

    def post_preapproval(self, preapproval_type, preapproval_data=None,
                         force_login_instructor=True):
        if preapproval_data is None:
            if preapproval_type == "email":
                preapproval_data = self.preapprove_data_emails
            elif preapproval_type == "institutional_id":
                preapproval_data = self.preapprove_data_institutional_ids

        assert preapproval_data is not None
        assert isinstance(preapproval_data, list)

        data = {
            "preapproval_type": [preapproval_type],
            "preapproval_data": ["\n".join(preapproval_data)],
            "roles": self.student_role_post_data,
            "submit": [""]
        }
        if not force_login_instructor:
            approver = self.get_logged_in_user()
        else:
            approver = self.instructor_participation.user
        with self.temporarily_switch_to_user(approver):
            return self.c.post(self.preapproval_url, data, follow=True)

    def get_preapproval_count(self):
        return ParticipationPreapproval.objects.all().count()


class CreatePreapprovalsTest(EnrollmentTestMixin,
                             SingleCourseTestMixin, TestCase):
    # test enrollment.create_preapprovals
    @classmethod
    def setUpTestData(cls):  # noqa
        super().setUpTestData()
        cls.course.enrollment_approval_required = True
        cls.course.preapproval_require_verified_inst_id = True
        cls.course.save()

    def setUp(self):
        super().setUp()
        fake_send_enrollment_decision = mock.patch(
            "course.enrollment.send_enrollment_decision")
        self.mock_send_enrollment_decision = (
            fake_send_enrollment_decision.start())
        self.addCleanup(fake_send_enrollment_decision.stop)

    def post_preapproval(
            self, preapproval_type, data, force_login_instructor=True):

        data = {
            "preapproval_type": [preapproval_type],
            "preapproval_data": data,
            "roles": self.student_role_post_data,
            "submit": [""]
        }
        if not force_login_instructor:
            approver = self.get_logged_in_user()
        else:
            approver = self.instructor_participation.user
        with self.temporarily_switch_to_user(approver):
            return self.c.post(self.preapproval_url, data, follow=True)

    def test_no_permission(self):
        with self.temporarily_switch_to_user(self.student_participation.user):
            resp = self.c.get(self.preapproval_url)
            self.assertEqual(resp.status_code, 403)
            resp = self.c.post(self.preapproval_url, data={})
            self.assertEqual(resp.status_code, 403)

    def test_login_required(self):
        with self.temporarily_switch_to_user(None):
            resp = self.c.get(self.preapproval_url)
            self.assertEqual(resp.status_code, 302)
            resp = self.c.post(self.preapproval_url, data={})
            self.assertEqual(resp.status_code, 302)

    def test_get(self):
        with self.temporarily_switch_to_user(self.instructor_participation.user):
            resp = self.c.get(self.preapproval_url)
            self.assertEqual(resp.status_code, 200)

    def test_post_form_not_valid(self):
        resp = self.post_preapproval("email", {})
        self.assertEqual(resp.status_code, 200)

        resp = self.post_preapproval("some_type", {})
        self.assertEqual(resp.status_code, 200)

        resp = self.post_preapproval("institutional_id", {})
        self.assertEqual(resp.status_code, 200)

    def test_create_preapproval_email(self):
        approval_data = "abc@foo.com\n  \n cde@foo.com\n  \n"
        resp = self.post_preapproval(
            "email",
            approval_data)
        self.assertRedirects(
            resp, self.course_page_url, fetch_redirect_response=False)
        self.assertEqual(
            self.get_preapproval_count(), 2)
        self.assertAddMessageCalledWith(
            [MESSAGE_BATCH_PREAPPROVED_RESULT_PATTERN
             % {
                 'n_created': 2,
                 'n_exist': 0,
                 'n_requested_approved': 0
             }])

        # repost same data
        resp = self.post_preapproval(
            "email",
            approval_data)
        self.assertRedirects(
            resp, self.course_page_url, fetch_redirect_response=False)
        self.assertEqual(
            self.get_preapproval_count(), 2)
        self.assertAddMessageCalledWith(
            [MESSAGE_BATCH_PREAPPROVED_RESULT_PATTERN
             % {
                 'n_created': 0,
                 'n_exist': 2,
                 'n_requested_approved': 0
             }])

    def test_create_preapproval_email_handle_pendinng(self):
        user = factories.UserFactory()
        factories.ParticipationFactory(
            course=self.course, user=user, status=p_status.requested
        )
        approval_data = "%s\n  \n cde@foo.com\n  \n" % user.email.upper()
        resp = self.post_preapproval(
            "email",
            approval_data)
        self.assertRedirects(
            resp, self.course_page_url, fetch_redirect_response=False)
        self.assertEqual(
            self.get_preapproval_count(), 2)
        self.assertAddMessageCalledWith(
            [MESSAGE_BATCH_PREAPPROVED_RESULT_PATTERN
             % {
                 'n_created': 2,
                 'n_exist': 0,
                 'n_requested_approved': 1
             }])
        self.assertEqual(
            self.mock_send_enrollment_decision.call_count, 1)
        self.mock_send_enrollment_decision.reset_mock()

        # repost same data
        resp = self.post_preapproval(
            "email",
            approval_data)
        self.assertRedirects(
            resp, self.course_page_url, fetch_redirect_response=False)
        self.assertEqual(
            self.get_preapproval_count(), 2)
        self.assertAddMessageCalledWith(
            [MESSAGE_BATCH_PREAPPROVED_RESULT_PATTERN
             % {
                 'n_created': 0,
                 'n_exist': 2,
                 'n_requested_approved': 0
             }])
        self.assertEqual(
            self.mock_send_enrollment_decision.call_count, 0)

    def test_create_preapproval_inst_id(self):
        approval_data = "abc\n  \ncde \n  \n"
        resp = self.post_preapproval(
            "institutional_id",
            approval_data)
        self.assertRedirects(
            resp, self.course_page_url, fetch_redirect_response=False)
        self.assertEqual(
            self.get_preapproval_count(), 2)
        self.assertAddMessageCalledWith(
            [MESSAGE_BATCH_PREAPPROVED_RESULT_PATTERN
             % {
                 'n_created': 2,
                 'n_exist': 0,
                 'n_requested_approved': 0
             }])

        # repost same data
        resp = self.post_preapproval(
            "institutional_id",
            approval_data)
        self.assertRedirects(
            resp, self.course_page_url, fetch_redirect_response=False)
        self.assertEqual(
            self.get_preapproval_count(), 2)
        self.assertAddMessageCalledWith(
            [MESSAGE_BATCH_PREAPPROVED_RESULT_PATTERN
             % {
                 'n_created': 0,
                 'n_exist': 2,
                 'n_requested_approved': 0
             }])

    def test_create_preapproval_inst_id_handle_pending_require_verified(self):
        self.course.preapproval_require_verified_inst_id = False
        self.course.save()
        user1 = factories.UserFactory(institutional_id_verified=True)
        user2 = factories.UserFactory(institutional_id_verified=False)
        factories.ParticipationFactory(
            course=self.course, user=user1, status=p_status.requested)
        factories.ParticipationFactory(
            course=self.course, user=user2, status=p_status.requested)
        approval_data = "{}\n  \ncde \n  {}\n".format(
            user1.institutional_id.upper(), user2.institutional_id)

        resp = self.post_preapproval(
            "institutional_id",
            approval_data)

        self.assertRedirects(
            resp, self.course_page_url, fetch_redirect_response=False)

        self.assertEqual(
            self.get_preapproval_count(), 3)

        self.assertAddMessageCalledWith(
            [MESSAGE_BATCH_PREAPPROVED_RESULT_PATTERN
             % {
                 'n_created': 3,
                 'n_exist': 0,
                 'n_requested_approved': 2
             }])

        self.assertEqual(
            self.mock_send_enrollment_decision.call_count, 2)
        self.mock_send_enrollment_decision.reset_mock()

        # repost same data
        resp = self.post_preapproval(
            "institutional_id",
            approval_data)
        self.assertRedirects(
            resp, self.course_page_url, fetch_redirect_response=False)
        self.assertEqual(
            self.get_preapproval_count(), 3)
        self.assertAddMessageCalledWith(
            [MESSAGE_BATCH_PREAPPROVED_RESULT_PATTERN
             % {
                 'n_created': 0,
                 'n_exist': 3,
                 'n_requested_approved': 0
             }])
        self.assertEqual(
            self.mock_send_enrollment_decision.call_count, 0)

    def test_create_preapproval_inst_id_handle_pending(self):
        user1 = factories.UserFactory(institutional_id_verified=True)
        user2 = factories.UserFactory(institutional_id_verified=False)
        factories.ParticipationFactory(
            course=self.course, user=user1, status=p_status.requested)
        factories.ParticipationFactory(
            course=self.course, user=user2, status=p_status.requested)
        approval_data = "{}\n  \ncde \n  {}\n".format(
            user1.institutional_id, user2.institutional_id)

        resp = self.post_preapproval(
            "institutional_id",
            approval_data)

        self.assertRedirects(
            resp, self.course_page_url, fetch_redirect_response=False)

        self.assertEqual(
            self.get_preapproval_count(), 3)

        self.assertAddMessageCalledWith(
            [MESSAGE_BATCH_PREAPPROVED_RESULT_PATTERN
             % {
                 'n_created': 3,
                 'n_exist': 0,
                 'n_requested_approved': 1
             }])
        self.assertEqual(
            self.mock_send_enrollment_decision.call_count, 1)
        self.mock_send_enrollment_decision.reset_mock()

        # repost same data
        resp = self.post_preapproval(
            "institutional_id",
            approval_data)
        self.assertRedirects(
            resp, self.course_page_url, fetch_redirect_response=False)
        self.assertEqual(
            self.get_preapproval_count(), 3)
        self.assertAddMessageCalledWith(
            [MESSAGE_BATCH_PREAPPROVED_RESULT_PATTERN
             % {
                 'n_created': 0,
                 'n_exist': 3,
                 'n_requested_approved': 0
             }])
        self.assertEqual(
            self.mock_send_enrollment_decision.call_count, 0)
        self.mock_send_enrollment_decision.reset_mock()

        # update_course
        self.course.preapproval_require_verified_inst_id = False
        self.course.save()

        # user2 is expected to be approved as active participation upon
        # course update.
        self.assertEqual(
            Participation.objects.get(user=user2).status, p_status.active)
        self.assertEqual(
            self.mock_send_enrollment_decision.call_count, 1)
        self.mock_send_enrollment_decision.reset_mock()


class EditParticipationFormTest(SingleCourseTestMixin, TestCase):
    # test enrollment.EditParticipationForm
    # (currently for cases not covered by other tests)
    # todo: full test

    def setUp(self):
        super().setUp()
        rf = RequestFactory()
        self.request = rf.get(self.get_course_page_url())

    def get_pctx_by_participation(self, participation):
        self.request.user = participation.user

        from course.utils import CoursePageContext
        return CoursePageContext(self.request, self.course.identifier)

    def test_role_button_disabled(self):
        pctx = self.get_pctx_by_participation(self.ta_participation)
        form = enrollment.EditParticipationForm(
            add_new=False, pctx=pctx, instance=self.student_participation)
        self.assertTrue(form.fields["roles"].disabled)

    def test_role_button_not_disabled(self):
        pctx = self.get_pctx_by_participation(self.instructor_participation)
        form = enrollment.EditParticipationForm(
            add_new=False, pctx=pctx, instance=self.student_participation)
        self.assertFalse(form.fields["roles"].disabled)

    def test_drop_button_not_added_for_dropped_participation(self):
        dropped = factories.ParticipationFactory(
            course=self.course, status=p_status.dropped)

        pctx = self.get_pctx_by_participation(self.instructor_participation)
        form = enrollment.EditParticipationForm(
            add_new=False, pctx=pctx, instance=dropped)

        names, _ = self.get_form_submit_inputs(form)
        self.assertNotIn("drop", names)


class EnrollmentEmailConnectionsTestMixin(LocmemBackendTestsMixin):
    # Ensure request/decision mail will be sent with/without EmailConnection
    # settings. https://github.com/inducer/relate/pull/366
    courses_attributes_extra_list = [{"enrollment_approval_required": True}]

    email_connections = {
        "enroll": {
            'host': 'smtp.gmail.com',
            'username': 'blah@blah.com',
            'password': 'password',
            'port': 587,
            'use_tls': True,
        },
    }

    email_connections_none = {}
    enrollment_email_from = "enroll@example.com"
    robot_email_from = "robot@example.com"


# Todo: refactor email connections in all views
class EnrollmentRequestEmailConnectionsTest(
            EnrollmentEmailConnectionsTestMixin, EnrollmentTestBaseMixin, TestCase):

    def test_email_with_email_connections1(self):
        # with EMAIL_CONNECTIONS and ENROLLMENT_EMAIL_FROM configured
        with self.settings(
                EMAIL_CONNECTIONS=self.email_connections,
                ROBOT_EMAIL_FROM=self.robot_email_from,
                ENROLLMENT_EMAIL_FROM=self.enrollment_email_from):

            expected_from_email = settings.ENROLLMENT_EMAIL_FROM
            with self.temporarily_switch_to_user(self.non_ptcp_active_user1):
                self.c.post(self.enroll_request_url, follow=True)

            msg = mail.outbox[0]
            self.assertEqual(msg.from_email, expected_from_email)

    def test_email_with_email_connections3(self):
        # with neither EMAIL_CONNECTIONS nor ENROLLMENT_EMAIL_FROM configured
        with self.settings(
                EMAIL_CONNECTIONS=self.email_connections,
                ROBOT_EMAIL_FROM=self.robot_email_from):
            if hasattr(settings, ENROLLMENT_EMAIL_FROM):
                del settings.ENROLLMENT_EMAIL_FROM

            expected_from_email = settings.ROBOT_EMAIL_FROM

            with self.temporarily_switch_to_user(self.non_ptcp_active_user1):
                self.c.post(self.enroll_request_url, follow=True)

            msg = mail.outbox[0]
            self.assertEqual(msg.from_email, expected_from_email)


class EnrollmentDecisionEmailConnectionsTest(
        EnrollmentEmailConnectionsTestMixin, EnrollmentDecisionTestMixin, TestCase):

    # {{{ with EMAIL_CONNECTIONS and ENROLLMENT_EMAIL_FROM configured
    def test_email_with_email_connections1(self):
        with self.settings(
                RELATE_EMAIL_SMTP_ALLOW_NONAUTHORIZED_SENDER=False,
                EMAIL_CONNECTIONS=self.email_connections,
                ROBOT_EMAIL_FROM=self.robot_email_from,
                ENROLLMENT_EMAIL_FROM=self.enrollment_email_from):

            expected_from_email = settings.ENROLLMENT_EMAIL_FROM

            with self.temporarily_switch_to_user(
                    self.instructor_participation.user):
                self.c.post(
                    self.my_participation_edit_url,
                    self.get_edit_participation_form_data("approve"))

            msg = mail.outbox[0]
            self.assertEqual(msg.from_email, expected_from_email)

    def test_email_with_email_connections2(self):
        with self.settings(
                RELATE_EMAIL_SMTP_ALLOW_NONAUTHORIZED_SENDER=True,
                EMAIL_CONNECTIONS=self.email_connections,
                ROBOT_EMAIL_FROM=self.robot_email_from,
                ENROLLMENT_EMAIL_FROM=self.enrollment_email_from):

            expected_from_email = self.course.from_email

            with self.temporarily_switch_to_user(
                    self.instructor_participation.user):
                self.c.post(
                    self.my_participation_edit_url,
                    self.get_edit_participation_form_data("approve"))

            msg = mail.outbox[0]
            self.assertEqual(msg.from_email, expected_from_email)
    # }}}

    # {{{ with neither EMAIL_CONNECTIONS nor ENROLLMENT_EMAIL_FROM configured
    def test_email_with_email_connections3(self):
        with self.settings(
                RELATE_EMAIL_SMTP_ALLOW_NONAUTHORIZED_SENDER=False,
                ROBOT_EMAIL_FROM=self.robot_email_from):
            if hasattr(settings, EMAIL_CONNECTIONS):
                del settings.EMAIL_CONNECTIONS
            if hasattr(settings, ENROLLMENT_EMAIL_FROM):
                del settings.ENROLLMENT_EMAIL_FROM

            expected_from_email = settings.ROBOT_EMAIL_FROM

            with self.temporarily_switch_to_user(
                    self.instructor_participation.user):
                self.c.post(
                    self.my_participation_edit_url,
                    self.get_edit_participation_form_data("approve"))

            msg = mail.outbox[0]
            self.assertEqual(msg.from_email, expected_from_email)

    def test_email_with_email_connections4(self):
        with self.settings(
                RELATE_EMAIL_SMTP_ALLOW_NONAUTHORIZED_SENDER=True,
                ROBOT_EMAIL_FROM=self.robot_email_from):
            if hasattr(settings, EMAIL_CONNECTIONS):
                del settings.EMAIL_CONNECTIONS
            if hasattr(settings, ENROLLMENT_EMAIL_FROM):
                del settings.ENROLLMENT_EMAIL_FROM

            expected_from_email = self.course.from_email

            with self.temporarily_switch_to_user(
                    self.instructor_participation.user):
                self.c.post(self.my_participation_edit_url,
                            self.get_edit_participation_form_data("approve"))
            msg = mail.outbox[0]
            self.assertEqual(msg.from_email, expected_from_email)
    # }}}


@pytest.mark.django_db
class ParticipationQueryFormTest(unittest.TestCase):
    #test enrollment.ParticipationQueryForm
    def setUp(self):
        self.course = factories.CourseFactory()

    def test_form_valid(self):
        data = {
            "queries": "id:1234",
            "op": "apply_tag",
            "tag": "hello"}

        form = enrollment.ParticipationQueryForm(data=data)
        self.assertTrue(form.is_valid())

    def test_form_valid_no_tag(self):
        data = {
            "queries": "id:1234",
            "op": "drop"}

        form = enrollment.ParticipationQueryForm(data=data)
        self.assertTrue(form.is_valid())

    def test_form_tag_invalid(self):
        data = {
            "queries": "id:1234",
            "op": "apply_tag",
            "tag": "~hello~"}

        form = enrollment.ParticipationQueryForm(data=data)
        self.assertIn("Name contains invalid characters.", form.errors["tag"])


class QueryParticipationsTestMixin(MockAddMessageMixing, SingleCoursePageTestMixin):

    @property
    def query_participation_url(self):
        return self.get_course_view_url(
            "relate-query_participations", self.course.identifier)

    @classmethod
    def setup_participation_data(cls):
        p_list = [factories.ParticipationFactory(
            course=cls.course,
            status=p_status.requested,
            tags=list(factories.ParticipationTagFactory.create_batch(
                size=2, course=cls.course)))]

        p_list.extend(factories.ParticipationFactory.create_batch(
            size=3, course=cls.course,
            tags=list(factories.ParticipationTagFactory.create_batch(
                size=2, course=cls.course))))

        return p_list

    def post_query_participation(self, queries, op=None, tag=None, apply=False,
                                 force_login_instructor=True):
        form_data = {"queries": queries}
        form_data["op"] = op or "apply_tag"
        form_data["tag"] = tag or ""
        if apply:
            form_data["apply"] = ""
        else:
            form_data["submit"] = ""

        if not force_login_instructor:
            u = self.get_logged_in_user()
        else:
            u = self.instructor_participation.user
        with self.temporarily_switch_to_user(u):
            return self.c.post(self.query_participation_url, data=form_data)


class QueryParticipationsParseQueryTest(QueryParticipationsTestMixin, TestCase):
    # test enrollment.query_participations for parse query
    @classmethod
    def setUpTestData(cls):  # noqa
        super().setUpTestData()

        # Participations are created here should not be modified
        cls.participations = cls.setup_participation_data()

    def test_no_query_permission(self):
        with self.temporarily_switch_to_user(self.participations[0].user):
            resp = self.c.get(self.query_participation_url)
            self.assertEqual(resp.status_code, 403)
            resp = self.c.post(self.query_participation_url, data={})
            self.assertEqual(resp.status_code, 403)

    def test_with_view_participant_masked_profile_permission(self):
        with self.temporarily_switch_to_user(self.ta_participation.user):
            resp = self.c.get(self.query_participation_url)
            self.assertEqual(resp.status_code, 200)
            resp = self.c.post(self.query_participation_url, data={})
            self.assertEqual(resp.status_code, 200)

        from course.constants import participation_permission as pperm
        from course.models import ParticipationPermission
        pp = ParticipationPermission(
            participation=self.ta_participation,
            permission=pperm.view_participant_masked_profile)
        pp.save()

        with self.temporarily_switch_to_user(self.ta_participation.user):
            resp = self.c.get(self.query_participation_url)
            self.assertEqual(resp.status_code, 403)
            resp = self.c.post(self.query_participation_url, data={})
            self.assertEqual(resp.status_code, 403)

    def test_user_id_equal(self):
        queries = "id:%d" % self.participations[0].user.id

        resp = self.post_query_participation(queries)
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(
            resp, "result", [self.participations[0]])

    def test_user_email_equal(self):
        queries = "email:%s" % self.participations[0].user.email

        resp = self.post_query_participation(queries)
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(
            resp, "result", [self.participations[0]])

    def test_user_email_contains(self):
        queries = "email-contains:test_factory_"

        resp = self.post_query_participation(queries)
        self.assertEqual(resp.status_code, 200)
        result = resp.context["result"]
        self.assertSetEqual(set(result), set(self.participations))

    def test_username_equal(self):
        queries = "username:%s" % self.participations[0].user.username

        resp = self.post_query_participation(queries)
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(
            resp, "result", [self.participations[0]])

    def test_username_contains(self):
        queries = "username-contains:testuser_"

        resp = self.post_query_participation(queries)
        self.assertEqual(resp.status_code, 200)
        result = resp.context["result"]
        self.assertSetEqual(set(result), set(self.participations))

    def test_inst_id_equal(self):
        queries = "institutional-id:%s" % (
            self.participations[0].user.institutional_id)

        resp = self.post_query_participation(queries)
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(
            resp, "result", [self.participations[0]])

    def test_inst_id_contains(self):
        queries = "institutional-id-contains:institutional_id"

        resp = self.post_query_participation(queries)
        self.assertEqual(resp.status_code, 200)
        result = resp.context["result"]
        self.assertSetEqual(set(result), set(self.participations))

    def test_tagged(self):
        queries = "tagged:%s" % (
            self.participations[0].tags.all()[0].name)

        resp = self.post_query_participation(queries)
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(
            resp, "result", [self.participations[0]])

    def test_tagged2(self):
        queries = "tagged:%s" % (
            self.participations[1].tags.all()[0].name)

        resp = self.post_query_participation(queries)
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(
            resp, "result", self.participations[1:])

    def test_role(self):
        queries = "role:%s" % (
            self.participations[1].roles.all()[0].identifier)
        resp = self.post_query_participation(queries)
        self.assertEqual(resp.status_code, 200)
        result = resp.context["result"]

        self.assertEqual(len(result), 5)

    def test_role_ta(self):
        queries = "role:teaching_assistant"
        resp = self.post_query_participation(queries)
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(
            resp, "result", [self.ta_participation])

    def test_status(self):
        queries = "status:requested"

        resp = self.post_query_participation(queries)
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(
            resp, "result", [self.participations[0]])

    def test_has_started_flow(self):
        with self.temporarily_switch_to_user(self.participations[1].user):
            self.start_flow(self.flow_id)
        with self.temporarily_switch_to_user(self.participations[2].user):
            self.start_flow(self.flow_id)

        queries = "has-started:%s" % self.flow_id

        resp = self.post_query_participation(queries)
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(
            resp, "result", [self.participations[1], self.participations[2]])

    def test_has_submitted_flow(self):
        with self.temporarily_switch_to_user(self.participations[1].user):
            self.start_flow(self.flow_id)
        with self.temporarily_switch_to_user(self.participations[2].user):
            self.start_flow(self.flow_id)
            self.end_flow()

        queries = "has-submitted:%s" % self.flow_id

        resp = self.post_query_participation(queries)
        self.assertEqual(resp.status_code, 200)
        self.assertResponseContextEqual(
            resp, "result", [self.participations[2]])

    def test_and(self):
        queries = "role:student and status:requested"
        resp = self.post_query_participation(queries)
        self.assertEqual(resp.status_code, 200)

        self.assertResponseContextEqual(
            resp, "result", [self.participations[0]])

    def test_not(self):
        queries = "role:student not status:active"
        resp = self.post_query_participation(queries)
        self.assertEqual(resp.status_code, 200)

        self.assertResponseContextEqual(
            resp, "result", [self.participations[0]])

    def test_or(self):
        queries = "id:%d or status:requested" % self.participations[0].user.id
        resp = self.post_query_participation(queries)
        self.assertEqual(resp.status_code, 200)

        self.assertResponseContextEqual(
            resp, "result", [self.participations[0]])

    def test_parentheses(self):
        queries = "role:student and (email-contains:.com not status:active)"
        resp = self.post_query_participation(queries)
        self.assertEqual(resp.status_code, 200)

        self.assertResponseContextEqual(
            resp, "result", [self.participations[0]])

    def test_multiple_line(self):
        queries = (
            "id:{}\n  \n  id:{}".format(
                self.participations[0].user.id, self.participations[2].user.id))

        resp = self.post_query_participation(queries)
        self.assertEqual(resp.status_code, 200)

        self.assertResponseContextEqual(
            resp, "result", [self.participations[0], self.participations[2]])

    def test_input_not_valid(self):
        queries = "unknown:"
        resp = self.post_query_participation(queries)
        self.assertEqual(resp.status_code, 200)

        self.assertFormErrorLoose(resp, None)
        self.assertResponseContextEqual(
            resp, "result", None)

        self.assertAddMessageCallCount(1)
        self.assertAddMessageCalledWith(
            "Error in line 1: InvalidTokenError: at index 0: ...unknown:...")

    # {{{ apply

    def test_apply_tag(self):
        p1 = factories.ParticipationFactory(
            course=self.course,
            user=factories.UserFactory(username="temp_user"))
        p2 = factories.ParticipationFactory(
            course=self.course,
            user=factories.UserFactory(username="temp_user2"))

        queries = "username:{} or username:{}".format(
            p1.user.username, p2.user.username)
        resp = self.post_query_participation(
            queries, apply=True, op="apply_tag", tag="temp_tag")
        self.assertEqual(resp.status_code, 200)

        p1.refresh_from_db()
        p2.refresh_from_db()
        self.assertEqual(p1.tags.all()[0].name, "temp_tag")
        self.assertEqual(p2.tags.all()[0].name, "temp_tag")
        self.assertAddMessageCallCount(1)
        self.assertAddMessageCalledWith(
            "Operation successful on 2 participations.")

    def test_remove_tag(self):
        to_remove_tag = "to_remove"
        p1 = factories.ParticipationFactory(
            course=self.course,
            user=factories.UserFactory(username="temp_user"),
            tags=[to_remove_tag, "abcd"]
        )
        p2 = factories.ParticipationFactory(
            course=self.course,
            user=factories.UserFactory(username="temp_user2"),
            tags=[to_remove_tag, "cdef"])

        queries = "username:{} or username:{}".format(
            p1.user.username, p2.user.username)
        resp = self.post_query_participation(
            queries, apply=True, op="remove_tag", tag=to_remove_tag)
        self.assertEqual(resp.status_code, 200)

        p1.refresh_from_db()
        p2.refresh_from_db()
        self.assertEqual(p1.tags.all()[0].name, "abcd")
        self.assertEqual(p2.tags.all()[0].name, "cdef")

        self.assertAddMessageCallCount(1)
        self.assertAddMessageCalledWith(
            "Operation successful on 2 participations.")

    def test_drop(self):
        to_remove_tag = "to_remove"
        p1 = factories.ParticipationFactory(
            course=self.course,
            user=factories.UserFactory(username="temp_user"),
            tags=[to_remove_tag, "abcd"])
        p2 = factories.ParticipationFactory(
            course=self.course,
            user=factories.UserFactory(username="temp_user2"),
            tags=[to_remove_tag, "cdef"])

        queries = "tagged:%s" % to_remove_tag
        resp = self.post_query_participation(
            queries, apply=True, op="drop")
        self.assertEqual(resp.status_code, 200)

        self.assertAddMessageCallCount(1)
        self.assertAddMessageCalledWith(
            "Operation successful on 2 participations.")

        p1.refresh_from_db()
        p2.refresh_from_db()
        self.assertEqual(p1.status, p_status.dropped)
        self.assertEqual(p2.status, p_status.dropped)

    # }}}


# vim: foldmethod=marker