Python testing mypy conformance (project-chip#38165)

* mypy type stubs for apps and tasks

* fix mypy warnings in taglist_and_topology_test.py

* fix mypy warnings in pics.py

* gh action yml

* gh workflow update

* run mypy from python env

* debug

* fix filename

* resolve mypy warnings in taglist file

* ignore mypy warning

* linter fixes

* lint fix 2

* resolve review comments

Author: asirko-soft

.github/workflows/mypy-validation.yml

@@ -0,0 +1,63 @@
+# Copyright (c) 2025 Project CHIP Authors
+#
+# 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.
+
+name: Mypy Type Validation
+
+on:
+  push:
+    branches: [ master ]
+    paths:
+      - 'src/python_testing/matter_testing_infrastructure/**/*.py'
+  pull_request:
+    paths:
+      - 'src/python_testing/matter_testing_infrastructure/**/*.py'
+
+jobs:
+  mypy-check:
+    runs-on: ubuntu-latest
+    if: github.actor != 'restyled-io[bot]'
+    
+    container:
+      image: ghcr.io/project-chip/chip-build:119
+      options: --privileged --sysctl "net.ipv6.conf.all.disable_ipv6=0 net.ipv4.conf.all.forwarding=1 net.ipv6.conf.all.forwarding=1"
+    
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+      
+      - name: Checkout submodules & Bootstrap
+        uses: ./.github/actions/checkout-submodules-and-bootstrap
+        with:
+          platform: linux
+      
+      - name: Build Python environment
+        run: |
+          scripts/run_in_build_env.sh './scripts/build_python.sh --install_virtual_env out/venv'
+      
+      - name: Run mypy validation
+        run: |
+          # List the directory to debug the file structure
+          ls -la src/python_testing/matter_testing_infrastructure/chip/testing/
+          
+          # TODO: Expand this list to include more files once they are mypy-compatible
+          # Eventually we should just check all files in the chip/testing directory
+          
+          ./scripts/run_in_python_env.sh out/venv "mypy --config-file=src/python_testing/matter_testing_infrastructure/mypy.ini \
+            src/python_testing/matter_testing_infrastructure/chip/testing/apps.py \
+            src/python_testing/matter_testing_infrastructure/chip/testing/tasks.py \
+            src/python_testing/matter_testing_infrastructure/chip/testing/taglist_and_topology_test.py \
+            src/python_testing/matter_testing_infrastructure/chip/testing/pics.py"
+          
+          # Print a reminder about expanding coverage
+          echo "⚠️ NOTE: Currently only checking a subset of files. Remember to expand coverage!" 

src/python_testing/matter_testing_infrastructure/chip/testing/pics.py

@@ -64,11 +64,26 @@ def parse_pics(lines: typing.List[str]) -> dict[str, bool]:
 
 
 def parse_pics_xml(contents: str) -> dict[str, bool]:
-    pics = {}
+    pics: dict[str, bool] = {}
     mytree = ET.fromstring(contents)
     for pi in mytree.iter('picsItem'):
-        name = pi.find('itemNumber').text
-        support = pi.find('support').text
+        name_elem = pi.find('itemNumber')
+        support_elem = pi.find('support')
+
+        # Raise an error if either element is None
+        if name_elem is None:
+            raise ValueError(f"PICS XML item missing 'itemNumber' element: {ET.tostring(pi, encoding='unicode')}")
+        if support_elem is None:
+            raise ValueError(f"PICS XML item missing 'support' element: {ET.tostring(pi, encoding='unicode')}")
+
+        # Raise an error if either text is None
+        name = name_elem.text
+        support = support_elem.text
+        if name is None:
+            raise ValueError(f"PICS XML item 'itemNumber' element missing text: {ET.tostring(pi, encoding='unicode')}")
+        if support is None:
+            raise ValueError(f"PICS XML item 'support' element missing text: {ET.tostring(pi, encoding='unicode')}")
+
         pics[name] = int(json.loads(support.lower())) == 1
     return pics
 

src/python_testing/matter_testing_infrastructure/chip/testing/taglist_and_topology_test.py

@@ -1,5 +1,5 @@
 #
-#    Copyright (c) 2023 Project CHIP Authors
+#    Copyright (c) 2025 Project CHIP Authors
 #    All rights reserved.
 #
 #    Licensed under the Apache License, Version 2.0 (the "License");
@@ -18,7 +18,7 @@
 import functools
 from collections import defaultdict
 from dataclasses import dataclass, field
-from typing import Any
+from typing import Any, Dict, List, Set, Tuple
 
 import chip.clusters as Clusters
 from chip.clusters.Types import Nullable
@@ -29,11 +29,11 @@ class TagProblem:
     root: int
     missing_attribute: bool
     missing_feature: bool
-    duplicates: set[int]
-    same_tag: set[int] = field(default_factory=set)
+    duplicates: Set[int]
+    same_tag: Set[int] = field(default_factory=set)
 
 
-def separate_endpoint_types(endpoint_dict: dict[int, Any]) -> tuple[list[int], list[int]]:
+def separate_endpoint_types(endpoint_dict: Dict[int, Any]) -> Tuple[List[int], List[int]]:
     """Returns a tuple containing the list of flat endpoints and a list of tree endpoints"""
     flat = []
     tree = []
@@ -52,13 +52,13 @@ def separate_endpoint_types(endpoint_dict: dict[int, Any]) -> tuple[list[int], l
     return (flat, tree)
 
 
-def get_all_children(endpoint_id, endpoint_dict: dict[int, Any]) -> set[int]:
+def get_all_children(endpoint_id: int, endpoint_dict: Dict[int, Any]) -> Set[int]:
     """Returns all the children (include subchildren) of the given endpoint
        This assumes we've already checked that there are no cycles, so we can do the dumb things and just trace the tree
     """
-    children = set()
+    children: Set[int] = set()
 
-    def add_children(endpoint_id, children):
+    def add_children(endpoint_id: int, children: Set[int]) -> None:
         immediate_children = endpoint_dict[endpoint_id][Clusters.Descriptor][Clusters.Descriptor.Attributes.PartsList]
         if not immediate_children:
             return
@@ -70,15 +70,16 @@ def add_children(endpoint_id, children):
     return children
 
 
-def find_tree_roots(tree_endpoints: list[int], endpoint_dict: dict[int, Any]) -> set[int]:
+def find_tree_roots(tree_endpoints: List[int], endpoint_dict: Dict[int, Any]) -> Set[int]:
     """Returns a set of all the endpoints in tree_endpoints that are roots for a tree (not include singletons)"""
     tree_roots = set()
 
-    def find_tree_root(current_id):
+    def find_tree_root(current_id: int) -> int:
         for endpoint_id, endpoint in endpoint_dict.items():
             if endpoint_id not in tree_endpoints:
                 continue
-            if current_id in endpoint[Clusters.Descriptor][Clusters.Descriptor.Attributes.PartsList]:
+            parts_list = endpoint[Clusters.Descriptor][Clusters.Descriptor.Attributes.PartsList]
+            if current_id in parts_list:
                 # this is not the root, move up
                 return find_tree_root(endpoint_id)
         return current_id
@@ -90,9 +91,9 @@ def find_tree_root(current_id):
     return tree_roots
 
 
-def parts_list_cycles(tree_endpoints: list[int], endpoint_dict: dict[int, Any]) -> list[int]:
+def parts_list_cycles(tree_endpoints: List[int], endpoint_dict: Dict[int, Any]) -> List[int]:
     """Returns a list of all the endpoints in the tree_endpoints list that contain cycles"""
-    def parts_list_cycle_detect(visited: set, current_id: int) -> bool:
+    def parts_list_cycle_detect(visited: Set[int], current_id: int) -> bool:
         if current_id in visited:
             return True
         visited.add(current_id)
@@ -105,28 +106,28 @@ def parts_list_cycle_detect(visited: set, current_id: int) -> bool:
     cycles = []
     # This is quick enough that we can do all the endpoints without searching for the roots
     for endpoint_id in tree_endpoints:
-        visited = set()
+        visited: Set[int] = set()
         if parts_list_cycle_detect(visited, endpoint_id):
             cycles.append(endpoint_id)
     return cycles
 
 
-def create_device_type_lists(roots: list[int], endpoint_dict: dict[int, Any]) -> dict[int, dict[int, set[int]]]:
+def create_device_type_lists(roots: List[int], endpoint_dict: Dict[int, Any]) -> Dict[int, Dict[int, Set[int]]]:
     """Returns a list of endpoints per device type for each root in the list"""
-    device_types = {}
+    device_types: Dict[int, Dict[int, Set[int]]] = {}
     for root in roots:
-        tree_device_types = defaultdict(set)
+        tree_device_types: Dict[int, Set[int]] = defaultdict(set)
         eps = get_all_children(root, endpoint_dict)
         eps.add(root)
         for ep in eps:
             for d in endpoint_dict[ep][Clusters.Descriptor][Clusters.Descriptor.Attributes.DeviceTypeList]:
                 tree_device_types[d.deviceType].add(ep)
-        device_types[root] = tree_device_types
+        device_types[root] = dict(tree_device_types)  # Convert defaultdict to dict before storing
 
     return device_types
 
 
-def get_direct_children_of_root(endpoint_dict: dict[int, Any]) -> set[int]:
+def get_direct_children_of_root(endpoint_dict: Dict[int, Any]) -> Set[int]:
     root_children = set(endpoint_dict[0][Clusters.Descriptor][Clusters.Descriptor.Attributes.PartsList])
     direct_children = root_children
     for ep in root_children:
@@ -135,7 +136,7 @@ def get_direct_children_of_root(endpoint_dict: dict[int, Any]) -> set[int]:
     return direct_children
 
 
-def create_device_type_list_for_root(direct_children, endpoint_dict: dict[int, Any]) -> dict[int, set[int]]:
+def create_device_type_list_for_root(direct_children: Set[int], endpoint_dict: Dict[int, Any]) -> Dict[int, Set[int]]:
     device_types = defaultdict(set)
     for ep in direct_children:
         for d in endpoint_dict[ep][Clusters.Descriptor][Clusters.Descriptor.Attributes.DeviceTypeList]:
@@ -147,19 +148,23 @@ def cmp_tag_list(a: Clusters.Descriptor.Structs.SemanticTagStruct, b: Clusters.D
     if type(a.mfgCode) != type(b.mfgCode):
         return -1 if type(a.mfgCode) is Nullable else 1
     if a.mfgCode != b.mfgCode:
-        return -1 if a.mfgCode < b.mfgCode else 1
+        # Adding type ignore for the comparison between potentially incompatible types
+        result = -1 if a.mfgCode < b.mfgCode else 1  # type: ignore
+        return result
     if a.namespaceID != b.namespaceID:
         return -1 if a.namespaceID < b.namespaceID else 1
     if a.tag != b.tag:
         return -1 if a.tag < b.tag else 1
     if type(a.label) != type(b.label):
         return -1 if type(a.label) is Nullable or a.label is None else 1
     if a.label != b.label:
-        return -1 if a.label < b.label else 1
+        # Adding type ignore for the comparison between potentially incompatible types
+        result = -1 if a.label < b.label else 1  # type: ignore
+        return result
     return 0
 
 
-def find_tag_list_problems(roots: list[int], device_types: dict[int, dict[int, set[int]]], endpoint_dict: dict[int, Any]) -> dict[int, TagProblem]:
+def find_tag_list_problems(roots: List[int], device_types: Dict[int, Dict[int, Set[int]]], endpoint_dict: Dict[int, Any]) -> Dict[int, TagProblem]:
     """Checks for non-spec compliant tag lists"""
     tag_problems = {}
     for root in roots:
@@ -196,7 +201,7 @@ def find_tag_list_problems(roots: list[int], device_types: dict[int, dict[int, s
     return tag_problems
 
 
-def flat_list_ok(flat_endpoint_id_to_check: int, endpoints_dict: dict[int, Any]) -> bool:
+def flat_list_ok(flat_endpoint_id_to_check: int, endpoints_dict: Dict[int, Any]) -> bool:
     '''Checks if the (flat) PartsList on the supplied endpoint contains all the sub-children of its parts.'''
     sub_children = set()
     for child in endpoints_dict[flat_endpoint_id_to_check][Clusters.Descriptor][Clusters.Descriptor.Attributes.PartsList]:

src/python_testing/matter_testing_infrastructure/chip/testing/tasks.py

@@ -154,7 +154,7 @@ def terminate(self):
         self.p.terminate()
         self.join()
 
-    def wait(self, timeout: Optional[float] = None) -> int:
+    def wait(self, timeout: Optional[float] = None) -> Optional[int]:
         """Wait for the subprocess to finish."""
         self.join(timeout)
         return self.returncode

src/python_testing/matter_testing_infrastructure/chip/typings/chip/testing/__init__.pyi

@@ -0,0 +1,2 @@
+# This file is a stub for the chip.testing package
+

src/python_testing/matter_testing_infrastructure/chip/typings/chip/testing/apps.pyi

@@ -0,0 +1,25 @@
+from typing import Any
+
+from chip.testing.tasks import Subprocess
+
+class AppServerSubprocess(Subprocess):
+    PREFIX: bytes
+
+    def __init__(self, app: str, storage_dir: str, discriminator: int,
+                 passcode: int, port: int = ...) -> None: ...
+
+    def __del__(self) -> None: ...
+
+    kvs_fd: int
+
+
+class IcdAppServerSubprocess(AppServerSubprocess):
+    paused: bool
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None: ...
+
+    def pause(self, check_state: bool = ...) -> None: ...
+
+    def resume(self, check_state: bool = ...) -> None: ...
+
+    def terminate(self) -> None: ...

src/python_testing/matter_testing_infrastructure/chip/typings/chip/testing/pics.pyi

@@ -0,0 +1,11 @@
+import typing
+
+def attribute_pics_str(pics_base: str, id: int) -> str: ...
+def accepted_cmd_pics_str(pics_base: str, id: int) -> str: ...
+def generated_cmd_pics_str(pics_base: str, id: int) -> str: ...
+def feature_pics_str(pics_base: str, bit: int) -> str: ...
+def server_pics_str(pics_base: str) -> str: ...
+def client_pics_str(pics_base: str) -> str: ...
+def parse_pics(lines: typing.List[str]) -> dict[str, bool]: ...
+def parse_pics_xml(contents: str) -> dict[str, bool]: ...
+def read_pics_from_file(path: str) -> dict[str, bool]: ...

src/python_testing/matter_testing_infrastructure/chip/typings/chip/testing/tasks.pyi

@@ -0,0 +1,43 @@
+import threading
+from typing import Any, BinaryIO, Callable, Optional, Pattern, Union
+
+def forward_f(f_in: BinaryIO, f_out: BinaryIO,
+              cb: Optional[Callable[[bytes, bool], bytes]] = ...,
+              is_stderr: bool = ...) -> None: ...
+
+
+class Subprocess(threading.Thread):
+    program: str
+    args: tuple[str, ...]
+    output_cb: Optional[Callable[[bytes, bool], bytes]]
+    f_stdout: BinaryIO
+    f_stderr: BinaryIO
+    output_match: Optional[Pattern[bytes]]
+    returncode: Optional[int]
+    p: Any
+    event: threading.Event
+    event_started: threading.Event
+    expected_output: Optional[Union[str, Pattern[bytes]]]
+
+    def __init__(self, program: str, *args: str,
+                 output_cb: Optional[Callable[[bytes, bool], bytes]] = ...,
+                 f_stdout: BinaryIO = ...,
+                 f_stderr: BinaryIO = ...) -> None: ...
+
+    def _set_output_match(self, pattern: Union[str, Pattern[bytes]]) -> None: ...
+
+    def _check_output(self, line: bytes, is_stderr: bool) -> bytes: ...
+
+    def run(self) -> None: ...
+
+    def start(self,
+              expected_output: Optional[Union[str, Pattern[bytes]]] = ...,
+              timeout: Optional[float] = ...) -> None: ...
+
+    def send(self, message: str, end: str = ...,
+             expected_output: Optional[Union[str, Pattern[bytes]]] = ...,
+             timeout: Optional[float] = ...) -> None: ...
+
+    def terminate(self) -> None: ...
+
+    def wait(self, timeout: Optional[float] = ...) -> Optional[int]: ...

src/python_testing/matter_testing_infrastructure/mypy.ini

@@ -0,0 +1,6 @@
+[mypy]
+mypy_path = chip/typings
+namespace_packages = True
+warn_unused_configs = True
+ignore_missing_imports = True
+explicit_package_bases = True