DockerSensorTestCase.test_poll()

Complexity

Conditions

1

Size

Total Lines

26

Duplication

Lines	0
Ratio	0 %

import mock
    def test_poll(self):
        # TODO: Move to base test class
        sensor = DockerSensor(sensor_service=self.sensor_service)

        # No existing and no running containers (e.g. after initial sensor poll)
        sensor._running_containers = {}
        sensor._get_active_containers = mock.Mock()
        sensor._get_active_containers.return_value = {}

        sensor.poll()
        self.assertEqual(self.get_dispatched_triggers(), [])

        # One container started
        sensor._get_active_containers.return_value = {'1': MOCK_CONTAINER_DATA}

        sensor.poll()
        self.assertEqual(len(self.get_dispatched_triggers()), 1)
        self.assertTriggerDispatched(trigger='docker.container_tracker.started',
                                     payload={'container_info': MOCK_CONTAINER_DATA})

        # One container stopped
        sensor._get_active_containers.return_value = {}
        sensor.poll()
        self.assertEqual(len(self.get_dispatched_triggers()), 2)
        self.assertTriggerDispatched(trigger='docker.container_tracker.stopped',
                                     payload={'container_info': MOCK_CONTAINER_DATA})