diff options
author | Bartosz Gardziejewski <bartosz.gardziejewski@nokia.com> | 2020-08-04 14:29:10 +0200 |
---|---|---|
committer | Bartosz Gardziejewski <bartosz.gardziejewski@nokia.com> | 2020-08-12 10:59:08 +0200 |
commit | 131d4dda822b637e816ef225d8a4c9981ccf56a7 (patch) | |
tree | a4ea4f3b350276b1da45033618f7ff7b74481a00 /tests/dcaegen2/testcases/resources/robot_library/dmaap_test/test_DMaaPSQueue.py | |
parent | bbd96bd93c8d3c773aee53a69c44b3ac9cc3696d (diff) |
Refactor DMaaP simulator and add tests.
Issue-ID: DCAEGEN2-1771
Signed-off-by: Bartosz Gardziejewski <bartosz.gardziejewski@nokia.com>
Change-Id: I65772f9cdaf546171941253abdf3977b41a3e50e
Diffstat (limited to 'tests/dcaegen2/testcases/resources/robot_library/dmaap_test/test_DMaaPSQueue.py')
-rw-r--r-- | tests/dcaegen2/testcases/resources/robot_library/dmaap_test/test_DMaaPSQueue.py | 52 |
1 files changed, 52 insertions, 0 deletions
diff --git a/tests/dcaegen2/testcases/resources/robot_library/dmaap_test/test_DMaaPSQueue.py b/tests/dcaegen2/testcases/resources/robot_library/dmaap_test/test_DMaaPSQueue.py new file mode 100644 index 00000000..f278a391 --- /dev/null +++ b/tests/dcaegen2/testcases/resources/robot_library/dmaap_test/test_DMaaPSQueue.py @@ -0,0 +1,52 @@ +from Queue import Queue +import pytest +from robot_library.dmaap_simulator.DMaaPQueue import DMaaPQueue + +wait_sec_for_dequeing_event = 0.1 +test_event = "\"topic\":{\"test\":123}" + + +class TestDMaaPQueue: + + dmaap_simulator = None + + @pytest.fixture(autouse=True, scope="function") + def initiate_dmaap_simulator(self): + TestDMaaPQueue.dmaap_simulator = DMaaPQueue(Queue()) + TestDMaaPQueue.dmaap_simulator.set_deque_event_timeout(wait_sec_for_dequeing_event) + yield + + def test_when_queue_is_empty_then_deque_returns_none(self): + # when + event = TestDMaaPQueue.dmaap_simulator.deque_event() + + # then + assert event is None + + def test_when_enque_event_then_dequeue_return_same_event(self): + # when + TestDMaaPQueue.dmaap_simulator.enque_event(test_event) + event = TestDMaaPQueue.dmaap_simulator.deque_event() + + # then + assert event == test_event + + def test_when_enque_and_dequeue_event_then_deque_return_none(self): + # when + TestDMaaPQueue.dmaap_simulator.enque_event(test_event) + TestDMaaPQueue.dmaap_simulator.deque_event() + event = TestDMaaPQueue.dmaap_simulator.deque_event() + + # then + assert event is None + + def test_when_enque_few_events_and_clean_up_then_dequeu_return_none(self): + # when + TestDMaaPQueue.dmaap_simulator.enque_event(test_event) + TestDMaaPQueue.dmaap_simulator.enque_event(test_event) + TestDMaaPQueue.dmaap_simulator.enque_event(test_event) + TestDMaaPQueue.dmaap_simulator.clean_up_event() + event = TestDMaaPQueue.dmaap_simulator.deque_event() + + # then + assert event is None |