(args: argparse.Namespace)
| 215 | |
| 216 | |
| 217 | def cmd_watch(args: argparse.Namespace) -> int: |
| 218 | interval = max(0.5, float(args.interval)) |
| 219 | timeout_seconds = max(0, int(args.timeout)) if args.timeout is not None else None |
| 220 | deadline = time.time() + timeout_seconds if timeout_seconds is not None else None |
| 221 | |
| 222 | last_status = None |
| 223 | while True: |
| 224 | job = load_job_state(args.job_id) |
| 225 | if job is None: |
| 226 | _print_json({'status': 'error', 'message': f'Job not found: {args.job_id}'}) |
| 227 | return 2 |
| 228 | |
| 229 | if last_status != job.status.value: |
| 230 | _print_json(_status_snapshot(job)) |
| 231 | last_status = job.status.value |
| 232 | |
| 233 | if job.status in {JobStatus.completed, JobStatus.failed}: |
| 234 | return 0 |
| 235 | |
| 236 | if deadline is not None and time.time() > deadline: |
| 237 | _print_json({'status': 'timeout', 'job_id': str(job.id), 'current_status': job.status.value}) |
| 238 | return 0 |
| 239 | |
| 240 | time.sleep(interval) |
| 241 | |
| 242 | |
| 243 | def cmd_run_job(args: argparse.Namespace) -> int: |
nothing calls this directly
no test coverage detected