(
*,
name: str,
service_name: str,
child_argv: List[str],
root: Path,
apply_id: Optional[str],
)
| 119 | |
| 120 | |
| 121 | def _runtime_state_json( |
| 122 | *, |
| 123 | name: str, |
| 124 | service_name: str, |
| 125 | child_argv: List[str], |
| 126 | root: Path, |
| 127 | apply_id: Optional[str], |
| 128 | ) -> str: |
| 129 | payload = { |
| 130 | "kind": "DaemonSet", |
| 131 | "name": name, |
| 132 | "service_name": service_name, |
| 133 | "argv": child_argv, |
| 134 | "cwd": str(root), |
| 135 | "log_path": str(root / f"{service_name}.log"), |
| 136 | } |
| 137 | if apply_id is not None: |
| 138 | payload["apply_id"] = apply_id |
| 139 | return json.dumps(payload, sort_keys=True) |
| 140 | |
| 141 | |
| 142 | def _run_supervisor_command( |
no outgoing calls
no test coverage detected