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

request_messenger_mock.py (3506B)

      1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\packages\sovereignty\mercenaryden\client\mocks\request_messenger_mock.py
      2 import uuid
      3 import datetime
      4 from collections import namedtuple
      5 from eveProto.generated.eve_public.sovereignty.mercenaryden.api.requests_pb2 import GetAsOwnerRequest
      6 from eveProto.generated.eve_public.sovereignty.mercenaryden.mercenaryden_pb2 import Identifier as MercenaryDenIdentifier
      7 from six.moves.http_client import OK, NOT_FOUND
      8 from logging import getLogger
      9 from sovereignty.mercenaryden.client.mocks.data import create_mercenary_den_mock
     10 from sovereignty.mercenaryden.client.data.activity import MercenaryDenActivity
     11 from sovereignty.mercenaryden.common.errors import GenericError, UnknownMercenaryDen
     12 logger = getLogger('mercenary_den')
     13 
     14 class MockMercenaryDenRequestMessenger(object):
     15 
     16     def get_mercenary_den_request(self, item_id):
     17         response = OK
     18         request = GetAsOwnerRequest(id=MercenaryDenIdentifier(sequential=item_id))
     19         info_log = 'Request: Get Mercenary Den {item_id}: {request}\n'.format(item_id=item_id, request=request)
     20         try:
     21             if response == OK:
     22                 info_log += 'Response (MOCKED): Get Mercenary Den {item_id}: ({status_code}) {response}'.format(item_id=item_id, status_code=OK, response=response)
     23                 return create_mercenary_den_mock(item_id, session.charid, session.solarsystemid2, session.corpid)
     24             if response == NOT_FOUND:
     25                 raise UnknownMercenaryDen('(MOCKED) Mercenary Den does not exist or caller is not owner and therefore has no visibility')
     26             else:
     27                 raise GenericError('(MOCKED) Unspecified error when fetching data for Mercenary Den: {code}'.format(code=response))
     28         finally:
     29             logger.info(info_log)
     30 
     31     def get_all_activities_for_den_request(self, item_id):
     32         activities = [MercenaryDenActivity(activity_id=uuid.uuid4(), den_id=item_id, dungeon_id=12355, name_id=871696, description_id=871697, started=False, expiry=datetime.datetime.utcnow() + datetime.timedelta(minutes=60), solar_system_id=30000208, development_impact=10, anarchy_impact=-10, infomorph_bonus=100)]
     33         next_generation = datetime.datetime.utcnow() + datetime.timedelta(days=1, hours=1, minutes=15)
     34         return (activities, next_generation)
     35 
     36     def _get_all_activities_request(self):
     37         if not hasattr(self, '_activities'):
     38             self._activities = [MercenaryDenActivity(activity_id=uuid.uuid4(), den_id=1000000013115L, dungeon_id=12355, name_id=871696, description_id=871697, started=False, expiry=datetime.datetime.utcnow() + datetime.timedelta(minutes=60), solar_system_id=30000208, development_impact=10, anarchy_impact=-10, infomorph_bonus=100)]
     39         return self._activities
     40 
     41     def get_all_activities_request_without_retries(self):
     42         return self._get_all_activities_request()
     43 
     44     def get_all_activities_request_with_retries(self):
     45         return self._get_all_activities_request()
     46 
     47     def start_activity_request(self, activity_id):
     48         logger.info('(Mocked) start_activity_request')
     49         result = namedtuple('Result', ['is_started'])
     50         return result(is_started=True)
     51 
     52     def get_activity_capacity_request(self):
     53         pass
     54 
     55     def get_all_owned_mercenary_dens_request(self):
     56         return [1000000270300L]
     57 
     58     def get_maximum_dens_info_dens_request(self):
     59         current_maximum = 5
     60         absolute_maximum = 5
     61         return (current_maximum, absolute_maximum)