| from unittest.mock import Mock, patch | |
| import requests | |
| from src.task_progress import _snapshots, refresh_progress, render_progress | |
| def test_timeline_counts_tasks_and_escapes_names(): | |
| response = Mock() | |
| response.json.return_value = {"records": [ | |
| {"id": "job", "type": "Job", "state": "inProgress"}, | |
| {"id": "1", "type": "Task", "state": "completed"}, | |
| {"id": "2", "type": "Task", "state": "inProgress", "name": "<script>"}, | |
| {"id": "3", "type": "Task", "state": "pending"}, | |
| ]} | |
| with patch('src.task_progress.requests.get', return_value=response): | |
| refresh_progress(101, 'https://example.com/timeline', {}) | |
| html = render_progress({"status": "Running", "ci_run_id": 101}) | |
| assert 'max="3" value="1"' in html | |
| assert '<script>' in html | |
| assert '<script>' not in html | |
| assert 'Pipeline steps: 1/3' in html | |
| def test_missing_failed_and_stale_progress(): | |
| with patch('src.task_progress.requests.get', side_effect=requests.Timeout): | |
| refresh_progress(102, 'https://example.com/timeline', {}) | |
| entry = {"status": "Running", "ci_run_id": 102} | |
| assert 'Waiting for progress information' in render_progress(entry) | |
| _snapshots['102'] = (0, 1, 2, 'Old step') | |
| assert 'Old step' not in render_progress(entry) | |
| assert 'value="100"' in render_progress({"status": "Finished"}) | |
| assert render_progress({"status": "Eval Failed"}) == 'Eval Failed' | |
| def test_retry_uses_latest_attempt(): | |
| response = Mock() | |
| response.json.return_value = {"records": [ | |
| {"identifier": "quant", "type": "Task", "attempt": 1, "state": "completed"}, | |
| {"identifier": "quant", "type": "Task", "attempt": 2, "state": "inProgress"}, | |
| ]} | |
| with patch('src.task_progress.requests.get', return_value=response): | |
| refresh_progress(103, 'https://example.com/timeline', {}) | |
| assert 'max="1" value="0"' in render_progress({"status": "Running", "ci_run_id": 103}) | |
| def test_progress_survives_queue_loading(tmp_path): | |
| import json | |
| from src.display.utils import QUANT_COLS, EVAL_COLS | |
| from src.populate import get_evaluation_queue_df | |
| for kind, cols in [('quant', QUANT_COLS), ('eval', EVAL_COLS)]: | |
| (tmp_path / f'model_{kind}_request_test.json').write_text(json.dumps({ | |
| 'model': 'org/model', 'status': 'Running', 'ci_run_id': 104, | |
| })) | |
| finished, running, pending, failed = get_evaluation_queue_df( | |
| str(tmp_path), cols, request_type=kind, | |
| ) | |
| assert len(running) == 1 | |
| assert '<progress' in running.iloc[0]['progress'] | |
| assert 'Waiting for progress information' in running.iloc[0]['progress'] | |
| assert finished.empty and pending.empty and failed.empty | |