(self, event_loop, fake_event_svc, adversary)
| 382 | mock_emit_state_change_method.assert_called_with(from_state='running', to_state='finished') |
| 383 | |
| 384 | def test_emit_state_change_event(self, event_loop, fake_event_svc, adversary): |
| 385 | op = Operation(name='test', agents=[], adversary=adversary, state='running') |
| 386 | fake_event_svc.reset() |
| 387 | |
| 388 | event_loop.run_until_complete( |
| 389 | op._emit_state_change_event( |
| 390 | from_state='running', |
| 391 | to_state='finished' |
| 392 | ) |
| 393 | ) |
| 394 | |
| 395 | expected_key = (Operation.EVENT_EXCHANGE, Operation.EVENT_QUEUE_STATE_CHANGED) |
| 396 | assert expected_key in fake_event_svc.fired |
| 397 | |
| 398 | event_kwargs = fake_event_svc.fired[expected_key] |
| 399 | assert event_kwargs['op'] == op.id |
| 400 | assert event_kwargs['from_state'] == 'running' |
| 401 | assert event_kwargs['to_state'] == 'finished' |
| 402 | |
| 403 | def test_with_learning_parser(self, event_loop, app_svc, file_svc, contact_svc, data_svc, learning_svc, event_svc, op_with_learning_parser, |
| 404 | make_test_link, make_test_result, knowledge_svc, fire_event_mock): |
nothing calls this directly
no test coverage detected