tests/unit/test_job_domain.py

@@ -1,10 +1,4 @@
-"""
-Job 领域模型单元测试
-"""
-
-import time
-from datetime import datetime, timezone
-from unittest.mock import patch
+"""Job 领域层单元测试 - job.py"""
 
 import pytest
 
@@ -19,524 +13,494 @@ from packages.domain.job import (
 class TestJobType:
     """JobType 枚举测试"""
 
-    def test_job_type_values(self):
-        """测试所有 JobType 值"""
-        assert JobType.VIDEO_COMPOSE == "video_compose"
-        assert JobType.RENDER_EDIT_PLAN == "render_edit_plan"
-        assert JobType.ASSET_INGEST == "asset_ingest"
-        assert JobType.CLASSIFICATION == "classification"
-        assert JobType.VOICE_EXTRACTION == "voice_extraction"
-        assert JobType.GENERATION == "generation"
+    def test_all_types_have_values(self):
+        """所有枚举成员都有字符串值"""
+        for jt in JobType:
+            assert isinstance(jt.value, str)
+            assert jt.value
 
-    def test_job_type_is_string(self):
-        """测试 StrEnum 行为"""
-        assert isinstance(JobType.VIDEO_COMPOSE, str)
+    def test_str_enum_behavior(self):
+        """是 str 枚举"""
         assert JobType.VIDEO_COMPOSE == "video_compose"
+        assert isinstance(JobType.VIDEO_COMPOSE, str)
+
+    def test_known_types_exist(self):
+        """核心任务类型都存在"""
+        assert JobType.VIDEO_COMPOSE
+        assert JobType.RENDER_EDIT_PLAN
+        assert JobType.ASSET_INGEST
+        assert JobType.CLASSIFICATION
+        assert JobType.GENERATION
 
 
 class TestJobStatus:
     """JobStatus 枚举测试"""
 
-    def test_job_status_values(self):
-        """测试所有 JobStatus 值"""
+    def test_all_statuses_have_values(self):
+        for js in JobStatus:
+            assert isinstance(js.value, str)
+            assert js.value
+
+    def test_str_enum_behavior(self):
         assert JobStatus.PENDING == "pending"
-        assert JobStatus.RUNNING == "running"
-        assert JobStatus.SUCCESS == "success"
-        assert JobStatus.FAILED == "failed"
-        assert JobStatus.CANCELLED == "cancelled"
+        assert isinstance(JobStatus.PENDING, str)
 
     def test_terminal_statuses(self):
-        """测试终态集合"""
+        """终态集合包含成功/失败/取消"""
         assert JobStatus.SUCCESS in TERMINAL_STATUSES
         assert JobStatus.FAILED in TERMINAL_STATUSES
         assert JobStatus.CANCELLED in TERMINAL_STATUSES
+
+    def test_pending_not_terminal(self):
         assert JobStatus.PENDING not in TERMINAL_STATUSES
+
+    def test_running_not_terminal(self):
         assert JobStatus.RUNNING not in TERMINAL_STATUSES
 
 
 class TestJobCreate:
-    """Job 创建测试"""
+    """Job.create 工厂方法测试"""
 
-    def test_create_basic_job(self):
-        """测试创建基本任务"""
+    def test_create_basic(self):
+        """基本创建"""
         job = Job.create(
-            project_id="proj-123",
+            project_id="proj-1",
             job_type=JobType.VIDEO_COMPOSE,
         )
-
-        assert job.id is not None
-        assert len(job.id) > 0
-        assert job.project_id == "proj-123"
+        assert job.id
+        assert len(job.id) == 32  # uuid4 hex
+        assert job.project_id == "proj-1"
         assert job.job_type == JobType.VIDEO_COMPOSE
         assert job.status == JobStatus.PENDING
         assert job.progress == 0.0
         assert job.payload == {}
         assert job.result == {}
-        assert job.error_message == ""
         assert job.retry_count == 0
         assert job.max_retries == 3
-        assert job.created_at is not None
-        assert job.updated_at is not None
-        assert job.started_at is None
-        assert job.completed_at is None
-
-    def test_create_with_all_params(self):
-        """测试创建带所有参数的任务"""
-        job = Job.create(
-            project_id="proj-456",
-            job_type=JobType.GENERATION,
-            payload={"key": "value"},
-            source_id="src-789",
-            created_by_user_id="user-001",
-            max_retries=5,
-        )
-
-        assert job.project_id == "proj-456"
-        assert job.job_type == JobType.GENERATION
-        assert job.payload == {"key": "value"}
-        assert job.source_id == "src-789"
-        assert job.created_by_user_id == "user-001"
-        assert job.max_retries == 5
+        assert job.created_at
+        assert job.updated_at
 
     def test_create_with_string_job_type(self):
-        """测试用字符串创建任务"""
+        """用字符串创建任务类型"""
         job = Job.create(
-            project_id="proj-123",
+            project_id="proj-1",
             job_type="video_compose",
         )
         assert job.job_type == JobType.VIDEO_COMPOSE
 
-    def test_create_with_invalid_job_type(self):
-        """测试无效任务类型"""
+    def test_create_invalid_string_job_type_raises(self):
+        """无效的任务类型字符串抛 ValueError"""
         with pytest.raises(ValueError, match="不支持的任务类型"):
-            Job.create(
-                project_id="proj-123",
-                job_type="invalid_type",
-            )
+            Job.create(project_id="proj-1", job_type="invalid_type")
 
-    def test_create_empty_project_id(self):
-        """测试空 project_id"""
+    def test_create_empty_project_id_raises(self):
+        """空 project_id 抛 ValueError"""
         with pytest.raises(ValueError, match="project_id 不能为空"):
-            Job.create(
-                project_id="",
-                job_type=JobType.VIDEO_COMPOSE,
-            )
+            Job.create(project_id="   ", job_type=JobType.VIDEO_COMPOSE)
 
-    def test_create_whitespace_project_id(self):
-        """测试空白 project_id 被 strip 后为空"""
-        with pytest.raises(ValueError, match="project_id 不能为空"):
-            Job.create(
-                project_id="   ",
-                job_type=JobType.VIDEO_COMPOSE,
-            )
-
-    def test_create_strips_strings(self):
-        """测试字符串字段被 strip"""
+    def test_create_with_payload(self):
+        """带 payload 创建"""
+        payload = {"video_id": "v1", "quality": "1080p"}
         job = Job.create(
-            project_id="  proj-123  ",
+            project_id="proj-1",
             job_type=JobType.VIDEO_COMPOSE,
-            source_id="  src-456  ",
-            created_by_user_id="  user-789  ",
+            payload=payload,
         )
-        assert job.project_id == "proj-123"
-        assert job.source_id == "src-456"
-        assert job.created_by_user_id == "user-789"
+        assert job.payload == payload
 
-    def test_create_default_payload(self):
-        """测试 None payload 默认化为空 dict"""
-        job = Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE, payload=None)
+    def test_create_with_source_id(self):
+        """带 source_id 创建"""
+        job = Job.create(
+            project_id="proj-1",
+            job_type=JobType.VIDEO_COMPOSE,
+            source_id="plan-123",
+        )
+        assert job.source_id == "plan-123"
+
+    def test_create_with_created_by(self):
+        """带创建人"""
+        job = Job.create(
+            project_id="proj-1",
+            job_type=JobType.VIDEO_COMPOSE,
+            created_by_user_id="user-1",
+        )
+        assert job.created_by_user_id == "user-1"
+
+    def test_create_with_custom_max_retries(self):
+        """自定义最大重试次数"""
+        job = Job.create(
+            project_id="proj-1",
+            job_type=JobType.VIDEO_COMPOSE,
+            max_retries=5,
+        )
+        assert job.max_retries == 5
+
+    def test_create_project_id_stripped(self):
+        """project_id 会被 strip"""
+        job = Job.create(
+            project_id="  proj-1  ",
+            job_type=JobType.VIDEO_COMPOSE,
+        )
+        assert job.project_id == "proj-1"
+
+    def test_create_source_id_stripped(self):
+        job = Job.create(
+            project_id="proj-1",
+            job_type=JobType.VIDEO_COMPOSE,
+            source_id="  src-1  ",
+        )
+        assert job.source_id == "src-1"
+
+    def test_create_created_by_stripped(self):
+        job = Job.create(
+            project_id="proj-1",
+            job_type=JobType.VIDEO_COMPOSE,
+            created_by_user_id="  user-1  ",
+        )
+        assert job.created_by_user_id == "user-1"
+
+    def test_create_none_payload_defaults_to_empty_dict(self):
+        """payload=None 时默认为空 dict"""
+        job = Job.create(
+            project_id="proj-1",
+            job_type=JobType.VIDEO_COMPOSE,
+            payload=None,
+        )
         assert job.payload == {}
 
-    def test_create_generates_unique_ids(self):
-        """测试每次创建生成不同的 ID"""
-        job1 = Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE)
-        job2 = Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE)
-        assert job1.id != job2.id
 
-    def test_create_sets_timestamps(self):
-        """测试创建时设置时间戳"""
-        before = datetime.now(timezone.utc)
-        time.sleep(0.01)
-        job = Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE)
-        time.sleep(0.01)
-        after = datetime.now(timezone.utc)
+class TestJobIsTerminal:
+    """is_terminal 属性测试"""
 
-        assert before < job.created_at < after
-        assert before < job.updated_at < after
+    def test_pending_not_terminal(self):
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        assert job.is_terminal is False
+
+    def test_running_not_terminal(self):
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.transition_to(JobStatus.RUNNING)
+        assert job.is_terminal is False
+
+    def test_success_is_terminal(self):
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.transition_to(JobStatus.RUNNING)
+        job.transition_to(JobStatus.SUCCESS)
+        assert job.is_terminal is True
+
+    def test_failed_is_terminal(self):
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.transition_to(JobStatus.RUNNING)
+        job.transition_to(JobStatus.FAILED)
+        assert job.is_terminal is True
+
+    def test_cancelled_is_terminal(self):
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.transition_to(JobStatus.CANCELLED)
+        assert job.is_terminal is True
 
 
-class TestJobStateTransitions:
-    """Job 状态转换测试"""
+class TestJobTransitions:
+    """状态转换测试"""
 
-    @pytest.fixture
-    def new_job(self):
-        return Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE)
+    def test_pending_to_running(self):
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.transition_to(JobStatus.RUNNING)
+        assert job.status == JobStatus.RUNNING
+        assert job.started_at is not None
 
-    # ===== Pending → Running =====
+    def test_pending_to_success(self):
+        """pending 可以直接到 success（快速成功）"""
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.transition_to(JobStatus.SUCCESS)
+        assert job.status == JobStatus.SUCCESS
+        assert job.completed_at is not None
 
-    def test_pending_to_running(self, new_job):
-        """测试 pending → running"""
-        assert new_job.status == JobStatus.PENDING
+    def test_pending_to_cancelled(self):
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.transition_to(JobStatus.CANCELLED)
+        assert job.status == JobStatus.CANCELLED
 
-        new_job.mark_running()
+    def test_running_to_success(self):
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.transition_to(JobStatus.RUNNING)
+        job.transition_to(JobStatus.SUCCESS)
+        assert job.status == JobStatus.SUCCESS
+        assert job.completed_at is not None
 
-        assert new_job.status == JobStatus.RUNNING
-        assert new_job.started_at is not None
-        assert new_job.completed_at is None
-        assert not new_job.is_terminal
+    def test_running_to_failed(self):
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.transition_to(JobStatus.RUNNING)
+        job.transition_to(JobStatus.FAILED)
+        assert job.status == JobStatus.FAILED
+        assert job.completed_at is not None
 
-    def test_pending_to_running_with_stage(self, new_job):
-        """测试 pending → running 带阶段描述"""
-        new_job.mark_running(stage="初始化")
-        assert new_job.current_stage == "初始化"
+    def test_running_to_cancelled(self):
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.transition_to(JobStatus.RUNNING)
+        job.transition_to(JobStatus.CANCELLED)
+        assert job.status == JobStatus.CANCELLED
 
-    # ===== Pending → Success =====
+    def test_failed_to_pending_retry(self):
+        """失败后可以回到 pending（重试）"""
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.transition_to(JobStatus.RUNNING)
+        job.transition_to(JobStatus.FAILED)
+        job.transition_to(JobStatus.PENDING)
+        assert job.status == JobStatus.PENDING
 
-    def test_pending_to_success(self, new_job):
-        """测试 pending → success（直接成功）"""
-        new_job.mark_success()
-
-        assert new_job.status == JobStatus.SUCCESS
-        assert new_job.progress == 100.0
-        assert new_job.current_stage == "完成"
-        assert new_job.completed_at is not None
-        assert new_job.is_terminal
-
-    def test_pending_to_success_with_result(self, new_job):
-        """测试 pending → success 带结果"""
-        result = {"output_url": "http://example.com/video.mp4"}
-        new_job.mark_success(result=result)
-
-        assert new_job.result == result
-
-    # ===== Pending → Cancelled =====
-
-    def test_pending_to_cancelled(self, new_job):
-        """测试 pending → cancelled"""
-        new_job.mark_cancelled()
-
-        assert new_job.status == JobStatus.CANCELLED
-        assert new_job.current_stage == "已取消"
-        assert new_job.is_terminal
-
-    # ===== Running → Success =====
-
-    def test_running_to_success(self, new_job):
-        """测试 running → success"""
-        new_job.mark_running()
-        new_job.mark_success()
-
-        assert new_job.status == JobStatus.SUCCESS
-        assert new_job.completed_at is not None
-        assert new_job.progress == 100.0
-        assert new_job.is_terminal
-
-    def test_running_to_success_preserves_started_at(self, new_job):
-        """测试 running → success 保留 started_at"""
-        new_job.mark_running()
-        started_at = new_job.started_at
-        new_job.mark_success()
-
-        assert new_job.started_at == started_at
-
-    # ===== Running → Failed =====
-
-    def test_running_to_failed(self, new_job):
-        """测试 running → failed"""
-        new_job.mark_running()
-        new_job.mark_failed("Something went wrong")
-
-        assert new_job.status == JobStatus.FAILED
-        assert new_job.error_message == "Something went wrong"
-        assert new_job.current_stage == "失败"
-        assert new_job.completed_at is not None
-        assert new_job.is_terminal
-
-    # ===== Running → Cancelled =====
-
-    def test_running_to_cancelled(self, new_job):
-        """测试 running → cancelled"""
-        new_job.mark_running()
-        new_job.mark_cancelled()
-
-        assert new_job.status == JobStatus.CANCELLED
-        assert new_job.is_terminal
-
-    # ===== Failed → Pending (Retry) =====
-
-    def test_failed_to_pending_retry(self, new_job):
-        """测试 failed → pending（重试）"""
-        new_job.mark_running()
-        new_job.mark_failed("error")
-        assert new_job.retry_count == 0
-
-        new_job.prepare_retry()
-
-        assert new_job.status == JobStatus.PENDING
-        assert new_job.retry_count == 1
-        assert new_job.progress == 0.0
-        assert new_job.error_message == ""
-        assert new_job.started_at is None
-        assert new_job.completed_at is None
-        assert new_job.celery_task_id == ""
-        assert "第 1 次重试" in new_job.current_stage
-
-    def test_retry_up_to_max_retries(self, new_job):
-        """测试最多重试 max_retries 次"""
-        new_job.max_retries = 2
-        new_job.mark_running()
-
-        # 第一次失败重试
-        new_job.mark_failed("error 1")
-        assert new_job.is_retryable  # 失败后可重试
-        new_job.prepare_retry()
-        assert new_job.retry_count == 1
-
-        # 第二次失败重试
-        new_job.mark_running()
-        new_job.mark_failed("error 2")
-        assert new_job.is_retryable  # retry_count=1 < max_retries=2
-        new_job.prepare_retry()
-        assert new_job.retry_count == 2
-
-        # 第三次失败后不可重试（retry_count == max_retries）
-        new_job.mark_running()
-        new_job.mark_failed("error 3")
-        assert not new_job.is_retryable  # 达到上限
-        with pytest.raises(ValueError, match="任务不可重试"):
-            new_job.prepare_retry()
-
-    def test_retry_not_from_failed(self, new_job):
-        """测试非 failed 状态不可重试"""
-        with pytest.raises(ValueError, match="任务不可重试"):
-            new_job.prepare_retry()  # pending 状态
-
-    # ===== 非法状态转换 =====
-
-    def test_invalid_transition_success_to_running(self, new_job):
-        """测试 success → running 非法"""
-        new_job.mark_success()
+    def test_invalid_transition_raises(self):
+        """非法状态转换抛 ValueError"""
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        # pending 不能直接到 failed
         with pytest.raises(ValueError, match="非法状态转换"):
-            new_job.mark_running()
+            job.transition_to(JobStatus.FAILED)
 
-    def test_invalid_transition_cancelled_to_running(self, new_job):
-        """测试 cancelled → running 非法"""
-        new_job.mark_cancelled()
-        with pytest.raises(ValueError, match="非法状态转换"):
-            new_job.mark_running()
+    def test_success_to_pending_raises(self):
+        """成功后不能回到 pending"""
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.transition_to(JobStatus.RUNNING)
+        job.transition_to(JobStatus.SUCCESS)
+        with pytest.raises(ValueError):
+            job.transition_to(JobStatus.PENDING)
 
-    def test_invalid_transition_pending_to_failed(self, new_job):
-        """测试 pending → failed 非法（必须经过 running）"""
-        with pytest.raises(ValueError, match="非法状态转换"):
-            new_job.mark_failed("test error")
+    def test_transition_with_string_status(self):
+        """用字符串做状态转换"""
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.transition_to("running")
+        assert job.status == JobStatus.RUNNING
 
-    def test_invalid_status_string(self, new_job):
-        """测试无效状态字符串"""
+    def test_transition_invalid_string_raises(self):
+        """无效状态字符串抛 ValueError"""
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
         with pytest.raises(ValueError, match="无效状态"):
-            new_job.transition_to("invalid_status")
+            job.transition_to("invalid_status")
+
+    def test_transition_updates_updated_at(self):
+        """状态转换更新 updated_at"""
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        old_updated = job.updated_at
+        import time
+        time.sleep(0.001)
+        job.transition_to(JobStatus.RUNNING)
+        assert job.updated_at >= old_updated
+
+    def test_started_at_only_set_once(self):
+        """started_at 只在第一次 RUNNING 时设置"""
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.transition_to(JobStatus.RUNNING)
+        first_started = job.started_at
+        job.transition_to(JobStatus.SUCCESS)
+        # 回到 pending 再 running（模拟重试场景，但started_at是None时才设置）
+        # 注意：正常重试是通过 prepare_retry 重置的
+        assert first_started is not None
 
 
-class TestJobProperties:
-    """Job 属性测试"""
+class TestJobMarkMethods:
+    """便捷标记方法测试"""
 
-    @pytest.fixture
-    def new_job(self):
-        return Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE)
+    def test_mark_running(self):
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.mark_running("合成中")
+        assert job.status == JobStatus.RUNNING
+        assert job.current_stage == "合成中"
 
-    def test_is_terminal_pending(self, new_job):
-        """测试 pending 不是终态"""
-        assert not new_job.is_terminal
+    def test_mark_running_no_stage(self):
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.mark_running()
+        assert job.status == JobStatus.RUNNING
+        assert job.current_stage == ""
 
-    def test_is_terminal_running(self, new_job):
-        """测试 running 不是终态"""
-        new_job.mark_running()
-        assert not new_job.is_terminal
+    def test_mark_success(self):
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.mark_running()
+        job.mark_success({"output_url": "http://..."})
+        assert job.status == JobStatus.SUCCESS
+        assert job.progress == 100.0
+        assert job.current_stage == "完成"
+        assert job.result == {"output_url": "http://..."}
 
-    def test_is_terminal_success(self, new_job):
-        """测试 success 是终态"""
-        new_job.mark_success()
-        assert new_job.is_terminal
+    def test_mark_success_no_result(self):
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.mark_running()
+        job.mark_success()
+        assert job.status == JobStatus.SUCCESS
+        assert job.result == {}
 
-    def test_is_terminal_failed(self, new_job):
-        """测试 failed 是终态"""
-        new_job.mark_running()
-        new_job.mark_failed("error")
-        assert new_job.is_terminal
+    def test_mark_failed(self):
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.mark_running()
+        job.mark_failed("网络超时")
+        assert job.status == JobStatus.FAILED
+        assert job.error_message == "网络超时"
+        assert job.current_stage == "失败"
 
-    def test_is_terminal_cancelled(self, new_job):
-        """测试 cancelled 是终态"""
-        new_job.mark_cancelled()
-        assert new_job.is_terminal
-
-    def test_is_retryable_failed_under_limit(self, new_job):
-        """测试失败且未达上限时可重试"""
-        new_job.mark_running()
-        new_job.mark_failed("error")
-        assert new_job.is_retryable
-
-    def test_is_retryable_failed_at_limit(self, new_job):
-        """测试失败且达上限时不可重试"""
-        new_job.max_retries = 0
-        new_job.mark_running()
-        new_job.mark_failed("error")
-        assert not new_job.is_retryable
-
-    def test_is_retryable_not_failed(self, new_job):
-        """测试非失败状态不可重试"""
-        assert not new_job.is_retryable  # pending
-        new_job.mark_running()
-        assert not new_job.is_retryable  # running
-        new_job.mark_success()
-        assert not new_job.is_retryable  # success
+    def test_mark_cancelled(self):
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.mark_cancelled()
+        assert job.status == JobStatus.CANCELLED
+        assert job.current_stage == "已取消"
 
 
 class TestJobProgress:
-    """Job 进度更新测试"""
+    """进度更新测试"""
 
-    @pytest.fixture
-    def running_job(self):
-        job = Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE)
+    def test_update_progress(self):
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.update_progress(50.0, "渲染中")
+        assert job.progress == 50.0
+        assert job.current_stage == "渲染中"
+
+    def test_update_progress_zero(self):
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.update_progress(0.0)
+        assert job.progress == 0.0
+
+    def test_update_progress_100(self):
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.update_progress(100.0)
+        assert job.progress == 100.0
+
+    def test_update_progress_negative_raises(self):
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        with pytest.raises(ValueError, match="进度必须在 0~100 之间"):
+            job.update_progress(-1.0)
+
+    def test_update_progress_over_100_raises(self):
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        with pytest.raises(ValueError, match="进度必须在 0~100 之间"):
+            job.update_progress(101.0)
+
+    def test_update_progress_without_stage(self):
+        """不传 stage 时不修改 current_stage"""
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.current_stage = "初始阶段"
+        job.update_progress(30.0)
+        assert job.progress == 30.0
+        assert job.current_stage == "初始阶段"
+
+    def test_update_progress_updates_updated_at(self):
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        old_updated = job.updated_at
+        import time
+        time.sleep(0.001)
+        job.update_progress(50.0)
+        assert job.updated_at >= old_updated
+
+
+class TestJobRetry:
+    """重试逻辑测试"""
+
+    def test_is_retryable_failed_within_limit(self):
+        """失败且未超过重试上限时可重试"""
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE, max_retries=3)
         job.mark_running()
-        return job
+        job.mark_failed("错误")
+        assert job.is_retryable is True
 
-    def test_update_progress_normal(self, running_job):
-        """测试正常更新进度"""
-        running_job.update_progress(50.0, stage="处理中")
-        assert running_job.progress == 50.0
-        assert running_job.current_stage == "处理中"
+    def test_is_retryable_failed_at_limit(self):
+        """达到重试上限时不可重试"""
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE, max_retries=1)
+        job.mark_running()
+        job.mark_failed("错误")
+        job.retry_count = 1
+        assert job.is_retryable is False
 
-    def test_update_progress_zero(self, running_job):
-        """测试更新进度为 0"""
-        running_job.update_progress(0.0)
-        assert running_job.progress == 0.0
+    def test_is_retryable_pending_false(self):
+        """pending 状态不可重试"""
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        assert job.is_retryable is False
 
-    def test_update_progress_hundred(self, running_job):
-        """测试更新进度为 100"""
-        running_job.update_progress(100.0)
-        assert running_job.progress == 100.0
+    def test_is_retryable_success_false(self):
+        """成功状态不可重试"""
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        job.mark_running()
+        job.mark_success()
+        assert job.is_retryable is False
 
-    def test_update_progress_negative(self, running_job):
-        """测试负进度报错"""
-        with pytest.raises(ValueError, match="进度必须在 0~100 之间"):
-            running_job.update_progress(-1.0)
+    def test_prepare_retry(self):
+        """准备重试"""
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE, max_retries=3)
+        job.mark_running()
+        job.mark_failed("网络错误")
+        job.celery_task_id = "task-123"
 
-    def test_update_progress_over_hundred(self, running_job):
-        """测试超过 100 的进度报错"""
-        with pytest.raises(ValueError, match="进度必须在 0~100 之间"):
-            running_job.update_progress(101.0)
+        job.prepare_retry()
 
-    def test_update_progress_without_stage(self, running_job):
-        """测试更新进度但不改变阶段"""
-        running_job.current_stage = "初始阶段"
-        running_job.update_progress(30.0)
-        assert running_job.progress == 30.0
-        assert running_job.current_stage == "初始阶段"  # 保留原值
+        assert job.status == JobStatus.PENDING
+        assert job.retry_count == 1
+        assert job.progress == 0.0
+        assert "第 1 次重试" in job.current_stage
+        assert job.error_message == ""
+        assert job.started_at is None
+        assert job.completed_at is None
+        assert job.celery_task_id == ""
 
-    def test_update_progress_updates_updated_at(self, running_job):
-        """测试更新进度会更新 updated_at"""
-        old_updated = running_job.updated_at
-        time.sleep(0.01)
-        running_job.update_progress(50.0)
-        assert running_job.updated_at > old_updated
+    def test_prepare_retry_not_retryable_raises(self):
+        """不可重试时抛 ValueError"""
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE, max_retries=0)
+        job.mark_running()
+        job.mark_failed("错误")
+        with pytest.raises(ValueError, match="任务不可重试"):
+            job.prepare_retry()
+
+    def test_prepare_retry_increments_correctly(self):
+        """多次重试计数正确"""
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE, max_retries=3)
+        job.mark_running()
+        job.mark_failed("错误1")
+        job.prepare_retry()
+        assert job.retry_count == 1
+
+        job.mark_running()
+        job.mark_failed("错误2")
+        job.prepare_retry()
+        assert job.retry_count == 2
 
 
 class TestJobToDict:
-    """Job 序列化测试"""
+    """to_dict 序列化测试"""
 
-    def test_to_dict_pending_job(self):
-        """测试 pending 状态的 Job 序列化为字典"""
+    def test_to_dict_contains_all_fields(self):
         job = Job.create(
-            project_id="proj-123",
+            project_id="p1",
             job_type=JobType.VIDEO_COMPOSE,
-            payload={"input": "data"},
-            source_id="src-456",
+            payload={"key": "value"},
+            source_id="src-1",
+            created_by_user_id="user-1",
         )
         d = job.to_dict()
-
         assert d["id"] == job.id
-        assert d["project_id"] == "proj-123"
+        assert d["project_id"] == "p1"
         assert d["job_type"] == "video_compose"
         assert d["status"] == "pending"
         assert d["progress"] == 0.0
-        assert d["payload"] == {"input": "data"}
-        assert d["result"] == {}
-        assert d["error_message"] == ""
-        assert d["retry_count"] == 0
-        assert d["max_retries"] == 3
-        assert d["source_id"] == "src-456"
+        assert d["payload"] == {"key": "value"}
+        assert d["source_id"] == "src-1"
+        assert d["created_by_user_id"] == "user-1"
         assert d["is_retryable"] is False
+
+    def test_to_dict_datetime_fields_are_strings(self):
+        """时间字段序列化为 ISO 字符串"""
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        d = job.to_dict()
+        assert isinstance(d["created_at"], str)
+        assert isinstance(d["updated_at"], str)
+
+    def test_to_dict_none_datetime_fields(self):
+        """未设置的时间字段为 None"""
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
+        d = job.to_dict()
         assert d["started_at"] is None
         assert d["completed_at"] is None
-        assert d["created_at"] is not None
-        assert d["updated_at"] is not None
 
-    def test_to_dict_completed_job(self):
-        """测试完成状态的 Job 序列化为字典"""
-        job = Job.create(project_id="proj-123", job_type=JobType.GENERATION)
+    def test_to_dict_after_success(self):
+        """成功后 to_dict 状态正确"""
+        job = Job.create(project_id="p1", job_type=JobType.VIDEO_COMPOSE)
         job.mark_running()
-        job.mark_success(result={"output": "result"})
+        job.mark_success({"url": "http://..."})
         d = job.to_dict()
-
         assert d["status"] == "success"
         assert d["progress"] == 100.0
-        assert d["result"] == {"output": "result"}
+        assert d["result"] == {"url": "http://..."}
         assert d["started_at"] is not None
         assert d["completed_at"] is not None
-
-    def test_to_dict_failed_job(self):
-        """测试失败状态的 Job 序列化为字典"""
-        job = Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE)
-        job.mark_running()
-        job.mark_failed("timeout error")
-        d = job.to_dict()
-
-        assert d["status"] == "failed"
-        assert d["error_message"] == "timeout error"
-        assert d["is_retryable"] is True
-
-
-class TestTransitionTimestamps:
-    """状态转换时间戳测试"""
-
-    @pytest.fixture
-    def new_job(self):
-        return Job.create(project_id="proj-123", job_type=JobType.VIDEO_COMPOSE)
-
-    def test_mark_running_sets_started_at(self, new_job):
-        """测试 mark_running 设置 started_at"""
-        assert new_job.started_at is None
-        new_job.mark_running()
-        assert new_job.started_at is not None
-        assert isinstance(new_job.started_at, datetime)
-        assert new_job.started_at.tzinfo is not None
-
-    def test_mark_running_twice_preserves_started_at(self, new_job):
-        """测试再次 mark_running 不覆盖 started_at"""
-        # 先手动转换到 running
-        new_job.transition_to(JobStatus.RUNNING)
-        first_started = new_job.started_at
-
-        # 不能直接再调 mark_running（会报错），但可以验证 started_at 不被重复设置
-        # transition_to 已经处理了 started_at is None 的逻辑
-        assert first_started == new_job.started_at
-
-    def test_mark_success_sets_completed_at(self, new_job):
-        """测试 mark_success 设置 completed_at"""
-        new_job.mark_running()
-        assert new_job.completed_at is None
-        new_job.mark_success()
-        assert new_job.completed_at is not None
-
-    def test_mark_failed_sets_completed_at(self, new_job):
-        """测试 mark_failed 设置 completed_at"""
-        new_job.mark_running()
-        assert new_job.completed_at is None
-        new_job.mark_failed("error")
-        assert new_job.completed_at is not None
-
-    def test_transition_updates_updated_at(self, new_job):
-        """测试每次状态转换都更新 updated_at"""
-        old_updated = new_job.updated_at
-        time.sleep(0.01)
-        new_job.mark_running()
-        assert new_job.updated_at > old_updated

tests/unit/test_quota_domain.py

@@ -1,14 +1,13 @@
-"""
-Quota 配额系统单元测试
-"""
+"""Quota 领域层单元测试 - quota.py"""
 
 import math
 
 import pytest
 
 from packages.domain.quota import (
-    QuotaChecker,
+    QUOTA_TIERS,
     QuotaCheckResult,
+    QuotaChecker,
     QuotaDimension,
     QuotaRegistry,
     QuotaTier,
@@ -20,134 +19,103 @@ from packages.domain.quota import (
 
 
 class TestQuotaDimension:
-    """配额维度枚举测试"""
+    """QuotaDimension 枚举测试"""
 
-    def test_builtin_dimensions_exist(self):
-        """测试内置维度存在"""
+    def test_all_dimensions_have_values(self):
+        """所有枚举成员都有字符串值"""
+        for dim in QuotaDimension:
+            assert isinstance(dim.value, str)
+            assert dim.value
+
+    def test_dimension_count(self):
+        """配额维度数量 >= 内置维度"""
+        # 至少有 storage_gb, videos_per_month, max_concurrent, max_templates 等
+        assert len(QuotaDimension) >= 7
+
+    def test_str_enum_behavior(self):
+        """是 str 枚举，可直接当字符串用"""
         assert QuotaDimension.STORAGE_GB == "storage_gb"
-        assert QuotaDimension.VIDEOS_PER_MONTH == "videos_per_month"
-        assert QuotaDimension.MAX_CONCURRENT == "max_concurrent"
-        assert QuotaDimension.MAX_TEMPLATES == "max_templates"
-        assert QuotaDimension.MAX_TITLES == "max_titles"
-        assert QuotaDimension.MAX_VOICEOVERS == "max_voiceovers"
-        assert QuotaDimension.AI_VOICE_ENABLED == "ai_voice_enabled"
-
-    def test_extended_dimensions_exist(self):
-        """测试扩展维度存在"""
-        assert QuotaDimension.AI_VOICE_CREDITS == "ai_voice_credits"
-        assert QuotaDimension.BATCH_EXPORT_ENABLED == "batch_export_enabled"
-        assert QuotaDimension.MULTI_PLATFORM_ENABLED == "multi_platform_enabled"
-        assert QuotaDimension.DEDUP_REPORT_ENABLED == "dedup_report_enabled"
-
-    def test_dimension_is_string(self):
-        """测试枚举值是字符串"""
         assert isinstance(QuotaDimension.STORAGE_GB, str)
-        assert QuotaDimension.STORAGE_GB == "storage_gb"
 
 
 class TestQuotaTier:
-    """配额等级测试"""
+    """QuotaTier 测试"""
 
     def test_get_limit_defined(self):
-        """测试获取已定义的配额限制"""
-        tier = QuotaTier(name="test", limits={"storage_gb": 10, "videos_per_month": 50})
-        assert tier.get_limit("storage_gb") == 10
-        assert tier.get_limit("videos_per_month") == 50
+        """已定义的维度返回正确值"""
+        tier = QuotaTier(name="test", limits={"storage": 10, "videos": 5})
+        assert tier.get_limit("storage") == 10
+        assert tier.get_limit("videos") == 5
 
     def test_get_limit_undefined_returns_zero(self):
-        """测试未定义维度返回 0"""
-        tier = QuotaTier(name="test", limits={"storage_gb": 10})
-        assert tier.get_limit("unknown_dim") == 0
+        """未定义的维度返回 0"""
+        tier = QuotaTier(name="test", limits={"storage": 10})
+        assert tier.get_limit("unknown") == 0
 
-    def test_is_unlimited_with_inf(self):
-        """测试不限量判断（inf）"""
+    def test_is_unlimited_true(self):
+        """不限量判断 - inf"""
         tier = QuotaTier(name="test", limits={"templates": float("inf")})
         assert tier.is_unlimited("templates") is True
 
-    def test_is_unlimited_with_finite(self):
-        """测试有限量判断"""
-        tier = QuotaTier(name="test", limits={"storage_gb": 10})
-        assert tier.is_unlimited("storage_gb") is False
+    def test_is_unlimited_false(self):
+        """限量判断"""
+        tier = QuotaTier(name="test", limits={"storage": 10})
+        assert tier.is_unlimited("storage") is False
 
-    def test_is_unlimited_undefined(self):
-        """测试未定义维度默认不限量（因为默认值是 inf）"""
+    def test_is_unlimited_undefined_returns_true(self):
+        """未定义的维度默认 inf，is_unlimited 返回 True"""
         tier = QuotaTier(name="test", limits={})
-        # is_unlimited 使用 limits.get(dim, float("inf")) == float("inf")
-        # 未定义时默认是 inf，所以返回 True
-        assert tier.is_unlimited("undefined") is True
-
-    def test_default_limits_empty(self):
-        """测试默认 limits 为空 dict"""
-        tier = QuotaTier(name="test")
-        assert tier.limits == {}
+        # get_limit 用 dict.get 默认 0，但 is_unlimited 用 dict.get 默认 inf
+        assert tier.is_unlimited("unknown") is True
 
 
 class TestQuotaTiers:
-    """预定义配额等级测试"""
-
-    def test_free_tier_limits(self):
-        """测试 free 套餐限制"""
-        from packages.domain.quota import QUOTA_TIERS
-
-        free = QUOTA_TIERS["free"]
-        assert free.name == "free"
-        assert free.get_limit("storage_gb") == 2
-        assert free.get_limit("videos_per_month") == 5
-        assert free.get_limit("max_concurrent") == 3
-        assert free.get_limit("max_templates") == 3
-        assert free.get_limit("max_titles") == 50
-        assert free.get_limit("max_voiceovers") == 10
-        assert free.get_limit("ai_voice_enabled") == 0
-        assert free.get_limit("ai_voice_credits") == 0
-
-    def test_basic_tier_limits(self):
-        """测试 basic 套餐限制"""
-        from packages.domain.quota import QUOTA_TIERS
-
-        basic = QUOTA_TIERS["basic"]
-        assert basic.name == "basic"
-        assert basic.get_limit("storage_gb") == 20
-        assert basic.get_limit("videos_per_month") == 30
-        assert basic.get_limit("max_concurrent") == 10
-        assert basic.get_limit("max_templates") == 15
-        assert basic.get_limit("max_titles") == 500
-        assert basic.get_limit("max_voiceovers") == 100
-        assert basic.get_limit("ai_voice_enabled") == 1
-        assert basic.get_limit("ai_voice_credits") == 100
-        assert basic.get_limit("batch_export_enabled") == 1
-
-    def test_premium_tier_limits(self):
-        """测试 premium 套餐限制"""
-        from packages.domain.quota import QUOTA_TIERS
-
-        premium = QUOTA_TIERS["premium"]
-        assert premium.name == "premium"
-        assert premium.get_limit("storage_gb") == 100
-        assert premium.get_limit("videos_per_month") == 100
-        assert premium.get_limit("max_concurrent") == 20
-        assert premium.is_unlimited("max_templates") is True
-        assert premium.get_limit("max_titles") == 500
-        assert premium.get_limit("max_voiceovers") == 100
-        assert premium.get_limit("ai_voice_enabled") == 1
-        assert premium.get_limit("ai_voice_credits") == 500
-        assert premium.get_limit("batch_export_enabled") == 1
-        assert premium.get_limit("multi_platform_enabled") == 1
-        assert premium.get_limit("dedup_report_enabled") == 1
+    """内置套餐配额测试"""
 
     def test_three_tiers_exist(self):
-        """测试三个套餐等级都存在"""
-        from packages.domain.quota import QUOTA_TIERS
-
+        """三个套餐等级都存在"""
         assert "free" in QUOTA_TIERS
         assert "basic" in QUOTA_TIERS
         assert "premium" in QUOTA_TIERS
 
+    def test_free_tier_storage(self):
+        """free 套餐 2GB 存储"""
+        assert QUOTA_TIERS["free"].get_limit(QuotaDimension.STORAGE_GB) == 2
+
+    def test_basic_tier_storage(self):
+        """basic 套餐 20GB 存储"""
+        assert QUOTA_TIERS["basic"].get_limit(QuotaDimension.STORAGE_GB) == 20
+
+    def test_premium_tier_storage(self):
+        """premium 套餐 100GB 存储"""
+        assert QUOTA_TIERS["premium"].get_limit(QuotaDimension.STORAGE_GB) == 100
+
+    def test_free_no_ai_voice(self):
+        """free 套餐没有 AI 配音"""
+        assert QUOTA_TIERS["free"].get_limit(QuotaDimension.AI_VOICE_ENABLED) == 0
+
+    def test_basic_has_ai_voice(self):
+        """basic 套餐有 AI 配音"""
+        assert QUOTA_TIERS["basic"].get_limit(QuotaDimension.AI_VOICE_ENABLED) == 1
+
+    def test_premium_templates_unlimited(self):
+        """premium 套餐模板不限量"""
+        assert QUOTA_TIERS["premium"].is_unlimited(QuotaDimension.MAX_TEMPLATES) is True
+
+    def test_free_videos_per_month(self):
+        """free 每月 5 个视频"""
+        assert QUOTA_TIERS["free"].get_limit(QuotaDimension.VIDEOS_PER_MONTH) == 5
+
+    def test_premium_multi_platform_enabled(self):
+        """premium 支持多平台发布"""
+        assert QUOTA_TIERS["premium"].get_limit(QuotaDimension.MULTI_PLATFORM_ENABLED) == 1
+
 
 class TestQuotaWarningLevel:
-    """告警级别测试"""
+    """告警级别常量测试"""
 
-    def test_warning_level_values(self):
-        """测试告警级别常量值"""
+    def test_level_values(self):
+        """四个告警级别都有定义"""
         assert QuotaWarningLevel.NORMAL == "normal"
         assert QuotaWarningLevel.WARNING == "warning"
         assert QuotaWarningLevel.CRITICAL == "critical"
@@ -155,25 +123,25 @@ class TestQuotaWarningLevel:
 
 
 class TestQuotaCheckResult:
-    """配额检查结果测试"""
+    """QuotaCheckResult 测试"""
 
     def test_usage_percent_normal(self):
-        """测试正常使用率计算"""
+        """正常使用百分比计算"""
         result = QuotaCheckResult(
             allowed=True,
-            dimension="storage_gb",
+            dimension="storage",
             limit=100,
-            used=50,
-            remaining=50,
+            used=30,
+            remaining=70,
             warning_level=QuotaWarningLevel.NORMAL,
         )
-        assert result.usage_percent == 50.0
+        assert result.usage_percent == 30.0
 
-    def test_usage_percent_over_limit(self):
-        """测试超出限制时 capped at 100%"""
+    def test_usage_percent_capped_at_100(self):
+        """超过 100% 时截断为 100%"""
         result = QuotaCheckResult(
             allowed=False,
-            dimension="storage_gb",
+            dimension="storage",
             limit=100,
             used=150,
             remaining=0,
@@ -182,10 +150,10 @@ class TestQuotaCheckResult:
         assert result.usage_percent == 100.0
 
     def test_usage_percent_zero_limit_with_usage(self):
-        """测试限制为 0 但有使用量时返回 100%"""
+        """limit=0 但有使用量，返回 100%"""
         result = QuotaCheckResult(
             allowed=False,
-            dimension="ai_voice",
+            dimension="storage",
             limit=0,
             used=5,
             remaining=0,
@@ -194,10 +162,10 @@ class TestQuotaCheckResult:
         assert result.usage_percent == 100.0
 
     def test_usage_percent_zero_limit_no_usage(self):
-        """测试限制为 0 且无使用量时返回 0%"""
+        """limit=0 且无使用量，返回 0%"""
         result = QuotaCheckResult(
             allowed=True,
-            dimension="ai_voice",
+            dimension="storage",
             limit=0,
             used=0,
             remaining=0,
@@ -206,283 +174,223 @@ class TestQuotaCheckResult:
         assert result.usage_percent == 0.0
 
     def test_usage_percent_unlimited(self):
-        """测试不限量时返回 0%"""
+        """不限量时使用百分比为 0"""
         result = QuotaCheckResult(
             allowed=True,
             dimension="templates",
             limit=float("inf"),
-            used=1000,
+            used=50,
             remaining=float("inf"),
             warning_level=QuotaWarningLevel.NORMAL,
         )
         assert result.usage_percent == 0.0
 
-    def test_usage_percent_exactly_100(self):
-        """测试刚好 100% 使用"""
-        result = QuotaCheckResult(
-            allowed=False,
-            dimension="storage_gb",
-            limit=100,
-            used=100,
-            remaining=0,
-            warning_level=QuotaWarningLevel.EXCEEDED,
-        )
-        assert result.usage_percent == 100.0
-
 
 class TestQuotaRegistry:
-    """配额注册表测试"""
+    """QuotaRegistry 测试"""
 
-    def test_initial_builtin_dimensions(self):
-        """测试初始化后内置维度已注册"""
+    def test_initial_dimensions(self):
+        """初始化时内置维度已注册"""
         registry = QuotaRegistry()
         dims = registry.list_dimensions()
+        assert QuotaDimension.STORAGE_GB in dims
+        assert QuotaDimension.VIDEOS_PER_MONTH in dims
 
-        assert "storage_gb" in dims
-        assert "videos_per_month" in dims
-        assert "max_concurrent" in dims
-        assert "max_templates" in dims
-        assert "max_titles" in dims
-        assert "max_voiceovers" in dims
-        assert "ai_voice_enabled" in dims
+    def test_initial_tiers(self):
+        """初始化时三个套餐已注册"""
+        registry = QuotaRegistry()
+        tiers = registry.list_tiers()
+        assert "free" in tiers
+        assert "basic" in tiers
+        assert "premium" in tiers
 
     def test_register_new_dimension(self):
-        """测试注册新维度"""
+        """注册新的配额维度"""
         registry = QuotaRegistry()
         registry.register_dimension("custom_dim", "自定义维度")
-
         dims = registry.list_dimensions()
         assert "custom_dim" in dims
         assert dims["custom_dim"] == "自定义维度"
 
-    def test_register_dimension_with_default_limits(self):
-        """测试注册带默认限制的新维度"""
-        registry = QuotaRegistry()
-        registry.register_dimension(
-            "custom_feature",
-            "自定义功能",
-            default_limits={"free": 0, "basic": 1, "premium": 5},
-        )
-
-        assert registry.get_limit("free", "custom_feature") == 0
-        assert registry.get_limit("basic", "custom_feature") == 1
-        assert registry.get_limit("premium", "custom_feature") == 5
-
-    def test_register_dimension_without_default_limits(self):
-        """测试注册不带默认限制的新维度（所有套餐默认 0）"""
-        registry = QuotaRegistry()
-        registry.register_dimension("new_feature", "新功能")
-
-        assert registry.get_limit("free", "new_feature") == 0
-        assert registry.get_limit("basic", "new_feature") == 0
-        assert registry.get_limit("premium", "new_feature") == 0
-
     def test_register_dimension_idempotent(self):
-        """测试重复注册是幂等的"""
+        """重复注册是幂等的"""
         registry = QuotaRegistry()
-        registry.register_dimension("test_dim", "测试维度", default_limits={"free": 10})
-        # 第二次注册不应该改变任何东西
-        registry.register_dimension("test_dim", "另一个描述", default_limits={"free": 999})
+        registry.register_dimension("custom", "描述1")
+        registry.register_dimension("custom", "描述2")
+        # 保留第一次注册的描述
+        assert registry.list_dimensions()["custom"] == "描述1"
 
-        dims = registry.list_dimensions()
-        assert dims["test_dim"] == "测试维度"  # 保留第一次的描述
-        assert registry.get_limit("free", "test_dim") == 10  # 保留第一次的限制
-
-    def test_register_unknown_plan_ignored(self):
-        """测试未知套餐的默认限制被忽略"""
+    def test_register_with_default_limits(self):
+        """注册时指定各套餐的默认限制"""
         registry = QuotaRegistry()
         registry.register_dimension(
-            "test_dim",
-            "测试",
-            default_limits={"free": 1, "enterprise": 100},
+            "custom",
+            "自定义",
+            default_limits={"free": 1, "basic": 10, "premium": 100},
         )
+        assert registry.get_limit("free", "custom") == 1
+        assert registry.get_limit("basic", "custom") == 10
+        assert registry.get_limit("premium", "custom") == 100
 
-        assert registry.get_limit("free", "test_dim") == 1
-        # enterprise 套餐不存在，不影响
-        assert "enterprise" not in registry.list_tiers()
+    def test_register_without_default_limits_defaults_to_zero(self):
+        """不指定默认限制时各套餐该维度为 0"""
+        registry = QuotaRegistry()
+        registry.register_dimension("custom_no_limit", "自定义")
+        assert registry.get_limit("free", "custom_no_limit") == 0
+        assert registry.get_limit("basic", "custom_no_limit") == 0
+
+    def test_register_default_limits_ignores_unknown_plan(self):
+        """默认限制中未知的套餐名被忽略"""
+        registry = QuotaRegistry()
+        registry.register_dimension(
+            "custom",
+            "自定义",
+            default_limits={"nonexistent": 999},
+        )
+        # 不报错，但也不会创建新套餐
+        assert registry.get_tier("nonexistent") is None
 
     def test_get_tier_existing(self):
-        """测试获取存在的套餐"""
+        """获取存在的套餐"""
         registry = QuotaRegistry()
         tier = registry.get_tier("free")
         assert tier is not None
         assert tier.name == "free"
 
     def test_get_tier_nonexistent(self):
-        """测试获取不存在的套餐返回 None"""
+        """获取不存在的套餐返回 None"""
         registry = QuotaRegistry()
-        assert registry.get_tier("nonexistent") is None
+        assert registry.get_tier("enterprise") is None
+
+    def test_get_limit_existing(self):
+        """获取存在的套餐和维度的限制"""
+        registry = QuotaRegistry()
+        assert registry.get_limit("free", QuotaDimension.STORAGE_GB) == 2
 
     def test_get_limit_nonexistent_plan(self):
-        """测试不存在套餐的限制返回 0"""
+        """不存在的套餐返回 0"""
         registry = QuotaRegistry()
-        assert registry.get_limit("enterprise", "storage_gb") == 0
-
-    def test_list_tiers(self):
-        """测试列出所有套餐"""
-        registry = QuotaRegistry()
-        tiers = registry.list_tiers()
-        assert "free" in tiers
-        assert "basic" in tiers
-        assert "premium" in tiers
-        assert len(tiers) == 3
+        assert registry.get_limit("unknown", QuotaDimension.STORAGE_GB) == 0
 
     def test_list_dimensions_returns_copy(self):
-        """测试 list_dimensions 返回副本（修改不影响内部）"""
+        """list_dimensions 返回副本，修改不影响内部"""
         registry = QuotaRegistry()
         dims = registry.list_dimensions()
-        dims["fake_dim"] = "fake"
+        dims["fake"] = "fake"
+        assert "fake" not in registry.list_dimensions()
 
-        # 原始注册表不应被修改
-        assert "fake_dim" not in registry.list_dimensions()
+    def test_list_tiers_returns_all_three(self):
+        """列出所有套餐"""
+        registry = QuotaRegistry()
+        tiers = registry.list_tiers()
+        assert len(tiers) == 3
+        assert set(tiers) == {"free", "basic", "premium"}
 
 
 class TestQuotaChecker:
-    """配额检查器测试"""
-
-    @pytest.fixture
-    def checker(self):
-        return QuotaChecker()
-
-    # ===== 基础检查 =====
-
-    def test_check_free_storage_under_limit(self, checker):
-        """测试 free 套餐存储未超限"""
-        result = checker.check("free", "storage_gb", 1.0)
+    """QuotaChecker 测试"""
 
+    def test_check_under_limit_allowed(self):
+        """使用量低于限制，允许"""
+        checker = QuotaChecker()
+        result = checker.check("free", QuotaDimension.STORAGE_GB, 1.0)
         assert result.allowed is True
-        assert result.limit == 2
-        assert result.used == 1.0
         assert result.remaining == 1.0
         assert result.warning_level == QuotaWarningLevel.NORMAL
-        assert result.dimension == "storage_gb"
-
-    def test_check_free_storage_over_limit(self, checker):
-        """测试 free 套餐存储超限"""
-        result = checker.check("free", "storage_gb", 3.0)
 
+    def test_check_at_limit_not_allowed(self):
+        """使用量等于限制，不允许（used < limit 判定）"""
+        checker = QuotaChecker()
+        result = checker.check("free", QuotaDimension.STORAGE_GB, 2.0)
         assert result.allowed is False
         assert result.remaining == 0
         assert result.warning_level == QuotaWarningLevel.EXCEEDED
 
-    def test_check_free_storage_exactly_at_limit(self, checker):
-        """测试刚好达到限制（不允许）"""
-        result = checker.check("free", "storage_gb", 2.0)
-
-        # used < limit → 2 < 2 → False
+    def test_check_over_limit(self):
+        """使用量超过限制"""
+        checker = QuotaChecker()
+        result = checker.check("free", QuotaDimension.STORAGE_GB, 3.0)
         assert result.allowed is False
+        assert result.remaining == 0
         assert result.warning_level == QuotaWarningLevel.EXCEEDED
 
-    # ===== 告警级别 =====
-
-    def test_warning_level_normal(self, checker):
-        """测试正常级别（< 80%）"""
-        result = checker.check("free", "storage_gb", 1.0)  # 50%
-        assert result.warning_level == QuotaWarningLevel.NORMAL
-
-    def test_warning_level_warning(self, checker):
-        """测试警告级别（80% ~ 95%）"""
-        result = checker.check("free", "storage_gb", 1.7)  # 85%
+    def test_check_warning_level_80_percent(self):
+        """80% 触发 WARNING"""
+        checker = QuotaChecker()
+        # 100GB 的 80% = 80GB
+        result = checker.check("premium", QuotaDimension.STORAGE_GB, 80.0)
         assert result.warning_level == QuotaWarningLevel.WARNING
 
-    def test_warning_level_critical(self, checker):
-        """测试严重级别（95% ~ 100%）"""
-        result = checker.check("free", "storage_gb", 1.95)  # 97.5%
+    def test_check_warning_level_95_percent(self):
+        """95% 触发 CRITICAL"""
+        checker = QuotaChecker()
+        result = checker.check("premium", QuotaDimension.STORAGE_GB, 95.0)
         assert result.warning_level == QuotaWarningLevel.CRITICAL
 
-    def test_warning_level_exceeded(self, checker):
-        """测试超限级别（>= 100%）"""
-        result = checker.check("free", "storage_gb", 2.0)  # 100%
+    def test_check_warning_level_exceeded(self):
+        """100% 及以上触发 EXCEEDED"""
+        checker = QuotaChecker()
+        result = checker.check("premium", QuotaDimension.STORAGE_GB, 100.0)
         assert result.warning_level == QuotaWarningLevel.EXCEEDED
 
-    # ===== 不限量 =====
-
-    def test_check_unlimited_templates_premium(self, checker):
-        """测试 premium 套餐模板不限量"""
-        result = checker.check("premium", "max_templates", 9999)
-
+    def test_check_unlimited_always_allowed(self):
+        """不限量的维度始终允许"""
+        checker = QuotaChecker()
+        result = checker.check("premium", QuotaDimension.MAX_TEMPLATES, 9999)
         assert result.allowed is True
-        assert result.limit == float("inf")
-        assert result.remaining == float("inf")
+        assert math.isinf(result.remaining)
         assert result.warning_level == QuotaWarningLevel.NORMAL
 
-    # ===== 0 限制 =====
-
-    def test_check_zero_limit_with_usage(self, checker):
-        """测试限制为 0 但有使用量"""
-        result = checker.check("free", "ai_voice_enabled", 1)
-
+    def test_check_unknown_plan_zero_limit(self):
+        """未知套餐限制为 0，used=0 时不允许（0 < 0 为 False）"""
+        checker = QuotaChecker()
+        result = checker.check("unknown", QuotaDimension.STORAGE_GB, 0)
+        assert result.limit == 0
         assert result.allowed is False
-        assert result.warning_level == QuotaWarningLevel.EXCEEDED
 
-    def test_check_zero_limit_no_usage(self, checker):
-        """测试限制为 0 且无使用量"""
-        result = checker.check("free", "ai_voice_enabled", 0)
-
-        # used < limit → 0 < 0 → False? 让我们看看...
-        # 实际上 0 < 0 是 False，所以 allowed = False
-        # 但 warning_level: limit <= 0 and used == 0 → NORMAL
-        # 等一下，看看代码逻辑：
-        # if limit <= 0: return EXCEEDED if used > 0 else NORMAL
-        assert result.warning_level == QuotaWarningLevel.NORMAL
-
-    # ===== 多维度检查 =====
-
-    def test_check_multiple(self, checker):
-        """测试批量检查多个维度"""
-        usage = {
-            "storage_gb": 1.0,
-            "videos_per_month": 3,
-            "max_concurrent": 2,
-        }
-        results = checker.check_multiple("free", usage)
-
-        assert len(results) == 3
-        dims = {r.dimension: r for r in results}
-        assert dims["storage_gb"].allowed is True
-        assert dims["videos_per_month"].allowed is True
-        assert dims["max_concurrent"].allowed is True
-
-    def test_check_multiple_some_exceeded(self, checker):
-        """测试批量检查中有超限的"""
-        usage = {
-            "storage_gb": 5.0,  # 超限
-            "videos_per_month": 3,  # 正常
-        }
-        results = checker.check_multiple("free", usage)
-
-        dims = {r.dimension: r for r in results}
-        assert dims["storage_gb"].allowed is False
-        assert dims["videos_per_month"].allowed is True
-
-    # ===== 自定义 registry =====
+    def test_check_multiple(self):
+        """批量检查多个维度"""
+        checker = QuotaChecker()
+        results = checker.check_multiple(
+            "free",
+            {
+                QuotaDimension.STORAGE_GB: 1.0,
+                QuotaDimension.VIDEOS_PER_MONTH: 3,
+            },
+        )
+        assert len(results) == 2
+        assert all(r.allowed for r in results)
+        dims = {r.dimension for r in results}
+        assert QuotaDimension.STORAGE_GB in dims
+        assert QuotaDimension.VIDEOS_PER_MONTH in dims
 
     def test_check_with_custom_registry(self):
-        """测试使用自定义 registry"""
+        """使用自定义注册表"""
         registry = QuotaRegistry()
-        registry.register_dimension(
-            "custom_feature",
-            "自定义",
-            default_limits={"free": 5, "basic": 20},
-        )
+        registry.register_dimension("custom", "自定义", default_limits={"free": 5})
         checker = QuotaChecker(registry)
-
-        result = checker.check("free", "custom_feature", 3)
+        result = checker.check("free", "custom", 3)
         assert result.allowed is True
         assert result.limit == 5
 
-        result = checker.check("basic", "custom_feature", 25)
-        assert result.allowed is False
+    def test_compute_warning_level_zero_limit_no_usage(self):
+        """limit=0, used=0 → NORMAL"""
+        level = QuotaChecker._compute_warning_level(0, 0)
+        assert level == QuotaWarningLevel.NORMAL
 
-    def test_check_unknown_plan(self, checker):
-        """测试未知套餐（限制为 0）"""
-        result = checker.check("enterprise", "storage_gb", 1)
-        assert result.allowed is False
-        assert result.limit == 0
+    def test_compute_warning_level_zero_limit_with_usage(self):
+        """limit=0, used>0 → EXCEEDED"""
+        level = QuotaChecker._compute_warning_level(1, 0)
+        assert level == QuotaWarningLevel.EXCEEDED
+
+    def test_compute_warning_level_negative_limit(self):
+        """limit<0 视同 0 处理"""
+        level = QuotaChecker._compute_warning_level(1, -1)
+        assert level == QuotaWarningLevel.EXCEEDED
 
 
 class TestGetWarningLevel:
-    """便捷函数 get_warning_level 测试"""
+    """get_warning_level 便捷函数测试"""
 
     def test_normal(self):
         assert get_warning_level(50, 100) == QuotaWarningLevel.NORMAL
@@ -491,61 +399,26 @@ class TestGetWarningLevel:
         assert get_warning_level(85, 100) == QuotaWarningLevel.WARNING
 
     def test_critical(self):
-        assert get_warning_level(96, 100) == QuotaWarningLevel.CRITICAL
+        assert get_warning_level(97, 100) == QuotaWarningLevel.CRITICAL
 
     def test_exceeded(self):
         assert get_warning_level(100, 100) == QuotaWarningLevel.EXCEEDED
-        assert get_warning_level(150, 100) == QuotaWarningLevel.EXCEEDED
-
-    def test_zero_limit_with_usage(self):
-        assert get_warning_level(5, 0) == QuotaWarningLevel.EXCEEDED
-
-    def test_zero_limit_no_usage(self):
-        assert get_warning_level(0, 0) == QuotaWarningLevel.NORMAL
 
     def test_unlimited(self):
         assert get_warning_level(9999, float("inf")) == QuotaWarningLevel.NORMAL
 
-    def test_boundary_79_percent(self):
-        """测试 79% 仍是 normal"""
-        assert get_warning_level(79, 100) == QuotaWarningLevel.NORMAL
-
-    def test_boundary_80_percent(self):
-        """测试 80% 是 warning"""
-        assert get_warning_level(80, 100) == QuotaWarningLevel.WARNING
-
-    def test_boundary_94_percent(self):
-        """测试 94% 仍是 warning"""
-        assert get_warning_level(94, 100) == QuotaWarningLevel.WARNING
-
-    def test_boundary_95_percent(self):
-        """测试 95% 是 critical"""
-        assert get_warning_level(95, 100) == QuotaWarningLevel.CRITICAL
-
-    def test_boundary_99_percent(self):
-        """测试 99% 仍是 critical"""
-        assert get_warning_level(99, 100) == QuotaWarningLevel.CRITICAL
-
-    def test_zero_usage(self):
-        """测试 0 使用量"""
-        assert get_warning_level(0, 100) == QuotaWarningLevel.NORMAL
-
 
 class TestGlobalSingletons:
     """全局单例测试"""
 
-    def test_quota_registry_exists(self):
-        """测试全局 quota_registry 存在"""
-        assert quota_registry is not None
+    def test_quota_registry_is_instance(self):
         assert isinstance(quota_registry, QuotaRegistry)
-        assert "free" in quota_registry.list_tiers()
 
-    def test_quota_checker_exists(self):
-        """测试全局 quota_checker 存在"""
-        assert quota_checker is not None
+    def test_quota_checker_is_instance(self):
         assert isinstance(quota_checker, QuotaChecker)
 
     def test_global_checker_uses_global_registry(self):
-        """测试全局 checker 使用全局 registry"""
-        result = quota_checker.check("free", "storage_gb", 1.0)
-        assert result.limit == 2
+        """全局 checker 使用全局 registry"""
+        # 验证能正常工作
+        result = quota_checker.check("free", QuotaDimension.STORAGE_GB, 1.0)
+        assert result.allowed is True