Coverage for kea/droidbot.py: 78%
152 statements
« prev ^ index » next coverage.py v7.6.9, created at 2024-12-22 16:05 +0800
1# This file contains the main class of droidbot
2# It can be used after AVD was started, app was installed, and adb had been set up properly
3# By configuring and creating a droidbot instance,
4# droidbot will start interacting with Android in AVD like a human
5import logging
6import os
7import sys
8import pkg_resources
9import shutil
10from threading import Timer
12from .env_manager import AppEnvManager
13from .input_manager import InputManager
15from typing import TYPE_CHECKING
16if TYPE_CHECKING:
17 from .start import Setting
19class DroidBot(object):
20 """
21 The main class of droidbot
22 """
24 # this is a single instance class
25 instance = None
27 def __init__(
28 self,
29 app_path=None,
30 device_serial=None,
31 is_emulator=False,
32 output_dir=None,
33 env_policy=None,
34 policy_name=None,
35 random_input=True,
36 event_interval=None,
37 event_count=0,
38 timeout=None,
39 keep_app=None,
40 keep_env=False,
41 cv_mode=False,
42 debug_mode=False,
43 profiling_method=None,
44 grant_perm=False,
45 send_document=True,
46 enable_accessibility_hard=False,
47 master=None,
48 humanoid=None,
49 ignore_ad=False,
50 replay_output=None,
51 kea=None,
52 number_of_events_that_restart_app=100,
53 run_initial_rules_after_every_mutation=True,
54 is_harmonyos=False,
55 is_package=False,
56 generate_utg=False,
57 settings:"Setting"=None
58 ):
59 """
60 initiate droidbot with configurations
61 :return:
62 """
63 if debug_mode:
64 logging.basicConfig(level=logging.DEBUG,format='%(asctime)s [%(levelname)s] [%(filename)s:%(lineno)d] %(message)s',
65 datefmt='%Y-%m-%d %H:%M:%S')
66 else:
67 logging.basicConfig(level= logging.INFO)
69 self.logger = logging.getLogger('DroidBot')
70 DroidBot.instance = self
71 self.output_dir = output_dir
72 if output_dir is not None:
73 #Clear the output left by the previous run to prevent JSON matching errors.
74 if os.path.isdir(output_dir):
75 shutil.rmtree(output_dir)
76 os.makedirs(output_dir)
77 html_index_path = pkg_resources.resource_filename(
78 "kea", "resources/index.html"
79 )
80 stylesheets_path = pkg_resources.resource_filename(
81 "kea", "resources/stylesheets"
82 )
83 target_stylesheets_dir = os.path.join(output_dir, "stylesheets")
84 if os.path.exists(target_stylesheets_dir):
85 shutil.rmtree(target_stylesheets_dir)
86 shutil.copy(html_index_path, output_dir)
87 shutil.copytree(stylesheets_path, target_stylesheets_dir)
89 self.timeout = timeout
90 self.timer = None
91 self.keep_env = keep_env
92 self.keep_app = keep_app
94 self.device = None
95 self.app = None
96 self.droidbox = None
97 self.env_manager = None
98 self.input_manager = None
99 self.enable_accessibility_hard = enable_accessibility_hard
100 self.humanoid = humanoid
101 self.ignore_ad = ignore_ad
102 self.replay_output = replay_output
104 self.enabled = True
105 self.kea = kea
107 # param initializer
108 self.app_path = app_path
109 self.device_serial = device_serial
110 self.is_emulator = is_emulator
111 self.env_policy = env_policy
112 self.policy_name = policy_name
113 self.random_input = random_input
114 self.event_interval = event_interval
115 self.event_count = event_count
116 self.cv_mode = cv_mode
117 self.profiling_method = profiling_method
118 self.grant_perm = grant_perm
119 self.send_document = send_document
120 self.master = master
121 self.number_of_events_that_restart_app = number_of_events_that_restart_app
122 self.run_initial_rules_after_every_mutation = run_initial_rules_after_every_mutation
123 self.is_package = is_package
124 self.generate_utg = generate_utg
125 self.settings = settings
126 try:
127 self.init_droidbot(is_harmonyos)
128 except Exception:
129 import traceback
131 traceback.print_exc()
132 self.stop()
133 sys.exit(-1)
135 def init_droidbot(self, is_harmonyos):
136 # initializer for Android system
137 if not is_harmonyos:
138 from .app import App
139 from .device import Device
140 self.app = App(self.app_path, output_dir=self.output_dir, settings=self.settings)
141 self.device = Device(
142 device_serial=self.device_serial,
143 is_emulator=self.is_emulator,
144 output_dir=self.output_dir,
145 cv_mode=self.cv_mode,
146 grant_perm=self.grant_perm,
147 send_document=self.send_document,
148 enable_accessibility_hard=self.enable_accessibility_hard,
149 humanoid=self.humanoid,
150 ignore_ad=self.ignore_ad,
151 app_package_name=self.app.package_name,
152 is_harmonyos=is_harmonyos
153 )
155 self.env_manager = AppEnvManager(
156 device=self.device, app=self.app, env_policy=self.env_policy
157 )
158 self.input_manager = InputManager(
159 device=self.device,
160 app=self.app,
161 policy_name=self.policy_name,
162 random_input=self.random_input,
163 event_interval=self.event_interval,
164 event_count=self.event_count,
165 profiling_method=self.profiling_method,
166 master=self.master,
167 replay_output=self.replay_output,
168 kea=self.kea,
169 number_of_events_that_restart_app=self.number_of_events_that_restart_app,
170 generate_utg=self.generate_utg
171 )
172 # self.send_documents()
173 # initializer for HarmonyOS system
174 else:
175 from .device_hm import DeviceHM
176 from .app_hm import AppHM
177 self.device = DeviceHM(
178 device_serial=self.device_serial,
179 is_emulator=self.is_emulator,
180 output_dir=self.output_dir,
181 cv_mode=self.cv_mode,
182 grant_perm=self.grant_perm,
183 enable_accessibility_hard=self.enable_accessibility_hard,
184 humanoid=self.humanoid,
185 ignore_ad=self.ignore_ad,
186 is_harmonyos=is_harmonyos,
187 save_log=False,
188 settings=self.settings)
189 self.app = AppHM(self.app_path, output_dir=self.output_dir, settings=self.settings)
191 self.env_manager = AppEnvManager(
192 device=self.device,
193 app=self.app,
194 env_policy=self.env_policy)
195 self.input_manager = InputManager(
196 device=self.device,
197 app=self.app,
198 policy_name=self.policy_name,
199 random_input=self.random_input,
200 event_count=self.event_count,
201 event_interval=self.event_interval,
202 profiling_method=self.profiling_method,
203 master=self.master,
204 replay_output=self.replay_output,
205 kea=self.kea)
207 @staticmethod
208 def get_instance():
209 if DroidBot.instance is None:
210 print("Error: DroidBot is not initiated!")
211 sys.exit(-1)
212 return DroidBot.instance
214 def start(self):
215 """
216 start interacting
217 :return:
218 """
219 if not self.enabled:
220 return
221 self.logger.info("Starting DroidBot")
222 try:
223 if self.timeout > 0:
224 self.logger.info("Will stop in %d seconds.", self.timeout)
225 self.timer = Timer(self.timeout, self.stop)
226 self.timer.start()
228 self.device.set_up()
230 if not self.enabled:
231 return
232 self.device.connect()
234 if not self.enabled:
235 return
236 self.device.send_documents(self.app)
237 self.device.install_app(self.app)
239 if not self.enabled:
240 return
241 self.env_manager.deploy()
243 if not self.enabled:
244 return
245 if self.droidbox is not None:
246 self.droidbox.set_apk(self.app.app_path)
247 self.droidbox.start_unblocked()
248 self.input_manager.start()
249 self.droidbox.stop()
250 self.droidbox.get_output()
251 else:
252 self.input_manager.start()
253 except KeyboardInterrupt:
254 self.logger.info("Keyboard interrupt.")
255 pass
256 except Exception:
257 import traceback
259 traceback.print_exc()
260 self.stop()
261 sys.exit(-1)
263 self.stop()
264 self.logger.info("DroidBot Stopped")
266 def stop(self):
267 self.enabled = False
268 if self.timer and self.timer.is_alive():
269 self.timer.cancel()
270 if self.env_manager:
271 self.env_manager.stop()
272 if self.input_manager:
273 self.input_manager.stop()
274 if self.droidbox:
275 self.droidbox.stop()
276 if self.device:
277 self.device.disconnect()
278 if not self.keep_env:
279 self.device.tear_down()
280 if not self.keep_app:
281 self.device.uninstall_app(self.app)
282 if (
283 hasattr(self.input_manager.policy, "master")
284 and self.input_manager.policy.master
285 ):
286 import xmlrpc.client
288 proxy = xmlrpc.client.ServerProxy(self.input_manager.policy.master)
289 proxy.stop_worker(self.device.serial)
291class DroidBotException(Exception):
292 pass