File size: 1,234 Bytes
201dc77 | 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 | """U10-1: resuming a Tracker after a torn tail must not glue records."""
import pytest
from align.utils import tracking
from align.utils.tracking import Tracker, read_history
def _torn_then_resume(tmp_path, tail):
t = Tracker(str(tmp_path), "sft")
t.log(1, loss=0.5)
t.log(2, loss=0.6)
t.finish()
with open(t.history_path, "a") as fh:
fh.write(tail) # kill mid-append: no trailing newline
t2 = Tracker(str(tmp_path), "sft")
t2.log(3, loss=0.8)
t2.log(4, loss=0.9)
t2.finish()
def test_resume_after_torn_tail_keeps_history_readable(tmp_path):
_torn_then_resume(tmp_path, '{"step": 3, "stage": "sft", "lo')
assert [r["step"] for r in read_history(str(tmp_path))] == [1, 2, 3, 4]
def test_resume_after_complete_record_missing_newline_keeps_it(tmp_path):
_torn_then_resume(tmp_path, '{"step": 9, "stage": "sft", "loss": 0.1}')
assert [r["step"] for r in read_history(str(tmp_path))] == [1, 2, 9, 3, 4]
def test_resume_when_whole_file_is_one_torn_line(tmp_path):
(tmp_path / "history.jsonl").write_text('{"step": 1, "lo')
t = Tracker(str(tmp_path), "sft")
t.log(2, loss=0.1)
t.finish()
assert [r["step"] for r in read_history(str(tmp_path))] == [2]
|