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
14 class RPCServer(SimpleXMLRPCServer
):
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')
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 info(f
'RPCServer::__screnshot {argument}')
32 taskMgr
.do_method_later(.01, self
.__callbacks
.screenshot
, 'rpc_screenshot', [argument
])
34 def __enforce_result(self
, argument
):
35 info(f
'RPCServer::__enforce_result {argument}')
36 taskMgr
.do_method_later(.01, self
.__callbacks
.enforce_result
, 'rpc_enforce_result', [argument
])
39 info(f
'RPCServer::__verify')
40 taskMgr
.do_method_later(.01, self
.__callbacks
.verify
, 'rpc_verify')
42 def __set_index(self
, argument
):
43 info(f
'RPCServer::__set_index {argument}')
44 taskMgr
.do_method_later(.01, self
.__callbacks
.set_index
, 'rpc_set_index', [argument
])
46 def __enforce_resolution(self
, argument
):
47 info(f
'RPCServer::__enforce_resolution {argument}')
48 taskMgr
.do_method_later(.01, self
.__callbacks
.enforce_resolution
, 'rpc_enforce_resolution', [argument
])
50 def get_position(self
, arg
):
51 info(f
'RPCServer::get_position {arg}')
52 return self
.__callbacks
.get_position(arg
)
55 info('RPCServer::destroy')
56 self
._BaseServer
__shutdown
_request
= True
59 class RPCServerThread(Thread
):
61 def __init__(self
, callbacks
):
62 Thread
.__init
__(self
, name
='rpc_server')
63 self
.__callbacks
= callbacks
66 self
.__server
= RPCServer(self
.__callbacks
)
67 self
.__server
.serve_forever()
72 def __init__(self
, creating_references
, position_manager
, app_name
):
73 self
.__creating
_references
= creating_references
74 self
.__position
_manager
= position_manager
75 self
.__app
_name
= app_name
76 Callbacks
= namedtuple('Callbacks', 'screenshot enforce_result verify set_index enforce_resolution get_position')
77 callbacks
= Callbacks(self
.__screenshot
,
78 self
.__enforce
_result
,
81 self
.__enforce
_resolution
,
83 RPCServerThread(callbacks
).start()
84 self
.__debug
_text
= OnscreenText('', fg
=(1, 0, 0, 1), scale
=.16)
85 self
.__debug
_text
.name
= 'test_txt'
87 self
.__file
_names
= []
90 self
.__set
_path
_base
_case
()
91 self
.__set
_path
_wine
()
92 self
.__set
_path
_references
()
94 def __set_path_base_case(self
):
95 self
.__path
= str(Filename().get_user_appdata_directory())
96 self
.__path
+= f
'/{self.__app_name}/'
97 self
.__path
+= f
'tests/functional{"_ref" if self.__creating_references else ""}/'
99 def __set_path_wine(self
):
100 home
= str(Path
.home())
101 p
= self
.__path
[1] + ':' + self
.__path
[2:].replace('/', '\\')
102 info(f
"__set_path_wine: {self.__path} {home} {p}")
103 #if self.__path.startswith('/c/users/') and exists(home + '/.wine/'):
104 # self.__path = home + '/.wine/drive_' + self.__path[1:]
105 if self
.__path
.startswith('/c/users/'):
108 def __set_path_references(self
):
109 if self
.__creating
_references
:
111 Filename().get_user_appdata_directory(),
112 f
'{self.__app_name}/tests/functional_ref_%s/' % _branch())
114 def __set_index(self
, idx
):
115 if int(idx
) == 1: rmtree(self
.__path
, ignore_errors
=True)
116 info(f
'creating dir: {self.__path}')
117 makedirs(self
.__path
, exist_ok
=True)
119 def __get_position(self
, target
):
120 info(f
'FunctionalTest::__get_position {target}')
121 if target
not in self
.__position
_manager
or not target
:
122 info(f
'{self.__position_manager=}')
123 return self
.__position
_manager
[target
]
125 def __screenshot(self
, name
):
126 self
.__file
_names
+= [self
.__path
+ name
]
127 r
= base
.screenshot(self
.__path
+ name
, False)
128 info(f
'screenshot {self.__path + name} ({r}; {getcwd()})')
130 def __enforce_result(self
, result
):
131 info(f
'enforce_result {result}')
132 messenger
.send('enforce_result', [result
])
134 def __enforce_resolution(self
, resolution
):
135 info(f
'enforce resolution {resolution} (callback)')
136 messenger
.send('enforce_resolution', [resolution
])
138 def __verify(self
, task
):
139 for f
in self
.__file
_names
:
140 info(f
'verifying {f}')