blob: d8acc6362e000d3438bc1b57ec83c38478e35664 [file] [log] [blame]
#!/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}")