Coverage for kea/input_manager.py: 82%

1import json 

2import logging 

3import time 

4 

5from .similarity import Similarity 

6 

7from .input_event import EventLog 

8from .input_policy import ( 

9 GuidedPolicy, 

10 POLICY_GUIDED, 

11 POLICY_RANDOM, 

12 KeaInputPolicy, 

13 RandomPolicy, 

14 POLICY_NONE, 

15 POLICY_LLM, 

16 LLMPolicy 

17) 

18 

19DEFAULT_POLICY = POLICY_RANDOM 

20RANDOM_POLICY = POLICY_RANDOM 

21DEFAULT_EVENT_INTERVAL = 1 

22DEFAULT_EVENT_COUNT = 100000000 

23DEFAULT_TIMEOUT = 3600 

24DEFAULT_DEVICE_SERIAL = "emulator-5554" 

25DEFAULT_UI_TARPIT_NUM = 2 

26 

27class UnknownInputException(Exception): 

28 pass 

29 

30 

31class InputManager(object): 

32 """ 

33 This class manages all events to send during app running 

34 """ 

35 

36 def __init__( 

37 self, 

38 device, 

39 app, 

40 policy_name, 

41 random_input, 

42 event_interval, 

43 event_count=DEFAULT_EVENT_COUNT, # the number of event generated in the explore phase. 

44 profiling_method=None, 

45 master=None, 

46 replay_output=None, 

47 kea=None, 

48 number_of_events_that_restart_app=100, 

49 generate_utg=False 

50 ): 

51 """ 

52 manage input event sent to the target device 

53 :param device: instance of Device 

54 :param app: instance of App 

55 :param policy_name: policy of generating events, string 

56 :return: 

57 """ 

58 self.logger = logging.getLogger('InputEventManager') 

59 self.enabled = True 

60 self.device = device 

61 self.app = app 

62 self.policy_name = policy_name 

63 self.random_input = random_input 

64 self.events = [] 

65 self.event_count = event_count 

66 self.event_interval = event_interval 

67 self.replay_output = replay_output 

68 self.monkey = None 

69 self.kea = kea 

70 self.profiling_method = profiling_method 

71 self.number_of_events_that_restart_app = number_of_events_that_restart_app 

72 self.generate_utg = generate_utg 

73 self.policy = self.get_input_policy(device, app, master) 

74 self.sim_calculator = Similarity(DEFAULT_UI_TARPIT_NUM) 

75 

76 def get_input_policy(self, device, app, master): 

77 if self.policy_name == POLICY_NONE: 

78 input_policy = None 

79 elif self.policy_name == POLICY_GUIDED: 

80 input_policy = GuidedPolicy( 

81 device, 

82 app, 

83 self.kea, 

84 self.generate_utg 

85 ) 

86 elif self.policy_name == POLICY_RANDOM: 

87 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) 

88 elif self.policy_name == POLICY_LLM: 

89 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) 

90 else: 

91 self.logger.warning( 

92 "No valid input policy specified. Using policy \"none\"." 

93 ) 

94 input_policy = None 

95 if isinstance(input_policy, KeaInputPolicy): 

96 input_policy.master = master 

97 return input_policy 

98 

99 def add_event(self, event): 

100 """ 

101 add one event to the event list 

102 :param event: the event to be added, should be subclass of AppEvent 

103 :return: 

104 """ 

105 if event is None: 

106 return 

107 self.events.append(event) 

108 

109 #Record and send events to the device. 

110 event_log = EventLog(self.device, self.app, event, self.profiling_method) 

111 event_log.start() 

112 while True: 

113 time.sleep(self.event_interval) 

114 if not self.device.pause_sending_event: 

115 break 

116 

117 event_log.stop() 

118 

119 def start(self): 

120 """ 

121 start sending event 

122 """ 

123 self.logger.info("start sending events, policy is %s" % self.policy_name) 

124 

125 try: 

126 if self.policy is not None: 

127 self.policy.start(self) 

128 

129 except KeyboardInterrupt: 

130 pass 

131 

132 self.stop() 

133 self.logger.info("Finish sending events") 

134 

135 def stop(self): 

136 """ 

137 stop sending event 

138 """ 

139 if self.monkey: 

140 if self.monkey.returncode is None: 

141 self.monkey.terminate() 

142 self.monkey = None 

143 pid = self.device.get_app_pid("com.android.commands.monkey") 

144 if pid is not None: 

145 self.device.adb.shell("kill -9 %d" % pid) 

146 self.enabled = False