-
Notifications
You must be signed in to change notification settings - Fork 2.1k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
b50d277
commit 3b9e0e2
Showing
4 changed files
with
181 additions
and
21 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,120 @@ | ||
| #!/usr/bin/env python3 | ||
|
|
||
| # | ||
| # Copyright (c) 2022 Project CHIP Authors | ||
| # All rights reserved. | ||
| # | ||
| # 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. | ||
| # | ||
|
|
||
| import argparse | ||
| import logging | ||
| import queue | ||
| import subprocess | ||
| import threading | ||
| import typing | ||
|
|
||
| from colorama import Fore, Style | ||
| from java.base import DumpProgramOutputToQueue | ||
|
|
||
|
|
||
| class IMTest: | ||
| def __init__(self, thread_list: typing.List[threading.Thread], queue: queue.Queue, cmd: [], args: str): | ||
| self.thread_list = thread_list | ||
| self.queue = queue | ||
| self.command = cmd | ||
|
|
||
| parser = argparse.ArgumentParser(description='Process pairing arguments.') | ||
|
|
||
| parser.add_argument('command', help="Command name") | ||
| parser.add_argument('-t', '--timeout', help="The program will return with timeout after specified seconds", default='200') | ||
| parser.add_argument('-a', '--address', help="Address of the device") | ||
| parser.add_argument('-p', '--port', help="Port of the remote device", default='5540') | ||
| parser.add_argument('-s', '--setup-payload', dest='setup_payload', | ||
| help="Setup Payload (manual pairing code or QR code content)") | ||
| parser.add_argument('-c', '--setup-pin-code', dest='setup_pin_code', | ||
| help=("Setup PIN code which can be used for password-authenticated " | ||
| "session establishment (PASE) with the Commissionee")) | ||
| parser.add_argument('-n', '--nodeid', help="The Node ID issued to the device", default='1') | ||
| parser.add_argument('-d', '--discriminator', help="Discriminator of the device", default='3840') | ||
| parser.add_argument('-u', '--paa-trust-store-path', dest='paa_trust_store_path', | ||
| help="Path that contains valid and trusted PAA Root Certificates") | ||
|
|
||
| args = parser.parse_args(args.split()) | ||
|
|
||
| self.command_name = args.command | ||
| self.nodeid = args.nodeid | ||
| self.address = args.address | ||
| self.port = args.port | ||
| self.setup_payload = args.setup_payload | ||
| self.setup_pin_code = args.setup_pin_code | ||
| self.discriminator = args.discriminator | ||
| self.timeout = args.timeout | ||
|
|
||
| logging.basicConfig(level=logging.INFO) | ||
|
|
||
| def TestCmdOnnetworkLongImInvoke(self, nodeid, setuppin, discriminator, timeout): | ||
| java_command = self.command + ['im', 'onnetwork-long-im-invoke', nodeid, setuppin, discriminator, timeout] | ||
| logging.info(f"Execute: {java_command}") | ||
| java_process = subprocess.Popen( | ||
| java_command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) | ||
| DumpProgramOutputToQueue(self.thread_list, Fore.GREEN + "JAVA " + Style.RESET_ALL, java_process, self.queue) | ||
| return java_process.wait() | ||
|
|
||
| def TestCmdOnnetworkLongImWrite(self, nodeid, setuppin, discriminator, timeout): | ||
| java_command = self.command + ['im', 'onnetwork-long-im-write', nodeid, setuppin, discriminator, timeout] | ||
| logging.info(f"Execute: {java_command}") | ||
| java_process = subprocess.Popen( | ||
| java_command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) | ||
| DumpProgramOutputToQueue(self.thread_list, Fore.GREEN + "JAVA " + Style.RESET_ALL, java_process, self.queue) | ||
| return java_process.wait() | ||
|
|
||
| def TestCmdOnnetworkLongImRead(self, nodeid, setuppin, discriminator, timeout): | ||
| java_command = self.command + ['im', 'onnetwork-long-im-read', nodeid, setuppin, discriminator, timeout] | ||
| logging.info(f"Execute: {java_command}") | ||
| java_process = subprocess.Popen( | ||
| java_command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) | ||
| DumpProgramOutputToQueue(self.thread_list, Fore.GREEN + "JAVA " + Style.RESET_ALL, java_process, self.queue) | ||
| return java_process.wait() | ||
|
|
||
| def TestCmdOnnetworkLongImSubscribe(self, nodeid, setuppin, discriminator, timeout): | ||
| java_command = self.command + ['im', 'onnetwork-long-im-subscribe', nodeid, setuppin, discriminator, timeout] | ||
| logging.info(f"Execute: {java_command}") | ||
| java_process = subprocess.Popen( | ||
| java_command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) | ||
| DumpProgramOutputToQueue(self.thread_list, Fore.GREEN + "JAVA " + Style.RESET_ALL, java_process, self.queue) | ||
| return java_process.wait() | ||
|
|
||
| def RunTest(self): | ||
| if self.command_name == 'onnetwork-long-im-invoke': | ||
| logging.info("Testing pairing onnetwork-long-im-invoke") | ||
| code = self.TestCmdOnnetworkLongImInvoke(self.nodeid, self.setup_pin_code, self.discriminator, self.timeout) | ||
| if code != 0: | ||
| raise Exception(f"Testing pairing onnetwork-long-im-invoke failed with error {code}") | ||
| elif self.command_name == 'onnetwork-long-im-write': | ||
| logging.info("Testing pairing onnetwork-long-im-write") | ||
| code = self.TestCmdOnnetworkLongImWrite(self.nodeid, self.setup_pin_code, self.discriminator, self.timeout) | ||
| if code != 0: | ||
| raise Exception(f"Testing pairing onnetwork-long-im-write failed with error {code}") | ||
| elif self.command_name == 'onnetwork-long-im-read': | ||
| logging.info("Testing pairing onnetwork-long-im-read") | ||
| code = self.TestCmdOnnetworkLongImRead(self.nodeid, self.setup_pin_code, self.discriminator, self.timeout) | ||
| if code != 0: | ||
| raise Exception(f"Testing pairing onnetwork-long-im-read failed with error {code}") | ||
| elif self.command_name == 'onnetwork-long-im-subscribe': | ||
| logging.info("Testing pairing onnetwork-long-im-subscribe") | ||
| code = self.TestCmdOnnetworkLongImSubscribe(self.nodeid, self.setup_pin_code, self.discriminator, self.timeout) | ||
| if code != 0: | ||
| raise Exception(f"Testing pairing onnetwork-long-im-subscribe failed with error {code}") | ||
| else: | ||
| raise Exception(f"Unsupported command {self.command_name}") |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters