(*, cluster_nodes: List[Dict[str, Any]])
| 500 | "ATOMIC_GROUP_NODE_RESOLUTION_TAIL": _render_atomic_group_node_resolution_tail(group_cfg["nodes"]), |
| 501 | "SELECTION_SUPERVISOR_PATH_BLOCK": _render_selection_supervisor_path_from_script_dir(), |
| 502 | "ATOMIC_GROUP_STOP_FN": _render_atomic_group_stop_fn( |
| 503 | runtime_specs=[ |
| 504 | { |
| 505 | "service_name": service_name, |
| 506 | "supervisor_label": _bare_atomic_group_member_selection_supervisor_label( |
| 507 | name_prefix=name_prefix, |
| 508 | group_name=group_name, |
| 509 | service_name=service_name, |
| 510 | ), |
| 511 | } |
| 512 | for service_name in stop_services |
| 513 | ], |
| 514 | ), |
| 515 | }, |
| 516 | ) |
| 517 | |
| 518 | |
| 519 | def _render_host_prelude(*, cluster_nodes: List[Dict[str, Any]]) -> str: |
| 520 | all_nodes = [_require_str(node.get("hostname"), "cluster_nodes[].hostname") for node in cluster_nodes] |
| 521 | ip_case_lines: list[str] = [] |
| 522 | host_case_lines: list[str] = [] |
| 523 | for node in cluster_nodes: |
| 524 | node_name = _require_str(node.get("hostname"), "cluster_nodes[].hostname") |
| 525 | node_ip = _require_str(node.get("ip"), f"cluster_nodes[{node_name}].ip") |
no test coverage detected