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

qaToolsForPublicGateway.py (5020B)

      1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\eve\client\script\sys\qaToolsForPublicGateway.py
      2 from uthread2 import Sleep
      3 
      4 class MockResponsePrimitive(object):
      5 
      6     def __init__(self, status_code):
      7         self.status_code = status_code
      8         self.status_message = 'Error: %s' % status_code
      9 
     10 
     11 class MockResponseChannel(object):
     12 
     13     def __init__(self, status_code):
     14         self.status_code = status_code
     15 
     16     def receive(self):
     17         response_primitive = MockResponsePrimitive(self.status_code)
     18         response_payload = None
     19         return (response_primitive, response_payload)
     20 
     21 
     22 class PublicGatewayRequestHijacker(object):
     23     SETTING_PUBLIC_REQUEST_HIJACKER_ENABLED = 'public_request_hijacker_enabled'
     24     SETTING_PUBLIC_REQUEST_HIJACKER_ERROR = 'public_request_hijacker_error'
     25     SETTING_PUBLIC_REQUEST_HIJACKER_LATENCY = 'public_request_hijacker_latency'
     26 
     27     def __init__(self, public_gateway):
     28         self.public_gateway = public_gateway
     29         self._original_user_sender = public_gateway.send_user_request
     30         self._original_character_sender = public_gateway.send_character_request
     31         self._original_user_blocking_sender = public_gateway.send_blocking_user_request_and_receive_response
     32         self._original_character_blocking_sender = public_gateway.send_blocking_character_request_and_receive_response
     33         self._is_enabled = None
     34         self._error = None
     35         self._latency = None
     36 
     37     def load_data(self):
     38         self._load_data_from_settings()
     39         self._apply()
     40 
     41     def get_data(self):
     42         return (self._is_enabled, self._error, self._latency)
     43 
     44     def enable(self, error, latency):
     45         self._is_enabled = True
     46         self._error = error
     47         self._latency = latency
     48         self._save_data_and_apply()
     49 
     50     def disable(self):
     51         self._is_enabled = False
     52         self._error = None
     53         self._latency = None
     54         self._save_data_and_apply()
     55 
     56     def _apply(self):
     57         if self._is_enabled:
     58             self._hijack_methods()
     59         else:
     60             self._revert_methods_to_original()
     61 
     62     def _load_data_from_settings(self):
     63         self._is_enabled = settings.user.ui.Get(self.SETTING_PUBLIC_REQUEST_HIJACKER_ENABLED, False)
     64         self._error = settings.user.ui.Get(self.SETTING_PUBLIC_REQUEST_HIJACKER_ERROR, None)
     65         self._latency = settings.user.ui.Get(self.SETTING_PUBLIC_REQUEST_HIJACKER_LATENCY, None)
     66 
     67     def _save_data_and_apply(self):
     68         self._save_data_in_settings()
     69         self._apply()
     70 
     71     def _save_data_in_settings(self):
     72         settings.user.ui.Set(self.SETTING_PUBLIC_REQUEST_HIJACKER_ENABLED, self._is_enabled)
     73         settings.user.ui.Set(self.SETTING_PUBLIC_REQUEST_HIJACKER_ERROR, self._error)
     74         settings.user.ui.Set(self.SETTING_PUBLIC_REQUEST_HIJACKER_LATENCY, self._latency)
     75         settings.user.WriteToDiskImmediate()
     76 
     77     def _revert_methods_to_original(self):
     78         self.public_gateway.send_user_request = self._original_user_sender
     79         self.public_gateway.send_character_request = self._original_character_sender
     80         self.public_gateway.send_blocking_user_request_and_receive_response = self._original_user_blocking_sender
     81         self.public_gateway.send_blocking_character_request_and_receive_response = self._original_character_blocking_sender
     82 
     83     def _hijack_methods(self):
     84         self.public_gateway.send_user_request = lambda *args, **kwargs: self._hijacked_method(self._error, self._latency, self._original_user_sender, *args, **kwargs)
     85         self.public_gateway.send_character_request = lambda *args, **kwargs: self._hijacked_method(self._error, self._latency, self._original_character_sender, *args, **kwargs)
     86         self.public_gateway.send_blocking_user_request_and_receive_response = lambda *args, **kwargs: self._hijacked_blocking_method(self._error, self._latency, self._original_user_blocking_sender, *args, **kwargs)
     87         self.public_gateway.send_blocking_character_request_and_receive_response = lambda *args, **kwargs: self._hijacked_blocking_method(self._error, self._latency, self._original_character_blocking_sender, *args, **kwargs)
     88 
     89     def _hijacked_method(self, error, latency, original_method, *args, **kwargs):
     90         if latency:
     91             Sleep(latency)
     92         if not error:
     93             return original_method(*args, **kwargs)
     94         if isinstance(error, type) and issubclass(error, Exception):
     95             raise error()
     96         request_primitive = None
     97         response_channel = MockResponseChannel(error)
     98         return (request_primitive, response_channel)
     99 
    100     def _hijacked_blocking_method(self, error, latency, original_method, *args, **kwargs):
    101         if latency:
    102             Sleep(latency)
    103         if not error:
    104             return original_method(*args, **kwargs)
    105         if isinstance(error, type) and issubclass(error, Exception):
    106             raise error()
    107         response_primitive = MockResponsePrimitive(error)
    108         response_payload = None
    109         return (response_primitive, response_payload)