test_messenger.py (8225B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\eve\client\script\ui\services\insurgency\test\test_messenger.py 2 from testsuites.testcases import ClientTestCase 3 import mock 4 from eve.client.script.ui.services.insurgency.messenger import InsurgencyMessenger, SystemData, NoticeData 5 from eveProto.generated.eve_public.math.fraction_pb2 import Fraction 6 from eveProto.generated.eve_public.pirate.corruption.api.requests_pb2 import GetSystemInfoRequest as GetSystemCorruptionRequest, GetSystemInfoResponse as GetSystemCorruptionResponse, GetStageThresholdsRequest as GetStageThresholdsCorruptionRequest, GetStageThresholdsResponse as GetStageThresholdsCorruptionResponse 7 from eveProto.generated.eve_public.pirate.suppression.api.requests_pb2 import GetSystemInfoRequest as GetSystemSuppressionRequest, GetSystemInfoResponse as GetSystemSuppressionResponse, GetStageThresholdsRequest as GetStageThresholdsSuppressionRequest, GetStageThresholdsResponse as GetStageThresholdsSuppressionResponse 8 from publicGateway.test.public_gateway_mock import new_request_test_row, character_request_test 9 from eveProto.generated.eve_public.solarsystem.solarsystem_pb2 import Identifier as SolarSystemId 10 from eveProto.generated.eve_public.pirate.corruption.api.notices_pb2 import IncreasedNotice as CorruptionIncreasedNotice, SetNotice as CorruptionSetNotice, ResetNotice as CorruptionResetNotice 11 from eveProto.generated.eve_public.pirate.suppression.api.notices_pb2 import IncreasedNotice as SuppressionIncreasedNotice, SetNotice as SuppressionSetNotice, ResetNotice as SuppressionResetNotice 12 SOLARSYSTEM_ID = 30000001 13 14 class TestInsurgencyMessenger(ClientTestCase): 15 16 def test_get_corruption_stages(self): 17 respose = GetStageThresholdsCorruptionResponse(thresholds=[Fraction(numerator=1, denominator=2)]) 18 test_rows = [new_request_test_row(messenger_function_name='_BlockingRequestStageThreasholds', messenger_function_kwargs={'requestClass': GetStageThresholdsCorruptionRequest, 19 'responseClass': GetStageThresholdsCorruptionResponse}, expected_request_payload=GetStageThresholdsCorruptionRequest(), response_primitive_dict={'status_code': 200}, response_payload=respose, expected_results=[0.5], expected_exception=None)] 20 for test_row in test_rows: 21 character_request_test(self, InsurgencyMessenger, test_row) 22 23 def test_get_suppression_stages(self): 24 respose = GetStageThresholdsSuppressionResponse(thresholds=[Fraction(numerator=1, denominator=2)]) 25 test_rows = [new_request_test_row(messenger_function_name='_BlockingRequestStageThreasholds', messenger_function_kwargs={'requestClass': GetStageThresholdsSuppressionRequest, 26 'responseClass': GetStageThresholdsSuppressionResponse}, expected_request_payload=GetStageThresholdsSuppressionRequest(), response_primitive_dict={'status_code': 200}, response_payload=respose, expected_results=[0.5], expected_exception=None)] 27 for test_row in test_rows: 28 character_request_test(self, InsurgencyMessenger, test_row) 29 30 def test_get_corruption_state(self): 31 request = GetSystemCorruptionRequest(system=SolarSystemId(sequential=SOLARSYSTEM_ID)) 32 respose = GetSystemCorruptionResponse(total_progress=Fraction(numerator=10, denominator=20), vanguard_contribution=Fraction(numerator=30, denominator=40), stage=1) 33 expectedReturn = SystemData(SOLARSYSTEM_ID, 10, 20, 1, 30, 40) 34 test_rows = [new_request_test_row(messenger_function_name='_BlockingRequestSystemCorruption', messenger_function_kwargs={'systemID': SOLARSYSTEM_ID}, expected_request_payload=request, response_primitive_dict={'status_code': 200}, response_payload=respose, expected_results=expectedReturn, expected_exception=None)] 35 for test_row in test_rows: 36 character_request_test(self, InsurgencyMessenger, test_row) 37 38 def test_get_suppression_state(self): 39 request = GetSystemSuppressionRequest(system=SolarSystemId(sequential=SOLARSYSTEM_ID)) 40 respose = GetSystemSuppressionResponse(total_progress=Fraction(numerator=10, denominator=20), vanguard_contribution=Fraction(numerator=30, denominator=40), stage=1) 41 expectedReturn = SystemData(SOLARSYSTEM_ID, 10, 20, 1, 30, 40) 42 test_rows = [new_request_test_row(messenger_function_name='_BlockingRequestSystemSuppression', messenger_function_kwargs={'systemID': SOLARSYSTEM_ID}, expected_request_payload=request, response_primitive_dict={'status_code': 200}, response_payload=respose, expected_results=expectedReturn, expected_exception=None)] 43 for test_row in test_rows: 44 character_request_test(self, InsurgencyMessenger, test_row) 45 46 def test_suppression_subscriptions_called(self): 47 messenger = InsurgencyMessenger(mock.Mock()) 48 messenger.public_gateway.subscribe_to_notice = mock.Mock() 49 messenger._GetSuppressionIncreasedCallback = mock.Mock(return_value=1) 50 messenger._GetSetCallback = mock.Mock(return_value=2) 51 messenger._GetResetCallback = mock.Mock(return_value=3) 52 messenger.SubscribeToSuppressionValueUpdates([]) 53 messenger.public_gateway.subscribe_to_notice.assert_has_calls([mock.call(SuppressionIncreasedNotice, 1), mock.call(SuppressionSetNotice, 2), mock.call(SuppressionResetNotice, 3)], any_order=True) 54 55 def test_corruption_subscriptions_called(self): 56 messenger = InsurgencyMessenger(mock.Mock()) 57 messenger.public_gateway.subscribe_to_notice = mock.Mock() 58 messenger._GetCorruptionIncreasedCallback = mock.Mock(return_value=1) 59 messenger._GetSetCallback = mock.Mock(return_value=2) 60 messenger._GetResetCallback = mock.Mock(return_value=3) 61 messenger.SubscribeToCorruptionValueUpdates([]) 62 messenger.public_gateway.subscribe_to_notice.assert_has_calls([mock.call(CorruptionIncreasedNotice, 1), mock.call(CorruptionSetNotice, 2), mock.call(CorruptionResetNotice, 3)], any_order=True) 63 64 def test_suppression_subscriptions_callback(self): 65 messenger = InsurgencyMessenger(mock.Mock()) 66 notice_to_initalized_notice = {SuppressionIncreasedNotice: SuppressionIncreasedNotice(system=SolarSystemId(sequential=SOLARSYSTEM_ID), new_suppression=Fraction(numerator=1, denominator=5)), 67 SuppressionSetNotice: SuppressionSetNotice(system=SolarSystemId(sequential=SOLARSYSTEM_ID), progress=Fraction(numerator=1, denominator=5)), 68 SuppressionResetNotice: SuppressionResetNotice(system=SolarSystemId(sequential=SOLARSYSTEM_ID), progress=Fraction(numerator=1, denominator=5))} 69 70 def subscribe_fun(notice, callback): 71 callback(notice_to_initalized_notice[notice], notice) 72 73 messenger.public_gateway.subscribe_to_notice = mock.Mock(side_effect=subscribe_fun) 74 75 def assertion_callback(systemID, value): 76 expected = NoticeData(newNumerator=1, newDenominator=5, vanguardInstigated=False) 77 self.assertEqual(expected, value) 78 self.assertEqual(systemID, SOLARSYSTEM_ID) 79 80 messenger.SubscribeToSuppressionValueUpdates(assertion_callback) 81 82 def test_corruption_subscriptions_callback(self): 83 messenger = InsurgencyMessenger(mock.Mock()) 84 notice_to_initalized_notice = {CorruptionIncreasedNotice: CorruptionIncreasedNotice(system=SolarSystemId(sequential=SOLARSYSTEM_ID), new_corruption=Fraction(numerator=1, denominator=5), vanguard=False), 85 CorruptionSetNotice: CorruptionSetNotice(system=SolarSystemId(sequential=SOLARSYSTEM_ID), progress=Fraction(numerator=1, denominator=5)), 86 CorruptionResetNotice: CorruptionResetNotice(system=SolarSystemId(sequential=SOLARSYSTEM_ID), progress=Fraction(numerator=1, denominator=5))} 87 88 def subscribe_fun(notice, callback): 89 callback(notice_to_initalized_notice[notice], notice) 90 91 messenger.public_gateway.subscribe_to_notice = mock.Mock(side_effect=subscribe_fun) 92 93 def assertion_callback(systemID, value): 94 expected = NoticeData(newNumerator=1, newDenominator=5, vanguardInstigated=False) 95 self.assertEqual(expected, value) 96 self.assertEqual(systemID, SOLARSYSTEM_ID) 97 98 messenger.SubscribeToCorruptionValueUpdates(assertion_callback)