test_recommendationMessenger.py (1761B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\eve\client\script\ui\shared\recommendation\message_bus\tests\test_recommendationMessenger.py 2 import mock 3 import time 4 import unittest 5 import uuid 6 from eve.client.script.ui.shared.recommendation.message_bus.recommendationMessenger import RecommendationMessenger 7 from testsuites.testcases import ClientTestCase 8 from publicGateway.grpc.test import _safe_tasklet_sleep 9 from publicGateway.grpc.test.mock_publicGatewaySvc import MockPublicGatewaySvc 10 has_stackless = False 11 try: 12 import stackless 13 has_stackless = True 14 except ImportError: 15 has_stackless = False 16 17 class TestHelpPointerMessengerCompatibility(ClientTestCase): 18 19 def setUp(self): 20 super(TestHelpPointerMessengerCompatibility, self).setUp() 21 self.mb = RecommendationMessenger(mock.MagicMock()) 22 self.uuid = uuid.uuid4().get_bytes() 23 24 def test_accepted(self): 25 operation_id = 1 26 self.mb.accepted(operation_id, self.uuid) 27 28 def test_dismissed(self): 29 operation_id = 1 30 self.mb.dismissed(operation_id, self.uuid) 31 32 def test_displayed(self): 33 operation_id = 1 34 self.mb.displayed(operation_id, self.uuid) 35 36 @unittest.skip('test fails and needs fixing') 37 def test_request_recommendations(self): 38 if not has_stackless: 39 self.skipTest('there can be no stackless router without stackless') 40 mock_add_func = mock.MagicMock() 41 self.mb = RecommendationMessenger(MockPublicGatewaySvc()) 42 self.mb.request_recommendations(1, True, mock_add_func, mock.MagicMock()) 43 start = time.time() 44 while not mock_add_func.called: 45 _safe_tasklet_sleep(start, 2) 46 47 mock_add_func.assert_called_once()