ya2/utils/functional.py

   1 from os import getcwd
   2 from logging import info
   3 from shutil import rmtree
   4 from os import makedirs
   5 from os.path import join, exists
   6 from xmlrpc.server import SimpleXMLRPCServer
   7 from threading import Thread
   8 from panda3d.core import Filename
   9 from direct.gui.OnscreenText import OnscreenText
  10 from ya2.build.build import _branch
  11
  12
  13 class RPCServer(SimpleXMLRPCServer):
  14
  15     def __init__(self, callbacks):
  16         super().__init__(('localhost', 6000), allow_none=True)
  17         self._callbacks = callbacks
  18         self.register_introspection_functions()
  19         self.register_function(self.screenshot, 'screenshot')
  20         self.register_function(self.enforce_res, 'enforce_res')
  21         self.register_function(self.verify, 'verify')
  22         self.register_function(self.set_idx, 'set_idx')
  23         self.register_function(self.enforce_resolution, 'enforce_resolution')
  24         self.register_function(self.get_pos, 'get_pos')
  25         self.register_function(self.destroy, 'destroy')
  26
  27     def screenshot(self, arg):
  28         taskMgr.doMethodLater(.01, self._callbacks[0], 'cb0', [arg])
  29
  30     def enforce_res(self, arg):
  31         taskMgr.doMethodLater(.01, self._callbacks[1], 'cb1', [arg])
  32
  33     def verify(self):
  34         taskMgr.doMethodLater(.01, self._callbacks[2], 'cb2')
  35
  36     def set_idx(self, arg):
  37         taskMgr.doMethodLater(.01, self._callbacks[3], 'cb3', [arg])
  38
  39     def enforce_resolution(self, arg):
  40         taskMgr.doMethodLater(.01, self._callbacks[4], 'cb4', [arg])
  41
  42     def get_pos(self, arg):
  43         return self._callbacks[5](arg)
  44
  45     def destroy(self):
  46         self._BaseServer__shutdown_request = True
  47
  48
  49 class RPCServerThread(Thread):
  50
  51     def __init__(self, callbacks):
  52         Thread.__init__(self)
  53         self._callbacks = callbacks
  54
  55     def run(self):
  56         self.server = RPCServer(self._callbacks)
  57         self.server.serve_forever()
  58
  59
  60 class FunctionalTest:
  61
  62     def __init__(self, ref, pos_mgr):
  63         super().__init__()
  64         self._pos_mgr = pos_mgr
  65         RPCServerThread([self._do_screenshot, self._do_enforce_res,
  66                          self.__verify, self._set_idx,
  67                          self._do_enforce_resolution, self.__get_pos]).start()
  68         self.txt = OnscreenText('', fg=(1, 0, 0, 1), scale=.16)
  69         # self._path = ''
  70         # if self.eng.is_appimage:
  71         self._path = str(Filename().get_user_appdata_directory())
  72         self._path += '/pmachines/'
  73         self._path += 'tests/functional%s/' % ('_ref' if ref else '')
  74         home = '/home/flavio'  # we must force this for wine
  75         # if self._path.startswith('/c/users/') and exists(str(Path.home()) +
  76         #     '/.local/share/flatpak-wine601/default/'):
  77         #     self._path = str(Path.home()) +
  78         #       '/.local/share/flatpak-wine601/default/drive_' + self._path[1:]
  79         if self._path.startswith('/c/users/') and exists(home + '/.wine/'):
  80             self._path = home + '/.wine/drive_' + self._path[1:]
  81         if ref:
  82             self._path = join(
  83                 Filename().get_user_appdata_directory(),
  84                 'pmachines/tests/functional_ref_%s/' % _branch())
  85         self._fnames = []
  86         # taskMgr.add(self.on_frame_unpausable, 'on-frame-unpausable')
  87         # self._do_screenshots(idx)
  88
  89     def _set_idx(self, idx):
  90         if int(idx) == 1:
  91             rmtree(self._path, ignore_errors=True)
  92         info('creating dir: %s' % self._path)
  93         makedirs(self._path, exist_ok=True)
  94
  95     def __get_pos(self, tgt):
  96         return self._pos_mgr.get(tgt)
  97
  98     def _do_screenshot(self, name):
  99         self._fnames += [self._path + name]
 100         # time = datetime.datetime.now().strftime('%y%m%d%H%M%S')
 101         # res = base.win.save_screenshot(
 102         #     Filename(path or ("yocto%s.png" % time)))
 103         # debug('screenshot %s (%s)' % (path or ("yocto%s.png" % time), res))
 104         res = base.screenshot(self._path + name, False)
 105         info('screenshot %s (%s; %s)' % (self._path + name, res, getcwd()))
 106
 107     def _do_enforce_res(self, res):
 108         info('enforce_res %s' % res)
 109         messenger.send('enforce_res', [res])
 110
 111     def _do_enforce_resolution(self, res):
 112         info('enforce resolution %s (callback)' % res)
 113         messenger.send('enforce_resolution', [res])
 114
 115     def __verify(self, task):
 116         # files = glob(self._path + '*')
 117         for fname in self._fnames:
 118             info('verifying %s' % fname)
 119             assert exists(fname)