Follow-up Persistent Subscription tests

BUILD.gn

@@ -357,6 +357,10 @@ if (current_toolchain != "${dir_pw_toolchain}/default:default") {
     enable_linux_lock_app_build =
         enable_default_builds && (host_os == "linux" || host_os == "mac")
 
+    # Build the Linux LIT ICD example.
+    enable_linux_lit_icd_app_build =
+        enable_default_builds && (host_os == "linux" || host_os == "mac")
+
     # Build the cc13x2x7_26x2x7 lock app example.
     enable_cc13x2x7_26x2x7_lock_app_build = enable_ti_simplelink_builds
 
@@ -610,6 +614,15 @@ if (current_toolchain != "${dir_pw_toolchain}/default:default") {
     extra_build_deps += [ ":linux_lock_app" ]
   }
 
+  if (enable_linux_lit_icd_app_build) {
+    group("linux_lit_icd_app") {
+      deps =
+          [ "${chip_root}/examples/lit-icd-app/linux(${standalone_toolchain})" ]
+    }
+
+    extra_build_deps += [ ":linux_lit_icd_app" ]
+  }
+
   if (enable_efr32_lock_app_build) {
     group("efr32_lock_app") {
       deps = [ "${chip_root}/examples/lock-app/efr32(${chip_root}/config/efr32/toolchain:efr32_lock_app)" ]

examples/lit-icd-app/lit-icd-common/lit-icd-server-app.matter

@@ -1670,6 +1670,7 @@ endpoint 0 {
     ram      attribute clusterRevision default = 1;
 
     handle command OpenCommissioningWindow;
+    handle command OpenBasicCommissioningWindow;
     handle command RevokeCommissioning;
   }
 

examples/lit-icd-app/lit-icd-common/lit-icd-server-app.zap

@@ -2549,6 +2549,14 @@
               "isIncoming": 1,
               "isEnabled": 1
             },
+            {
+              "name": "OpenBasicCommissioningWindow",
+              "code": 1,
+              "mfgCode": null,
+              "source": "client",
+              "isIncoming": 1,
+              "isEnabled": 1
+            },
             {
               "name": "RevokeCommissioning",
               "code": 2,

examples/platform/linux/AppMain.cpp

@@ -568,6 +568,13 @@ void ChipLinuxAppMainLoop(AppMainLoopImplementation * impl)
     // Init ZCL Data Model and CHIP App Server
     Server::GetInstance().Init(initParams);
 
+#if CONFIG_BUILD_FOR_HOST_UNIT_TEST
+    // Set ReadHandler Capacity for Subscriptions
+    chip::app::InteractionModelEngine::GetInstance()->SetHandlerCapacityForSubscriptions(
+        LinuxDeviceOptions::GetInstance().subscriptionCapacity);
+    chip::app::InteractionModelEngine::GetInstance()->SetForceHandlerQuota(true);
+#endif
+
     // Now that the server has started and we are done with our startup logging,
     // log our discovery/onboarding information again so it's not lost in the
     // noise.

examples/platform/linux/Options.cpp

@@ -87,6 +87,9 @@ enum
 #if defined(PW_RPC_ENABLED)
     kOptionRpcServerPort = 0x1023,
 #endif
+#if CONFIG_BUILD_FOR_HOST_UNIT_TEST
+    kDeviceOption_SubscriptionCapacity = 0x1024,
+#endif
 };
 
 constexpr unsigned kAppUsageLength = 64;
@@ -143,6 +146,9 @@ OptionDef sDeviceOptionDefs[] = {
     { "simulate-no-internal-time", kNoArgument, kOptionSimulateNoInternalTime },
 #if defined(PW_RPC_ENABLED)
     { "rpc-server-port", kArgumentRequired, kOptionRpcServerPort },
+#endif
+#if CONFIG_BUILD_FOR_HOST_UNIT_TEST
+    { "subscription-capacity", kArgumentRequired, kDeviceOption_SubscriptionCapacity },
 #endif
     {}
 };
@@ -263,6 +269,10 @@ const char * sDeviceOptionHelp =
 #if defined(PW_RPC_ENABLED)
     "  --rpc-server-port\n"
     "       Start RPC server on specified port\n"
+#endif
+#if CONFIG_BUILD_FOR_HOST_UNIT_TEST
+    "  --subscription-capacity\n"
+    "       Max number of subscriptions the device will allow\n"
 #endif
     "\n";
 
@@ -521,6 +531,11 @@ bool HandleOption(const char * aProgram, OptionSet * aOptions, int aIdentifier,
     case kOptionRpcServerPort:
         LinuxDeviceOptions::GetInstance().rpcServerPort = static_cast<uint16_t>(atoi(aValue));
         break;
+#endif
+#if CONFIG_BUILD_FOR_HOST_UNIT_TEST
+    case kDeviceOption_SubscriptionCapacity:
+        LinuxDeviceOptions::GetInstance().subscriptionCapacity = static_cast<int32_t>(atoi(aValue));
+        break;
 #endif
     default:
         PrintArgError("%s: INTERNAL ERROR: Unhandled option: %s\n", aProgram, aName);

examples/platform/linux/Options.h

@@ -70,6 +70,9 @@ struct LinuxDeviceOptions
     bool mSimulateNoInternalTime = false;
 #if defined(PW_RPC_ENABLED)
     uint16_t rpcServerPort = 33000;
+#endif
+#if CONFIG_BUILD_FOR_HOST_UNIT_TEST
+    int32_t subscriptionCapacity = CHIP_IM_MAX_NUM_SUBSCRIPTIONS;
 #endif
     static LinuxDeviceOptions & GetInstance();
 };

integrations/docker/images/stage-2/chip-cirque-device-base/Dockerfile

@@ -30,6 +30,7 @@ RUN apt-get update \
   libgirepository1.0-dev \
   libglib2.0-dev \
   libjpeg-dev \
+  openssh-server \
   psmisc \
   python3-dev \
   python3-pip \
@@ -55,7 +56,12 @@ RUN apt-get update \
   && echo "ctrl_interface=/run/wpa_supplicant" >> /etc/wpa_supplicant/wpa_supplicant.conf \
   && echo "update_config=1" >> /etc/wpa_supplicant/wpa_supplicant.conf \
   && rm -rf /var/lib/apt/lists/* \
-  && pip3 install --no-cache-dir click==8.0.3
+  && pip3 install --no-cache-dir click==8.0.3 paramiko \
+  && mkdir /var/run/sshd \
+  && echo 'root:admin' | chpasswd \
+  && sed -i 's/#Port 22/Port 2222/' /etc/ssh/sshd_config \
+  && sed -i 's/#PermitRootLogin prohibit-password/PermitRootLogin yes/' /etc/ssh/sshd_config \
+  && sed 's@session\s*required\s*pam_loginuid.so@session optional pam_loginuid.so@g' -i /etc/pam.d/sshd
 
 COPY CHIPCirqueDaemon.py /bin/CHIPCirqueDaemon.py
 COPY entrypoint.sh /opt/entrypoint.sh
@@ -65,3 +71,4 @@ WORKDIR /
 ENTRYPOINT ["/opt/entrypoint.sh"]
 
 EXPOSE 80
+EXPOSE 2222

scripts/build/gn_gen_cirque.sh

@@ -36,7 +36,7 @@ echo "Setup build environment"
 source "./scripts/activate.sh"
 
 echo "Build: GN configure"
-gn --root="$CHIP_ROOT" gen --check --fail-on-unused-args out/debug --args='target_os="all"'"chip_build_tests=false chip_enable_wifi=false chip_im_force_fabric_quota_check=true enable_default_builds=false enable_host_gcc_build=true enable_standalone_chip_tool_build=true enable_linux_all_clusters_app_build=true enable_linux_lighting_app_build=true"
+gn --root="$CHIP_ROOT" gen --check --fail-on-unused-args out/debug --args='target_os="all"'"chip_build_tests=false chip_enable_wifi=false chip_im_force_fabric_quota_check=true enable_default_builds=false enable_host_gcc_build=true enable_standalone_chip_tool_build=true enable_linux_all_clusters_app_build=true enable_linux_lighting_app_build=true enable_linux_lit_icd_app_build=true"
 
 echo "Build: Ninja build"
 time ninja -C out/debug all check

scripts/tests/cirque_tests.sh

@@ -49,6 +49,8 @@ CIRQUE_TESTS=(
     "CommissioningFailureOnReportTest"
     "PythonCommissioningTest"
     "CommissioningWindowTest"
+    "SubscriptionResumptionTest"
+    "SubscriptionResumptionCapacityTest"
 )
 
 BOLD_GREEN_TEXT="\033[1;32m"

src/controller/python/chip/ChipDeviceCtrl.py

@@ -1417,7 +1417,8 @@ def ZCLWriteAttribute(self, cluster: str, attribute: str, nodeid, endpoint, grou
 
         return asyncio.run(self.WriteAttribute(nodeid, [(endpoint, req, dataVersion)]))
 
-    def ZCLSubscribeAttribute(self, cluster, attribute, nodeid, endpoint, minInterval, maxInterval, blocking=True):
+    def ZCLSubscribeAttribute(self, cluster, attribute, nodeid, endpoint, minInterval, maxInterval, blocking=True,
+                              keepSubscriptions=False, autoResubscribe=True):
         ''' Wrapper over ReadAttribute for a single attribute
             Returns a SubscriptionTransaction. See ReadAttribute for more information.
         '''
@@ -1428,7 +1429,8 @@ def ZCLSubscribeAttribute(self, cluster, attribute, nodeid, endpoint, minInterva
             req = eval(f"GeneratedObjects.{cluster}.Attributes.{attribute}")
         except BaseException:
             raise UnknownAttribute(cluster, attribute)
-        return asyncio.run(self.ReadAttribute(nodeid, [(endpoint, req)], None, False, reportInterval=(minInterval, maxInterval)))
+        return asyncio.run(self.ReadAttribute(nodeid, [(endpoint, req)], None, False, reportInterval=(minInterval, maxInterval),
+                                              keepSubscriptions=keepSubscriptions, autoResubscribe=autoResubscribe))
 
     def ZCLCommandList(self):
         self.CheckIsActive()

src/controller/python/test/test_scripts/base.py

@@ -40,6 +40,7 @@
 from chip.ChipStack import ChipStack
 from chip.crypto import p256keypair
 from chip.utils import CommissioningBuildingBlocks
+from cirque_restart_remote_device import restartRemoteDevice
 
 logger = logging.getLogger('PythonMatterControllerTEST')
 logger.setLevel(logging.INFO)
@@ -336,6 +337,21 @@ def TestCommissioningWithSetupPayload(self, setupPayload: str, nodeid: int):
         self.logger.info("Commissioning finished.")
         return True
 
+    def TestOnNetworkCommissioning(self, discriminator: int, setuppin: int, nodeid: int, ip_override: str = None):
+        self.logger.info("Testing discovery")
+        device = self.TestDiscovery(discriminator=discriminator)
+        if not device:
+            self.logger.info("Failed to discover any devices.")
+            return False
+        address = device.addresses[0]
+        if ip_override:
+            address = ip_override
+        self.logger.info("Testing commissioning")
+        if not self.TestCommissioning(address, setuppin, nodeid):
+            self.logger.info("Failed to finish commissioning")
+            return False
+        return True
+
     def TestUsedTestCommissioner(self):
         return self.devCtrl.GetTestCommissionerUsed()
 
@@ -1316,3 +1332,148 @@ def TestFabricScopedCommandDuringPase(self, nodeid: int):
             status = ex.status
 
         return status == IM.Status.UnsupportedAccess
+
+    def TestSubscriptionResumption(self, nodeid: int, endpoint: int, remote_ip: str, ssh_port: int, remote_server_app: str):
+        '''
+        This test validates that the device can resume the subscriptions after restarting.
+        It is executed in Linux Cirque tests and the steps of this test are:
+        1. Subscription the NodeLable attribute on BasicInformation cluster with the controller
+        2. Restart the remote server app
+        3. Validate that the controller can receive a report from the remote server app
+        '''
+        desiredPath = None
+        receivedUpdate = False
+        updateLock = threading.Lock()
+        updateCv = threading.Condition(updateLock)
+
+        def OnValueReport(path: Attribute.TypedAttributePath, transaction: Attribute.SubscriptionTransaction) -> None:
+            nonlocal desiredPath, updateCv, updateLock, receivedUpdate
+            if path.Path != desiredPath:
+                return
+
+            data = transaction.GetAttribute(path)
+            logger.info(
+                f"Received report from server: path: {path.Path}, value: {data}")
+            with updateLock:
+                receivedUpdate = True
+                updateCv.notify_all()
+
+        try:
+            desiredPath = Clusters.Attribute.AttributePath(
+                EndpointId=0, ClusterId=0x28, AttributeId=5)
+            # BasicInformation Cluster, NodeLabel Attribute
+            subscription = self.devCtrl.ZCLSubscribeAttribute(
+                "BasicInformation", "NodeLabel", nodeid, endpoint, 1, 50, keepSubscriptions=True, autoResubscribe=False)
+            subscription.SetAttributeUpdateCallback(OnValueReport)
+
+            self.logger.info("Restart remote deivce")
+            restartRemoteThread = restartRemoteDevice(
+                remote_ip, ssh_port, "root", "admin", remote_server_app, "--thread --discriminator 3840")
+            restartRemoteThread.start()
+            # After device restarts, the attribute will be set dirty so the subscription can receive
+            # the update
+            with updateCv:
+                while receivedUpdate is False:
+                    if not updateCv.wait(10.0):
+                        self.logger.error(
+                            "Failed to receive subscription resumption report")
+                        break
+
+            restartRemoteThread.join(10.0)
+
+            #
+            # Clean-up by shutting down the sub. Otherwise, we're going to get callbacks through
+            # OnValueChange on what will soon become an invalid execution context above.
+            #
+            subscription.Shutdown()
+
+            if restartRemoteThread.is_alive():
+                # Thread join timed out
+                self.logger.error("Failed to join change thread")
+                return False
+
+            return receivedUpdate
+
+        except Exception as ex:
+            self.logger.exception(f"Failed to finish API test: {ex}")
+            return False
+
+        return True
+
+    '''
+    The SubscriptionResumptionCapacity Cirque Test is to verify that the device can still handle new subscription
+    requests when resuming the maximum subscriptions. The steps for this test are:
+    1. Commission the server app to the first fabric and send maximum subscription requests from the controller in
+    the first fabric to establish maximum subscriptions.
+    2. Open the commissioning window to make the server app can be commissioned to the second fabric.
+    3. Shutdown the controller in the first fabric to extend the time of resuming subscriptions. The server app will
+    keep resolving the address of the first controller for a while after rebooting.
+    4. Commission the server app to the second fabric.
+    5. Restart the server app and the server app will start resuming subscriptions. Since the first controller is
+    shutdown, the server app will keep resolving the address of the first controller for a while and the subscription
+    resumption will not fail so quickly.
+    6. When the server app is resuming subscriptions, send a new subscription request from the second controller.
+    Verify that the device can still handle this subscription.
+
+    BaseTestHelper provides two controllers. However, if using the two controller (devCtrl and devCtrl2) in one
+    MobileDevice to execute this Cirque test, the CHIPEndDevice can still resolve the address for first controller
+    even if the first controller is shutdown by 'self.devCtrl.Shutdown()'. And the server will fail to establish the
+    subscriptions immediately, which makes it hard to send the new subscription request from the second controller
+    at the time of server app resuming maximum subscriptions.
+    So we will use two controller containers for this test and divide the test to two steps. The Step1 is executed in
+    controller 1 in container 1 while the Step2 is executed in controller 2 in container 2
+    '''
+
+    def TestSubscriptionResumptionCapacityStep1(self, nodeid: int, endpoint: int, subscription_capacity: int):
+        try:
+            # BasicInformation Cluster, NodeLabel Attribute
+            for i in range(subscription_capacity):
+                self.devCtrl.ZCLSubscribeAttribute(
+                    "BasicInformation", "NodeLabel", nodeid, endpoint, 1, 50, keepSubscriptions=True, autoResubscribe=False)
+
+            logger.info("Send OpenBasicCommissioningWindow command on fist controller")
+            asyncio.run(
+                self.devCtrl.SendCommand(
+                    nodeid,
+                    0,
+                    Clusters.AdministratorCommissioning.Commands.OpenBasicCommissioningWindow(180),
+                    timedRequestTimeoutMs=10000
+                ))
+            return True
+
+        except Exception as ex:
+            self.logger.exception(f"Failed to finish API test: {ex}")
+            return False
+
+        return True
+
+    def TestSubscriptionResumptionCapacityStep2(self, nodeid: int, endpoint: int, remote_ip: str, ssh_port: int,
+                                                remote_server_app: str, subscription_capacity: int):
+        try:
+            self.logger.info("Restart remote deivce")
+            extra_agrs = f"--thread --discriminator 3840 --subscription-capacity {subscription_capacity}"
+            restartRemoteThread = restartRemoteDevice(remote_ip, ssh_port, "root", "admin", remote_server_app, extra_agrs)
+            restartRemoteThread.start()
+
+            # Wait for some time so that the device will be resolving the address of the first controller after restarting
+            time.sleep(8)
+            restartRemoteThread.join(10.0)
+
+            self.logger.info("Send a new subscription request from the second controller")
+            # Close previous session so that the second controller will res-establish the session with the remote device
+            self.devCtrl.CloseSession(nodeid)
+            self.devCtrl.ZCLSubscribeAttribute(
+                "BasicInformation", "NodeLabel", nodeid, endpoint, 1, 50, keepSubscriptions=True, autoResubscribe=False)
+
+            if restartRemoteThread.is_alive():
+                # Thread join timed out
+                self.logger.error("Failed to join change thread")
+                return False
+
+            return True
+
+        except Exception as ex:
+            self.logger.exception(f"Failed to finish API test: {ex}")
+            return False
+
+        return True