Merge branch 'main' of github.com:ecnusse/Kea

kea/device.py

@@ -19,7 +19,7 @@
 from .adapter.process_monitor import ProcessMonitor
 from .adapter.telnet import TelnetConsole
 from .adapter.user_input_monitor import UserInputMonitor
-
+from .utils import save_log
 from .app import App
 from .intent import Intent
 
@@ -31,7 +31,6 @@
 DEFAULT_NUM = '1234567890'
 DEFAULT_CONTENT = 'Hello world!'
 
-
 class Device(object):
     """
     this class describes a connected device
@@ -60,6 +59,8 @@ def __init__(
         """
 
         self.logger = logging.getLogger(self.__class__.__name__)
+        self.output_dir = output_dir
+        save_log(self.logger, self.output_dir)
         self.app_package_name = app_package_name
         if device_serial is None:
             from .utils import get_available_devices
@@ -76,7 +77,6 @@ def __init__(
         self.serial = device_serial
         self.is_emulator = is_emulator
         self.cv_mode = cv_mode
-        self.output_dir = output_dir
         if output_dir is not None:
             if not os.path.isdir(output_dir):
                 os.makedirs(output_dir)
@@ -872,6 +872,7 @@ def save_screenshot_for_report(self, event_name=None, event=None, current_state=
         """
 
         self.cur_event_count += 1
+        self.logger.info("Total event count: %s" % self.cur_event_count)
         if current_state is None:
             self.current_state = self.get_current_state()
         else:

kea/device_state.py

@@ -9,8 +9,8 @@
 if TYPE_CHECKING:
     from .device import Device
 
-from .utils import md5, deprecated, COLOR
-from .input_event import SearchEvent, SetTextAndSearchEvent, TouchEvent, LongTouchEvent, ScrollEvent, SetTextEvent, KeyEvent, UIEvent
+from .utils import md5, deprecated
+from .input_event import SearchEvent, SetTextAndSearchEvent, TouchEvent, LongTouchEvent, ScrollEvent, SetTextEvent, KeyEvent, UIEvent, SwipeEvent
 
 
 class DeviceState(object):
@@ -600,6 +600,20 @@ def get_possible_input(self):
         # For old Android navigation bars
         # possible_events.append(KeyEvent(name="MENU"))
 
+        for view_id in enabled_view_ids:
+            if self.__safe_dict_get(self.views[view_id], 'package') == 'com.tencent.mm' and self.__safe_dict_get(self.views[view_id], 'resource_id') == "com.tencent.mm:id/media_container":
+                bounds = self.__safe_dict_get(self.views[view_id], 'bounds')
+                x0 = bounds[0][0]
+                y0 = bounds[0][1]
+                x1 = bounds[1][0]
+                y1 = bounds[1][1]
+                swipe_x = (x0 + x1) / 2.0
+                swipe_y0 = y0 + 3.0 * (y1 - y0) / 4.0
+                swipe_y1 = y0 + (y1- y0) / 4.0
+                possible_events.append(SwipeEvent(start_x=swipe_x, start_y=swipe_y0, end_x=swipe_x, end_y=swipe_y1, duration = 500))
+                possible_events.append(SwipeEvent(start_x=swipe_x, start_y=swipe_y1, end_x=swipe_x, end_y=swipe_y0, duration = 500))
+
+
         self.possible_events = possible_events
         return [] + possible_events
 

kea/droidbot.py

@@ -55,6 +55,7 @@ def __init__(
         is_harmonyos=False,
         is_package=False,
         generate_utg=False,
+        disable_rotate=False,
         settings:"Setting"=None
     ):
         """
@@ -123,6 +124,7 @@ def __init__(
         self.run_initial_rules_after_every_mutation = run_initial_rules_after_every_mutation
         self.is_package = is_package
         self.generate_utg = generate_utg
+        self.disable_rotate = disable_rotate
         self.settings = settings
         try:
             self.init_droidbot(is_harmonyos)
@@ -168,7 +170,10 @@ def init_droidbot(self, is_harmonyos):
                 replay_output=self.replay_output,
                 kea=self.kea,
                 number_of_events_that_restart_app=self.number_of_events_that_restart_app,
-                generate_utg=self.generate_utg
+                generate_utg=self.generate_utg,
+                output_dir=self.output_dir,
+                is_package=self.is_package,
+                disable_rotate=self.disable_rotate
             )
             # self.send_documents()
         # initializer for HarmonyOS system
@@ -203,7 +208,10 @@ def init_droidbot(self, is_harmonyos):
                 profiling_method=self.profiling_method,
                 master=self.master,
                 replay_output=self.replay_output,
-                kea=self.kea)
+                kea=self.kea,
+                output_dir=self.output_dir,
+                is_package=self.is_package,
+                disable_rotate=self.disable_rotate)
 
     @staticmethod
     def get_instance():

kea/input_event.py

@@ -789,7 +789,7 @@ def get_views(self):
         return views
 
     def get_event_name(self):
-        return self.get_event_str()
+        return "Swipe"
 
 class ScrollEvent(UIEvent):
     """

kea/input_manager.py

@@ -3,7 +3,7 @@
 import time
 
 from .similarity import Similarity
-
+from .utils import save_log
 from .input_event import EventLog
 from .input_policy import (
     GuidedPolicy,
@@ -46,7 +46,10 @@ def __init__(
         replay_output=None,
         kea=None,
         number_of_events_that_restart_app=100,
-        generate_utg=False
+        generate_utg=False,
+        output_dir=None,
+        is_package=False,
+        disable_rotate=False
     ):
         """
         manage input event sent to the target device
@@ -56,6 +59,8 @@ def __init__(
         :return:
         """
         self.logger = logging.getLogger('InputEventManager')
+        self.output_dir = output_dir
+        save_log(self.logger, self.output_dir)
         self.enabled = True
         self.device = device
         self.app = app
@@ -70,8 +75,10 @@ def __init__(
         self.profiling_method = profiling_method
         self.number_of_events_that_restart_app = number_of_events_that_restart_app
         self.generate_utg = generate_utg
-        self.policy = self.get_input_policy(device, app, master)
         self.sim_calculator = Similarity(DEFAULT_UI_TARPIT_NUM)
+        self.disable_rotate=disable_rotate
+        self.is_package = is_package
+        self.policy = self.get_input_policy(device, app, master)
 
     def get_input_policy(self, device, app, master):
         if self.policy_name == POLICY_NONE:
@@ -81,12 +88,14 @@ def get_input_policy(self, device, app, master):
                 device,
                 app,
                 self.kea,
-                self.generate_utg
+                self.generate_utg,
+                self.disable_rotate,
+                self.output_dir
             )
         elif self.policy_name == POLICY_RANDOM:
-            input_policy = RandomPolicy(device, app, kea=self.kea, number_of_events_that_restart_app = self.number_of_events_that_restart_app, clear_and_reinstall_app=True, allow_to_generate_utg = self.generate_utg)
+            input_policy = RandomPolicy(device, app, kea=self.kea, number_of_events_that_restart_app = self.number_of_events_that_restart_app, clear_and_reinstall_app= not self.is_package, allow_to_generate_utg = self.generate_utg,disable_rotate=self.disable_rotate,output_dir=self.output_dir)
         elif self.policy_name == POLICY_LLM:
-            input_policy = LLMPolicy(device, app, kea=self.kea, number_of_events_that_restart_app = self.number_of_events_that_restart_app, clear_and_restart_app_data_after_100_events=True, allow_to_generate_utg = self.generate_utg)
+            input_policy = LLMPolicy(device, app, kea=self.kea, number_of_events_that_restart_app = self.number_of_events_that_restart_app, clear_and_restart_app_data_after_100_events=True, allow_to_generate_utg = self.generate_utg, output_dir=self.output_dir)
         else:
             self.logger.warning(
                 "No valid input policy specified. Using policy \"none\"."

kea/input_policy.py

@@ -1,11 +1,10 @@
-from dataclasses import dataclass
 import os
 import logging
 import random
 import copy
 import re
 import time
-from .utils import Time, generate_report
+from .utils import Time, generate_report, save_log, RULE_STATE
 from abc import abstractmethod
 from .input_event import (
     KEY_RotateDeviceToPortraitEvent,
@@ -57,14 +56,6 @@
 POLICY_NONE = "none"
 POLICY_LLM = "llm"
 
-
-@dataclass
-class RULE_STATE:
-    PRECONDITION_SATISFIED = "#satisfy pre"
-    PROPERTY_CHECKED = "#check property"
-    POSTCONDITION_VIOLATED = "#postcondition is violated"
-
-
 class InputInterruptedException(Exception):
     pass
 
@@ -89,6 +80,7 @@ def __init__(self, device: "Device", app: "App", allow_to_generate_utg=False):
         self.allow_to_generate_utg = allow_to_generate_utg
         self.triggered_bug_information = []
         self.time_needed_to_satisfy_precondition = []
+        self.statistics_of_rules = {}
 
         self._num_restarts = 0
         self._num_steps_outside = 0
@@ -148,6 +140,7 @@ def start(self, input_manager: "InputManager"):
                     self.time_needed_to_satisfy_precondition,
                     self.device.cur_event_count,
                     self.time_recoder.get_time_duration(),
+                    self.statistics_of_rules
                 )
 
             except KeyboardInterrupt:
@@ -262,12 +255,12 @@ def __init__(self, device, app, kea: "Kea" = None, allow_to_generate_utg=False):
         # self.to_state = None
 
         # retrive all the rules from the provided properties
-        self.statistics_of_rules = {}
         for rule in self.kea.all_rules:
-            self.statistics_of_rules[str(rule)] = {
+            self.statistics_of_rules[str(rule.function.__name__)] = {
                 RULE_STATE.PRECONDITION_SATISFIED: 0,
                 RULE_STATE.PROPERTY_CHECKED: 0,
                 RULE_STATE.POSTCONDITION_VIOLATED: 0,
+                RULE_STATE.UI_OBJECT_NOT_FOUND: 0
             }
 
     def run_initializer(self):
@@ -298,16 +291,12 @@ def check_rule_whose_precondition_are_satisfied(self):
             return
 
         candidate_rules_list = list(rules_ready_to_be_checked.keys())
-        for candidate_rule in candidate_rules_list:
-            self.statistics_of_rules[str(candidate_rule)][
-                RULE_STATE.PRECONDITION_SATISFIED
-            ] += 1
         # randomly select a rule to check
         rule_to_check = random.choice(candidate_rules_list)
 
         if rule_to_check is not None:
             self.logger.info(f"-------Check Property : {rule_to_check}------")
-            self.statistics_of_rules[str(rule_to_check)][
+            self.statistics_of_rules[str(rule_to_check.function.__name__)][
                 RULE_STATE.PROPERTY_CHECKED
             ] += 1
             precondition_page_index = self.device.cur_event_count
@@ -323,7 +312,7 @@ def check_rule_whose_precondition_are_satisfied(self):
                     "-------time from start : %s-----------"
                     % str(self.time_recoder.get_time_duration())
                 )
-                self.statistics_of_rules[str(rule_to_check)][
+                self.statistics_of_rules[str(rule_to_check.function.__name__)][
                     RULE_STATE.POSTCONDITION_VIOLATED
                 ] += 1
                 postcondition_page__index = self.device.cur_event_count
@@ -347,6 +336,9 @@ def check_rule_whose_precondition_are_satisfied(self):
                 self.logger.error(
                     f"-------Execution failed: UiObjectNotFound during exectution. Property:{rule_to_check}-----------"
                 )
+                self.statistics_of_rules[str(rule_to_check.function.__name__)][
+                    RULE_STATE.UI_OBJECT_NOT_FOUND
+                ] += 1
             elif result == CHECK_RESULT.PRECON_NOT_SATISFIED:
                 self.logger.info("-------Precondition not satisfied-----------")
             else:
@@ -377,12 +369,17 @@ def __init__(
             number_of_events_that_restart_app=100,
             clear_and_reinstall_app=False,
             allow_to_generate_utg=False,
+            disable_rotate=False,
+            output_dir=None
     ):
         super(RandomPolicy, self).__init__(device, app, kea, allow_to_generate_utg)
         self.restart_app_after_check_property = restart_app_after_check_property
         self.number_of_events_that_restart_app = number_of_events_that_restart_app
         self.clear_and_reinstall_app = clear_and_reinstall_app
         self.logger = logging.getLogger(self.__class__.__name__)
+        self.output_dir=output_dir
+        save_log(self.logger, self.output_dir)
+        self.disable_rotate=disable_rotate
         self.last_rotate_events = KEY_RotateDeviceToPortraitEvent
 
     def generate_event(self):
@@ -414,6 +411,10 @@ def generate_event(self):
             return KillAndRestartAppEvent(app=self.app)
 
         rules_to_check = self.kea.get_rules_whose_preconditions_are_satisfied()
+        for rule_to_check in rules_to_check:
+            self.statistics_of_rules[str(rule_to_check.function.__name__)][
+                RULE_STATE.PRECONDITION_SATISFIED
+            ] += 1
 
         if len(rules_to_check) > 0:
             t = self.time_recoder.get_time_duration()
@@ -449,7 +450,8 @@ def generate_random_event_based_on_current_state(self):
 
         possible_events = current_state.get_possible_input()
         possible_events.append(KeyEvent(name="BACK"))
-        possible_events.append(RotateDevice())
+        if not self.disable_rotate:
+            possible_events.append(RotateDevice())
 
         self._event_trace += EVENT_FLAG_EXPLORE
 
@@ -472,10 +474,12 @@ class GuidedPolicy(KeaInputPolicy):
     generate events around the main path
     """
 
-    def __init__(self, device, app, kea=None, allow_to_generate_utg=False):
+    def __init__(self, device, app, kea=None, allow_to_generate_utg=False,disable_rotate=False,output_dir=None):
         super(GuidedPolicy, self).__init__(device, app, kea, allow_to_generate_utg)
         self.logger = logging.getLogger(self.__class__.__name__)
-
+        self.output_dir = output_dir
+        save_log(self.logger,self.output_dir)
+        self.disable_rotate = disable_rotate
         if len(self.kea.all_mainPaths):
             self.logger.info("Found %d mainPaths" % len(self.kea.all_mainPaths))
         else:
@@ -571,9 +575,11 @@ def mutate_the_main_path(self):
                     self.logger.info(
                         "reach the end of the main path that could satisfy the precondition"
                     )
-                    rules_to_check = (
-                        self.kea.get_rules_whose_preconditions_are_satisfied()
-                    )
+                    rules_to_check = self.kea.get_rules_whose_preconditions_are_satisfied()
+                    for rule_to_check in rules_to_check:
+                        self.statistics_of_rules[str(rule_to_check.function.__name__)][
+                            RULE_STATE.PRECONDITION_SATISFIED
+                        ] += 1
                     if len(rules_to_check) > 0:
                         t = self.time_recoder.get_time_duration()
                         self.time_needed_to_satisfy_precondition.append(t)
@@ -704,7 +710,8 @@ def generate_random_event_based_on_current_state(self):
         # if self.random_input:
         #     random.shuffle(possible_events)
         possible_events.append(KeyEvent(name="BACK"))
-        possible_events.append(RotateDevice())
+        if not self.disable_rotate:
+            possible_events.append(RotateDevice())
 
         self._event_trace += EVENT_FLAG_EXPLORE
 
@@ -734,9 +741,12 @@ def __init__(
             number_of_events_that_restart_app=100,
             clear_and_restart_app_data_after_100_events=False,
             allow_to_generate_utg=False,
+            output_dir=None
     ):
         super(LLMPolicy, self).__init__(device, app, kea)
         self.logger = logging.getLogger(self.__class__.__name__)
+        self.output_dir = output_dir
+        save_log(self.logger,self.output_dir)
         self.__action_history = []
         self.__all_action_history = set()
         self.__activity_history = set()
@@ -849,6 +859,10 @@ def generate_llm_event(self):
             )
             return ReInstallAppEvent(self.app)
         rules_to_check = self.kea.get_rules_whose_preconditions_are_satisfied()
+        for rule_to_check in rules_to_check:
+            self.statistics_of_rules[str(rule_to_check.function.__name__)][
+                RULE_STATE.PRECONDITION_SATISFIED
+            ] += 1
 
         if len(rules_to_check) > 0:
             t = self.time_recoder.get_time_duration()

kea/kea.py

@@ -344,7 +344,7 @@ def get_rules_whose_preconditions_are_satisfied(self) -> Dict["Rule", "KeaTest"]
 
     def single_thread_precondition_checker(self):
         """
-        check precondition with sigle-thread
+        check precondition with single-thread
         """
         rules_passed_precondition: Dict["Rule", "KeaTest"] = dict()