(self, outputs, metric_name, help_text, increments, compressed)
| 95 | self.assertNotIn(metric_name + "_total " + str(increments) + ".0\n", output) |
| 96 | |
| 97 | def assert_outputs(self, outputs, metric_name, help_text, increments, compressed): |
| 98 | self.assertEqual(len(outputs), 2) |
| 99 | response_start = outputs[0] |
| 100 | self.assertEqual(response_start['type'], 'http.response.start') |
| 101 | response_body = outputs[1] |
| 102 | self.assertEqual(response_body['type'], 'http.response.body') |
| 103 | # Status code |
| 104 | self.assertEqual(response_start['status'], 200) |
| 105 | # Headers |
| 106 | num_of_headers = 2 if compressed else 1 |
| 107 | self.assertEqual(len(response_start['headers']), num_of_headers) |
| 108 | self.assertIn((b"Content-Type", CONTENT_TYPE_PLAIN_0_0_4.encode('utf8')), response_start['headers']) |
| 109 | if compressed: |
| 110 | self.assertIn((b"Content-Encoding", b"gzip"), response_start['headers']) |
| 111 | # Body |
| 112 | if compressed: |
| 113 | output = gzip.decompress(response_body['body']).decode('utf8') |
| 114 | else: |
| 115 | output = response_body['body'].decode('utf8') |
| 116 | |
| 117 | self.assert_metrics(output, metric_name, help_text, increments) |
| 118 | |
| 119 | def validate_metrics(self, metric_name, help_text, increments): |
| 120 | """ |
no test coverage detected