Skip to content

samples: openthread: align nRF_Connect_SDK_13_14.py with GRLv64 #25848

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

samples: openthread: align nRF_Connect_SDK_13_14.py with GRLv64 #25848

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
178 changes: 145 additions & 33 deletions samples/openthread/cli/harness-thci/nRF_Connect_SDK_13_14.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
#!/usr/bin/env python
#
# Copyright (c) 2016, The OpenThread Authors.
Expand Down Expand Up @@ -48,6 +48,7 @@

import serial
from serial.serialutil import SerialException
from typing import Literal

import commissioner
from GRLLibs.ThreadPacket.PlatformPackets import (
Expand Down Expand Up @@ -335,9 +336,10 @@
if line is None:
time.sleep(0.01)
continue
line = line.replace("\x1b[J", "")

# self.log("readline: %s", line)
# skip empty lines
line = line.replace("\x1b[J", "")
if line:
response.append(line)

Expand All @@ -347,6 +349,8 @@
m = OpenThreadTHCI._COMMAND_OUTPUT_ERROR_PATTERN.match(line)
if m is not None:
code, msg = m.groups()
if ignore_exec:
return True
raise CommandError(int(code), msg)
else:
if ignore_exec:
Expand Down Expand Up @@ -964,7 +968,9 @@
specific type of MAC address
"""
# if power down happens, return extended address assigned previously
if self.isPowerDown:
if self.isPowerDown and bType == MacType.EthMac:
return self._deviceGetEtherMac()
elif self.isPowerDown:
macAddr64 = self.mac
else:
if bType == MacType.FactoryMac:
Expand Down Expand Up @@ -2234,9 +2240,9 @@

@API
def vendorInfo(self, name="", model="", version=""):
vn = "vendor name %s" % name
vm = "vendor model %s" % model
vv = "vendor swversion %s" % version
vn = "vendor name %s" % name.replace(" ", "")
vm = "vendor model %s" % model.replace(" ", "")
vv = "vendor swversion %s" % version.replace(" ", "")
self.__executeCommand(vn)
self.__executeCommand(vm)
self.__executeCommand(vv)
Expand Down Expand Up @@ -3050,10 +3056,10 @@
# elif self.DeviceCapability == OT12_CAPBS:
# return OT12_VERSION in self.UIStatusMsg
if self.DeviceCapability == OT13_CAPBS:
return [
True if OT13_VER in self.UIStatusMsg else False
for OT13_VER in OT13_VERSION
]
OT13_FirmwareCheck = any(

Check warning on line 3059 in samples/openthread/cli/harness-thci/nRF_Connect_SDK_13_14.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Rename this local variable "OT13_FirmwareCheck" to match the regular expression ^[_a-z][a-z0-9_]*$. 


See more on https://sonarcloud.io/project/issues?id=nrfconnect_sdk-nrf&issues=AZrEumc7caPpW9znHFgR&open=AZrEumc7caPpW9znHFgR&pullRequest=25848
OT13_VER in self.UIStatusMsg for OT13_VER in OT13_VERSION
)
return OT13_FirmwareCheck
else:
return False

Expand Down Expand Up @@ -3262,26 +3268,45 @@
return self.__executeCommand(cmd)[-1] == "Done"

@API
def send_udp(self, interface, destination, port, payload="12ABcd"):
def send_udp(
self,
interface,
destination,
port,
payload="12ABcd",
return_result=False,
spiff_udp_ip="",
):
"""payload hexstring"""
assert payload is not None, "payload should not be none"
assert interface == 0, "non-BR must send UDP to Thread interface"
self.__udpOpen()
self.__udpOpen(spiff_udp_ip, port)
time.sleep(0.5)
cmd_ = "test tmforiginfilter disable"
self.__executeCommand(cmd_)
cmd = "udp send %s %s -x %s" % (destination, port, payload)
result = self.__executeCommand(cmd)[-1] == "Done"

if not return_result:
result = self.__executeCommand(cmd)[-1] == "Done"
else:
result = self.__executeCommand(cmd)[0]
_cmd = "test tmforiginfilter enable"
self.__executeCommand(_cmd)
return result

def __udpOpen(self):
def __udpOpen(self, spiff_udp_ip="", port=""):

Check warning on line 3296 in samples/openthread/cli/harness-thci/nRF_Connect_SDK_13_14.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Rename method "__udpOpen" to match the regular expression ^[a-z_][a-z0-9_]*$. 


See more on https://sonarcloud.io/project/issues?id=nrfconnect_sdk-nrf&issues=AZrEumc7caPpW9znHFgS&open=AZrEumc7caPpW9znHFgS&pullRequest=25848
if spiff_udp_ip != "":
self.__isUdpOpened = False
self.__executeCommand("udp close")
if not self.__isUdpOpened:
cmd = "udp open"
self.__executeCommand(cmd)

# Bind to RLOC address and first dynamic port
rlocAddr = self.getRloc()

cmd = "udp bind %s 49152" % rlocAddr
if spiff_udp_ip == "":
cmd = "udp bind %s 49152" % rlocAddr
else:
cmd = "udp connect %s %s" % (spiff_udp_ip, port)
self.__executeCommand(cmd)

self.__isUdpOpened = True
Expand Down Expand Up @@ -3404,6 +3429,10 @@
cmd = "discover %s" % channel
return self.__executeCommand(cmd)

@API
def icmp_ping_req(self, addr):
return self.__executeCommand("ping %s 2" % addr)

########################################################################################################################
# 1.3 THCI Commands
#
Expand Down Expand Up @@ -3736,7 +3765,9 @@

def srp_client_remove(self, instancename, servicename):
cmd = "srp client service remove %s %s" % (instancename, servicename)
# cmd = 'srp client service remove service-test-1 _thread-test._udp'

Check warning on line 3768 in samples/openthread/cli/harness-thci/nRF_Connect_SDK_13_14.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Remove this commented out code. 


See more on https://sonarcloud.io/project/issues?id=nrfconnect_sdk-nrf&issues=AZrEumc7caPpW9znHFgT&open=AZrEumc7caPpW9znHFgT&pullRequest=25848
Copy link

Copilot AI Nov 27, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Remove commented-out code. If this is an example, consider adding it to documentation instead.

Suggested change
# cmd = 'srp client service remove service-test-1 _thread-test._udp'

Copilot uses AI. Check for mistakes.
self.__executeCommand(cmd)
# self.__executeCommand("netdata register")
Copy link

Copilot AI Nov 27, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Remove commented-out code. If this command should be executed conditionally, implement proper conditional logic instead.

Suggested change
# self.__executeCommand("netdata register")

Copilot uses AI. Check for mistakes.
time.sleep(3)

def srpCallBackEnable(self):
Expand All @@ -3758,10 +3789,6 @@
cmd = "srp client host clear"
self.__executeCommand(cmd)

def client_key_lease(self, keylease):
cmd = "srp client keyleaseinterval %s" % keylease
self.__executeCommand(cmd)

def srpCallBackDisable(self):
cmd = "srp client callback disable"
self.__executeCommand(cmd)
Expand Down Expand Up @@ -3858,7 +3885,7 @@
cmd = "srp client leaseinterval %s" % (lease)
self.__executeCommand(cmd)
if keylease != None:
cmd = "srp client keyleaseinterval %s" % (keylease)

Check failure on line 3888 in samples/openthread/cli/harness-thci/nRF_Connect_SDK_13_14.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Define a constant instead of duplicating this literal "srp client keyleaseinterval %s" 3 times. 


See more on https://sonarcloud.io/project/issues?id=nrfconnect_sdk-nrf&issues=AZrEumc7caPpW9znHFgQ&open=AZrEumc7caPpW9znHFgQ&pullRequest=25848
self.__executeCommand(cmd)

# Not required to execute this command for port 12346
Expand All @@ -3872,17 +3899,17 @@
srv,
tcp_service_type,
port,
weight,
priority,
weight,
txt,
)
else:
cmd = "srp client service add %s %s %s %s %s %s" % (
srv,
service_type,
port,
weight,
priority,
weight,
txt,
)
self.__executeCommand(cmd)
Expand Down Expand Up @@ -4019,17 +4046,17 @@
srv,
tcp_service_type,
port,
weight,
priority,
weight,
txt,
)
else:
cmd = "srp client service add %s %s %s %s %s %s" % (
srv,
service_type,
port,
weight,
priority,
weight,
txt,
)
self.__executeCommand(cmd)
Expand All @@ -4045,6 +4072,41 @@
cmd = "srp client start %s %s" % (dst_addr, port)
self.__executeCommand(cmd)

def srp_service_registration(
self,
host_name,
host_addr,
lease_info=(),
dst_addr_and_port=(),
services=[],
service_key: Literal["", "enable", "disable"] = "",
name_compression: Literal["", "enable", "disable"] = "",
):
self.__executeCommand("srp client host clear")
self.__executeCommand(f"srp client host name {host_name}")
self.__executeCommand(f"srp client host address {host_addr}")
self.__executeCommand(f"srp client leaseinterval {lease_info[0]}")
self.__executeCommand(f"srp client keyleaseinterval {lease_info[1]}")

if service_key != "":
self.__executeCommand(f"srp client service key {service_key}")

if name_compression != "":
self.__executeCommand(f"dns compression {name_compression}")

if dst_addr_and_port:
self.__executeCommand("srp client stop")
self.__executeCommand(f"dns config {dst_addr_and_port[0]} 53 6000 3 1")
self.__executeCommand(
f"srp client start {dst_addr_and_port[0]} {dst_addr_and_port[1]}"
)
else:
self.__executeCommand("srp client autostart enable")

for instance_name, service_name, port, priority, weight, txt in services:
cmd = f"srp client service add {instance_name} {service_name} {port} {priority} {weight} {txt}"
self.__executeCommand(cmd)

def dns_query(self, addr="", service_type="_thread-test._udp"):
"""Send unicast DNS query

Expand Down Expand Up @@ -4073,9 +4135,9 @@
cmd = "dns resolve4 %s.default.service.arpa" % host_name
self.__executeCommand(cmd)

def dns_resolve_ip(self, host_name, server_ip=""):
def dns_resolve_ip(self, host_name, server_ip="", ignore_exec=False):
cmd = "dns resolve %s %s" % (host_name, server_ip)
self.__executeCommand(cmd)
self.__executeCommand(cmd, ignore_exec=ignore_exec)

def dns_resolve_ipv4(self, host_name, server_ip="", ignore_exec=False):
cmd = "dns resolve4 %s %s" % (host_name, server_ip)
Expand Down Expand Up @@ -4154,14 +4216,20 @@
self.__executeCommand(cmd)

def dns_query_instance(
self, ip_addr, service_name, service_type="_thread-test._udp"
self,
service_name,
service_type="_thread-test._udp",
ip_addr="",
ignore_exec=False,
):
cmd = "dns service %s %s.default.service.arpa %s 53 6000 3 1" % (
cmd = "dns service %s %s.default.service.arpa" % (
service_name,
service_type,
ip_addr,
)
self.__executeCommand(cmd)
if ip_addr:
cmd += f" {ip_addr} 53 6000 3 1"
print(cmd)
Copy link

Copilot AI Nov 27, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Replace print statement with proper logging using self.log() as used elsewhere in the file.

Copilot uses AI. Check for mistakes.
self.__executeCommand(cmd, ignore_exec=ignore_exec)

def setMliid(self, sMlIid):
self.__setMlIid(sMlIid)
Expand All @@ -4174,6 +4242,10 @@
cmd = "srp client host remove"
self.__executeCommand(cmd)

def client_service_clear(self, instancename, servicename):
cmd = "srp client service clear %s %s" % (instancename, servicename)
self.__executeCommand(cmd)

def add_new_service(self):
rloc = self.__executeCommand("rloc16")[0]
mleid = self.__executeCommand("ipaddr mleid")[0]
Expand Down Expand Up @@ -4250,9 +4322,49 @@
# time.sleep(15)

@API
def add_service(self, service, server):
self.__executeCommand("service add 44970 %s %s" % (service, server))
self.__executeCommand("netdata register")
def add_service(self, service="", server="", service_data=False):
if service_data:
self.__executeCommand("service add 44970 %s%s" % (service, server))
self.__executeCommand("netdata register")
else:
self.__executeCommand("service add 44970 %s %s" % (service, server))
self.__executeCommand("netdata register")

def tcp_http_request(self, addr="", port=80, hexa_http_request=""):
self.__executeCommand("tcp init")
self.__executeCommand("tcp connect %s %s" % (addr, port))
time.sleep(3)
tcp_request = self.__executeCommand(
"tcp send -x %s" % hexa_http_request, ignore_exec=True
)
print(tcp_request)
Copy link

Copilot AI Nov 27, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Replace print statement with proper logging using self.log() as used elsewhere in the file.

Suggested change
print(tcp_request)
self.log(tcp_request)

Copilot uses AI. Check for mistakes.
time.sleep(3)
self.__executeCommand("tcp sendend", ignore_exec=True)
self.__executeCommand("tcp deinit", ignore_exec=True)
return "Failed" if tcp_request is True else True
Copy link

Copilot AI Nov 27, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The logic is inverted and confusing. When tcp_request is True (success with ignore_exec), it returns 'Failed' string. When it's not True, it returns True boolean. This should return consistent boolean values: return tcp_request is not True.

Suggested change
return "Failed" if tcp_request is True else True
return tcp_request is not True

Copilot uses AI. Check for mistakes.

def nat64_prefix(self):
netdata = self.__executeCommand("netdata show")
nat64 = ""
for addr in netdata:
if "/96" in addr:
nat64 = addr
break
return nat64

def pd_omr(self, prefix):
pd = self.__executeCommand("ipaddr")
result = ""
for pd_omr in pd[:-1]:
if ModuleHelper.GetFullIpv6Address(pd_omr).startswith(prefix):
result = pd_omr
break
else:
continue
Comment on lines +4362 to +4363
Copy link

Copilot AI Nov 27, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The else-continue is unnecessary. Remove the else clause since continue is only needed when the condition is false.

Suggested change
else:
continue
continue

Copilot uses AI. Check for mistakes.
if result == "":
prefix = ModuleHelper.GetFullIpv6Address(prefix + "::")
result = prefix
return result

########################################################################################################################
# 1.3 THCI Commands
Expand Down