test_socks.py 文件源码

def test_handle_stream_closed(self):
        self.layer.socks_conn = Mock()
        self.layer.socks_conn.send = Mock(side_effect=self.collect_send_event)

        socks_request = Request(
            REQ_COMMAND["CONNECT"], ADDR_TYPE["IPV4"],
            u"1.2.3.4", self.port)

        addr_not_support_status = RESP_STATUS["ADDRESS_TYPE_NOT_SUPPORTED"]
        network_unreach_status = RESP_STATUS["NETWORK_UNREACHABLE"]
        general_fail_status = RESP_STATUS["GENRAL_FAILURE"]

        error_cases = [
            (errno.ENOEXEC, addr_not_support_status),
            (errno.EBADF, addr_not_support_status),
            (errno.ETIMEDOUT, network_unreach_status),
            (errno.EADDRINUSE, general_fail_status),
            (55566, general_fail_status)]

        for code, expect_status in error_cases:
            self.layer.create_dest_stream = Mock(
                side_effect=self.create_raise_exception_function(
                    StreamClosedError((code, ))))
            result_future = self.layer.handle_request_and_create_destination(
                socks_request)
            with self.assertRaises(DestNotConnectedError):
                yield result_future

            self.assertIsNotNone(self.event)
            self.assertIsInstance(self.event, Response)
            self.assertEqual(self.event.status, expect_status)
            self.assertEqual(self.event.atyp, ADDR_TYPE["IPV4"])
            self.assertEqual(str(self.event.addr), "1.2.3.4")
            self.assertEqual(self.event.port, self.port)