test_healthchecks.py 文件源码

def test_should_report_healthy_if_kafka_healthy(self, kafka_check):
        kafka_check.healthcheck.return_value = healthcheck.CheckResult(True,
                                                                       'OK')
        self.resource._kafka_check = kafka_check

        ret = self.simulate_request(ENDPOINT,
                                    headers={
                                        'Content-Type': 'application/json'
                                    },
                                    decode='utf8',
                                    method='GET')
        self.assertEqual(falcon.HTTP_OK, self.srmock.status)

        ret = json.loads(ret)
        self.assertIn('kafka', ret)
        self.assertEqual('OK', ret.get('kafka'))