tests_regular.py 文件源码

def test_0005_bad_pipeline(self):
        "Chain 3 queries, second is bad, should stop the response stream."
        self.real_test = "{0}".format(inspect.stack()[0][3])
        self._prepare_pipe_test()
        self.req2.set_location(Tools.NULL,
                               random=True)
        self.req3 = Request(id(self))
        self.req3.set_location(self.config.get('SERVER_DEFAULT_LOCATION'),
                               random=True)
        with Client() as csock:
            csock.send(self.req1)
            csock.send(self.req2)
            csock.send(self.req3)
            responses = csock.read_all()
            outmsg(str(responses))

        self.analysis(responses,
                      expected_number=2,
                      regular_expected=False)

        self.assertTrue((responses.count <= 2))
        if (responses.count > 1):
            self.assertIn(self.status,
                          [self.STATUS_REJECTED, self.STATUS_ERR400],
                          'Bad response status {0}'.format(self.status))