code: def _run_yarn_test_with_env(env_path, timeout):
      cluster = None
  
      coverage_result = os.path.join(
          MARS_ROOT, ".dist-coverage", ".coverage"
      )
      cov_dir = os.path.join(MARS_ROOT, ".dist-coverage")
      os.makedirs(cov_dir, exist_ok=True)
      os.chmod(cov_dir, 0o777)
      try:
          log_config_file = os.path.join(
              os.path.dirname(os.path.abspath(__file__)),
              "yarn-logging.conf",
          )
  
          cmd_tmpl = (
              '"{executable}" -m coverage run --source=%s/mars --rcfile=%s/setup.cfg'
              % (MARS_ROOT, MARS_ROOT)
          )
          extra_env = {
              "COVERAGE_FILE": coverage_result,
              "COVERAGE_PROCESS_START": f"{MARS_ROOT}/setup.cfg",
          }
          cluster = new_cluster(
              env_path,
              timeout=timeout,
              worker_cpu=1,
              worker_mem="1G",
              extra_env=extra_env,
              log_config=log_config_file,
              extra_args=f"--config-file {MARS_ROOT}/mars/deploy/yarn/tests/test_yarn_config.yml",
              log_when_fail=True,
              cmd_tmpl=cmd_tmpl,
          )
          assert cluster.endpoint is not None
  
          check_time = time.time()
          while cluster.session.get_total_n_cpu() == 0:
              time.sleep(1)
              if time.time() - check_time > 5:
                  raise SystemError("Worker not ready")
  
          a = mt.ones((100, 100), chunk_size=30) * 2 * 1 + 1
          b = mt.ones((100, 100), chunk_size=30) * 2 * 1 + 1
          r = (a * b * 2 + 1).sum().execute().fetch()
  
          expected = (np.ones(a.shape) * 2 * 1 + 1) ** 2 * 2 + 1
          np.testing.assert_array_equal(r, expected.sum())
      finally:
          if cluster is not None:
              cluster.stop()
          _collect_coverage()
  
file_path: mars/mars/deploy/yarn/tests/test_yarn.py
imports: import glob
  import logging
  import os
  import shutil
  import sqlite3
  import subprocess
  import tempfile
  import time
  from distutils.spawn import find_executable
  import numpy
  import pytest
  from ....None import tensor
  from ....tests.core import flaky, require_hadoop
  from ...yarn import new_cluster
label_desc: Use image node_id 2 for calling a global function with example usage: _run_yarn_test_with_env(env_path, timeout) without return types