test_code.py

                my_socket_error = socket_error()
                my_socket_error.errno = errno.ECONNREFUSED
                mock_ctn_request.side_effect = my_socket_error

                # force timeout
                with mock.patch("course.page.code.DOCKER_TIMEOUT", 0.0001):
                    res = request_python_run_with_retries(
                        run_req={}, run_timeout=0.1, retry_count=0)
                    self.assertEqual(res["result"], "uncaught_error")
                    self.assertEqual(res['message'],
                                     "Timeout waiting for container.")
                    self.assertEqual(res["exec_host"], fake_host_ip)
                    self.assertIn(type(my_socket_error).__name__, res["traceback"])

            with self.subTest(
                    case="Docker ping socket error with erron EAFNOSUPPORT"):
                my_socket_error = socket_error()

                # This errno should raise error
                my_socket_error.errno = errno.EAFNOSUPPORT
                mock_ctn_request.side_effect = my_socket_error

                # force timeout
                with mock.patch("course.page.code.DOCKER_TIMEOUT", 0.0001):
                    with self.assertRaises(socket_error) as e:
                        request_python_run_with_retries(
                            run_req={}, run_timeout=0.1, retry_count=0)
                        self.assertEqual(e.exception.errno, my_socket_error.errno)

                with self.assertRaises(socket_error) as e:
                    request_python_run_with_retries(
                        run_req={}, run_timeout=0.1, retry_count=0)
                    self.assertEqual(e.exception.errno, my_socket_error.errno)

            # This should be the last subTest, because this will the behavior of
            # change mock_remove_ctn
            with self.subTest(
                    case="Docker ping timeout with InvalidPingResponse and "
                         "remove container failed with APIError"):
                invalid_ping_resp_msg = "my custom invalid ping response exception"
                fake_host_ip = "0.0.0.0"

                mock_inpect_ctn.return_value = {
                    "NetworkSettings": {
                        "Ports": {"%d/tcp" % RUNPY_PORT: (
                            {"HostIp": fake_host_ip, "HostPort": fake_host_port},
                        )}
                    }}

                mock_ctn_request.side_effect = (
                    InvalidPingResponse(invalid_ping_resp_msg))
                mock_remove_ctn.reset_mock()
                from django.http import HttpResponse
                fake_response_content = "this should not appear"
                mock_remove_ctn.side_effect = DockerAPIError(
                    message="my custom docker api error",
                    response=HttpResponse(content=fake_response_content))

                # force timeout
                with mock.patch("course.page.code.DOCKER_TIMEOUT", 0.0001):
                    res = request_python_run_with_retries(
                        run_req={}, run_timeout=0.1, retry_count=0)
                    self.assertEqual(res["result"], "uncaught_error")
                    self.assertEqual(res['message'],
                                     "Timeout waiting for container.")
                    self.assertEqual(res["exec_host"], "localhost")
                    self.assertIn(InvalidPingResponse.__name__, res["traceback"])
                    self.assertIn(invalid_ping_resp_msg, res["traceback"])

                    # No need to bother the students with this nonsense.
                    self.assertNotIn(DockerAPIError.__name__, res["traceback"])
                    self.assertNotIn(fake_response_content, res["traceback"])

    @skipIf(six.PY2, "PY2 doesn't support subTest")
    def test_docker_container_ping_return_not_ok(self):
        with (
                mock.patch("docker.client.Client.create_container")) as mock_create_ctn, (  # noqa
                mock.patch("docker.client.Client.start")) as mock_ctn_start, (
                mock.patch("docker.client.Client.logs")) as mock_ctn_logs, (
                mock.patch("docker.client.Client.remove_container")) as mock_remove_ctn, (  # noqa
                mock.patch("docker.client.Client.inspect_container")) as mock_inpect_ctn, (  # noqa
                mock.patch("six.moves.http_client.HTTPConnection.request")) as mock_ctn_request, (  # noqa
                mock.patch("six.moves.http_client.HTTPConnection.getresponse")) as mock_ctn_get_response:  # noqa

            mock_create_ctn.return_value = {"Id": "someid"}
            mock_ctn_start.side_effect = lambda x: None
            mock_ctn_logs.side_effect = lambda x: None
            mock_remove_ctn.return_value = None
            fake_host_ip = "192.168.1.100"
            fake_host_port = "69999"

            mock_inpect_ctn.return_value = {
                "NetworkSettings": {
                    "Ports": {"%d/tcp" % RUNPY_PORT: (
                        {"HostIp": fake_host_ip, "HostPort": fake_host_port},
                    )}
                }}

            # force timeout
            with mock.patch("course.page.code.DOCKER_TIMEOUT", 0.0001):
                with self.subTest(
                        case="Docker ping response not OK"):
                    mock_ctn_request.side_effect = lambda x, y: None
                    mock_ctn_get_response.return_value = six.BytesIO(b"NOT OK")

                    res = request_python_run_with_retries(
                        run_req={}, run_timeout=0.1, retry_count=0)
                    self.assertEqual(res["result"], "uncaught_error")
                    self.assertEqual(res['message'],
                                     "Timeout waiting for container.")
                    self.assertEqual(res["exec_host"], fake_host_ip)
                    self.assertIn(InvalidPingResponse.__name__, res["traceback"])

    @skipIf(six.PY2, "PY2 doesn't support subTest")
    def test_docker_container_runpy_timeout(self):
        with (
                mock.patch("docker.client.Client.create_container")) as mock_create_ctn, (  # noqa
                mock.patch("docker.client.Client.start")) as mock_ctn_start, (
                mock.patch("docker.client.Client.logs")) as mock_ctn_logs, (
                mock.patch("docker.client.Client.remove_container")) as mock_remove_ctn, (  # noqa
                mock.patch("docker.client.Client.inspect_container")) as mock_inpect_ctn, (  # noqa
                mock.patch("six.moves.http_client.HTTPConnection.request")) as mock_ctn_request, (  # noqa
                mock.patch("six.moves.http_client.HTTPConnection.getresponse")) as mock_ctn_get_response:  # noqa

            mock_create_ctn.return_value = {"Id": "someid"}
            mock_ctn_start.side_effect = lambda x: None
            mock_ctn_logs.side_effect = lambda x: None
            mock_remove_ctn.return_value = None
            fake_host_ip = "192.168.1.100"
            fake_host_port = "69999"

            mock_inpect_ctn.return_value = {
                "NetworkSettings": {
                    "Ports": {"%d/tcp" % RUNPY_PORT: (
                        {"HostIp": fake_host_ip, "HostPort": fake_host_port},
                    )}
                }}

            with self.subTest(
                    case="Docker ping passed by runpy timed out"):

                # first request is ping, second request raise socket.timeout
                mock_ctn_request.side_effect = [None, sock_timeout]
                mock_ctn_get_response.return_value = six.BytesIO(b"OK")

                res = request_python_run_with_retries(
                    run_req={}, run_timeout=0.1, retry_count=0)
                self.assertEqual(res["result"], "timeout")
                self.assertEqual(res["exec_host"], fake_host_ip)

    @skipIf(six.PY2, "PY2 doesn't support subTest")
    def test_docker_container_runpy_retries_count(self):
        with (
                mock.patch("course.page.code.request_python_run")) as mock_req_run, (  # noqa
                mock.patch("course.page.code.is_nuisance_failure")) as mock_is_nuisance_failure:  # noqa
            expected_result = "this is my custom result"
            mock_req_run.return_value = {"result": expected_result}
            with self.subTest(actual_retry_count=4):
                mock_is_nuisance_failure.side_effect = [True, True, True, False]
                res = request_python_run_with_retries(
                    run_req={}, run_timeout=0.1, retry_count=5)
                self.assertEqual(res["result"], expected_result)
                self.assertEqual(mock_req_run.call_count, 4)
                self.assertEqual(mock_is_nuisance_failure.call_count, 4)

            mock_req_run.reset_mock()
            mock_is_nuisance_failure.reset_mock()
            with self.subTest(actual_retry_count=2):
                mock_is_nuisance_failure.side_effect = [True, True, True, False]
                res = request_python_run_with_retries(
                    run_req={}, run_timeout=0.1, retry_count=1)
                self.assertEqual(res["result"], expected_result)
                self.assertEqual(mock_req_run.call_count, 2)
                self.assertEqual(mock_is_nuisance_failure.call_count, 1)


class IsNuisanceFailureTest(unittest.TestCase):
    # Testing is_nuisance_failure

    def test_not_uncaught_error(self):
        result = {"result": "not_uncaught_error"}
        self.assertFalse(is_nuisance_failure(result))

    def test_no_traceback(self):
        result = {"result": "uncaught_error"}
        self.assertFalse(is_nuisance_failure(result))

    def test_traceback_unkown(self):
        result = {"result": "uncaught_error",
                  "traceback": "unknow traceback"}
        self.assertFalse(is_nuisance_failure(result))

    def test_traceback_has_badstatusline(self):
        result = {"result": "uncaught_error",
                  "traceback": "BadStatusLine: \nfoo"}
        self.assertTrue(is_nuisance_failure(result))

    def test_traceback_address_already_in_use(self):
        result = {"result": "uncaught_error",
                  "traceback": "\nbind: address already in use \nfoo"}
        self.assertTrue(is_nuisance_failure(result))

    def test_traceback_new_connection_error(self):
        result = {"result": "uncaught_error",
                  "traceback":
                      "\nrequests.packages.urllib3.exceptions."
                      "NewConnectionError: \nfoo"}
        self.assertTrue(is_nuisance_failure(result))

    def test_traceback_remote_disconnected(self):
        result = {"result": "uncaught_error",
                  "traceback":
                      "\nhttp.client.RemoteDisconnected: \nfoo"}
        self.assertTrue(is_nuisance_failure(result))


class CodeQuestionWithHumanTextFeedbackSpecialCase(
        SingleCoursePageTestMixin, SubprocessRunpyContainerMixin, TestCase):
    """
    https://github.com/inducer/relate/issues/269
    https://github.com/inducer/relate/commit/2af0ad7aa053b735620b2cf0bae0b45822bfb87f  # noqa
    """

    flow_id = QUIZ_FLOW_ID

    @classmethod
    def setUpTestData(cls):  # noqa
        super(CodeQuestionWithHumanTextFeedbackSpecialCase, cls).setUpTestData()
        cls.c.force_login(cls.student_participation.user)
        cls.start_flow(cls.flow_id)

    def setUp(self):  # noqa
        super(CodeQuestionWithHumanTextFeedbackSpecialCase, self).setUp()
        self.c.force_login(self.student_participation.user)
        self.rf = RequestFactory()

    def get_grade_feedback(self, answer_data, page_value,
                           human_feedback_percentage, grade_data):
        page_id = "py_simple_list"
        course_identifier = self.course.identifier
        flow_session_id = self.get_default_flow_session_id(course_identifier)
        flow_session = FlowSession.objects.get(id=flow_session_id)

        page_ordinal = self.get_page_ordinal_via_page_id(
            page_id, course_identifier, flow_session_id)

        post_data = answer_data.copy()
        post_data.update({"submit": ""})

        request = self.rf.post(
            self.get_page_url_by_ordinal(
                page_ordinal, course_identifier, flow_session_id),
            post_data)
        request.user = self.student_participation.user

        pctx = CoursePageContext(request, course_identifier)
        fpctx = FlowPageContext(
            pctx.repo, pctx.course, self.flow_id, page_ordinal,
            self.student_participation, flow_session, request)
        page_desc = fpctx.page_desc
        page_desc.value = page_value
        page_desc.human_feedback_percentage = human_feedback_percentage

        page = PythonCodeQuestionWithHumanTextFeedback(None, None, page_desc)

        page_context = fpctx.page_context
        grade_data.setdefault('grade_percent', None)
        grade_data.setdefault('released', True)
        grade_data.setdefault('feedback_text', "")
        page_data = fpctx.page_data
        feedback = page.grade(
            page_context=page_context,
            answer_data=answer_data,
            page_data=page_data,
            grade_data=grade_data)

        return feedback

    def test_code_with_human_feedback(self):
        answer_data = {"answer": 'b = [a + 0] * 50'}
        grade_data = {"grade_percent": 100}
        page_value = 4
        human_feedback_percentage = 60
        feedback = self.get_grade_feedback(
            answer_data, page_value, human_feedback_percentage, grade_data)
        self.assertIn("The overall grade is 100%.", feedback.feedback)
        self.assertIn(
            "The autograder assigned 1.60/1.60 points.", feedback.feedback)
        self.assertIn(
            "The human grader assigned 2.40/2.40 points.", feedback.feedback)

    def test_code_with_human_feedback_full_percentage(self):
        answer_data = {"answer": 'b = [a + 0] * 50'}
        grade_data = {"grade_percent": 100}
        page_value = 0
        human_feedback_percentage = 100
        from course.page.base import AnswerFeedback
        with mock.patch(
                "course.page.code.PythonCodeQuestion.grade") as mock_py_grade:

            # In this way, code_feedback.correctness is None
            mock_py_grade.return_value = AnswerFeedback(correctness=None)
            feedback = self.get_grade_feedback(
                answer_data, page_value, human_feedback_percentage, grade_data)
            self.assertIn("The overall grade is 100%.", feedback.feedback)
            self.assertIn(
                "No information on correctness of answer.", feedback.feedback)
            self.assertIn(
                "The human grader assigned 0/0 points.", feedback.feedback)

    def test_code_with_human_feedback_zero_percentage(self):
        answer_data = {"answer": 'b = [a + 0] * 50'}
        grade_data = {}
        page_value = 0
        human_feedback_percentage = 0
        feedback = self.get_grade_feedback(
            answer_data, page_value, human_feedback_percentage, grade_data)
        self.assertIn("The overall grade is 100%.", feedback.feedback)
        self.assertIn(
            "Your answer is correct.", feedback.feedback)
        self.assertIn(
            "The autograder assigned 0/0 points.", feedback.feedback)

# vim: fdm=marker