| # Copyright (c) 2013 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| import copy |
| import os |
| import re |
| import subprocess |
| import threading |
| import time |
| from autotest_lib.client.bin import test |
| from autotest_lib.client.common_lib import error |
| |
| POSITION_X_VALUATOR = '0' |
| POSITION_Y_VALUATOR = '1' |
| POSITION_VALUATORS = [POSITION_X_VALUATOR, POSITION_Y_VALUATOR] |
| PROPERTY_EVENT = 'PropertyEvent' |
| TOUCH_BEGIN_EVENT = 'RawTouchBegin' |
| TOUCH_END_EVENT = 'RawTouchEnd' |
| TOUCH_UPDATE_EVENT = 'RawTouchUpdate' |
| TOUCH_EVENTS = [TOUCH_BEGIN_EVENT, TOUCH_END_EVENT, TOUCH_UPDATE_EVENT] |
| |
| |
| class platform_EvdevSynDropTest(test.test): |
| """Test SYN_DROPPED event handling in xf86-input-evdev. |
| |
| When there is a SYN_DROPPED seen in xf86-input-evdev driver, it tries to |
| sync with the kernel evdev state and injects finger arriving, finger |
| leaving or axes update events accordingly to make sure upper layer |
| (such as X server) is seamless of SYN_DROPPED event. The test mainly |
| focuses on testing finger arriving and leaving events generated in |
| SYN_DROPPED handling. |
| |
| Currently we have four basic SYN_DROPPED tests to be performed : |
| a. Finger leaving, i.e., tracking id changes from a non-negative |
| number to -1 after SYN_DROPPED. |
| b. Finger arriving, i.e., tracking id changes from -1 to a |
| non-negative number. |
| c. Finger changing, i.e., original finger leaving and new finger |
| arriving, tracking id changes from a non-negative number to |
| another one. |
| d. Same finger, but axes change, i.e., no tracking id changes, but some |
| axes values have changed. |
| |
| For each test case, a pre-recorded event log is chopped into four |
| pieces: |
| |
| 1. original finger set |
| 2. finger events(leaving/arriving) to be dropped due to queue full |
| 3. events playbacked to push xinput events out |
| 4. cleanup events |
| |
| After playing-back the first piece, we block the evdev event reading in |
| order to generate SYN_DROPPED when we inject the second piece of events. |
| Then, we unblock the evdev event reading and inject the third piece of |
| events to collect the finger leaving/arriving XI2 events. These XI2 |
| events could be used to verify if SYN_DROPPED handling is processed |
| correctly. |
| """ |
| version = 1 |
| |
| def setUp(self): |
| self.stop_parser = False |
| self.syndrop_detected = False |
| self.stop_xi2_parser = False |
| self.finger_set = {} |
| |
| os.environ['DISPLAY'] = ':0' |
| os.environ['XAUTHORITY'] = '/home/chronos/.Xauthority' |
| self.data_dir = os.path.join(self.bindir, 'data') |
| device_file = os.path.join(self.data_dir, 'device') |
| self.device_ready = threading.Event() |
| self.dev_emu_proc = subprocess.Popen(['evemu-device', device_file], |
| stdout=subprocess.PIPE) |
| self.device_id = self.get_evemu_device_id(self.dev_emu_proc) |
| self.device = '/dev/input/event' + self.device_id |
| |
| # Xorg log parser for detection of SYN_DROPPED |
| self.xlog_parser_thread = threading.Thread(target=self.xlog_parser) |
| self.xlog_parser_thread.start() |
| |
| # Wait until the device is available in X server |
| self.device_ready.wait() |
| |
| # xinput 2 event parser for finger tracking above xf86-input-evdev |
| self.xi2_parser_ready = threading.Event() |
| self.xi2_proc = subprocess.Popen(['stdbuf', '-o0', 'xinput', 'test-xi2', |
| self.device_id], |
| stdout=subprocess.PIPE) |
| self.xi2_parser_thread = threading.Thread(target=self.xi2_parser, |
| args=(self.xi2_proc,)) |
| self.xi2_parser_thread.start() |
| |
| # Wait until XI2 parser is ready |
| self.xi2_parser_ready.wait() |
| |
| def tearDown(self): |
| self.dev_emu_proc.kill() |
| self.xi2_proc.kill() |
| self.stop_parser = True |
| self.xlog_parser_thread.join() |
| self.stop_xi2_parser = True |
| self.xi2_parser_thread.join() |
| |
| def xi2_parser(self, proc): |
| event_type = 0 |
| self.position = {} |
| while not self.stop_xi2_parser: |
| line = proc.stdout.readline() |
| if not self.xi2_parser_ready.is_set(): |
| if re.search('Reporting', line): |
| self.xi2_parser_ready.set() |
| # EVENT type 23 (RawTouchUpdate) |
| type_line = re.match('EVENT\stype\s\d+\s\((\w+)\)', line) |
| if type_line: |
| event_type = type_line.group(1) |
| if event_type == PROPERTY_EVENT: |
| self.finger_set = {} |
| if event_type in TOUCH_EVENTS: |
| detail = re.match('\s+detail:\s(\d+)', line) |
| if detail: |
| finger_id = detail.group(1) |
| if event_type not in self.finger_set.keys(): |
| self.finger_set[event_type] = [] |
| if finger_id not in self.finger_set[event_type]: |
| self.finger_set[event_type].append(finger_id) |
| # For valuators |
| # 0: 1902.00 (1902.00) |
| valuator = re.match('\s+(\d+):\s(\d+.\d+)\s+\(\d+.\d+\)', line) |
| if valuator and valuator.group(1) in POSITION_VALUATORS: |
| self.position[valuator.group(1)] = valuator.group(2) |
| |
| def xlog_parser(self): |
| log = open('/var/log/Xorg.0.log', 'r') |
| log.seek(0, 2) |
| while not self.stop_parser: |
| line = log.readline() |
| if not line: |
| time.sleep(0.001) |
| continue |
| if not self.device_ready.is_set(): |
| if re.search('ChromeOS-MT-Device:\sSync_State:', line): |
| self.device_ready.set() |
| if re.match('[^\+]+\++\sSYN_DROPPED\s\++', line): |
| self.syndrop_detected = True |
| |
| def get_evemu_device_id(self, proc): |
| output = proc.stdout.readline() |
| match = re.match('[^ ]+\s\/dev\/input\/event(\d+)', output) |
| return match.group(1) |
| |
| def test_condition(self, condition=None): |
| # Inject the original finger events |
| step1 = open(os.path.join(self.data_dir, condition + '.1'), 'r') |
| subprocess.call(['evemu-play', self.device], stdin=step1) |
| |
| self.original_set = copy.deepcopy(self.finger_set) |
| |
| # Blocking xf86-input-evdev event reading |
| step2 = open(os.path.join(self.data_dir, condition + '.2'), 'r') |
| subprocess.call(['xinput', 'set-prop', self.device_id, |
| 'Block Event Reading', '1']) |
| |
| # Inject events to fill up the evdev queue to generate SYN_DROPPED |
| subprocess.call(['evemu-play', self.device], stdin=step2) |
| |
| # Unblock xf86-input-evdev event processing |
| step3 = open(os.path.join(self.data_dir, condition + '.3'), 'r') |
| subprocess.call(['xinput', 'set-prop', self.device_id, |
| 'Block Event Reading', '0']) |
| subprocess.call(['evemu-play', self.device], stdin=step3) |
| |
| # Check if SYN_DROPPED event is seen in Xorg.0.log |
| if not self.syndrop_detected: |
| raise error.TestError('Did not see SYN_DROPPED event') |
| |
| # Store the finger event seen in xinput after SYN_DROPPED |
| self.syndrop_set = copy.deepcopy(self.finger_set) |
| |
| # Store the latest position valuators especially for test case (d) |
| self.syndrop_position = copy.deepcopy(self.position) |
| |
| # Continue on injecting remaining events for cleanup |
| step4 = open(os.path.join(self.data_dir, condition + '.4'), 'r') |
| subprocess.call(['evemu-play', self.device], stdin=step4) |
| |
| def verify_finger_leaving(self): |
| """Verify if finger leaving event is injected after SYN_DROPPED. |
| |
| Suppose there are two fingers with tracking id '1' and '2' in the |
| beginning. Finger '2' leaves after SYN_DROPPED, then we should get |
| the finger sets in xi2_parser: |
| original_set: |
| {'RawTouchBegin': ['1', '2'], 'RawTouchUpdate': ['1', '2']} |
| syndrop_set: |
| {'RawTouchEnd': ['2'], 'RawTouchUpdate': ['1']} |
| """ |
| syndrop_end_set = set(self.syndrop_set[TOUCH_END_EVENT]) |
| syndrop_update_set = set(self.syndrop_set[TOUCH_UPDATE_EVENT]) |
| union = syndrop_end_set | syndrop_update_set |
| intersection = syndrop_end_set & syndrop_update_set |
| if union != set(self.original_set[TOUCH_BEGIN_EVENT]): |
| raise error.TestFail('The union should include both fingers') |
| if intersection: |
| raise error.TestFail('The intersection of the set should be empty') |
| |
| def verify_finger_arriving(self): |
| """Verify if finger arriving event is injected after SYN_DROPPED. |
| |
| Suppose there is only one finger with tracking id '3' in the |
| beginning. Finger '4' appears after SYN_DROPPED, then we should get |
| the finger sets in xi2_parser: |
| original_set: |
| {'RawTouchBegin': ['3'], 'RawTouchUpdate': ['3']} |
| syndrop_set: |
| {'RawTouchBegin': ['4'], 'RawTouchUpdate': ['3', '4']} |
| """ |
| original_begin_set = set(self.original_set[TOUCH_BEGIN_EVENT]) |
| syndrop_begin_set = set(self.syndrop_set[TOUCH_BEGIN_EVENT]) |
| union = original_begin_set | syndrop_begin_set |
| intersection = original_begin_set & syndrop_begin_set |
| if union != set(self.syndrop_set[TOUCH_UPDATE_EVENT]): |
| raise error.TestFail('The union should include both fingers') |
| if intersection: |
| raise error.TestFail('The intersection of the set should be empty') |
| |
| def verify_finger_changing(self): |
| """Verify if finger changing events are injected after SYN_DROPPED. |
| |
| Suppose there are two fingers with tracking id '5' and '6' in the |
| beginning. Finger '6' leaves and Finger '7' appears after |
| SYN_DROPPED, then we should get the finger sets in xi2_parser: |
| original_set: |
| {'RawTouchBegin': ['5', '6'], 'RawTouchUpdate': ['5', '6']} |
| syndrop_set: |
| {'RawTouchEnd': ['6'], 'RawTouchBegin': ['7'], |
| 'RawTouchUpdate': ['5', '7']} |
| """ |
| original_update_set = set(self.original_set[TOUCH_UPDATE_EVENT]) |
| syndrop_update_set = set(self.syndrop_set[TOUCH_UPDATE_EVENT]) |
| finger_leaving = original_update_set - syndrop_update_set |
| if not finger_leaving: |
| raise error.TestFail('One original finger should disappear') |
| if finger_leaving != set(self.syndrop_set[TOUCH_END_EVENT]): |
| raise error.TestFail('Should see finger leaving event') |
| |
| finger_arriving = syndrop_update_set - original_update_set |
| if not finger_arriving: |
| raise error.TestFail('One new finger should arrive') |
| if finger_arriving != set(self.syndrop_set[TOUCH_BEGIN_EVENT]): |
| raise error.TestFail('Should see finger arriving event') |
| |
| def verify_same_finger(self): |
| """Verify if finger position are updated after SYN_DROPPED. |
| |
| Suppose there is one finger in the beginning. Between block reading |
| event in step 2 and unblock reading event in step3, the finger moves |
| to (1897, 725). And there is no POSITION_X and POSITION_Y update in |
| step 3. The test is to verify the position should be updated with |
| (1897, 725) in xi2_parser after SYN_DROPPED. |
| """ |
| if self.syndrop_position[POSITION_X_VALUATOR] != '1897.00': |
| raise error.TestFail('Incorrect POSITION_X after SYN_DROPPED') |
| if self.syndrop_position[POSITION_Y_VALUATOR] != '725.00': |
| raise error.TestFail('Incorrect POSITION_Y after SYN_DROPPED') |
| |
| def run_once(self): |
| self.setUp() |
| self.test_condition('finger_leaving') |
| self.tearDown() |
| self.verify_finger_leaving() |
| |
| self.setUp() |
| self.test_condition('finger_arriving') |
| self.tearDown() |
| self.verify_finger_arriving() |
| |
| self.setUp() |
| self.test_condition('finger_changing') |
| self.tearDown() |
| self.verify_finger_changing() |
| |
| self.setUp() |
| self.test_condition('same_finger') |
| self.tearDown() |
| self.verify_same_finger() |
