NovaBryan/EVE_Dump
Log | Files | 📦 下载最新源码 (.zip) | Refs | README

mock_gateway.py (2034B)

      1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\communitytracks\client\mock\mock_gateway.py
      2 from six.moves import http_client as httplib
      3 from communitytracks.client.data.definition_identifier import DefinitionIdentifier
      4 from communitytracks.client.data.instance_attributes import InstanceAttributes
      5 from communitytracks.client.data.instance_identifier import InstanceIdentifier
      6 from communitytracks.common.const import VANGUARD_NEMESIS_DEFINITION_ID, VANGUARD_NEMESIS_INSTANCE_ID, LOGGER_NAME
      7 from logging import getLogger
      8 from eveProto.generated.eve_public.communitytrack.api.requests_pb2 import GetRequest, GetResponse
      9 from mock import Mock
     10 logger = getLogger(LOGGER_NAME)
     11 
     12 class MockCommunityTracksGateway(object):
     13 
     14     def __init__(self):
     15         pass
     16 
     17     def send_blocking_character_request_and_receive_response(self, request_payload, expected_response_class, timeout_seconds, max_attempts, retry_delay_in_seconds):
     18         if isinstance(request_payload, GetRequest):
     19             status_code = httplib.OK
     20             instance_id = InstanceIdentifier.from_proto(request_payload.instance)
     21             if instance_id.sequential == VANGUARD_NEMESIS_INSTANCE_ID:
     22                 instance_attributes = InstanceAttributes(definition_identifier=DefinitionIdentifier(fsd_identifier=VANGUARD_NEMESIS_DEFINITION_ID), progress=12)
     23             else:
     24                 instance_attributes = None
     25             response_primitive = Mock()
     26             response_primitive.status_code = status_code
     27             if status_code == httplib.OK and instance_attributes:
     28                 response_payload = GetResponse(attributes=InstanceAttributes.to_proto(instance_attributes))
     29             else:
     30                 response_payload = GetResponse()
     31             return (response_primitive, response_payload)
     32         raise NotImplementedError('Attempted to use MockCommunityTracksGateway for request that has not been mocked: %s' % request_payload)
     33 
     34     def subscribe_to_notice(self, notice_payload, callback):
     35         pass