(self, outputs, metric_name, help_text, increments, compressed)
| 25 | c.inc() |
| 26 | |
| 27 | def assert_outputs(self, outputs, metric_name, help_text, increments, compressed): |
| 28 | self.assertEqual(len(outputs), 1) |
| 29 | if compressed: |
| 30 | output = gzip.decompress(outputs[0]).decode(encoding="utf-8") |
| 31 | else: |
| 32 | output = outputs[0].decode('utf8') |
| 33 | # Status code |
| 34 | self.assertEqual(self.captured_status, "200 OK") |
| 35 | # Headers |
| 36 | num_of_headers = 2 if compressed else 1 |
| 37 | self.assertEqual(len(self.captured_headers), num_of_headers) |
| 38 | self.assertIn(("Content-Type", CONTENT_TYPE_PLAIN_0_0_4), self.captured_headers) |
| 39 | if compressed: |
| 40 | self.assertIn(("Content-Encoding", "gzip"), self.captured_headers) |
| 41 | # Body |
| 42 | self.assertIn("# HELP " + metric_name + "_total " + help_text + "\n", output) |
| 43 | self.assertIn("# TYPE " + metric_name + "_total counter\n", output) |
| 44 | self.assertIn(metric_name + "_total " + str(increments) + ".0\n", output) |
| 45 | |
| 46 | def validate_metrics(self, metric_name, help_text, increments): |
| 47 | """ |
no outgoing calls
no test coverage detected