def METHOD_NAME(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()