Function test_coupling_alloc2

tests/test_coupling.py:107–153

(hq_env: HqEnv)

def test_coupling_alloc2(hq_env: HqEnv):
    hq_env.start_server()

    hq_env.command(
        [
            "submit",
            "--cpus=4",
            "--resource=foo=2",
            "--",
            *bash('echo "$HQ_RESOURCE_VALUES_cpus;$HQ_RESOURCE_VALUES_foo"'),
        ]
    )

    hq_env.command(
        [
            "submit",
            "--cpus=2",
            "--resource=foo=4",
            "--",
            *bash('echo "$HQ_RESOURCE_VALUES_cpus;$HQ_RESOURCE_VALUES_foo"'),
        ]
    )

    hq_env.start_worker(
        cpus="[[1, 2, 3], [10, 11, 12], [21, 22, 23]]",
        args=[
            "--coupling=cpus[0]:foo[0],cpus[1]:foo[1],cpus[2]:foo[2]",
            "--resource=ddd=sum(123)",
            "--resource=foo=[[1, 2, 3],[10, 11, 12],[20, 21, 23]]",
        ],
    )

    wait_for_job_state(hq_env, 1, "FINISHED")

    def get_groups(data):
        cpus, foos = (x.split(",") for x in data.rstrip().split(";"))
        g = set()
        for c in cpus:
            g.add(int(c) // 10)
        for f in foos:
            g.add(int(f) // 10)
        return g

    with open(default_task_output(job_id=1), "r") as f:
        assert len(get_groups(f.read())) == 2
    with open(default_task_output(job_id=2), "r") as f:
        assert len(get_groups(f.read())) == 2


def test_coupling_combined(hq_env: HqEnv):