code: 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.ai."""
      # 1. Validate the provide app_folder is correct.
      if not os.path.exists(os.path.join(app_folder, app_name)):
          raise Exception(
              f"The app folder should contain an {app_name} file."
          )
      if app_folder.endswith("/"):
          app_folder = app_folder[:-1]
  
      # 2. Create the right application name.
      basename = app_folder.split("/")[-1]
      PR_NUMBER = os.getenv("PR_NUMBER", None)
  
      is_editable_mode = get_dist_path_if_editable_install("lightning")
      if not is_editable_mode and PR_NUMBER is not None:
          raise Exception(
              "Lightning requires to be installed in editable mode in the CI."
          )
  
      TEST_APP_NAME = os.getenv("TEST_APP_NAME", basename)
      os.environ["TEST_APP_NAME"] = TEST_APP_NAME
  
      if PR_NUMBER:
          name = f"test-{PR_NUMBER}-{TEST_APP_NAME}-" + str(
              int(time.time())
          )
      else:
          name = f"test-{TEST_APP_NAME}-" + str(int(time.time()))
  
      os.environ["LIGHTNING_APP_NAME"] = name
  
      url = _Config.url
      if url.endswith("/"):
          url = url[:-1]
      payload = {
          "apiKey": _Config.api_key,
          "username": _Config.username,
      }
      url_login = url + "/v1/auth/login"
      res = requests.post(url_login, data=json.dumps(payload))
      if "token" not in res.json():
          raise RuntimeError(
              f"You haven't properly setup your environment variables with {url_login} and data: \n{payload}"
          )
  
      token = res.json()["token"]
  
      # 3. Disconnect from the App if any.
      Popen("lightning logout", shell=True).wait()
  
      # 4. Launch the application in the cloud from the Lightning CLI.
      with tempfile.TemporaryDirectory() as tmpdir:
          env_copy = os.environ.copy()
          env_copy["PACKAGE_LIGHTNING"] = "1"
          env_copy["LIGHTING_TESTING"] = "1"
          if debug:
              print("Debug mode is enabled")
              env_copy["LIGHTNING_DEBUG"] = "1"
          shutil.copytree(app_folder, tmpdir, dirs_exist_ok=True)
          # TODO - add -no-cache to the command line.
          stdout_path = get_logfile(f"run_app_in_cloud_{name}")
  
          cmd_extra_args = []
  
          with open(stdout_path, "w") as stdout:
              cmd = [
                  sys.executable,
                  "-m",
                  "lightning",
                  "run",
                  "app",
                  app_name,
                  "--cloud",
                  "--name",
                  name,
                  "--open-ui",
                  "false",
                  *cmd_extra_args,
              ]
              print(f"Command: {cmd}")
              process = Popen(
                  (cmd + extra_args),
                  cwd=tmpdir,
                  env=env_copy,
                  stdout=stdout,
                  stderr=sys.stderr,
              )
              process.wait()
  
          # Fallback URL to prevent failures in case we don't get the admin URL
          admin_url = _Config.url
          with open(stdout_path) as fo:
              for line in fo.readlines():
                  if line.startswith("APP_LOGS_URL: "):
                      admin_url = line.replace("APP_LOGS_URL: ", "")
                      break
  
          if is_editable_mode:
              # Added to ensure the current code is properly uploaded.
              # Otherwise, it could result in un-tested PRs.
              pkg_found = False
              with open(stdout_path) as fo:
                  for line in fo.readlines():
                      if (
                          "Packaged Lightning with your application"
                          in line
                      ):
                          pkg_found = True
                      print(line)  # TODO: use logging
              assert pkg_found
          os.remove(stdout_path)
  
      # 5. Print your application name
      print(
          f"The Lightning App Name is: [bold magenta]{name}[/bold magenta]"
      )
  
      # 6. Create chromium browser, auth to lightning.app.ai and yield the admin and view pages.
      with sync_playwright() as p:
          browser = p.chromium.launch(
              headless=bool(int(os.getenv("HEADLESS", "0")))
          )
          context = browser.new_context(
              # Eventually this will need to be deleted
              http_credentials=HttpCredentials(
                  {
                      "username": os.getenv("LAI_USER", "").strip(),
                      "password": os.getenv("LAI_PASS", ""),
                  }
              ),
              record_video_dir=os.path.join(
                  _Config.video_location, TEST_APP_NAME
              ),
              record_har_path=_Config.har_location,
          )
  
          client = LightningClient()
          project_id = _get_project(client).project_id
  
          app = _fetch_app_by_name(client, project_id, name)
          app_id = app.id
          print(
              f"The Lightning App ID is: {app.id}"
          )  # useful for Grafana
  
          if debug:
              process = Process(
                  target=_print_logs, kwargs={"app_id": app_id}
              )
              process.start()
  
          admin_page = context.new_page()
          admin_page.goto(admin_url)
          admin_page.evaluate(
              """data => {
              window.localStorage.setItem('gridUserId', data[0]);
              window.localStorage.setItem('gridUserKey', data[1]);
              window.localStorage.setItem('gridUserToken', data[2]);
          }
          """,
              [_Config.id, _Config.key, token],
          )
          if constants.LIGHTNING_CLOUD_PROJECT_ID:
              admin_page.evaluate(
                  """data => {
                  window.localStorage.setItem('gridDefaultProjectIdOverride', JSON.stringify(data[0]));
              }
              """,
                  [constants.LIGHTNING_CLOUD_PROJECT_ID],
              )
  
          admin_page.reload()
  
          view_page = context.new_page()
          i = 1
          while True:
              app = _fetch_app_by_name(client, project_id, name)
              msg = f"Still in phase {app.status.phase}"
  
              # wait until the app is running and openapi.json is ready
              if (
                  app.status.phase
                  == V1LightningappInstanceState.RUNNING
              ):
                  status_code = requests.get(
                      f"{app.status.url}/openapi.json"
                  ).status_code
                  if status_code == 200:
                      print(
                          "App is running, continuing with testing..."
                      )
                      view_page.goto(f"{app.status.url}/view")
                      break
                  msg = f"Received status code {status_code} at {app.status.url!r}"
              elif app.status.phase not in (
                  V1LightningappInstanceState.PENDING,
                  V1LightningappInstanceState.NOT_STARTED,
              ):
                  # there's a race condition if the app goes from pending to running to something else before we evaluate
                  # the condition above. avoid it by checking stopped explicitly
                  print(
                      f"App finished with phase {app.status.phase}, finished testing..."
                  )
                  break
              if debug and i % 30 == 0:
                  print(f"{msg}, continuing infinite loop...")
              i += 1
              sleep(1)
  
          logs_api_client = _LightningLogsSocketAPI(client.api_client)
  
          def fetch_logs(
              component_names: Optional[List[str]] = None,
          ) -> Generator:
              """This methods creates websockets connection in threads and returns the logs to the main thread."""
              if not component_names:
                  works = (
                      client.lightningwork_service_list_lightningwork(
                          project_id=project_id,
                          app_id=app_id,
                      ).lightningworks
                  )
  
                  component_names = ["flow"] + [w.name for w in works]
              else:
  
                  def add_prefix(c: str) -> str:
                      if c == "flow":
                          return c
                      if not c.startswith("root."):
                          return "root." + c
                      return c
  
                  component_names = [
                      add_prefix(c) for c in component_names
                  ]
  
              gen = _app_logs_reader(
                  logs_api_client=logs_api_client,
                  project_id=project_id,
                  app_id=app_id,
                  component_names=component_names,
                  follow=False,
                  on_error_callback=_on_error_callback,
              )
              for log_event in gen:
                  yield log_event.message
  
          try:
              yield admin_page, view_page, fetch_logs, name
          except KeyboardInterrupt:
              pass
          finally:
              if debug:
                  process.kill()
  
              context.close()
              browser.close()
              Popen("lightning disconnect", shell=True).wait()
  
              delete_cloud_lightning_apps(name=name)
  
imports: import asyncio
  import datetime
  import json
  import os
  import shutil
  import subprocess
  import sys
  import tempfile
  import time
  from contextlib import contextmanager
  from multiprocessing import Process
  from subprocess import Popen
  from time import sleep
  from typing import Any, Callable, Dict, Generator, List, Optional, Type
  import requests
  from lightning_cloud.openapi import V1LightningappInstanceState
  from lightning_cloud.openapi.rest import ApiException
  from requests import Session
  from rich import print
  from rich.color import ANSI_COLOR_NAMES
  from lightning.app.cli.lightning_cli import run_app
  from lightning.app.core import LightningApp, LightningFlow, constants
  from lightning.app.runners.multiprocess import MultiProcessRuntime
  from lightning.app.testing.config import _Config
  from lightning.app.utilities.app_logs import _app_logs_reader
  from lightning.app.utilities.cloud import _get_project
  from lightning.app.utilities.enum import CacheCallsKeys
  from lightning.app.utilities.imports import _is_playwright_available, requires
  from lightning.app.utilities.log import get_logfile
  from lightning.app.utilities.logs_socket_api import _LightningLogsSocketAPI
  from lightning.app.utilities.network import LightningClient, _configure_session
  from lightning.app.utilities.packaging.lightning_utils import get_dist_path_if_editable_install
  from lightning.app.utilities.proxies import ProxyWorkRun
label_desc: Use image node_id 8 for calling a global function with example usage: run_app_in_cloud(app_folder, app_name, extra_args, debug) and returns: c, c, unknown