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
13 class RPCServer(SimpleXMLRPCServer
):
15 def __init__(self
, callbacks
):
16 super().__init
__(('localhost', 7000), 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')
27 def screenshot(self
, arg
):
28 taskMgr
.doMethodLater(.01, self
._callbacks
[0], 'cb0', [arg
])
30 def enforce_res(self
, arg
):
31 taskMgr
.doMethodLater(.01, self
._callbacks
[1], 'cb1', [arg
])
34 taskMgr
.doMethodLater(.01, self
._callbacks
[2], 'cb2')
36 def set_idx(self
, arg
):
37 taskMgr
.doMethodLater(.01, self
._callbacks
[3], 'cb3', [arg
])
39 def enforce_resolution(self
, arg
):
40 taskMgr
.doMethodLater(.01, self
._callbacks
[4], 'cb4', [arg
])
42 def get_pos(self
, arg
):
43 return self
._callbacks
[5](arg
)
46 self
._BaseServer
__shutdown
_request
= True
49 class RPCServerThread(Thread
):
51 def __init__(self
, callbacks
):
52 Thread
.__init
__(self
, name
='rpc_server')
53 self
._callbacks
= callbacks
56 self
.server
= RPCServer(self
._callbacks
)
57 self
.server
.serve_forever()
62 def __init__(self
, ref
, pos_mgr
):
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
.txt
.name
= 'test_txt'
71 # if self.eng.is_appimage:
72 self
._path
= str(Filename().get_user_appdata_directory())
73 self
._path
+= '/pmachines/'
74 self
._path
+= 'tests/functional%s/' % ('_ref' if ref
else '')
75 home
= '/home/flavio' # we must force this for wine
76 # if self._path.startswith('/c/users/') and exists(str(Path.home()) +
77 # '/.local/share/flatpak-wine601/default/'):
78 # self._path = str(Path.home()) +
79 # '/.local/share/flatpak-wine601/default/drive_' + self._path[1:]
80 if self
._path
.startswith('/c/users/') and exists(home
+ '/.wine/'):
81 self
._path
= home
+ '/.wine/drive_' + self
._path
[1:]
84 Filename().get_user_appdata_directory(),
85 'pmachines/tests/functional_ref_%s/' % _branch())
87 # taskMgr.add(self.on_frame_unpausable, 'on-frame-unpausable')
88 # self._do_screenshots(idx)
90 def _set_idx(self
, idx
):
92 rmtree(self
._path
, ignore_errors
=True)
93 info('creating dir: %s' % self
._path
)
94 makedirs(self
._path
, exist_ok
=True)
96 def __get_pos(self
, tgt
):
97 return self
._pos
_mgr
.get(tgt
)
99 def _do_screenshot(self
, name
):
100 self
._fnames
+= [self
._path
+ name
]
101 # time = datetime.datetime.now().strftime('%y%m%d%H%M%S')
102 # res = base.win.save_screenshot(
103 # Filename(path or ("yocto%s.png" % time)))
104 # debug('screenshot %s (%s)' % (path or ("yocto%s.png" % time), res))
105 res
= base
.screenshot(self
._path
+ name
, False)
106 info('screenshot %s (%s; %s)' % (self
._path
+ name
, res
, getcwd()))
108 def _do_enforce_res(self
, res
):
109 info('enforce_res %s' % res
)
110 messenger
.send('enforce_res', [res
])
112 def _do_enforce_resolution(self
, res
):
113 info('enforce resolution %s (callback)' % res
)
114 messenger
.send('enforce_resolution', [res
])
116 def __verify(self
, task
):
117 # files = glob(self._path + '*')
118 for fname
in self
._fnames
:
119 info('verifying %s' % fname
)