diff --git a/docs/source-app/examples/dag/dag_from_scratch.rst b/docs/source-app/examples/dag/dag_from_scratch.rst
index cde46953328bd..4af39f1af794e 100644
--- a/docs/source-app/examples/dag/dag_from_scratch.rst
+++ b/docs/source-app/examples/dag/dag_from_scratch.rst
@@ -39,10 +39,9 @@ First, let's define the component we need:
     :lines: 55-79
 
 And its run method executes the steps described above.
-Additionally, ``work.stop`` is used to reduce cost when running in the cloud.
 
 .. literalinclude:: ../../../examples/app_dag/app.py
-    :lines: 81-108
+    :lines: 80-103
 
 ----
 
@@ -51,4 +50,4 @@ Step 2: Define the scheduling
 *****************************
 
 .. literalinclude:: ../../../examples/app_dag/app.py
-    :lines: 109-137
+    :lines: 106-135
diff --git a/examples/app_dag/app.py b/examples/app_dag/app.py
index 2c1cfb4309fd6..531e39028420e 100644
--- a/examples/app_dag/app.py
+++ b/examples/app_dag/app.py
@@ -56,7 +56,7 @@ class DAG(L.LightningFlow):
 
     """This component is a DAG."""
 
-    def __init__(self, models_paths):
+    def __init__(self, models_paths: list):
         super().__init__()
         # Step 1: Create a work to get the data.
         self.data_collector = GetDataWork()
@@ -80,12 +80,10 @@ def __init__(self, models_paths):
     def run(self):
         # Step 1 and 2: Download and process the data.
         self.data_collector.run()
-        self.data_collector.stop()  # Stop the data_collector to reduce cost
         self.processing.run(
             df_data=self.data_collector.df_data,
             df_target=self.data_collector.df_target,
         )
-        self.processing.stop()  # Stop the processing to reduce cost
 
         # Step 3: Launch n models training in parallel.
         for model, work in self.dict.items():
@@ -128,7 +126,7 @@ def run(self):
 app = L.LightningApp(
     ScheduledDAG(
         DAG,
-        models=[
+        models_paths=[
             "svm.SVR",
             "linear_model.LinearRegression",
             "tree.DecisionTreeRegressor",
diff --git a/src/lightning_app/CHANGELOG.md b/src/lightning_app/CHANGELOG.md
index 822ef9bcaaa61..dd4a638a76765 100644
--- a/src/lightning_app/CHANGELOG.md
+++ b/src/lightning_app/CHANGELOG.md
@@ -221,6 +221,9 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/).
 - Resolved a bug where the `install` command was not installing the latest version of an app/component by default ([#14181](https://github.com/Lightning-AI/lightning/pull/14181))
 
 
+- Fixed the `examples/app_dag` example ([#14359](https://github.com/Lightning-AI/lightning/pull/14359))
+
+
 ## [0.5.5] - 2022-08-9
 
 ### Deprecated
diff --git a/src/lightning_app/testing/testing.py b/src/lightning_app/testing/testing.py
index 43aa7c55be728..fb087a2d62b71 100644
--- a/src/lightning_app/testing/testing.py
+++ b/src/lightning_app/testing/testing.py
@@ -219,7 +219,7 @@ def _run_cli(args) -> Generator:
 def run_app_in_cloud(
     app_folder: str, app_name: str = "app.py", extra_args: List[str] = [], debug: bool = True
 ) -> Generator:
-    """This utility is used to automate testing e2e application with lightning_app.ai."""
+    """This utility is used to automate testing e2e application with lightning.ai."""
     # 1. Validate the provide app_folder is correct.
     if not os.path.exists(os.path.join(app_folder, "app.py")):
         raise Exception("The app folder should contain an app.py file.")
diff --git a/tests/tests_app_examples/test_app_dag.py b/tests/tests_app_examples/test_app_dag.py
new file mode 100644
index 0000000000000..6d9a865a0a000
--- /dev/null
+++ b/tests/tests_app_examples/test_app_dag.py
@@ -0,0 +1,21 @@
+import os
+from time import sleep
+
+import pytest
+from tests_app import _PROJECT_ROOT
+
+from lightning_app.testing.testing import run_app_in_cloud
+
+
+@pytest.mark.cloud
+def test_app_dag_example_cloud() -> None:
+    with run_app_in_cloud(os.path.join(_PROJECT_ROOT, "examples/app_dag")) as (_, _, fetch_logs, _):
+
+        launch_log, finish_log = False, False
+        while not (launch_log and finish_log):
+            for log in fetch_logs(["flow"]):
+                if "Launching a new DAG" in log:
+                    launch_log = True
+                elif "Finished training and evaluating" in log:
+                    finish_log = True
+            sleep(1)