Skip to content

Commit

Permalink
Updating method for attributes_guard functionality
Browse files Browse the repository at this point in the history
  • Loading branch information
@j-ororke
j-ororke committed Sep 30, 2024
1 parent fd714b8 commit f090844
Show file tree
Hide file tree
Showing 3 changed files with 27 additions and 27 deletions.
34 changes: 17 additions & 17 deletions src/python_testing/TC_OpstateCommon.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@
from chip.clusters.Attribute import EventReadResult, SubscriptionTransaction
from chip.clusters.Types import NullValue
from chip.interaction_model import InteractionModelError, Status
from matter_testing_support import ClusterAttributeChangeAccumulator, EventChangeCallback, TestStep
from matter_testing_support import ClusterAttributeChangeAccumulator, EventChangeCallback, TestStep, has_attribute
from mobly import asserts


Expand Down Expand Up @@ -248,7 +248,7 @@ async def TEST_TC_OPSTATE_BASE_1_1(self, endpoint=1, cluster_revision=1, feature
attributes.ClusterRevision.attribute_id
]

if self.attributes_guard(attributes.CountdownTime):
if self.attributes_guard(has_attribute(attributes.CountdownTime)):
expected_value.append(attributes.CountdownTime.attribute_id)

await self.read_and_expect_array_contains(endpoint=endpoint,
Expand Down Expand Up @@ -347,7 +347,7 @@ async def TEST_TC_OPSTATE_BASE_2_1(self, endpoint=1):

# STEP 2: TH reads from the DUT the PhaseList attribute
self.step(2)
if self.attributes_guard(attributes.PhaseList):
if self.attributes_guard(has_attribute(attributes.PhaseList)):
phase_list = await self.read_expect_success(endpoint=endpoint,
attribute=attributes.PhaseList)
if phase_list is not NullValue:
Expand All @@ -357,7 +357,7 @@ async def TEST_TC_OPSTATE_BASE_2_1(self, endpoint=1):

# STEP 3: TH reads from the DUT the CurrentPhase attribute
self.step(3)
if self.attributes_guard(attributes.CurrentPhase):
if self.attributes_guard(has_attribute(attributes.CurrentPhase)):
current_phase = await self.read_expect_success(endpoint=endpoint,
attribute=attributes.CurrentPhase)
if (phase_list == NullValue) or (not phase_list):
Expand All @@ -369,7 +369,7 @@ async def TEST_TC_OPSTATE_BASE_2_1(self, endpoint=1):

# STEP 4: TH reads from the DUT the CountdownTime attribute
self.step(4)
if self.attributes_guard(attributes.CountdownTime):
if self.attributes_guard(has_attribute(attributes.CountdownTime)):
countdown_time = await self.read_expect_success(endpoint=endpoint,
attribute=attributes.CountdownTime)
if countdown_time is not NullValue:
Expand All @@ -378,7 +378,7 @@ async def TEST_TC_OPSTATE_BASE_2_1(self, endpoint=1):

# STEP 5: TH reads from the DUT the OperationalStateList attribute
self.step(5)
if self.attributes_guard(attributes.OperationalStateList):
if self.attributes_guard(has_attribute(attributes.OperationalStateList)):
operational_state_list = await self.read_expect_success(endpoint=endpoint,
attribute=attributes.OperationalStateList)
defined_states = [state.value for state in cluster.Enums.OperationalStateEnum
Expand Down Expand Up @@ -464,7 +464,7 @@ async def TEST_TC_OPSTATE_BASE_2_1(self, endpoint=1):

# STEP 7: TH reads from the DUT the OperationalError attribute
self.step(7)
if self.attributes_guard(attributes.OperationalError):
if self.attributes_guard(has_attribute(attributes.OperationalError)):
operational_error = await self.read_expect_success(endpoint=endpoint,
attribute=attributes.OperationalError)
# Defined Errors
Expand Down Expand Up @@ -600,7 +600,7 @@ async def TEST_TC_OPSTATE_BASE_2_2(self, endpoint=1):

# STEP 3: TH reads from the DUT the OperationalStateList attribute
self.step(3)
if self.attributes_guard(attributes.OperationalStateList):
if self.attributes_guard(has_attribute(attributes.OperationalStateList)):
operational_state_list = await self.read_expect_success(endpoint=endpoint,
attribute=attributes.OperationalStateList)

Expand Down Expand Up @@ -628,15 +628,15 @@ async def TEST_TC_OPSTATE_BASE_2_2(self, endpoint=1):

# STEP 6: TH reads from the DUT the OperationalError attribute
self.step(6)
if self.attributes_guard(attributes.OperationalError):
if self.attributes_guard(has_attribute(attributes.OperationalError)):
await self.read_and_expect_property_value(endpoint=endpoint,
attribute=attributes.OperationalError,
attr_property="errorStateID",
expected_value=cluster.Enums.ErrorStateEnum.kNoError)

# STEP 7: TH reads from the DUT the CountdownTime attribute
self.step(7)
if self.attributes_guard(attributes.CountdownTime):
if self.attributes_guard(has_attribute(attributes.CountdownTime)):
initial_countdown_time = await self.read_expect_success(endpoint=endpoint,
attribute=attributes.CountdownTime)
if initial_countdown_time is not NullValue:
Expand All @@ -645,7 +645,7 @@ async def TEST_TC_OPSTATE_BASE_2_2(self, endpoint=1):

# STEP 8: TH reads from the DUT the PhaseList attribute
self.step(8)
if self.attributes_guard(attributes.PhaseList):
if self.attributes_guard(has_attribute(attributes.PhaseList)):
phase_list = await self.read_expect_success(endpoint=endpoint,
attribute=attributes.PhaseList)
phase_list_len = 0
Expand All @@ -656,7 +656,7 @@ async def TEST_TC_OPSTATE_BASE_2_2(self, endpoint=1):

# STEP 9: TH reads from the DUT the CurrentPhase attribute
self.step(9)
if self.attributes_guard(attributes.CurrentPhase):
if self.attributes_guard(has_attribute(attributes.CurrentPhase)):
current_phase = await self.read_expect_success(endpoint=endpoint,
attribute=attributes.CurrentPhase)
if (phase_list == NullValue) or (not phase_list):
Expand All @@ -669,12 +669,12 @@ async def TEST_TC_OPSTATE_BASE_2_2(self, endpoint=1):

# STEP 10: TH waits for {PIXIT.WAITTIME.COUNTDOWN}
self.step(10)
if self.attributes_guard(attributes.CountdownTime):
if self.attributes_guard(has_attribute(attributes.CountdownTime)):
time.sleep(wait_time)

# STEP 11: TH reads from the DUT the CountdownTime attribute
self.step(11)
if self.attributes_guard(attributes.CountdownTime):
if self.attributes_guard(has_attribute(attributes.CountdownTime)):
countdown_time = await self.read_expect_success(endpoint=endpoint,
attribute=attributes.CountdownTime)

Expand Down Expand Up @@ -815,7 +815,7 @@ async def TEST_TC_OPSTATE_BASE_2_3(self, endpoint=1):

# STEP 6: TH reads from the DUT the CountdownTime attribute
self.step(6)
if self.attributes_guard(attributes.CountdownTime):
if self.attributes_guard(has_attribute(attributes.CountdownTime)):
initial_countdown_time = await self.read_expect_success(endpoint=endpoint,
attribute=attributes.CountdownTime)
if initial_countdown_time is not NullValue:
Expand All @@ -829,7 +829,7 @@ async def TEST_TC_OPSTATE_BASE_2_3(self, endpoint=1):

# STEP 8: TH reads from the DUT the CountdownTime attribute
self.step(8)
if self.attributes_guard(attributes.CountdownTime):
if self.attributes_guard(has_attribute(attributes.CountdownTime)):
countdown_time = await self.read_expect_success(endpoint=endpoint,
attribute=attributes.CountdownTime)

Expand Down Expand Up @@ -1061,7 +1061,7 @@ async def TEST_TC_OPSTATE_BASE_2_5(self, endpoint=1):

# STEP 5: TH reads from the DUT the CountdownTime attribute
self.step(5)
if self.attributes_guard(attributes.CountdownTime):
if self.attributes_guard(has_attribute(attributes.CountdownTime)):
initial_countdown_time = await self.read_expect_success(endpoint=endpoint,
attribute=attributes.CountdownTime)

Expand Down
14 changes: 7 additions & 7 deletions src/python_testing/TC_RVCOPSTATE_2_1.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@

import chip.clusters as Clusters
from chip.clusters.Types import NullValue
from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main
from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main, has_attribute
from mobly import asserts


Expand Down Expand Up @@ -94,7 +94,7 @@ async def test_TC_RVCOPSTATE_2_1(self):
if self.is_ci:
self.write_to_app_pipe({"Name": "Reset"})

if self.attributes_guard(attributes.PhaseList):
if self.attributes_guard(has_attribute(attributes.PhaseList)):
self.print_step(2, "Read PhaseList attribute")
phase_list = await self.read_mod_attribute_expect_success(endpoint=self.endpoint, attribute=attributes.PhaseList)

Expand All @@ -107,7 +107,7 @@ async def test_TC_RVCOPSTATE_2_1(self):

asserts.assert_less_equal(phase_list_len, 32, "PhaseList length(%d) must be less than 32!" % phase_list_len)

if self.attributes_guard(attributes.CurrentPhase):
if self.attributes_guard(has_attribute(attributes.CurrentPhase)):
self.print_step(3, "Read CurrentPhase attribute")
current_phase = await self.read_mod_attribute_expect_success(endpoint=self.endpoint, attribute=attributes.CurrentPhase)
logging.info("CurrentPhase: %s" % (current_phase))
Expand All @@ -118,7 +118,7 @@ async def test_TC_RVCOPSTATE_2_1(self):
asserts.assert_true(0 <= current_phase < phase_list_len,
"CurrentPhase(%s) must be between 0 and %d" % (current_phase, (phase_list_len - 1)))

if self.attributes_guard(attributes.CountdownTime):
if self.attributes_guard(has_attribute(attributes.CountdownTime)):
self.print_step(4, "Read CountdownTime attribute")
countdown_time = await self.read_mod_attribute_expect_success(endpoint=self.endpoint,
attribute=attributes.CountdownTime)
Expand All @@ -128,7 +128,7 @@ async def test_TC_RVCOPSTATE_2_1(self):
asserts.assert_true(countdown_time >= 0 and countdown_time <= 259200,
"CountdownTime(%s) must be between 0 and 259200" % countdown_time)

if self.attributes_guard(attributes.OperationalStateList):
if self.attributes_guard(has_attribute(attributes.OperationalStateList)):
self.print_step(5, "Read OperationalStateList attribute")
operational_state_list = await self.read_mod_attribute_expect_success(endpoint=self.endpoint,
attribute=attributes.OperationalStateList)
Expand All @@ -151,7 +151,7 @@ async def test_TC_RVCOPSTATE_2_1(self):

asserts.assert_true(error_state_present, "The OperationalStateList does not have an ID entry of Error(0x03)")

if self.attributes_guard(attributes.OperationalState):
if self.attributes_guard(has_attribute(attributes.OperationalState)):
self.print_step(6, "Read OperationalState attribute")
operational_state = await self.read_mod_attribute_expect_success(endpoint=self.endpoint,
attribute=attributes.OperationalState)
Expand Down Expand Up @@ -218,7 +218,7 @@ async def test_TC_RVCOPSTATE_2_1(self):
self.wait_for_user_input(prompt_msg=f"{test_step}, and press Enter when done.\n")
await self.read_and_validate_opstate(step="6n", expected_state=Clusters.RvcOperationalState.Enums.OperationalStateEnum.kDocked)

if self.attributes_guard(attributes.OperationalError):
if self.attributes_guard(has_attribute(attributes.OperationalError)):
self.print_step(7, "Read OperationalError attribute")
operational_error = await self.read_mod_attribute_expect_success(endpoint=self.endpoint,
attribute=attributes.OperationalError)
Expand Down
6 changes: 3 additions & 3 deletions src/python_testing/matter_testing_support.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1410,7 +1410,7 @@ def pics_guard(self, pics_condition: bool):
self.mark_current_step_skipped()
return pics_condition

def attributes_guard(self, attribute):
def attributes_guard(self, attribute_condition: bool):
"""Similar to pics_guard above, except checks a condition and if False marks the test step as skipped and
returns False using attributes against attributes_list, otherwise returns True.
For example can be used to check if a test step should be run:
Expand All @@ -1423,9 +1423,9 @@ def attributes_guard(self, attribute):
if self.attribute_guard(condition2_needs_to_be_false_to_skip_step):
# skip step 2 if condition not met
"""
if not has_attribute(attribute):
if not attribute_condition:
self.mark_current_step_skipped()
return attribute
return attribute_condition

def mark_current_step_skipped(self):
try:
Expand Down

0 comments on commit f090844

Please sign in to comment.