def METHOD_NAME(client, flow_name, engine):
    """Assumption: nodes in the K8s cluster have more than 4 CPUs.
    We run a special operator that checks the number of CPUs that are available.
    We check the expected default number of cpus, as well as a custom number.
    """
    def _count_available_cpus():
        # Copied from: https://donghao.org/2022/01/20/how-to-get-the-number-of-cpu-cores-inside-a-container/
        with open("/sys/fs/cgroup/cpu/cpu.cfs_quota_us") as fp:
            cfs_quota_us = int(fp.read())
        with open("/sys/fs/cgroup/cpu/cpu.cfs_period_us") as fp:
            cfs_period_us = int(fp.read())
        container_cpus = cfs_quota_us // cfs_period_us
        # For physical machine, the `cfs_quota_us` could be '-1'
        cpus = cpu_count() if container_cpus < 1 else container_cpus
        return cpus
    global_config({"engine": engine})
    # Returns the default number of CPUs of the K8s cluster. (Currently 2)
    @op(requirements=[])
    def count_default_available_cpus():
        return _count_available_cpus()
    num_default_available_cpus = count_default_available_cpus()
    assert num_default_available_cpus.get() == 2
    # Returns 4, the custom number of CPUs on the K8s cluster.
    @op(requirements=[], resources={"num_cpus": 4})
    def count_with_custom_available_cpus():
        return _count_available_cpus()
    num_count_available_cpus = count_with_custom_available_cpus()
    assert num_count_available_cpus.get() == 4
    default_cpus_flow = publish_flow_test(
        client,
        name=flow_name(),
        artifacts=num_default_available_cpus,
        engine=engine,
    )
    custom_cpus_flow = publish_flow_test(
        client,
        name=flow_name(),
        artifacts=num_count_available_cpus,
        engine=engine,
    )
    assert default_cpus_flow.latest().artifact("count_default_available_cpus artifact").get() == 2
    assert (
        custom_cpus_flow.latest().artifact("count_with_custom_available_cpus artifact").get() == 4
    )