(self)
| 19 | return self.app |
| 20 | |
| 21 | def test_metrics(self): |
| 22 | state = EventsState() |
| 23 | worker_name = 'worker1' |
| 24 | task_name = 'task1' |
| 25 | state.get_or_create_worker(worker_name) |
| 26 | events = [ |
| 27 | Event('worker-online', hostname=worker_name), Event('worker-heartbeat', hostname=worker_name, active=1) |
| 28 | ] |
| 29 | events += task_succeeded_events(worker=worker_name, name=task_name, id='123') |
| 30 | |
| 31 | for i, e in enumerate(events): |
| 32 | e['clock'] = i |
| 33 | e['local_received'] = time.time() |
| 34 | state.event(e) |
| 35 | self.app.events.state = state |
| 36 | |
| 37 | metrics = self.get('/metrics').body.decode('utf-8') |
| 38 | events = dict(re.findall('flower_events_total{task="task1",type="(task-.*)",worker="worker1"} (.*)', metrics)) |
| 39 | |
| 40 | self.assertTrue('task-received' in events) |
| 41 | self.assertTrue('task-started' in events) |
| 42 | self.assertTrue('task-succeeded' in events) |
| 43 | |
| 44 | self.assertTrue(f'flower_worker_online{{worker="{worker_name}"}} 1.0' in metrics) |
| 45 | self.assertTrue(f'flower_worker_number_of_currently_executing_tasks{{worker="{worker_name}"}} 1.0' in metrics) |
| 46 | |
| 47 | def test_task_prefetch_time_metric(self): |
| 48 | state = EventsState() |
nothing calls this directly
no test coverage detected