style: auto-format with black + isort + prettier [skip ci-format-check]
CI/CD Pipeline / Check push changed paths (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 2m36s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 1m31s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 1m35s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 4m36s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 4m54s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 5m45s
AI Code Review / AI Code Review (pull_request) Failing after 7m6s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 9m5s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 10m19s
CI/CD Pipeline / Validate - Code Quality (pull_request) Has been cancelled
CI/CD Pipeline / Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled

This commit is contained in:
CI Bot
2026-08-29 12:26:01 +00:00
parent 2174e91c48
commit 5d705307c1
2 changed files with 30 additions and 33 deletions
@@ -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)
+29 -31
View File
@@ -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