Function test_coupling_alloc1

tests/test_coupling.py:54–104

(hq_env: HqEnv)

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

    hq_env.command(
        [
            "submit",
            "--array=1-50",
            "--cpus=1",
            "--resource=foo=3 compact!",
            "--",
            *bash('echo "$HQ_RESOURCE_VALUES_cpus;$HQ_RESOURCE_VALUES_foo"'),
        ]
    )

    hq_env.command(
        [
            "submit",
            "--array=1-50",
            "--cpus=1",
            "--resource=foo=1 compact!",
            "--",
            *bash('echo "$HQ_RESOURCE_VALUES_cpus;$HQ_RESOURCE_VALUES_foo"'),
        ]
    )

    hq_env.start_worker(
        cpus="[[1, 2, 3], [4, 5, 6]]",
        args=[
            "--coupling=cpus[0]:foo[0],cpus[1]:foo[1]",
            "--resource=ddd=sum(123)",
            "--resource=foo=[[10,20,30,40],[50,60,70,80]]",
        ],
    )

    wait_for_job_state(hq_env, [1, 2], "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) // 4)
        for f in foos:
            g.add(int(f) // 50)
        return g

    for i in range(1, 51):
        with open(default_task_output(job_id=1, task_id=i), "r") as f:
            assert len(get_groups(f.read())) == 1
    for i in range(1, 51):
        with open(default_task_output(job_id=2, task_id=i), "r") as f:
            assert len(get_groups(f.read())) == 1


def test_coupling_alloc2(hq_env: HqEnv):