abstract test positions in world units

[pmachines.git] / ya2 / utils / functional.py

from os import getcwd
from logging import info
from shutil import rmtree
from os import makedirs
from os.path import join, exists
from xmlrpc.server import SimpleXMLRPCServer
from threading import Thread
from panda3d.core import Filename
from direct.gui.OnscreenText import OnscreenText
from ya2.build.build import _branch


class RPCServer(SimpleXMLRPCServer):

    def __init__(self, callbacks):
        super().__init__(('localhost', 6000), allow_none=True)
        self._callbacks = callbacks
        self.register_introspection_functions()
        self.register_function(self.screenshot, 'screenshot')
        self.register_function(self.enforce_res, 'enforce_res')
        self.register_function(self.verify, 'verify')
        self.register_function(self.set_idx, 'set_idx')
        self.register_function(self.enforce_resolution, 'enforce_resolution')
        self.register_function(self.get_pos, 'get_pos')
        self.register_function(self.destroy, 'destroy')

    def screenshot(self, arg):
        taskMgr.doMethodLater(.01, self._callbacks[0], 'cb0', [arg])

    def enforce_res(self, arg):
        taskMgr.doMethodLater(.01, self._callbacks[1], 'cb1', [arg])

    def verify(self):
        taskMgr.doMethodLater(.01, self._callbacks[2], 'cb2')

    def set_idx(self, arg):
        taskMgr.doMethodLater(.01, self._callbacks[3], 'cb3', [arg])

    def enforce_resolution(self, arg):
        taskMgr.doMethodLater(.01, self._callbacks[4], 'cb4', [arg])

    def get_pos(self, arg):
        return self._callbacks[5](arg)

    def destroy(self):
        self._BaseServer__shutdown_request = True


class RPCServerThread(Thread):

    def __init__(self, callbacks):
        Thread.__init__(self, name='rpc_server')
        self._callbacks = callbacks

    def run(self):
        self.server = RPCServer(self._callbacks)
        self.server.serve_forever()


class FunctionalTest:

    def __init__(self, ref, pos_mgr):
        super().__init__()
        self._pos_mgr = pos_mgr
        RPCServerThread([self._do_screenshot, self._do_enforce_res,
                         self.__verify, self._set_idx,
                         self._do_enforce_resolution, self.__get_pos]).start()
        self.txt = OnscreenText('', fg=(1, 0, 0, 1), scale=.16)
        self.txt.name = 'test_txt'
        # self._path = ''
        # if self.eng.is_appimage:
        self._path = str(Filename().get_user_appdata_directory())
        self._path += '/pmachines/'
        self._path += 'tests/functional%s/' % ('_ref' if ref else '')
        home = '/home/flavio'  # we must force this for wine
        # if self._path.startswith('/c/users/') and exists(str(Path.home()) +
        #     '/.local/share/flatpak-wine601/default/'):
        #     self._path = str(Path.home()) +
        #       '/.local/share/flatpak-wine601/default/drive_' + self._path[1:]
        if self._path.startswith('/c/users/') and exists(home + '/.wine/'):
            self._path = home + '/.wine/drive_' + self._path[1:]
        if ref:
            self._path = join(
                Filename().get_user_appdata_directory(),
                'pmachines/tests/functional_ref_%s/' % _branch())
        self._fnames = []
        # taskMgr.add(self.on_frame_unpausable, 'on-frame-unpausable')
        # self._do_screenshots(idx)

    def _set_idx(self, idx):
        if int(idx) == 1:
            rmtree(self._path, ignore_errors=True)
        info('creating dir: %s' % self._path)
        makedirs(self._path, exist_ok=True)

    def __get_pos(self, tgt):
        return self._pos_mgr.get(tgt)

    def _do_screenshot(self, name):
        self._fnames += [self._path + name]
        # time = datetime.datetime.now().strftime('%y%m%d%H%M%S')
        # res = base.win.save_screenshot(
        #     Filename(path or ("yocto%s.png" % time)))
        # debug('screenshot %s (%s)' % (path or ("yocto%s.png" % time), res))
        res = base.screenshot(self._path + name, False)
        info('screenshot %s (%s; %s)' % (self._path + name, res, getcwd()))

    def _do_enforce_res(self, res):
        info('enforce_res %s' % res)
        messenger.send('enforce_res', [res])

    def _do_enforce_resolution(self, res):
        info('enforce resolution %s (callback)' % res)
        messenger.send('enforce_resolution', [res])

    def __verify(self, task):
        # files = glob(self._path + '*')
        for fname in self._fnames:
            info('verifying %s' % fname)
            assert exists(fname)