[Local Agent] Remove AMS test and update README.
Change-Id: I96fa0e215013372906ee922e780d09359a0cd695
diff --git a/README.md b/README.md
index c918f9f..e91eda6 100644
--- a/README.md
+++ b/README.md
@@ -5,7 +5,7 @@
To support partners in testing and certifying their products, **Test Suite for Smart Home with Matter** provides the tools and services to allow partners to comprehensively test their Matter products against the Google ecosystem in a scalable way.
-Local Agent, one of the major components of Smart Home Partner Test Suite, is a software package which can be installed in the host machines at partners' laboratory. A local agent is required for running Auto test suite to command and control DUTs and perform Matter protocol operations at the testing site. Note that partners won't need to use local agent if they only aim to run Manual test suites.
+Local Agent, one of the major components of Smart Home Partner Test Suite, is a Python-based application which can be installed in the host machines at partners' laboratory. The local agent is required for running automated test suite. It is used to command and control the devices under test (hereinafter called DUT) and perform Matter protocol operations at the testing site. Note that partners won't need to use local agent if they only aim to run Manual test suites.
In order to use local agent, partners will have to enable the Matter testability of their DUTs, please contact your **Google Developer Support**.
@@ -14,14 +14,36 @@
## Features
-1. Registration: The linking process to link user's local agent to the backend.
-2. Service Sync: Reporting status, polling RPC requests and sending RPC responses to the backend.
-3. Device Control: Interacting with DUT for command and control, including collecting any available logs
+1. Linking and unlinking the local agent from the test suite web app.
+2. Commanding and controlling the devices from the test suite web app.
+3. Reporting local agent and connected device statuses to the backend.
+4. Collecting artifacts and logs from the connected devices and uploading to the backend.
+
+
+## Prerequisites
+
+Make sure your machine meets the following prerequisites before setting up the local agent.
+
+### Host Minimum Requirements
+
+1. Ubuntu LTS (Other Linux-based environments may work but aren’t officially supported.)
+2. 4GB RAM
+3. 1GB Hard disk space
+4. Internet access
+5. Device debug interface (Typically UART over USB)
+
+### Matter extension packages
+
+The device control library we're using in local agent is GDM, also known as [Gazoo Device Manager](https://github.com/google/gazoo-device), which is also a python package for interacting with Matter devices.
+
+To allow local agent to talk to your Matter devices, you must build and add your own Matter device controllers to GDM. Please follow the extension [Matter controller codelab in GDM](https://github.com/google/gazoo-device/tree/master/examples/example_matter_extension_package) to build and install.
+
+Note that you'll have to install your extension packages under the same virtual environment that you create for setting up the local agent. Please contact your **Google Developer Support** if you are not sure how to implement your extension packages.
## Set up
-1. Clone the repo (at a desired place):
+1. Clone the repository (at a desired place):
```
$ git clone https://testsuite-smarthome-matter.googlesource.com/local-agent
```
@@ -44,9 +66,13 @@
$ pip install .
```
-> For development, we might want to use `pip install -e .` for the last step.
+> For development, we might want to use `pip install -e .`.
> (See documentation of the "-e" option [here](https://pip.pypa.io/en/stable/cli/pip_install/#cmdoption-e))
+5. Copy the config file [example_config.ini](https://testsuite-smarthome-matter.googlesource.com/local-agent/+/refs/heads/master/example_config.ini) to your host at ```~/.config/google/local_agent_config.ini```.
+
+6. Fill the names of your Matter controller extension packages (The extension you built in the above ```Matter extension packages``` section) in ```~/.config/google/local_agent_config.ini```. See [example_config.ini](https://testsuite-smarthome-matter.googlesource.com/local-agent/+/refs/heads/master/example_config.ini#20) for references.
+
## Usage
@@ -58,9 +84,98 @@
3. Once you’ve started your local agent, you should see a prompt that asks you for the linking code. Paste the linking code you copied from the web app. The local agent should start the linking process, and it will also start the device detection process, which should detect your DUTs that are connected to your host machine.
+4. A successful linking result will look like this:
-## How to extend your Matter device controllers in Gazoo Device Manager
+```
+$ local-agent
+20211207 14:39:39.785 I local_agent.py:444 Start the linking process
+Linking Code:1d8eb371
+20211207 14:41:10.240 I local_agent.py:149 Local Agent is linked successfully.
+20211207 14:41:10.241 I local_agent.py:224 Reporting status to AMS.
+20211207 14:41:10.242 I local_agent.py:258 Polling JSON-RPC requests from AMS.
+Moving config files to the backup directory /usr/local/google/home/chingtzuchen/gazoo/gdm/conf/backup/backup-20211207-064110
+
+##### Step 1/3: Detecting potential new communication addresses. #####
+
+ detecting potential AdbComms communication addresses
+ detecting potential DockerComms communication addresses
+ detecting potential JlinkSerialComms communication addresses
+ detecting potential PigweedSerialComms communication addresses
+ detecting potential PtyProcessComms communication addresses
+ detecting potential SshComms communication addresses
+ detecting potential SerialComms communication addresses
+No read/write permission for these serial address(es): ['/dev/bus/usb/001/001', '/dev/bus/usb/001/005', '/dev/bus/usb/002/001']
+ detecting potential UsbComms communication addresses
+ detecting potential YepkitComms communication addresses
+'ykushcmd' is not installed. Cannot get Yepkit serials.
+Found 1 possible pigweedserialcomms connections:
+ /dev/serial/by-id/usb-Silicon_Labs_J-Link_Pro_OB_000440186694-if00
+Found 2 possible serialcomms connections:
+ /dev/bus/usb/001/002
+ /dev/bus/usb/001/003
+Found 1 possible usbcomms connections:
+ 000440186694
+
+##### Step 2/3 Identify Device Type of Connections. #####
+
+Identifying pigweedserialcomms devices..
+_dev_serial_by-id_usb-Silicon_Labs_J-Link_Pro_OB_000440186694-if00_detect.txt logging to file /tmp/_dev_serial_by-id_usb-Silicon_Labs_J-Link_Pro_OB_000440186694-if00_detect.txt
+_dev_serial_by-id_usb-Silicon_Labs_J-Link_Pro_OB_000440186694-if00_detect.txt closing switchboard processes
+20211207 14:41:11.263 I local_agent.py:258 Polling JSON-RPC requests from AMS.
+ /dev/serial/by-id/usb-Silicon_Labs_J-Link_Pro_OB_000440186694-if00 is a efr32matterlighting.
+ pigweedserialcomms device_type detection complete.
+Identifying serialcomms devices..
+20211207 14:41:12.281 I local_agent.py:258 Polling JSON-RPC requests from AMS.
+ /dev/bus/usb/001/002 responses did not match a known device type.
+ /dev/bus/usb/001/003 responses did not match a known device type.
+ serialcomms device_type detection complete.
+Identifying usbcomms devices..
+ 000440186694 responses did not match a known device type.
+ usbcomms device_type detection complete.
+
+##### Step 3/3: Extract Persistent Info from Detected Devices. #####
+
+Getting info from communication port /dev/serial/by-id/usb-Silicon_Labs_J-Link_Pro_OB_000440186694-if00 for efr32matterlighting
+ efr32matterlighting_detect checking device readiness: attempt 1 of 1
+ efr32matterlighting_detect starting GazooDeviceBase.check_device_ready
+Power cycling properties not yet detected.
+ efr32matterlighting_detect health check 1/3 succeeded: Check power cycling ready.
+ efr32matterlighting_detect waiting up to 3s for device to be connected.
+ efr32matterlighting_detect health check 2/3 succeeded: Check device connected.
+ efr32matterlighting_detect logging to file /tmp/_dev_serial_by-id_usb-Silicon_Labs_J-Link_Pro_OB_000440186694-if00_detect.txt
+20211207 14:41:13.302 I local_agent.py:258 Polling JSON-RPC requests from AMS.
+ efr32matterlighting_detect health check 3/3 succeeded: Check create switchboard.
+ efr32matterlighting_detect GazooDeviceBase.check_device_ready successful. It took 0s.
+ efr32matterlighting_detect starting Efr32MatterDevice.get_detection_info
+ efr32matterlighting_detect Efr32MatterDevice.get_detection_info successful. It took 0s.
+ efr32matterlighting_detect closing switchboard processes
+20211207 14:41:14.320 I local_agent.py:258 Polling JSON-RPC requests from AMS.
+
+##### Detection Summary #####
+
+ 1 new devices detected:
+ efr32matterlighting-6694
+
+ 3 connections found but not detected:
+ /dev/bus/usb/001/002
+ /dev/bus/usb/001/003
+ 000440186694
+
+If a connection failed detection, check https://github.com/google/gazoo-device/blob/master/docs/device_setup for tips
+
+Device Alias Type Model Connected
+------------------------------ --------------- ------------------------ -------------------- ----------
+efr32matterlighting-6694 <undefined> efr32matterlighting PROTO connected
+
+Other Devices Alias Type Model Available
+------------------------------ --------------- ------------------------ -------------------- ----------
+
+1 total Gazoo device(s) available.
+20211207 14:41:15.339 I local_agent.py:258 Polling JSON-RPC requests from AMS.
+20211207 14:41:16.362 I local_agent.py:258 Polling JSON-RPC requests from AMS.
+20211207 14:41:20.455 W local_agent.py:269 The local agent is unlinked.
+20211207 14:41:20.456 I local_agent.py:285 Stopped polling RPC because stop event is set.
+20211207 14:41:20.456 I local_agent.py:249 Stopped reporting info because stop event is set.
+```
-The device control library we're using in local agent is GDM, also known as [Gazoo Device Manager](https://github.com/google/gazoo-device), which is also a python package for interacting with Matter devices.
-To allow local agent to talk to your Matter devices, you must build and add your own Matter device controllers to GDM. Please follow the extension [Matter controller codelab in GDM](https://github.com/google/gazoo-device/tree/master/examples/example_matter_extension_package) to build and install.
-Once your controllers are installed in your virtual environment, fill the names of your extension packages in ```~/.config/google/local_agent_config.ini```. See ```example_config.ini``` for references.
+Please contact your **Google Developer Support** or **smarthome-testsuite-contrib@google.com** if you encounter any unexpected problems.
\ No newline at end of file
diff --git a/example_config.ini b/example_config.ini
index d316118..00a3875 100644
--- a/example_config.ini
+++ b/example_config.ini
@@ -13,9 +13,13 @@
# limitations under the License.
[ServerConfig]
-AMS_HOST = localhost
-AMS_PORT = 8080
-AMS_SCHEME = http
+# No need to enable these configurations unless you're testing local agent with
+# local deployment backend.
+# AMS_HOST = localhost
+# AMS_PORT = 8080
+# AMS_SCHEME = http
[MatterDeviceControllerPackages]
-example_matter_extension_package
+# Fill the name of you extension package.
+# See https://github.com/google/gazoo-device/tree/master/examples/example_matter_extension_package for creation and installation.
+# example_matter_extension_package
diff --git a/local_agent/tests/README.md b/local_agent/tests/README.md
index c91aebd..459f463 100644
--- a/local_agent/tests/README.md
+++ b/local_agent/tests/README.md
@@ -1,6 +1,6 @@
# Local Agent Tests
-This folder contains unit and functional (integration with backend) tests for Local Agent.
+This folder contains unit tests for Local Agent.
You'll have to install the extra packages to run the tests:
```
@@ -23,37 +23,3 @@
```
$ pip show local-agent | grep -e "^Location" | awk -F " " '{print $2}'
```
-
-## Real AMS Test
-
-```Important```: If you're NOT an internal Googler or tester, you won't be able to run this test.
-
-Run the local agent with a real Test Suite Backend (AMS).
-
-
-1. Backend setup: If you're an internal Googler or tester, please contact the local agent author for the backend deployment; Otherwise, you won't be able to deploy the backend.
-
-2. Edit the local agent config file: filling the ```AMS_HOST``` and ```AMS_PORT``` fields with the AMS configurations. Contact the local agent author if you're not sure about the backend configurations. See ```example_config.ini``` for references.
-
-3. Activate the virtual environment.
-
-4. Connect your Matter devices with the above mentioned testabilities to the host machine. (You can still run the AMS test without connecting any devices).
-
-5. Start the fake front end process \
-Use ```-host [tsb_host]``` and ```-port [tsb_port]``` to indicate the host and port of deployed backend, the fake front end will use ```host=localhost``` and ```port=8000``` if the arguments are not provided. \
-For example: \
- Run with real devices
- ```
- python fake_front_end.py
- ```
-
-6. Start the local agent process:
- ```
- local-agent -u [config file location]
- ```
- or just put the config under ```~/.config/google/local_agent_config.ini```, then start the process parameterlessly:
- ```
- local-agent
- ```
-
-7. The Local Agent should prompt the user for entering the linking code, which is displayed in the console of ```fake_front_end``` process.
diff --git a/local_agent/tests/ams_tests/fake_front_end.py b/local_agent/tests/ams_tests/fake_front_end.py
deleted file mode 100644
index a5e036f..0000000
--- a/local_agent/tests/ams_tests/fake_front_end.py
+++ /dev/null
@@ -1,489 +0,0 @@
-# Copyright 2021 Google LLC
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-"""Fake front end for local agent and real AMS integration test."""
-import argparse
-import http
-import immutabledict
-import logging
-import requests
-import signal
-import sys
-import time
-import threading
-from typing import Any, Dict, List, Optional, Set, Tuple
-
-import fake_test_suite
-
-
-_LOCAL_TSB_HOST = '127.0.0.1'
-_LOCAL_TSB_PORT = 8080
-_TEST_PROJECT = 'test-project'
-_POLLING_PERIOD = 1 # seconds
-_STATUS_POLLING_PERIOD = 30 # seconds
-_COOL_DOWN_SEC = 1 # seconds
-_DETECTION_COOL_DOWN_SEC = 3 # To wait for GDM detection to complete.
-
-
-logger = logging.getLogger(__name__)
-
-# ======================== TSB endpoints ========================== #
-TEST_SUITE_AUTH = '/tsb/api/test-suite/auth'
-TEST_SUITE_PROJECTS = '/tsb/api/test-suite/projects'
-LINKING_CODE = '/tsb/api/test-suite/local-agent/linking-code'
-AGENT_STATUS = '/tsb/api/test-suite/local-agent/info'
-AGENT_RPC = '/tsb/api/test-suite/local-agent/rpc'
-UNLINK_AGENT = '/tsb/api/test-suite/local-agent/unlink'
-RPC_METADATA = f'{AGENT_RPC}/metadata'
-# ================================================================= #
-
-
-# ======================== Constants ========================== #
-ALL_SUITE_CLASSES = (
- fake_test_suite.BrightnessSuite,
- fake_test_suite.ColorSuite,
- fake_test_suite.DeviceCommonSuite,
- fake_test_suite.LightOnOffSuite,
- fake_test_suite.LockUnlockSuite,
-)
-GDM_CAPABILITY_TO_HG_TRAIT = immutabledict.immutabledict({
- 'pw_rpc_common': 'Common',
- 'pw_rpc_light': 'OnOff',
- 'pw_rpc_lock': 'LockUnlock',
-})
-# ============================================================= #
-
-
-# ======================== Module level functions ========================== #
-def setup_logger() -> None:
- """Sets up the logger for logging."""
- logger.setLevel(logging.DEBUG)
- handler = logging.StreamHandler()
- handler.setLevel(logging.DEBUG)
- handler.setFormatter(
- logging.Formatter('[%(asctime)s %(levelname)s] %(message)s'))
- logger.addHandler(handler)
-
-
-def parse_args() -> Tuple[str, Optional[int]]:
- """Sets up the parser for argument parsing.
-
- Returns:
- Tuple: TSB host, TSB port
- """
- parser = argparse.ArgumentParser()
- parser.add_argument(
- '-host', '--tsb_host', type=str, required=False,
- default=_LOCAL_TSB_HOST, help='TSB host')
- parser.add_argument(
- '-port', '--tsb_port', type=int, required=False,
- default=_LOCAL_TSB_PORT, help='TSB port')
- args, leftover = parser.parse_known_args(sys.argv[1:])
- sys.argv[1:] = leftover
-
- tsb_host = args.tsb_host
- tsb_port = args.tsb_port if tsb_host == _LOCAL_TSB_HOST else None
-
- return tsb_host, tsb_port
-
-
-def raise_exception(response: requests.models.Response, err_msg: str) -> None:
- """Raises exception for HTTP response status code != 200.
-
- Args:
- response: HTTP response.
- err_msg: Error message.
- """
- err_msg = f'{err_msg}. Status: {response.status_code}'
- try:
- ams_err_msg = response.json()['errorMessage']
- except:
- ams_err_msg = ''
- if ams_err_msg:
- err_msg += f', AMS error message: {ams_err_msg}'
- raise RuntimeError(err_msg)
-# ========================================================================== #
-
-
-class TSBService:
- """TSB endpoint service."""
-
- def __init__(self, tsb_host: str, tsb_port: Optional[int]):
- if tsb_port is None:
- self._base_url = f'http://{tsb_host}'
- else:
- self._base_url = f'http://{tsb_host}:{tsb_port}'
- auth_token = self.get_auth_token()
- self._auth = {'Authorization': auth_token}
-
- def get_auth_token(self) -> str:
- """Service to obtain the test suite user's auth token.
-
- Returns:
- Test suite user auth token.
-
- Raises:
- RuntimeError: HTTP response status code is not 200.
- """
- url = self._base_url + TEST_SUITE_AUTH
- resp = requests.post(url, json={'idToken': 'debug-token'})
- if resp.status_code != http.HTTPStatus.OK:
- raise_exception(resp, 'Failed to get auth token')
- auth_token = resp.json().get('authToken')
- return auth_token
-
- def get_or_create_test_project(self) -> None:
- """Service to create test project if needed.
-
- Raises:
- RuntimeError: HTTP response status code is not 200.
- """
- url = self._base_url + TEST_SUITE_PROJECTS
- resp = requests.get(url, headers=self._auth)
- if resp.status_code != http.HTTPStatus.OK:
- raise_exception(resp, 'Failed to get test project')
-
- for project in resp.json().get('result'):
- if project['id'] == _TEST_PROJECT:
- return
- resp = requests.post(
- url, headers=self._auth, json={'projectIds': [_TEST_PROJECT]})
- if resp.status_code != http.HTTPStatus.CREATED:
- raise_exception(resp, 'Failed to create test project')
-
- def get_linking_code(self) -> str:
- """Service to obtain the linking code.
-
- Returns:
- Linking code.
-
- Raises:
- RuntimeError: HTTP response status code is not 200.
- """
- url = self._base_url + LINKING_CODE
- resp = requests.post(
- url, headers=self._auth, json={'projectId': _TEST_PROJECT})
- if resp.status_code != http.HTTPStatus.OK:
- raise_exception(resp, 'Failed to get linking code')
- linking_code = resp.json()['result'].get('code')
- return linking_code
-
- def get_agent_status(self) -> Optional[Dict[str, Any]]:
- """Service to retrieve the local agent status.
-
- Returns:
- Local Agent status dict or None if agent is not linked.
-
- Raises:
- RuntimeError: HTTP response status code is not 200 nor 404.
- """
- url = self._base_url + AGENT_STATUS + f'?projectId={_TEST_PROJECT}'
- resp = requests.get(url, headers=self._auth)
- if resp.status_code == http.HTTPStatus.NOT_FOUND:
- return None
- elif resp.status_code == http.HTTPStatus.OK:
- return resp.json()['result']
- else:
- raise_exception(resp, 'Failed to get agent status')
-
- def send_rpc_request(self, rpc_request: Dict[str, Any]) -> str:
- """Service to send RPC request to the AMS.
-
- Args:
- rpc_request: JSON RPC request.
-
- Returns:
- JSON-RPC id.
-
- Raises:
- RuntimeError: HTTP response status code is not 200.
- """
- url = self._base_url + AGENT_RPC
- resp = requests.post(url,
- headers=self._auth, json=rpc_request)
- if resp.status_code != http.HTTPStatus.OK:
- raise_exception(resp, 'Failed to send RPC request')
- rpc_id = resp.json()['result']['id']
- return rpc_id
-
- def get_rpc_metadata(self, rpc_id: str) -> Dict[str, Optional[int]]:
- """Service to get RPC metadata with given rpc_id.
-
- Args:
- rpc_id: JSON-RPC id.
-
- Returns:
- RPC metadata.
-
- Raises:
- RuntimeError: HTTP response status code is not 200.
- """
- url = (self._base_url + RPC_METADATA +
- f'?projectId={_TEST_PROJECT}&rpcId={rpc_id}')
- resp = requests.get(url, headers=self._auth)
- if resp.status_code != http.HTTPStatus.OK:
- raise_exception(resp, 'Failed to get RPC metadata')
- metadata = resp.json()['result']
- return metadata
-
- def get_rpc_response(self, rpc_id: str) -> Dict[str, Any]:
- """Service to get RPC response with given rpc_id.
-
- Args:
- rpc_id: JSON-RPC id.
-
- Returns:
- RPC response.
-
- Raises:
- RuntimeError: HTTP response status code is not 200.
- """
- url = (self._base_url + AGENT_RPC +
- f'?projectId={_TEST_PROJECT}&rpcId={rpc_id}')
- resp = requests.get(url, headers=self._auth)
- if resp.status_code != http.HTTPStatus.OK:
- raise_exception(resp, 'Failed to get RPC response')
- rpc_response = resp.json()['result']
- return rpc_response
-
- def unlink_agent(self) -> None:
- """API to unlink local agent. Raises RuntimeError if failed."""
- url = self._base_url + UNLINK_AGENT
- post_data = {'projectId': _TEST_PROJECT}
- resp = requests.post(url, json=post_data, headers=self._auth)
- if resp.status_code != http.HTTPStatus.OK:
- raise_exception(resp, 'Failed to unlink local agent')
-
-
-class FakeFrontEnd:
- """Fake front end module."""
-
- def __init__(self, host: str, port: int):
- # Registers termination signal handler
- signal.signal(signal.SIGINT, self._terminate)
-
- # Retrieves auth token and creates test project
- self._tsb_service = TSBService(host, port)
- self._tsb_service.get_or_create_test_project()
-
- # Local Agent status
- self._local_agent_status = None
-
- # Worker threads: executing test plan and retrieving agent status.
- self._termination_event = threading.Event()
- self._test_plan_worker = threading.Thread(
- target=self._create_and_execute_test_plan, daemon=True)
- self._agent_status_worker = threading.Thread(
- target=self._retrieve_agent_status, daemon=True)
-
- def _terminate(self, sig_num: int, frame: 'frame') -> None:
- """Signal handler upon receiving a SIGINT.
-
- Args:
- sig_num: Signal number passed to the handler.
- frame: Current stack frame passed to the handler.
- """
- del sig_num, frame # Unused
- logger.warning('Terminates fake front end process.')
- self._termination_event.set()
-
- def run(self) -> None:
- """Runs fake front end.
-
- Simulate the front end behaviors:
- 1. Links the local agent.
- 2. Sends RPC requests to TSB, polls the metadata and gets response.
- 3. Retrieves the local agent status simultaneously.
- 4. Unlinks the agent after the test is completed.
- """
- if self._link_agent():
-
- time.sleep(_DETECTION_COOL_DOWN_SEC)
- self._test_plan_worker.start()
- self._agent_status_worker.start()
-
- while not self._termination_event.is_set():
- time.sleep(_POLLING_PERIOD)
-
- self._unlink_agent()
-
- def _checks_if_agent_is_linked(self) -> bool:
- """Returns if local agent is linked.
-
- Returns:
- True if agent is linked, false otherwise.
- """
- status = self._tsb_service.get_agent_status()
- return status is not None and status.get('status') != 'OFFLINE'
-
- def _checks_if_response_is_stored(self, rpc_id: str) -> bool:
- """Returns if the rpc response of rpc_id is stored.
-
- Returns:
- True if the rpc response is stored, false otherwise.
- """
- metadata = self._tsb_service.get_rpc_metadata(rpc_id=rpc_id)
- resp_timestamp = metadata.get('responseStoredTimestamp')
- return resp_timestamp is not None
-
- def _link_agent(self) -> bool:
- """Links local agent.
-
- Retrieves linking code and checks if agent is linked.
-
- Returns:
- True if agent is linked, false otherwise.
- """
- linking_code = self._tsb_service.get_linking_code()
- print(f'\033[1m******************** Linking Code: {linking_code} ****'
- '****************\033[0m')
- while (not self._termination_event.is_set() and
- not self._checks_if_agent_is_linked()):
- logger.info(f'No agent is linked, sleep {_POLLING_PERIOD} sec...')
- time.sleep(_POLLING_PERIOD)
- if not self._termination_event.is_set():
- logger.info('The local agent is linked.')
- return True
- return False
-
- def _run_rpc_requests(
- self, rpc_request: Dict[str, Any]) -> Optional[Dict[str, Any]]:
- """Runs RPC request.
-
- Simulates the FE behavior: sends the rpc request to BE, polls for the
- rpc metadata, once it gets updated, retrieves the rpc response from BE.
-
- Args:
- rpc_request: JSON-RPC request.
-
- Returns:
- JSON-RPC response or None if fake front end is interrupted by SIGINT.
- """
- rpc_id = self._tsb_service.send_rpc_request(rpc_request=rpc_request)
- logger.info(f'Sent RPC request: {rpc_request}.')
-
- while (not self._termination_event.is_set() and
- not self._checks_if_response_is_stored(rpc_id)):
- logger.info(
- f'RPC response not available, sleep {_POLLING_PERIOD} sec...')
- time.sleep(_POLLING_PERIOD)
-
- # Interrupted by SIGINT
- if self._termination_event.is_set():
- return None
-
- logger.info('Metadata is updated.')
- rpc_response = self._tsb_service.get_rpc_response(rpc_id=rpc_id)
-
- return rpc_response
-
- def _unlink_agent(self) -> None:
- """Unlinks local agent."""
- logger.info('Unlinking local agent.')
- self._tsb_service.unlink_agent()
- logger.info('Local agent unlinked.')
-
- def _create_and_execute_test_plan(self) -> None:
- """Creates and executes test plan."""
- while self._local_agent_status is None:
- logger.info(f'Local Agent status not available yet.')
- time.sleep(_POLLING_PERIOD)
- connected_devices = []
- for device_info in self._local_agent_status['devices']:
- device_id = device_info['deviceId']
- hg_traits = self._get_hg_traits(device_info['capabilities'])
- connected_devices.append((device_id, hg_traits))
-
- for device_id, hg_traits in connected_devices:
- suites = self._generate_suites(device_id, hg_traits)
- for suite in suites:
- logger.info(
- f'Executes suite {suite} for device {device_id} ...')
- self._run_suite(suite)
-
- def _get_hg_traits(self, capabilities: List[str]) -> List[str]:
- """Maps the GDM capability to the corresponding HG trait.
-
- Args:
- capabilities: List of GDM capability.
-
- Returns:
- List of HG traits.
- """
- hg_traits = []
- for capability in capabilities:
- hg_trait = GDM_CAPABILITY_TO_HG_TRAIT.get(capability)
- if hg_trait is not None:
- hg_traits.append(hg_trait)
- return hg_traits
-
- def _retrieve_agent_status(self) -> None:
- """Retrieves local agent status."""
- while not self._termination_event.is_set():
- status = self._tsb_service.get_agent_status()
- if status is not None:
- logger.info(f'Retrieves local agent status: {status}')
- self._local_agent_status = status
- time.sleep(_STATUS_POLLING_PERIOD)
-
- def _generate_suites(
- self, device_id: str, device_traits: Set[str]) -> List[Any]:
- """Generates test suites based on device trait.
-
- In reality, FE sends device info to BE for test plan/suite generation.
- To reduce the maintenance effort, the fake FE here is to simply map the
- corresponding fake-suite directly.
-
- Args:
- device_id: GDM device id.
- device_traits: Set of device traits on HG.
-
- Returns:
- The list of suites which are applicable to the device traits.
- """
- suites = []
- for suite_class in ALL_SUITE_CLASSES:
- if suite_class.is_applicable_to(device_traits):
- suites.append(suite_class(device_id))
- return suites
-
- def _run_suite(self, suite: Any) -> None:
- """Runs suite procedures.
-
- Args:
- suite: Suite instance.
- """
- device_ids = [suite.device_id]
- start_suite_rpc, end_suite_rpc = (
- fake_test_suite.generate_start_end_suite_rpc(device_ids))
-
- all_procedures = [start_suite_rpc] + suite.procedures + [end_suite_rpc]
-
- for rpc_request in all_procedures:
- logger.info(f'Runs RPC request {rpc_request}')
- rpc_response = self._run_rpc_requests(rpc_request=rpc_request)
- logger.info(f'Retrieves RPC response: {rpc_response}')
- time.sleep(_COOL_DOWN_SEC)
-
-
-def main() -> None:
- """Main entry of fake front end."""
- setup_logger()
- tsb_host, tsb_port = parse_args()
- fake_front_end = FakeFrontEnd(host=tsb_host, port=tsb_port)
- fake_front_end.run()
-
-
-if __name__ == '__main__':
- main()
diff --git a/local_agent/tests/ams_tests/fake_test_suite.py b/local_agent/tests/ams_tests/fake_test_suite.py
deleted file mode 100644
index fda9f4f..0000000
--- a/local_agent/tests/ams_tests/fake_test_suite.py
+++ /dev/null
@@ -1,272 +0,0 @@
-# Copyright 2021 Google LLC
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-"""Fake test suite for local agent and real AMS integration test."""
-import secrets
-from typing import Any, Dict, List, Optional, Set, Tuple
-
-
-_TEST_PROJECT = 'test-project'
-_TEST_BRIGHTNESS_LEVEL = 150
-_TEST_COLOR_HUE = 50
-_TEST_COLOR_SATURATION = 30
-
-# ======================== Supported Methods ========================== #
-# See go/rainier-amsprocedure-api for more details
-
-# Basic
-_REBOOT = 'setReboot'
-_FACTORY_RESET = 'setFactoryReset'
-
-# On / Off
-_TURN_ON_LIGHT = 'setOn'
-_TURN_OFF_LIGHT = 'setOff'
-_GET_LIGHT_STATE = 'getOnOff'
-
-# Lock / Unlock
-_LOCK_DEVICE = 'setLock'
-_UNLOCK_DEVICE = 'setUnlock'
-_GET_IS_LOCKED = 'getIsLocked'
-
-# Brightness
-_GET_LIGHT_BRIGHTNESS = 'getBrightness'
-_SET_LIGHT_BRIGHTNESS = 'setBrightness'
-
-# Color
-_GET_LIGHT_COLOR = 'getColor'
-_SET_LIGHT_COLOR = 'setColor'
-
-# Test suite management
-_START_SUITE = 'startTestSuite'
-_END_SUITE = 'endTestSuite'
-# ============================================================= #
-
-# ====================== Helper methods ======================= #
-# TODO(b/194029399) Retrieves correct test result id.
-def generate_start_end_suite_rpc(
- device_ids: List[str]) -> Tuple[Dict[str, Any], Dict[str, Any]]:
- """Generates startTestSuite and endTestSuite RPC request.
-
- Args:
- device_ids: List of GDM device ids.
-
- Returns:
- Tuple: startTestSuite request and endTestSuite request.
- """
- fake_suite_id = secrets.token_hex(4)
- start_suite_rpc = {
- 'projectId': _TEST_PROJECT,
- 'method': _START_SUITE,
- 'params': {'id': fake_suite_id,
- 'dutDeviceIds': device_ids}}
- end_suite_rpc = {
- 'projectId': _TEST_PROJECT,
- 'method': _END_SUITE,
- 'params': {'id': fake_suite_id}}
- return start_suite_rpc, end_suite_rpc
-# ============================================================= #
-
-
-class BaseSuite:
- """Base class for all suite templates."""
-
- NAME = ''
- REQUIRED_TRAITS = set()
-
- def __init__(self, device_id: str):
- """Suite constructor.
-
- Args:
- device_id: GDM device id.
- """
- self._device_id = device_id
- self._procedures = []
-
- @classmethod
- def is_applicable_to(cls, device_traits: Set[str]) -> bool:
- """Whether the suite is applicable to the device traits.
-
- Args:
- device_traits: The device traits on HG.
-
- Returns:
- True if it's applicable, false otherwise.
- """
- return cls.REQUIRED_TRAITS.issubset(device_traits)
-
- def __str__(self) -> str:
- """Returns the suite name."""
- return self.NAME
-
- @property
- def device_id(self) -> str:
- """Returns the device id."""
- return self._device_id
-
- @property
- def procedures(self) -> List[Dict[str, Any]]:
- """Returns the procedures."""
- return self._procedures
-
-
-class LightOnOffSuite(BaseSuite):
- """Light on off test suite."""
-
- NAME = 'Light OnOff Suite'
- REQUIRED_TRAITS = {'OnOff',}
-
- def __init__(self, device_id: str):
- """Light on off test suite constructor."""
-
- super().__init__(device_id=device_id)
-
- self._set_on_light_rpc = {
- 'projectId': _TEST_PROJECT,
- 'method': _TURN_ON_LIGHT,
- 'params': {'dutDeviceId': self._device_id}}
-
- self._set_off_light_rpc = {
- 'projectId': _TEST_PROJECT,
- 'method': _TURN_OFF_LIGHT,
- 'params': {'dutDeviceId': self._device_id}}
-
- self._get_state_light_rpc = {
- 'projectId': _TEST_PROJECT,
- 'method': _GET_LIGHT_STATE,
- 'params': {'dutDeviceId': self._device_id}}
-
- self._procedures = [
- self._set_on_light_rpc,
- self._get_state_light_rpc,
- self._set_off_light_rpc,
- self._get_state_light_rpc,
- ]
-
-
-class LockUnlockSuite(BaseSuite):
- """Lock unlock test suite."""
-
- NAME = 'Lock Unlock Suite'
- REQUIRED_TRAITS = {'LockUnlock',}
-
- def __init__(self, device_id: str):
- """Lock unlock test suite constructor."""
-
- super().__init__(device_id=device_id)
-
- self._set_lock_rpc = {
- 'projectId': _TEST_PROJECT,
- 'method': _LOCK_DEVICE,
- 'params': {'dutDeviceId': self._device_id}}
-
- self._set_unlock_rpc = {
- 'projectId': _TEST_PROJECT,
- 'method': _UNLOCK_DEVICE,
- 'params': {'dutDeviceId': self._device_id}}
-
- self._get_locked_state_rpc = {
- 'projectId': _TEST_PROJECT,
- 'method': _GET_IS_LOCKED,
- 'params': {'dutDeviceId': self._device_id}}
-
- self._procedures = [
- self._set_lock_rpc,
- self._get_locked_state_rpc,
- self._set_unlock_rpc,
- self._get_locked_state_rpc,
- ]
-
-
-class DeviceCommonSuite(BaseSuite):
- """Device common suite."""
-
- NAME = 'Device Common Suite'
- REQUIRED_TRAITS = {'Common',}
-
- def __init__(self, device_id: str):
- """Device common test suite constructor."""
-
- super().__init__(device_id=device_id)
-
- self._set_reboot_rpc = {
- 'projectId': _TEST_PROJECT,
- 'method': _REBOOT,
- 'params': {'dutDeviceId': self._device_id}}
-
- self._set_factory_reset_rpc = {
- 'projectId': _TEST_PROJECT,
- 'method': _FACTORY_RESET,
- 'params': {'dutDeviceId': self._device_id}}
-
- self._procedures = [
- self._set_reboot_rpc,
- self._set_factory_reset_rpc,
- ]
-
-
-class BrightnessSuite(BaseSuite):
- """Lighting brightness suite."""
-
- NAME = 'Brightness Suite'
- REQUIRED_TRAITS = {'OnOff',}
-
- def __init__(self, device_id: str):
- """Brightness test suite constructor."""
-
- super().__init__(device_id=device_id)
-
- self._set_brightness = {
- 'projectId': _TEST_PROJECT,
- 'method': _SET_LIGHT_BRIGHTNESS,
- 'params': {'dutDeviceId': self._device_id,
- 'level':_TEST_BRIGHTNESS_LEVEL}}
-
- self._get_brightness = {
- 'projectId': _TEST_PROJECT,
- 'method': _GET_LIGHT_BRIGHTNESS,
- 'params': {'dutDeviceId': self._device_id}}
-
- self._procedures = [
- self._set_brightness,
- self._get_brightness,
- ]
-
-
-class ColorSuite(BaseSuite):
- """Lighting color suite."""
-
- NAME = 'Color Suite'
- REQUIRED_TRAITS = {'OnOff',}
-
- def __init__(self, device_id: str):
- """Color test suite constructor."""
-
- super().__init__(device_id=device_id)
-
- self._set_color = {
- 'projectId': _TEST_PROJECT,
- 'method': _SET_LIGHT_COLOR,
- 'params': {'dutDeviceId': self._device_id,
- 'hue': _TEST_COLOR_HUE,
- 'saturation': _TEST_COLOR_SATURATION}}
-
- self._get_color = {
- 'projectId': _TEST_PROJECT,
- 'method': _GET_LIGHT_COLOR,
- 'params': {'dutDeviceId': self._device_id}}
-
- self._procedures = [
- self._set_color,
- self._get_color,
- ]
