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 subscriptions number for the device to manage\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

@@ -36,11 +36,15 @@
 import chip.FabricAdmin
 import chip.interaction_model as IM
 import chip.native
+import paramiko
 from chip import ChipDeviceCtrl
 from chip.ChipStack import ChipStack
 from chip.crypto import p256keypair
 from chip.utils import CommissioningBuildingBlocks
 
+CHIP_REPO = os.path.join(os.path.abspath(
+    os.path.dirname(__file__)), "..", "..", "..", "..", "..")
+
 logger = logging.getLogger('PythonMatterControllerTEST')
 logger.setLevel(logging.INFO)
 
@@ -1316,3 +1320,192 @@ 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):
+        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()
+
+        class _restartRemoteDevice(threading.Thread):
+            def __init__(self, remote_ip: str, ssh_port: int, remote_server_app: str):
+                super(_restartRemoteDevice, self).__init__()
+                self.remote_ip = remote_ip
+                self.ssh_port = ssh_port
+                self.remote_server_app = remote_server_app
+
+            def run(self):
+                client = paramiko.SSHClient()
+                client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+                try:
+                    client.connect(self.remote_ip, self.ssh_port, "root", "admin")
+                    client.exec_command(
+                        ("kill \"$(ps aux | grep -E \'out/debug/standalone/{}\' | grep -v grep | grep -v gdb | "
+                         "awk \'{{print $2}}\')\"").format(self.remote_server_app))
+                    time.sleep(1)
+                    stdin, stdout, stderr = client.exec_command(
+                        ("ps aux | grep -E \'out/debug/standalone/{}\' | grep -v grep | grep -v gdb | "
+                         "awk \'{{print $2}}\'").format(self.remote_server_app))
+                    if not stdout.read().decode().strip():
+                        logger.info(f"Succeed to kill remote process {self.remote_server_app}")
+                    else:
+                        logger.error(f"Failed to kill remote process {self.remote_server_app}")
+
+                    client.exec_command(
+                        ("CHIPCirqueDaemon.py -- run gdb -batch -return-child-result -q -ex \"set pagination off\" "
+                         "-ex run -ex \"thread apply all bt\" --args {} --thread --discriminator 3840").format(
+                             os.path.join(CHIP_REPO, "out/debug/standalone", self.remote_server_app)))
+
+                finally:
+                    client.close()
+
+        try:
+            desiredPath = Clusters.Attribute.AttributePath(
+                EndpointId=0, ClusterId=0x28, AttributeId=5)
+            # Basic Information 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, remote_server_app)
+            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
+
+    def TestSubscriptionResumptionCapacityStep1(self, nodeid: int, endpoint: int, subscription_capacity: int):
+        try:
+            # OnOff Cluster, OnOff 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):
+        updateLock = threading.Lock()
+        updateCv = threading.Condition(updateLock)
+
+        class _restartRemoteDevice(threading.Thread):
+            def __init__(self, remote_ip: str, ssh_port: int, remote_server_app: str, subscription_capacity: int):
+                super(_restartRemoteDevice, self).__init__()
+                self.remote_ip = remote_ip
+                self.ssh_port = ssh_port
+                self.remote_server_app = remote_server_app
+                self.subscription_capacity = subscription_capacity
+
+            def run(self):
+                client = paramiko.SSHClient()
+                client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+                try:
+                    client.connect(self.remote_ip, self.ssh_port, "root", "admin")
+                    client.exec_command(
+                        ("kill \"$(ps aux | grep -E \'out/debug/standalone/{}\' | grep -v grep | grep -v gdb | "
+                         "awk \'{{print $2}}\')\"").format(self.remote_server_app))
+                    time.sleep(1)
+                    stdin, stdout, stderr = client.exec_command(
+                        ("ps aux | grep -E \'out/debug/standalone/{}\' | grep -v grep | grep -v gdb | "
+                         "awk \'{{print $2}}\'").format(self.remote_server_app))
+                    if not stdout.read().decode().strip():
+                        logger.info(f"Succeed to kill remote process {self.remote_server_app}")
+                    else:
+                        logger.error(f"Failed to kill remote process {self.remote_server_app}")
+
+                    client.exec_command("systemctl restart avahi-deamon.service")
+                    client.exec_command(
+                        ("CHIPCirqueDaemon.py -- run gdb -batch -return-child-result -q -ex \"set pagination off\" "
+                         "-ex run -ex \"thread apply all bt\" --args {} --thread --discriminator 3840 "
+                         "--subscription-capacity {}").format(
+                             os.path.join(CHIP_REPO, "out/debug/standalone", self.remote_server_app),
+                             self.subscription_capacity))
+                    with updateLock:
+                        updateCv.notifyAll()
+
+                finally:
+                    client.close()
+
+        try:
+            self.logger.info("Restart remote deivce")
+            restartRemoteThread = _restartRemoteDevice(remote_ip, ssh_port, remote_server_app, subscription_capacity)
+            restartRemoteThread.start()
+            with updateCv:
+                if not updateCv.wait(8.0):
+                    self.logger.error(
+                        "Failed to restart the remote device")
+
+            # Wait for some time so that the device will be resolving the address of the first controller
+            time.sleep(3)
+
+            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)
+            restartRemoteThread.join(10.0)
+
+            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