(self)
| 351 | self.assertEqual(self.requests[0][1], b'') |
| 352 | |
| 353 | def test_push_with_handler(self): |
| 354 | def my_test_handler(url, method, timeout, headers, data): |
| 355 | headers.append(['X-Test-Header', 'foobar']) |
| 356 | # Handler should be passed sane default timeout |
| 357 | self.assertEqual(timeout, 30) |
| 358 | return default_handler(url, method, timeout, headers, data) |
| 359 | |
| 360 | push_to_gateway(self.address, "my_job", self.registry, handler=my_test_handler) |
| 361 | self.assertEqual(self.requests[0][0].command, 'PUT') |
| 362 | self.assertEqual(self.requests[0][0].path, '/metrics/job/my_job') |
| 363 | self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_PLAIN_0_0_4) |
| 364 | self.assertEqual(self.requests[0][0].headers.get('x-test-header'), 'foobar') |
| 365 | self.assertEqual(self.requests[0][1], b'# HELP g help\n# TYPE g gauge\ng 0.0\n') |
| 366 | |
| 367 | def test_push_with_basic_auth_handler(self): |
| 368 | def my_auth_handler(url, method, timeout, headers, data): |
nothing calls this directly
no test coverage detected