MCPcopy Create free account
hub / github.com/It4innovations/hyperqueue / wait_for_state

Function wait_for_state

tests/utils/wait.py:32–72  ·  view source on GitHub ↗
(
    env,
    ids: Union[int, List[int]],
    target_states: Union[str, List[str]],
    commands: List[str],
    state_index: int,
    check_running_processes=True,
    **kwargs,
)

Source from the content-addressed store, hash-verified

30
31
32def wait_for_state(
33 env,
34 ids: Union[int, List[int]],
35 target_states: Union[str, List[str]],
36 commands: List[str],
37 state_index: int,
38 check_running_processes=True,
39 **kwargs,
40):
41 if isinstance(ids, int):
42 ids = {str(ids)}
43 else:
44 ids = set(str(id) for id in ids)
45
46 if isinstance(target_states, str):
47 target_states = {target_states.lower()}
48 else:
49 target_states = set(state.lower() for state in target_states)
50
51 last_table = None
52
53 def check():
54 nonlocal last_table
55 if check_running_processes:
56 env.check_running_processes()
57 table = env.command(commands, as_table=True)
58 last_table = table
59 items = [row[state_index].lower() for row in table if row[0].lstrip("*") in ids]
60 if len(items) < len(ids):
61 return False
62 r = all(s in target_states for s in items)
63 if not r:
64 if all(s in TERMINAL_STATES for s in items):
65 raise Exception(f"Waiting for {target_states} but job(s) are already in terminal states: {items}")
66 return r
67
68 def on_timeout():
69 if last_table is not None:
70 return f"most recent table:\n{last_table}"
71
72 wait_until(check, on_timeout=on_timeout, **kwargs)
73
74
75def wait_for_job_state(env, ids: Union[int, List[int]], target_states: Union[str, List[str]], **kwargs):

Callers 2

wait_for_job_stateFunction · 0.85
wait_for_worker_stateFunction · 0.85

Calls 1

wait_untilFunction · 0.70

Tested by

no test coverage detected