Skip to content

Commit

Permalink
Iterating on T&C test scripts
Browse files Browse the repository at this point in the history 
./scripts/run_in_build_env.sh './scripts/build_python.sh --install_virtual_env out/venv'

export ZAP_DEVELOPMENT_PATH=/workspace/zap
./scripts/build/build_examples.py --target linux-x64-all-clusters build
./scripts/build/build_examples.py --target linux-x64-terms-and-conditions build

cat <<EOF > /tmp/test_env.yaml
ALL_CLUSTERS_APP: out/linux-x64-all-clusters/chip-all-clusters-app
TERMS_AND_CONDITIONS_APP: out/linux-x64-terms-and-conditions/chip-terms-and-conditions-app
EOF

rm /tmp/chip* ; ./scripts/run_in_python_env.sh ./out/venv './scripts/tests/run_python_test.py --load-from-env /tmp/test_env.yaml --script src/python_testing/TC_CGEN_2_6.py'
  • Loading branch information
@swan-amazon
swan-amazon committed Sep 23, 2024
1 parent 9283b64 commit 778d322
Show file tree
Hide file tree
Showing 4 changed files with 335 additions and 117 deletions.
206 changes: 110 additions & 96 deletions src/python_testing/TC_CGEN_2_5.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,113 +21,127 @@
# test-runner-run/run1/factoryreset: True
# test-runner-run/run1/quiet: True
# test-runner-run/run1/app-args: --KVS kvs1
# test-runner-run/run1/script-args: --commissioning-method on-network --qr-code MT:-24J0AFN00KA0648G00 --tc-version 1 --tc-user-response 1 --trace_file /tmp/chip_trace.log --trace_log 1 --trace_decode 1
# test-runner-run/run1/script-args: --in-test-commissioning-method on-network --qr-code MT:-24J0AFN00KA0648G00 --trace-to json:log
# === END CI TEST ARGUMENTS ===

import logging
import random

import asyncio
from typing import List
import chip.clusters as Clusters
from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main
from chip import ChipDeviceCtrl
from matter_testing_support import MatterBaseTest, TestStep, async_test_body, default_matter_test_main
from mobly import asserts


class TC_CGEN_2_5(MatterBaseTest):
async def verify_tc_attributes(self, expected_tc_version: int, expected_tc_acknowledgements: int):
attr = Clusters.GeneralCommissioning.Attributes.TCAcceptedVersion
tc_accepted_version = await self.read_single_attribute(dev_ctrl=self.th1, node_id=self.dut_node_id, endpoint=0, attribute=attr)
asserts.assert_equal(tc_accepted_version, expected_tc_version,
f"Expected TCAcceptedVersion to be {expected_tc_version}, but got {tc_accepted_version}")

attr = Clusters.GeneralCommissioning.Attributes.TCAcknowledgements
tc_acknowedgements = await self.read_single_attribute(dev_ctrl=self.th1, node_id=self.dut_node_id, endpoint=0, attribute=attr)
asserts.assert_equal(tc_acknowedgements, expected_tc_acknowledgements,
f"Expected TCAcknowledgements to be {expected_tc_acknowledgements}, but got {tc_acknowedgements}")
def desc_TC_CGEN_2_5(self) -> str:
return "[TC-CGEN-2.5] Verification For SetTCAcknowledgements [DUT as Server]"

def steps_TC_CGEN_2_5(self) -> list[TestStep]:
return [
TestStep("1", "TH starts commissioning the DUT. It performs all commissioning steps from ArmFailSafe, except SetTCAcknowledgements and CommissioningComplete.", is_commissioning=False),
TestStep("2", "TH reads TCAcknowledgementsRequired attribute from the DUT."),
TestStep("3", "TH sends SetTCAcknowledgements to DUT with the following values:\nTCVersion: Greater than or equal to TCMinRequiredVersion on DUT\nTCUserResponse: All terms required by DUT accepted"),
TestStep("4", "TH sends CommissioningComplete to DUT."),
TestStep("5", "TH reads TCAcceptedVersion attribute from the DUT."),
TestStep("6", "TH reads TCAcknowledgements attribute from the DUT."),
TestStep("7", "TH reads TCMinRequiredVersion attribute from the DUT."),
TestStep("8", "TH reads TCAcknowledgementsRequired attribute from the DUT."),
TestStep("9", "TH sends the SetTCAcknowledgements command to the DUT with the fields set as follows:\nTCVersion: 0\nTCUserResponse: 0"),
TestStep("10", "TH removes all fabrics from DUT with RemoveFabric."),
TestStep("11", "DUT is placed into a commissionable state."),
TestStep("12", "TH starts commissioning the DUT. It performs all commissioning steps, except for TC configuration with SetTCAcknowledgements."),
]

@async_test_body
async def test_TC_CGEN_2_5(self):
self.th1 = self.default_controller
self.discriminator = random.randint(0, 4095)

logging.info('Step 1 - TH reads the TCAcceptedVersion attribute')
attr = Clusters.GeneralCommissioning.Attributes.TCAcceptedVersion
tc_accepted_version = await self.read_single_attribute(dev_ctrl=self.th1, node_id=self.dut_node_id, endpoint=0, attribute=attr)

print(f"tc_accepted_version: {tc_accepted_version}")

logging.info('Step 2 - TH reads the TCAcknowledgements attribute')
attr = Clusters.GeneralCommissioning.Attributes.TCAcknowledgements
tc_acknowedgements = await self.read_single_attribute(dev_ctrl=self.th1, node_id=self.dut_node_id, endpoint=0, attribute=attr)
print(f"tc_acknowedgements: {tc_acknowedgements}")

logging.info('Step 3 - TH reads the TCMinRequiredVersion attribute')
attr = Clusters.GeneralCommissioning.Attributes.TCMinRequiredVersion
tc_min_requried_version = await self.read_single_attribute(dev_ctrl=self.th1, node_id=self.dut_node_id, endpoint=0, attribute=attr)
print(f"tc_min_requried_version: {tc_min_requried_version}")

logging.info('Step 4 - TH reads the TCAcknowledgementsRequired attribute')
attr = Clusters.GeneralCommissioning.Attributes.TCAcknowledgementsRequired
tc_required = await self.read_single_attribute(dev_ctrl=self.th1, node_id=self.dut_node_id, endpoint=0, attribute=attr)
print(f"tc_required: {tc_required}")

logging.info('Step 5 - TH sends SetTCAcknowledgements with greater values than current and verify set')
new_accepted_version = tc_accepted_version + 1
new_acknowledgements = 65535

cmd = Clusters.GeneralCommissioning.Commands.SetTCAcknowledgements(new_accepted_version, new_acknowledgements)
resp = await self.th1.SendCommand(nodeid=self.dut_node_id, endpoint=0, payload=cmd, timedRequestTimeoutMs=6000)
asserts.assert_equal(resp.errorCode, Clusters.GeneralCommissioning.Enums.CommissioningErrorEnum.kOk, 'Incorrect error code')
await self.verify_tc_attributes(new_accepted_version, new_acknowledgements)

logging.info('Step 6 - TH sends SetTCAcknowledgements with no accepted terms at version 0')
cmd = Clusters.GeneralCommissioning.Commands.SetTCAcknowledgements(0, 0)
resp = await self.th1.SendCommand(nodeid=self.dut_node_id, endpoint=0, payload=cmd, timedRequestTimeoutMs=6000)
asserts.assert_equal(
resp.errorCode, Clusters.GeneralCommissioning.Enums.CommissioningErrorEnum.kTCMinVersionNotMet, 'Incorrect error code'
)
await self.verify_tc_attributes(new_accepted_version, new_acknowledgements)

logging.info('Step 7 - TH sends SetTCAcknowledgements with no accepted terms at version 1')
cmd = Clusters.GeneralCommissioning.Commands.SetTCAcknowledgements(1, 0)
resp = await self.th1.SendCommand(nodeid=self.dut_node_id, endpoint=0, payload=cmd, timedRequestTimeoutMs=6000)
asserts.assert_equal(
resp.errorCode, Clusters.GeneralCommissioning.Enums.CommissioningErrorEnum.kRequiredTCNotAccepted, 'Incorrect error code'
)
await self.verify_tc_attributes(new_accepted_version, new_acknowledgements)

logging.info('Step 8 - TH sends ArmFailSafe with ExpiryLengthSeconds set to 60')
cmd = Clusters.GeneralCommissioning.Commands.ArmFailSafe(60)
resp = await self.th1.SendCommand(nodeid=self.dut_node_id, endpoint=0, payload=cmd, timedRequestTimeoutMs=6000)
asserts.assert_equal(resp.errorCode, Clusters.GeneralCommissioning.Enums.CommissioningErrorEnum.kOk,
'Incorrect error code')

logging.info('Step 9 - TH sends SetTCAcknowledgements with incremented TCVersion')
cmd = Clusters.GeneralCommissioning.Commands.SetTCAcknowledgements(new_accepted_version + 1, new_acknowledgements)
resp = await self.th1.SendCommand(nodeid=self.dut_node_id, endpoint=0, payload=cmd, timedRequestTimeoutMs=6000)
asserts.assert_equal(resp.errorCode, Clusters.GeneralCommissioning.Enums.CommissioningErrorEnum.kOk,
'Incorrect error code')

logging.info('Step 10 - TH sends ArmFailSafe with ExpiryLengthSeconds set to 0')
cmd = Clusters.GeneralCommissioning.Commands.ArmFailSafe(0)
resp = await self.th1.SendCommand(nodeid=self.dut_node_id, endpoint=0, payload=cmd, timedRequestTimeoutMs=6000)
asserts.assert_equal(resp.errorCode, Clusters.GeneralCommissioning.Enums.CommissioningErrorEnum.kOk,
'Incorrect error code')

# Verify that the TC attributes have been reset to the pre-failsafe state
await self.verify_tc_attributes(new_accepted_version, new_acknowledgements)

logging.info('Step 11 - TH removes all fabrics from the device')
fabrics = await self.read_single_attribute(dev_ctrl=self.th1, node_id=self.dut_node_id, endpoint=0, attribute=Clusters.OperationalCredentials.Attributes.Fabrics)
is_ci = self.check_pics('PICS_SDK_CI_ONLY')
commissioner: ChipDeviceCtrl.ChipDeviceController = self.default_controller

# Don't set TCs for the next commissioning and skip CommissioningComplete so we can manually call CommissioningComplete in order to check the response error code
self.step("1")
commissioner.SetTCRequired(False)
commissioner.SetSkipCommissioningComplete(True)
self.matter_test_config.commissioning_method = self.matter_test_config.in_test_commissioning_method
await self.commission_devices()

self.step("2")
# Verify that TCAcknowledgementsRequired value is representable in the 'Bool' type
# Verify that TCAcknowledgementsRequired value is True

self.step("3")
response: Clusters.GeneralCommissioning.Commands.SetTCAcknowledgementsResponse = await commissioner.SendCommand(
nodeid=self.dut_node_id,
endpoint=0,
payload=Clusters.GeneralCommissioning.Commands.SetTCAcknowledgements(TCVersion=2**16 - 1, TCUserResponse=2**16 - 1),
timedRequestTimeoutMs=1000)
# Verify that DUT sends SetTCAcknowledgementsResponse Command to TH With ErrorCode as 'OK'(0).
asserts.assert_equal(response.errorCode, Clusters.GeneralCommissioning.Enums.CommissioningErrorEnum.kOk, 'Incorrect error code')

self.step("4")
response: Clusters.GeneralCommissioning.Commands.CommissioningCompleteResponse = await commissioner.SendCommand(
nodeid=self.dut_node_id,
endpoint=0,
payload=Clusters.GeneralCommissioning.Commands.CommissioningComplete(),
timedRequestTimeoutMs=1000)
# Verify that DUT sends CommissioningCompleteResponse Command to TH With ErrorCode as 'OK'(0).
asserts.assert_equal(response.errorCode, Clusters.GeneralCommissioning.Enums.CommissioningErrorEnum.kOk, 'Incorrect error code')

self.step("5")
# Verify that TCAcceptedVersion value fits in the 'uint16' type
# Verify that TCAcceptedVersion is the value sent in step 3

self.step("6")
# Verify that TCAcknowledgements is a value representable in the map16 type
# Verify that TCAcknowledgements is the value sent in step 3

self.step("7")
# Verify that TCMinRequiredVersion value fits in the 'uint16' type

self.step("8")
# Verify that TCAcknowledgementsRequired value is False

self.step("9")
response: Clusters.GeneralCommissioning.Commands.SetTCAcknowledgementsResponse = await commissioner.SendCommand(
nodeid=self.dut_node_id,
endpoint=0,
payload=Clusters.GeneralCommissioning.Commands.SetTCAcknowledgements(TCVersion=0, TCUserResponse=0),
timedRequestTimeoutMs=1000)
# Verify that DUT sends SetTCAcknowledgementsResponse Command to TH With ErrorCode as 'TCMinVersionNotMet'(7)
asserts.assert_equal(response.errorCode, Clusters.GeneralCommissioning.Enums.CommissioningErrorEnum.kTCMinVersionNotMet, 'Incorrect error code')
# Verify that TCAcceptedVersion and TCAcknowledgements still contain the values sent in step 3.

self.step("10")
fabrics: List[Clusters.OperationalCredentials.Structs.FabricDescriptorStruct] = await self.read_single_attribute(
dev_ctrl=commissioner,
node_id=self.dut_node_id,
endpoint=0,
attribute=Clusters.OperationalCredentials.Attributes.Fabrics)

# Re-order the list of fabrics so that the test harness admin fabric is removed last
commissioner_fabric = next((fabric for fabric in fabrics if fabric.fabricIndex == commissioner.fabricId), None)
fabrics.remove(commissioner_fabric)
fabrics.append(commissioner_fabric)

for fabric in fabrics:
if fabric.fabricIndex == self.th1.fabricId:
continue
cmd = Clusters.OperationalCredentials.Commands.RemoveFabric(fabric.fabricIndex)
resp = await self.th1.SendCommand(nodeid=self.dut_node_id, endpoint=0, payload=cmd, timedRequestTimeoutMs=6000)
asserts.assert_equal(resp.statusCode, Clusters.OperationalCredentials.Enums.NodeOperationalCertStatusEnum.kOk)

# Remove TH1 fabric last
cmd = Clusters.OperationalCredentials.Commands.RemoveFabric(self.th1.fabricId)
resp = await self.th1.SendCommand(nodeid=self.dut_node_id, endpoint=0, payload=cmd, timedRequestTimeoutMs=6000)
response: Clusters.OperationalCredentials.Commands.NOCResponse = await commissioner.SendCommand(
nodeid=self.dut_node_id,
endpoint=0,
payload=Clusters.OperationalCredentials.Commands.RemoveFabric(fabric.fabricIndex),
timedRequestTimeoutMs=1000)
asserts.assert_equal(response.statusCode, Clusters.OperationalCredentials.Enums.NodeOperationalCertStatusEnum.kOk)

self.step("11")
self.wait_for_user_input(prompt_msg="Set the DUT into commissioning mode")
await asyncio.sleep(5)

self.step("12")
response: Clusters.GeneralCommissioning.Commands.CommissioningCompleteResponse = await commissioner.SendCommand(
nodeid=self.dut_node_id,
endpoint=0,
payload=Clusters.GeneralCommissioning.Commands.CommissioningComplete(),
timedRequestTimeoutMs=1000)
# Verify that DUT sends CommissioningCompleteResponse Command to TH With ErrorCode as 'TCAcknowledgementsNotReceived'(6).
asserts.assert_equal(response.errorCode, Clusters.GeneralCommissioning.Enums.CommissioningErrorEnum.kTCAcknowledgementsNotReceived, 'Incorrect error code')


if __name__ == "__main__":
Expand Down
Loading

0 comments on commit 778d322

Please sign in to comment.