diff --git a/apps/api/app/services/asset_segment_tracker.py b/apps/api/app/services/asset_segment_tracker.py index cb485816b..6bff40234 100644 --- a/apps/api/app/services/asset_segment_tracker.py +++ b/apps/api/app/services/asset_segment_tracker.py @@ -120,8 +120,7 @@ def remove_used_segment( for r in ranges: try: match = ( - abs(float(r["start"]) - float(start)) <= tolerance - and abs(float(r["end"]) - float(end)) <= tolerance + abs(float(r["start"]) - float(start)) <= tolerance and abs(float(r["end"]) - float(end)) <= tolerance ) except (KeyError, TypeError, ValueError): remaining.append(r) diff --git a/tests/unit/test_asset_segment_tracker.py b/tests/unit/test_asset_segment_tracker.py index 043e51b1f..258771c2f 100644 --- a/tests/unit/test_asset_segment_tracker.py +++ b/tests/unit/test_asset_segment_tracker.py @@ -22,7 +22,6 @@ os.environ.setdefault("DATABASE_URL", "sqlite:///test.db") sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api")) import pytest - from app.services import asset_segment_tracker as ast from app.services.asset_segment_tracker import ( get_used_segments, @@ -31,6 +30,7 @@ from app.services.asset_segment_tracker import ( remove_used_segment, reset_used_segments, ) + from packages.domain.plan_generator_utils import _calc_random_start_time @@ -121,12 +121,17 @@ def _db(models): def test_get_used_segments_aggregates_ranges(patched_model): models = { - "a1": FakeModel("a1", {"used_time_ranges": [ - {"start": 1.0, "end": 5.0, "plan_id": "p1"}, - {"start": 9.0, "end": 12.0, "plan_id": "p2"}, - ]}), - "a2": FakeModel("a2", {"other": 1}), # 无区间记录 - "a3": FakeModel("a3"), # metadata 为空 + "a1": FakeModel( + "a1", + { + "used_time_ranges": [ + {"start": 1.0, "end": 5.0, "plan_id": "p1"}, + {"start": 9.0, "end": 12.0, "plan_id": "p2"}, + ] + }, + ), + "a2": FakeModel("a2", {"other": 1}), # 无区间记录 + "a3": FakeModel("a3"), # metadata 为空 } db = _db(models) @@ -202,12 +207,8 @@ def test_remove_not_found_returns_false(patched_model): def test_remove_respects_tolerance(patched_model): models = { - "a1": FakeModel("a1", {"used_time_ranges": [ - {"start": 5.0, "end": 9.0, "plan_id": "p1"} - ]}), - "a2": FakeModel("a2", {"used_time_ranges": [ - {"start": 5.0, "end": 9.0, "plan_id": "p1"} - ]}), + "a1": FakeModel("a1", {"used_time_ranges": [{"start": 5.0, "end": 9.0, "plan_id": "p1"}]}), + "a2": FakeModel("a2", {"used_time_ranges": [{"start": 5.0, "end": 9.0, "plan_id": "p1"}]}), } db = _db(models) @@ -218,9 +219,7 @@ def test_remove_respects_tolerance(patched_model): def test_remove_plan_id_must_match(patched_model): - models = {"a1": FakeModel("a1", {"used_time_ranges": [ - {"start": 5.0, "end": 9.0, "plan_id": "plan-A"} - ]})} + models = {"a1": FakeModel("a1", {"used_time_ranges": [{"start": 5.0, "end": 9.0, "plan_id": "plan-A"}]})} db = _db(models) # 时间匹配但 plan_id 不同 → 不删除 @@ -232,10 +231,15 @@ def test_remove_plan_id_must_match(patched_model): def test_reset_clears_ranges_keeps_other_fields(patched_model): - models = {"a1": FakeModel("a1", { - "generation_use_count": 9, - "used_time_ranges": [{"start": 1, "end": 2}], - })} + models = { + "a1": FakeModel( + "a1", + { + "generation_use_count": 9, + "used_time_ranges": [{"start": 1, "end": 2}], + }, + ) + } db = _db(models) reset_used_segments(db, "a1") @@ -250,17 +254,15 @@ def test_reset_clears_ranges_keeps_other_fields(patched_model): def test_reset_callback_clears_persisted_and_memory(patched_model): - models = {"a1": FakeModel("a1", {"used_time_ranges": [ - {"start": 0, "end": 30} - ]})} + models = {"a1": FakeModel("a1", {"used_time_ranges": [{"start": 0, "end": 30}]})} db = _db(models) used_segments = {"a1": [(0.0, 30.0)], "a2": [(1.0, 2.0)]} cb = make_reset_callback(db, used_segments) cb("a1") - assert "a1" not in used_segments # 内存清空 - assert "a2" in used_segments # 其他素材不受影响 + assert "a1" not in used_segments # 内存清空 + assert "a2" in used_segments # 其他素材不受影响 assert models["a1"].meta()["used_time_ranges"] == [] @@ -277,9 +279,7 @@ def test_calc_random_start_invokes_reset_when_exhausted(): reset_called.append(asset_id) used.pop(asset_id, None) # 模拟轮回清空 - result = _calc_random_start_time( - "a1", 10.0, durations, used, on_exhausted=_on_exhausted - ) + result = _calc_random_start_time("a1", 10.0, durations, used, on_exhausted=_on_exhausted) assert reset_called == ["a1"] assert result is not None @@ -302,9 +302,7 @@ def test_calc_random_start_with_space_does_not_reset(): used = {"a1": [(0.0, 50.0)]} reset_called = [] - result = _calc_random_start_time( - "a1", 5.0, durations, used, on_exhausted=lambda aid: reset_called.append(aid) - ) + result = _calc_random_start_time("a1", 5.0, durations, used, on_exhausted=lambda aid: reset_called.append(aid)) assert reset_called == [] assert result is not None