text: __copyright__ = "Copyright (C) 2018 Dong Zhuang"
  
  __license__ = """
  Permission is hereby granted, free of charge, to any person obtaining a copy
  of this software and associated documentation files (the "Software"), to deal
  in the Software without restriction, including without limitation the rights
  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  copies of the Software, and to permit persons to whom the Software is
  furnished to do so, subject to the following conditions:
  
  The above copyright notice and this permission notice shall be included in
  all copies or substantial portions of the Software.
  
  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  THE SOFTWARE.
  """
  
  import pytest
  from django.test import TestCase, Client
  from django.core.exceptions import ObjectDoesNotExist
  from django.urls import reverse
  
  from relate.utils import dict_to_struct
  
  from course.models import FlowSession
  from course import analytics
  
  from tests.base_test_mixins import (  # noqa
      SingleCourseTestMixin, CoursesTestMixinBase, SingleCoursePageTestMixin,
      SingleCourseQuizPageTestMixin, MockAddMessageMixing, HackRepoMixin)
  from tests.utils import mock
  from tests import factories
  
  
  def _dummy_histogram(*args):
      from course.analytics import Histogram
      hist = Histogram()
      hist.add_data_point(1)
      hist.add_data_point(2)
      return hist
  
  
  @pytest.mark.slow
  class FlowListTest(SingleCourseTestMixin, TestCase):
      """test analytics.flow_list"""
      def get_flow_list_url(self, course_identifier=None):
          course_identifier = course_identifier or self.get_default_course_identifier()
          kwargs = {"course_identifier": course_identifier}
          return reverse("relate-flow_list", kwargs=kwargs)
  
      def get_flow_list_view(self, course_identifier=None,
                             force_login_instructor=True):
          course_identifier = course_identifier or self.get_default_course_identifier()
          if not force_login_instructor:
              user = self.get_logged_in_user()
          else:
              user = self.instructor_participation.user
  
          with self.temporarily_switch_to_user(user):
              return self.client.get(
                  self.get_flow_list_url(course_identifier))
  
      def test_not_authenticated(self):
          with self.temporarily_switch_to_user(None):
              resp = self.get_flow_list_view(force_login_instructor=False)
              self.assertEqual(resp.status_code, 302)
  
      def test_no_pperm(self):
          with self.temporarily_switch_to_user(self.student_participation.user):
              resp = self.get_flow_list_view(force_login_instructor=False)
              self.assertEqual(resp.status_code, 403)
  
      def test_result(self):
          flow_ids = ["c", "b", "a"]
  
          for flow_id in flow_ids:
              factories.FlowSessionFactory.create_batch(
                  size=2,
                  participation=self.student_participation,
                  flow_id=flow_id)
  
          another_course = factories.CourseFactory(identifier="another-course")
          another_participation = factories.ParticipationFactory(course=another_course)
  
          # This make sure other courses' flow_id won't be included
          factories.FlowSessionFactory(participation=another_participation,
                                       flow_id="d")
  
          resp = self.get_flow_list_view()
          self.assertEqual(resp.status_code, 200)
  
          self.assertResponseContextEqual(resp, "flow_ids", sorted(flow_ids))
  
  
  class HistogramTest(CoursesTestMixinBase, TestCase):
      """test analytics.Histogram (for cases not covered by other tests)"""
      def test_add_data_point_num_max_value(self):
          """self.num_max_value is not None and value > self.num_max_value"""
          his = analytics.Histogram(num_max_value=100)
          his.add_data_point(value=110, weight=1.1)
          self.assertEqual(his.string_weights["(value greater than max)"], 1.1)
  
      def test_add_data_point_num_min_value(self):
          """self.num_min_value is not None and value < self.num_min_value"""
          his = analytics.Histogram(num_min_value=60)
          his.add_data_point(value=50, weight=1.2)
          self.assertEqual(his.string_weights["(value smaller than min)"], 1.2)
  
      def test_get_bin_info_list_num_bin_starts_is_not_none(self):
          # just make sure it works
          his = analytics.Histogram(num_bin_starts=[2])
          his.get_bin_info_list()
  
      def test_get_bin_info_nothing_configured(self):
          # just make sure it works
          his = analytics.Histogram()
          his.get_bin_info_list()
          his.html()
          self.assertTemplateUsed("course/histogram-wide.html")
  
      def test_non_wide_html_template_render(self):
          his = analytics.Histogram()
          faked_get_bin_info_list = [analytics.BinInfo(
              title="foo"*30, raw_weight="bar", percentage=100)]
          with mock.patch(
                  "course.analytics.Histogram.get_bin_info_list"
          ) as mock_get_bin_info_list:
              mock_get_bin_info_list.return_value = faked_get_bin_info_list
  
              his.html()
              self.assertTemplateUsed("course/histogram.html")
  
  
  @pytest.mark.slow
  class IsFlowMultipleSubmitTest(SingleCourseTestMixin, TestCase):
      """test course.analytics.is_flow_multiple_submit"""
      def test_flow_desc_has_no_rule(self):
          flow_desc = self.get_hacked_flow_desc(del_rules=True)
          self.assertFalse(analytics.is_flow_multiple_submit(flow_desc))
  
      def test_flow_desc_access_rule_has_no_change_answer_perm(self):
          flow_desc_dict = self.get_hacked_flow_desc(as_dict=True)
          rules = flow_desc_dict["rules"]
          rules.access = [dict_to_struct(
              {"permissions": ["submit_answer"]})]
          flow_desc = dict_to_struct(flow_desc_dict)
          self.assertFalse(analytics.is_flow_multiple_submit(flow_desc))
  
      def test_flow_desc_access_rule_has_change_answer_perm(self):
          flow_desc_dict = self.get_hacked_flow_desc(as_dict=True)
          rules = flow_desc_dict["rules"]
          rules.access = [dict_to_struct(
              {"permissions": ["submit_answer", "change_answer"]})]
          flow_desc = dict_to_struct(flow_desc_dict)
          self.assertTrue(analytics.is_flow_multiple_submit(flow_desc))
  
  
  @pytest.mark.slow
  class IsPageMultipleSubmitTest(SingleCoursePageTestMixin, HackRepoMixin, TestCase):
      """test course.analytics.is_page_multiple_submit"""
      @classmethod
      def setUpTestData(cls):  # noqa
          super().setUpTestData()
          cls.course.active_git_commit_sha = "my_fake_commit_sha_for_page_analytics"
          cls.course.save()
  
          client = Client()
          client.force_login(cls.student_participation.user)
          # cache the page_descs
          cls.start_flow(client, cls.flow_id)
          cls.flow_desc = cls.get_hacked_flow_desc()
  
      def setUp(self):
          super().setUp()
          faked_is_flow_multiple_submit = \
              mock.patch("course.analytics.is_flow_multiple_submit")
          self.mock_is_flow_multiple_submit = faked_is_flow_multiple_submit.start()
          self.mock_is_flow_multiple_submit.return_value = False
          self.addCleanup(faked_is_flow_multiple_submit.stop)
  
      def get_page_desc_by_page_id(self, page_id):
          for group_desc in self.flow_desc.groups:
              for page_desc in group_desc.pages:
                  if page_desc.id == page_id:
                      return page_desc
  
      def test_page_has_no_access_rules(self):
          page_id = "fear"
          page_desc = self.get_page_desc_by_page_id(page_id)
  
          self.assertFalse(analytics.is_page_multiple_submit(
              self.flow_desc, page_desc))
  
          self.mock_is_flow_multiple_submit.return_value = True
          self.assertTrue(analytics.is_page_multiple_submit(
              self.flow_desc, page_desc))
  
      def test_page_access_rules_remove_permissions_not_remove_change_permission(self):
          page_id = "half"
          page_desc = self.get_page_desc_by_page_id(page_id)
  
          self.mock_is_flow_multiple_submit.return_value = True
          self.assertTrue(analytics.is_page_multiple_submit(
              self.flow_desc, page_desc))
  
      def test_page_access_rules_remove_permissions_removed_change_permission(self):
          page_id = "ice_cream_toppings"
          page_desc = self.get_page_desc_by_page_id(page_id)
  
          self.mock_is_flow_multiple_submit.return_value = True
          self.assertFalse(analytics.is_page_multiple_submit(
              self.flow_desc, page_desc))
  
      def test_page_access_rules_add_permissions_not_add_change_permission(self):
          page_id = "lsq"
          page_desc = self.get_page_desc_by_page_id(page_id)
          self.assertFalse(analytics.is_page_multiple_submit(
              self.flow_desc, page_desc))
  
      def test_page_access_rules_add_permissions_added_change_permission(self):
          page_id = "krylov"
          page_desc = self.get_page_desc_by_page_id(page_id)
          self.assertTrue(analytics.is_page_multiple_submit(
              self.flow_desc, page_desc))
  
      def test_page_access_rules_neither_add_nor_remove_permission(self):
          page_id = "age_group"
          page_desc = self.get_page_desc_by_page_id(page_id)
          self.assertFalse(analytics.is_page_multiple_submit(
              self.flow_desc, page_desc))
  
          self.mock_is_flow_multiple_submit.return_value = True
          page_desc = self.get_page_desc_by_page_id(page_id)
          self.assertTrue(analytics.is_page_multiple_submit(
              self.flow_desc, page_desc))
  
  
  @pytest.mark.slow
  class PageAnalyticsTest(SingleCourseTestMixin, TestCase):
      """test analytics.page_analytics, (for cases not covered by other tests)"""
      def test_not_authenticated(self):
          with self.temporarily_switch_to_user(None):
              resp = self.get_flow_page_analytics(
                  flow_id="blabla", group_id="foo", page_id="bar",
                  force_login_instructor=False)
              self.assertEqual(resp.status_code, 302)
  
      def test_no_pperm(self):
          # student user is logged in
          resp = self.get_flow_page_analytics(
              flow_id="blabla", group_id="foo", page_id="bar",
              force_login_instructor=False)
          self.assertEqual(resp.status_code, 403)
  
  
  @pytest.mark.slow
  class FlowAnalyticsTest(SingleCourseQuizPageTestMixin, HackRepoMixin,
                          MockAddMessageMixing, TestCase):
      """analytics.flow_analytics"""
  
      @classmethod
      def setUpTestData(cls):  # noqa
          super().setUpTestData()
          cls.course.active_git_commit_sha = "my_fake_commit_sha_for_flow_analytics"
          cls.course.save()
  
          client = Client()
          client.force_login(cls.student_participation.user)
          cls.start_flow(client, cls.flow_id)
          fs = FlowSession.objects.last()
          for page_ordinal in range(fs.page_count):
              cls.submit_page_answer_by_ordinal_and_test(
                      client,
                      page_ordinal=page_ordinal, do_grading=False,
                      do_human_grade=False)
          cls.end_flow(client)
  
          # start another in-progress session with answers
          cls.start_flow(client, cls.flow_id)
          fs = FlowSession.objects.last()
          for page_ordinal in range(fs.page_count):
              cls.submit_page_answer_by_ordinal_and_test(
                      client,
                      page_ordinal=page_ordinal, do_grading=False,
                      do_human_grade=False)
  
          # start another ended session with out answers
          cls.start_flow(client, cls.flow_id)
          cls.end_flow(client)
  
          # create another participation and with a flow session
          another_partcpt = factories.ParticipationFactory(course=cls.course)
          with cls.temporarily_switch_to_user(client, another_partcpt.user):
              cls.start_flow(client, cls.flow_id)
              cls.end_flow(client)
  
      def test_not_authenticated(self):
          with self.temporarily_switch_to_user(None):
              resp = self.get_flow_analytics_view(
                  flow_id="blabla", force_login_instructor=False)
              self.assertEqual(resp.status_code, 302)
  
      def test_no_pperm(self):
          # student user is logged in
          resp = self.get_flow_analytics_view(
              flow_id="blabla", force_login_instructor=False)
          self.assertEqual(resp.status_code, 403)
  
      def test_success(self):
          resp = self.get_flow_analytics_view(flow_id=self.flow_id)
          self.assertEqual(resp.status_code, 200)
          self.assertResponseContextEqual(
              resp, "flow_identifier", self.flow_id)
  
          resp = self.get_flow_analytics_view(flow_id=self.flow_id,
                                              restrict_to_first_attempt=1)
          self.assertEqual(resp.status_code, 200)
          self.assertResponseContextEqual(
              resp, "flow_identifier", self.flow_id)
  
      def test_success_check_func_call(self):
          with mock.patch(
              "course.analytics.make_grade_histogram",
              return_value=_dummy_histogram()
          ) as mock_make_g_his, mock.patch(
              "course.analytics.make_page_answer_stats_list"
          ) as mock_make_stats_list, mock.patch(
              "course.analytics.make_time_histogram",
              return_value=_dummy_histogram()
          ) as mock_make_t_his, mock.patch(
              "course.analytics.count_participants",
              return_value=2,
          ) as mock_count_particpt:
              resp = self.get_flow_analytics_view(flow_id=self.flow_id)
              self.assertEqual(resp.status_code, 200)
              self.assertResponseContextEqual(
                  resp, "flow_identifier", self.flow_id)
              self.assertResponseContextEqual(resp, "participant_count", 2)
              self.assertResponseContextEqual(resp, "restrict_to_first_attempt", 0)
  
              self.assertEqual(mock_make_g_his.call_count, 1)
              self.assertEqual(mock_make_stats_list.call_count, 1)
              self.assertEqual(mock_make_t_his.call_count, 1)
              self.assertEqual(mock_count_particpt.call_count, 1)
  
      def test_success_test_restrict_to_first_attempt(self):
          with mock.patch(
              "course.analytics.make_grade_histogram",
              return_value=_dummy_histogram()
          ) as mock_make_g_his, mock.patch(
              "course.analytics.make_page_answer_stats_list"
          ) as mock_make_stats_list, mock.patch(
              "course.analytics.make_time_histogram",
              return_value=_dummy_histogram(),
          ) as mock_make_t_his, mock.patch(
              "course.analytics.count_participants",
              return_value=2,
          ) as mock_count_particpt:
              resp = self.get_flow_analytics_view(flow_id=self.flow_id,
                                                  restrict_to_first_attempt=1)
              self.assertEqual(resp.status_code, 200)
              self.assertResponseContextEqual(
                  resp, "flow_identifier", self.flow_id)
              self.assertResponseContextEqual(resp, "participant_count", 2)
  
              # make_page_answer_stats_list is called
              # using restrict_to_first_attempt = 1
              self.assertIn(1, mock_make_stats_list.call_args[0])
              self.assertResponseContextEqual(resp, "restrict_to_first_attempt", 1)
  
              self.assertEqual(mock_make_g_his.call_count, 1)
              self.assertEqual(mock_make_stats_list.call_count, 1)
              self.assertEqual(mock_make_t_his.call_count, 1)
              self.assertEqual(mock_count_particpt.call_count, 1)
  
      def test_success_test_restrict_to_first_attempt_invalid(self):
          with mock.patch(
              "course.analytics.make_grade_histogram",
              return_value=_dummy_histogram()
          ) as mock_make_g_his, mock.patch(
              "course.analytics.make_page_answer_stats_list"
          ) as mock_make_stats_list, mock.patch(
              "course.analytics.make_time_histogram",
              return_value=_dummy_histogram()
          ) as mock_make_t_his, mock.patch(
              "course.analytics.count_participants",
              return_value=2,
          ) as mock_count_particpt:
              resp = self.get_flow_analytics_view(flow_id=self.flow_id,
                                                  restrict_to_first_attempt="foo")
              self.assertEqual(resp.status_code, 200)
              self.assertResponseContextEqual(
                  resp, "flow_identifier", self.flow_id)
  
              # make_page_answer_stats_list is called
              # using restrict_to_first_attempt = 0
              self.assertIn(0, mock_make_stats_list.call_args[0])
              self.assertResponseContextEqual(resp, "restrict_to_first_attempt", 0)
  
              self.assertEqual(mock_make_g_his.call_count, 1)
              self.assertEqual(mock_make_stats_list.call_count, 1)
              self.assertEqual(mock_make_t_his.call_count, 1)
              self.assertEqual(mock_count_particpt.call_count, 1)
  
      def test_flow_id_does_not_exist(self):
          with mock.patch(
              "course.analytics.make_page_answer_stats_list"
          ) as mock_make_stats_list:
              mock_make_stats_list.side_effect = ObjectDoesNotExist()
              resp = self.get_flow_analytics_view(flow_id=self.flow_id)
              self.assertEqual(resp.status_code, 404)
              self.assertAddMessageCalledWith(
                  ("Flow '%s' was not found in the repository, but it exists in "
                      "the database--maybe it was deleted?")
                  % self.flow_id)
  
  # vim: fdm=marker
  