(app) Resolve a bug where the state changes isn't detected properly

src/lightning_app/CHANGELOG.md

@@ -21,6 +21,9 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/).
 
 -
 
+### Fixed
+
+- Resolved a bug where the state change detection using DeepDiff won't worked with Path, Drive objects ([#14465](https://github.com/Lightning-AI/lightning/pull/14465))
 
 ### Removed
 

src/lightning_app/core/app.py

@@ -9,6 +9,7 @@
 from time import time
 
 from deepdiff import DeepDiff, Delta
+from lightning_utilities.core.apply_func import apply_to_collection
 
 import lightning_app
 from lightning_app import _console
@@ -21,6 +22,7 @@
 )
 from lightning_app.core.queues import BaseQueue, SingleProcessQueue
 from lightning_app.frontend import Frontend
+from lightning_app.storage import Drive, Path
 from lightning_app.storage.path import storage_root_dir
 from lightning_app.utilities.app_helpers import _delta_to_app_state_delta, _LightningAppRef, Logger
 from lightning_app.utilities.commands.base import _process_requests
@@ -319,9 +321,12 @@ def maybe_apply_changes(self) -> bool:
         deltas = self._collect_deltas_from_ui_and_work_queues()
 
         if not deltas:
+            # Path and Drive aren't processed by DeepDiff, so we need to convert them to dict.
+            last_state = apply_to_collection(self.last_state, (Path, Drive), lambda x: x.to_dict())
+            state = apply_to_collection(self.state, (Path, Drive), lambda x: x.to_dict())
             # When no deltas are received from the Rest API or work queues,
             # we need to check if the flow modified the state and populate changes.
-            deep_diff = DeepDiff(self.last_state, self.state, verbose_level=2)
+            deep_diff = DeepDiff(last_state, state, verbose_level=2)
             if deep_diff:
                 # TODO: Resolve changes with ``CacheMissException``.
                 # new_state = self.populate_changes(self.last_state, self.state)

tests/tests_app/core/test_lightning_app.py

@@ -20,6 +20,7 @@
 from lightning_app.core.queues import BaseQueue, MultiProcessQueue, RedisQueue, SingleProcessQueue
 from lightning_app.frontend import StreamlitFrontend
 from lightning_app.runners import MultiProcessRuntime, SingleProcessRuntime
+from lightning_app.storage import Path
 from lightning_app.storage.path import storage_root_dir
 from lightning_app.testing.helpers import RunIf
 from lightning_app.testing.testing import LightningTestApp
@@ -975,3 +976,44 @@ def test_debug_mode_logging():
     app = LightningApp(A4())
     assert _console.level == logging.INFO
     MultiProcessRuntime(app, start_server=False).dispatch()
+
+
+class WorkPath(LightningWork):
+    def __init__(self):
+        super().__init__()
+        self.path = None
+
+    def run(self):
+        self.path = Path(__file__)
+
+
+class FlowPath(LightningFlow):
+    def __init__(self):
+        super().__init__()
+        self.w = WorkPath()
+
+    def run(self):
+        self.w.run()
+
+
+class TestLightningHasUpdatedApp(LightningApp):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.counter = 0
+
+    def run_once(self):
+        res = super().run_once()
+
+        if self.root.w.has_succeeded:
+            self.counter += 1
+
+        # TODO: Resolve bug where it should work with self.counter == 2
+        if self.counter > 5:
+            assert not self._has_updated
+            return True
+        return res
+
+
+def test_lightning_app_has_updated():
+    app = TestLightningHasUpdatedApp(FlowPath())
+    MultiProcessRuntime(app, start_server=False).dispatch()

tests/tests_app_examples/test_v0_app.py

@@ -11,7 +11,7 @@
 
 class LightningAppTestInt(LightningTestApp):
     def run_once(self) -> Tuple[bool, float]:
-        if self.root.counter > 1:
+        if self.root.counter == 1:
             print("V0 App End")
             self.stage = AppStage.STOPPING
             return True, 0.0