File size: 2,729 Bytes
970c0d3 92d50ea 970c0d3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 | 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
|