def METHOD_NAME(
    dagster_instance_for_k8s_run_launcher,
    user_code_namespace_for_k8s_run_launcher,
    webserver_url_for_k8s_run_launcher,
):
    job_name = "slow_job_k8s"
    run_config = load_yaml_from_path(
        os.path.join(get_test_project_environments_path(), "env_s3.yaml")
    )
    run_id = launch_run_over_graphql(
        webserver_url_for_k8s_run_launcher, run_config=run_config, job_name=job_name
    )
    DagsterKubernetesClient.production_client().wait_for_job(
        job_name="dagster-run-%s" % run_id, namespace=user_code_namespace_for_k8s_run_launcher
    )
    timeout = datetime.timedelta(0, 30)
    start_time = datetime.datetime.now()
    while True:
        assert datetime.datetime.now() < start_time + timeout, "Timed out waiting for can_terminate"
        if can_terminate_run_over_graphql(webserver_url_for_k8s_run_launcher, run_id):
            break
        time.sleep(5)
    terminate_run_over_graphql(webserver_url_for_k8s_run_launcher, run_id=run_id)
    start_time = datetime.datetime.now()
    dagster_run = None
    while True:
        assert datetime.datetime.now() < start_time + timeout, "Timed out waiting for termination"
        dagster_run = dagster_instance_for_k8s_run_launcher.get_run_by_id(run_id)
        if dagster_run.status == DagsterRunStatus.CANCELED:
            break
        time.sleep(5)
    assert dagster_run.status == DagsterRunStatus.CANCELED
    assert not can_terminate_run_over_graphql(webserver_url_for_k8s_run_launcher, run_id)