ya2/utils/functional.py

   1 from os import getcwd, makedirs
   2 from logging import info
   3 from shutil import rmtree
   4 from os.path import join, exists
   5 from pathlib import Path
   6 from xmlrpc.server import SimpleXMLRPCServer
   7 from threading import Thread
   8 from collections import namedtuple
   9 from panda3d.core import Filename
  10 from direct.gui.OnscreenText import OnscreenText
  11 from ya2.build.build import _branch
  12
  13
  14 class RPCServer(SimpleXMLRPCServer):
  15
  16     def __init__(self, callbacks):
  17         super().__init__(('localhost', 7000), allow_none=True)
  18         self.__callbacks = callbacks
  19         self.register_introspection_functions()
  20         self.register_function(self.__screenshot, 'screenshot')
  21         self.register_function(self.__enforce_result, 'enforce_result')
  22         self.register_function(self.__verify, 'verify')
  23         self.register_function(self.__set_index, 'set_index')
  24         self.register_function(self.__enforce_resolution, 'enforce_resolution')
  25         self.register_function(self.get_position, 'get_position')
  26         self.register_function(self.destroy, 'destroy')
  27
  28     def __screenshot(self, argument):
  29         # refactor: these methods should not depend on taskMgr
  30         # since i can't write unit tests for them
  31         taskMgr.do_method_later(.01, self.__callbacks.screenshot, 'rpc_screenshot', [argument])
  32
  33     def __enforce_result(self, argument):
  34         taskMgr.do_method_later(.01, self.__callbacks.enforce_result, 'rpc_enforce_result', [argument])
  35
  36     def __verify(self):
  37         taskMgr.do_method_later(.01, self.__callbacks.verify, 'rpc_verify')
  38
  39     def __set_index(self, argument):
  40         taskMgr.do_method_later(.01, self.__callbacks.set_index, 'rpc_set_index', [argument])
  41
  42     def __enforce_resolution(self, argument):
  43         taskMgr.do_method_later(.01, self.__callbacks.enforce_resolution, 'rpc_enforce_resolution', [argument])
  44
  45     def get_position(self, arg):
  46         return self.__callbacks.get_position(arg)
  47
  48     def destroy(self):
  49         self._BaseServer__shutdown_request = True
  50
  51
  52 class RPCServerThread(Thread):
  53
  54     def __init__(self, callbacks):
  55         Thread.__init__(self, name='rpc_server')
  56         self.__callbacks = callbacks
  57
  58     def run(self):
  59         self.__server = RPCServer(self.__callbacks)
  60         self.__server.serve_forever()
  61
  62
  63 class FunctionalTest:
  64
  65     def __init__(self, creating_references, position_manager, app_name):
  66         self.__creating_references = creating_references
  67         self.__position_manager = position_manager
  68         self.__app_name = app_name
  69         Callbacks = namedtuple('Callbacks', 'screenshot enforce_result verify set_index enforce_resolution get_position')
  70         callbacks = Callbacks(self.__screenshot,
  71                               self.__enforce_result,
  72                               self.__verify,
  73                               self.__set_index,
  74                               self.__enforce_resolution,
  75                               self.__get_position)
  76         RPCServerThread(callbacks).start()
  77         self.__debug_text = OnscreenText('', fg=(1, 0, 0, 1), scale=.16)
  78         self.__debug_text.name = 'test_txt'
  79         self.__set_path()
  80         self.__file_names = []
  81
  82     def __set_path(self):
  83         self.__set_path_base_case()
  84         self.__set_path_wine()
  85         self.__set_path_references()
  86
  87     def __set_path_base_case(self):
  88         self.__path = str(Filename().get_user_appdata_directory())
  89         self.__path += f'/{self.__app_name}/'
  90         self.__path += f'tests/functional{"_ref" if self.__creating_references else ""}/'
  91
  92     def __set_path_wine(self):
  93         home = Path.home()
  94         if self.__path.startswith('/c/users/') and exists(home + '/.wine/'):
  95             self.__path = home + '/.wine/drive_' + self.__path[1:]
  96
  97     def __set_path_references(self):
  98         if self.__creating_references:
  99             self.__path = join(
 100                 Filename().get_user_appdata_directory(),
 101                 f'{self.__app_name}/tests/functional_ref_%s/' % _branch())
 102
 103     def __set_index(self, idx):
 104         if int(idx) == 1: rmtree(self.__path, ignore_errors=True)
 105         info(f'creating dir: {self.__path}')
 106         makedirs(self.__path, exist_ok=True)
 107
 108     def __get_position(self, target):
 109         return self.__position_manager[target]
 110
 111     def __screenshot(self, name):
 112         self.__file_names += [self.__path + name]
 113         r = base.screenshot(self.__path + name, False)
 114         info(f'screenshot {self.__path + name} ({r}; {getcwd()})')
 115
 116     def __enforce_result(self, result):
 117         info(f'enforce_result {result}')
 118         messenger.send('enforce_result', [result])
 119
 120     def __enforce_resolution(self, resolution):
 121         info(f'enforce resolution {resolution} (callback)')
 122         messenger.send('enforce_resolution', [resolution])
 123
 124     def __verify(self, task):
 125         for f in self.__file_names:
 126             info(f'verifying {f}')
 127             assert exists(f)