test(e2e): 补充P1级前端E2E测试（剪辑策划/音色库/声音克隆/模板库/标题库/设置/订阅）

.gitea/workflows/auto-merge.yml

@@ -16,7 +16,46 @@ jobs:
           GITHUB_TOKEN: ${{ github.token }}
         run: |
           set -eu
-          bash scripts/ci_checkout.sh
+          python3 - <<'PY'
+          import io, os, tarfile, time, urllib.request, urllib.error
+          url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/{os.environ['GITHUB_SHA']}.tar.gz"
+          request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
+          last_err = None
+          for attempt in range(5):
+              try:
+                  with urllib.request.urlopen(request, timeout=120) as response:
+                      archive = response.read()
+                  break
+              except urllib.error.HTTPError as e:
+                  last_err = e
+                  if e.code >= 500 and attempt < 4:
+                      wait = 2 ** attempt
+                      print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                      time.sleep(wait)
+                      continue
+                  raise
+              except Exception as e:
+                  last_err = e
+                  if attempt < 4:
+                      wait = 2 ** attempt
+                      print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                      time.sleep(wait)
+                      continue
+                  raise
+          else:
+              raise last_err
+          with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
+              top_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
+              for member in tar.getmembers():
+                  name = member.name
+                  if name == top_prefix[:-1]:
+                      continue
+                  if name.startswith(top_prefix):
+                      member.name = name[len(top_prefix):]
+                  if member.name:
+                      tar.extract(member, '.')
+          PY
+
       - name: Auto merge develop PRs
         run: |
           bash scripts/auto_merge_prs.sh develop

.gitea/workflows/ci-cd.yml

@@ -15,7 +15,6 @@ on:
     branches:
       - main
       - develop
-  workflow_dispatch:
 
 permissions:
   contents: read
@@ -37,7 +36,46 @@ jobs:
           GITHUB_TOKEN: ${{ github.token }}
         run: |
           set -eu
-          bash scripts/ci_checkout.sh
+          python3 - <<'PY'
+          import io, os, tarfile, time, urllib.request, urllib.error
+          url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
+          request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
+          last_err = None
+          for attempt in range(5):
+              try:
+                  with urllib.request.urlopen(request, timeout=120) as response:
+                      archive = response.read()
+                  break
+              except urllib.error.HTTPError as e:
+                  last_err = e
+                  if e.code >= 500 and attempt < 4:
+                      wait = 2 ** attempt
+                      print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                      time.sleep(wait)
+                      continue
+                  raise
+              except Exception as e:
+                  last_err = e
+                  if attempt < 4:
+                      wait = 2 ** attempt
+                      print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                      time.sleep(wait)
+                      continue
+                  raise
+          else:
+              raise last_err
+          with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
+              root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
+              for member in tar.getmembers():
+                  name = member.name
+                  if name == root_prefix[:-1]:
+                      continue
+                  if name.startswith(root_prefix):
+                      member.name = name[len(root_prefix):]
+                  if member.name:
+                      tar.extract(member, '.')
+          PY
+
       - name: Verify CI environment
         shell: sh
         run: |
@@ -46,14 +84,6 @@ jobs:
           python3 -m pip --version
           echo "CI environment is ready"
 
-      - name: Cache pip dependencies
-        uses: actions/cache@v3
-        with:
-          path: ~/.cache/pip
-          key: ${{ runner.os }}-pip-${{ hashFiles('**/requirements*.txt') }}
-          restore-keys: |
-            ${{ runner.os }}-pip-
-
       - name: Install dependencies
         shell: sh
         run: |
@@ -179,7 +209,7 @@ jobs:
         run: |
           set -eu
           pip install -q pytest-rerunfailures
-          PYTHONPATH="$PWD/apps/api:$PWD" python3 -m pytest tests/integration -q --timeout=60 -x --reruns 2 --reruns-delay 1 -m "not performance"
+          PYTHONPATH="$PWD/apps/api:$PWD" python3 -m pytest tests/integration -q --timeout=60 -x --reruns 2 --reruns-delay 1
 
       - name: Run API performance baseline tests
         shell: sh
@@ -251,14 +281,45 @@ jobs:
           GITHUB_TOKEN: ${{ github.token }}
         run: |
           set -eu
-          bash scripts/ci_checkout.sh
-      - name: Cache npm dependencies
-        uses: actions/cache@v3
-        with:
-          path: ~/.npm
-          key: ${{ runner.os }}-npm-${{ hashFiles('apps/web/package-lock.json') }}
-          restore-keys: |
-            ${{ runner.os }}-npm-
+          python3 - <<'PY'
+          import io, os, tarfile, time, urllib.request, urllib.error
+          url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
+          request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
+          last_err = None
+          for attempt in range(5):
+              try:
+                  with urllib.request.urlopen(request, timeout=120) as response:
+                      archive = response.read()
+                  break
+              except urllib.error.HTTPError as e:
+                  last_err = e
+                  if e.code >= 500 and attempt < 4:
+                      wait = 2 ** attempt
+                      print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                      time.sleep(wait)
+                      continue
+                  raise
+              except Exception as e:
+                  last_err = e
+                  if attempt < 4:
+                      wait = 2 ** attempt
+                      print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                      time.sleep(wait)
+                      continue
+                  raise
+          else:
+              raise last_err
+          with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
+              root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
+              for member in tar.getmembers():
+                  name = member.name
+                  if name == root_prefix[:-1]:
+                      continue
+                  if name.startswith(root_prefix):
+                      member.name = name[len(root_prefix):]
+                  if member.name:
+                      tar.extract(member, '.')
+          PY
 
       - name: Install dependencies
         shell: sh
@@ -266,7 +327,6 @@ jobs:
           set -eu
           docker run --rm \
             -v "$PWD:/workspace" \
-            -v "$HOME/.npm:/root/.npm" \
             -w /workspace/apps/web \
             docker.m.daocloud.io/library/node:20 \
             sh -lc 'npm ci'
@@ -325,7 +385,45 @@ jobs:
         GITHUB_TOKEN: ${{ github.token }}
       run: |
         set -eu
-        bash scripts/ci_checkout.sh
+        python3 - <<'INNERPY'
+        import io, os, tarfile, time, urllib.request, urllib.error
+        url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
+        request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
+        last_err = None
+        for attempt in range(5):
+            try:
+                with urllib.request.urlopen(request, timeout=120) as response:
+                    archive = response.read()
+                break
+            except urllib.error.HTTPError as e:
+                last_err = e
+                if e.code >= 500 and attempt < 4:
+                    wait = 2 ** attempt
+                    print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                    time.sleep(wait)
+                    continue
+                raise
+            except Exception as e:
+                last_err = e
+                if attempt < 4:
+                    wait = 2 ** attempt
+                    print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                    time.sleep(wait)
+                    continue
+        else:
+            raise last_err
+        with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
+            root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
+            for member in tar.getmembers():
+                name = member.name
+                if name == root_prefix[:-1]:
+                    continue
+                if name.startswith(root_prefix):
+                    member.name = name[len(root_prefix):]
+                if member.name:
+                    tar.extract(member, '.')
+        INNERPY
+
     - name: Build and push all images to Gitea Registry
       shell: sh
       env:
@@ -418,15 +516,45 @@ jobs:
         GITHUB_TOKEN: ${{ github.token }}
       run: |
         set -eu
-        bash scripts/ci_checkout.sh
-    - name: Cache npm dependencies
-      uses: actions/cache@v3
-      with:
-        path: ~/.npm
-        key: ${{ runner.os }}-npm-${{ hashFiles('apps/web/package-lock.json') }}
-        restore-keys: |
-          ${{ runner.os }}-npm-
-
+        python3 - <<'PY'
+        import io, os, tarfile, time, urllib.request, urllib.error
+        url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
+        request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
+        last_err = None
+        for attempt in range(5):
+            try:
+                with urllib.request.urlopen(request, timeout=120) as response:
+                    archive = response.read()
+                break
+            except urllib.error.HTTPError as e:
+                last_err = e
+                if e.code >= 500 and attempt < 4:
+                    wait = 2 ** attempt
+                    print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                    time.sleep(wait)
+                    continue
+                raise
+            except Exception as e:
+                last_err = e
+                if attempt < 4:
+                    wait = 2 ** attempt
+                    print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                    time.sleep(wait)
+                    continue
+                raise
+        else:
+            raise last_err
+        with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
+            root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
+            for member in tar.getmembers():
+                name = member.name
+                if name == root_prefix[:-1]:
+                    continue
+                if name.startswith(root_prefix):
+                    member.name = name[len(root_prefix):]
+                if member.name:
+                    tar.extract(member, '.')
+        PY
     - name: Run Playwright E2E on staging
       shell: sh
       run: |
@@ -437,7 +565,6 @@ jobs:
           -e E2E_BROWSER_CHANNEL=chromium \
           -e PLAYWRIGHT_HEADLESS=1 \
           -v "$PWD:/workspace" \
-          -v "$HOME/.npm:/root/.npm" \
           -w /workspace/apps/web \
           git.xiaoxiajianji.com/xiaoxia/base/playwright:v1.45.0-jammy \
           sh -lc "npm ci && npx playwright test --reporter=line --project=chromium e2e/auth.spec.ts e2e/auth-guard.spec.ts e2e/core-upload.spec.ts e2e/core-generation.spec.ts"
@@ -456,15 +583,45 @@ jobs:
         GITHUB_TOKEN: ${{ github.token }}
       run: |
         set -eu
-        bash scripts/ci_checkout.sh
-    - name: Cache npm dependencies
-      uses: actions/cache@v3
-      with:
-        path: ~/.npm
-        key: ${{ runner.os }}-npm-${{ hashFiles('apps/web/package-lock.json') }}
-        restore-keys: |
-          ${{ runner.os }}-npm-
-
+        python3 - <<'PY'
+        import io, os, tarfile, time, urllib.request, urllib.error
+        url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
+        request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
+        last_err = None
+        for attempt in range(5):
+            try:
+                with urllib.request.urlopen(request, timeout=120) as response:
+                    archive = response.read()
+                break
+            except urllib.error.HTTPError as e:
+                last_err = e
+                if e.code >= 500 and attempt < 4:
+                    wait = 2 ** attempt
+                    print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                    time.sleep(wait)
+                    continue
+                raise
+            except Exception as e:
+                last_err = e
+                if attempt < 4:
+                    wait = 2 ** attempt
+                    print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                    time.sleep(wait)
+                    continue
+                raise
+        else:
+            raise last_err
+        with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
+            root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
+            for member in tar.getmembers():
+                name = member.name
+                if name == root_prefix[:-1]:
+                    continue
+                if name.startswith(root_prefix):
+                    member.name = name[len(root_prefix):]
+                if member.name:
+                    tar.extract(member, '.')
+        PY
     - name: Run API integration tests on staging
       shell: sh
       run: |
@@ -473,7 +630,6 @@ jobs:
           -e E2E_BASE_URL=https://staging.xiaoxiajianji.com \
           -e E2E_API_BASE=https://staging-api.xiaoxiajianji.com/api/v1 \
           -v "$PWD:/workspace" \
-          -v "$HOME/.npm:/root/.npm" \
           -w /workspace/apps/web \
           git.xiaoxiajianji.com/xiaoxia/base/playwright:v1.45.0-jammy \
           sh -lc 'npm ci && npx playwright test --reporter=line e2e/test_auth.spec.ts e2e/test_asset.spec.ts e2e/test_project.spec.ts'
@@ -494,7 +650,46 @@ jobs:
         GITHUB_TOKEN: ${{ github.token }}
       run: |
         set -eu
-        bash scripts/ci_checkout.sh
+        python3 - <<'PY'
+        import io, os, tarfile, time, urllib.request, urllib.error
+        url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
+        request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
+        last_err = None
+        for attempt in range(5):
+            try:
+                with urllib.request.urlopen(request, timeout=120) as response:
+                    archive = response.read()
+                break
+            except urllib.error.HTTPError as e:
+                last_err = e
+                if e.code >= 500 and attempt < 4:
+                    wait = 2 ** attempt
+                    print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                    time.sleep(wait)
+                    continue
+                raise
+            except Exception as e:
+                last_err = e
+                if attempt < 4:
+                    wait = 2 ** attempt
+                    print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                    time.sleep(wait)
+                    continue
+                raise
+        else:
+            raise last_err
+        with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
+            root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
+            for member in tar.getmembers():
+                name = member.name
+                if name == root_prefix[:-1]:
+                    continue
+                if name.startswith(root_prefix):
+                    member.name = name[len(root_prefix):]
+                if member.name:
+                    tar.extract(member, '.')
+        PY
+
     - name: Build and push all images (api + worker + web, with buildx cache)
       shell: sh
       env:
@@ -598,14 +793,46 @@ jobs:
         GITHUB_TOKEN: ${{ github.token }}
       run: |
         set -eu
-        bash scripts/ci_checkout.sh
-    - name: Cache npm dependencies
-      uses: actions/cache@v3
-      with:
-        path: ~/.npm
-        key: ${{ runner.os }}-npm-${{ hashFiles('apps/web/package-lock.json') }}
-        restore-keys: |
-          ${{ runner.os }}-npm-
+        python3 - <<'PY'
+        import io, os, tarfile, time, urllib.request, urllib.error
+        url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
+        request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
+        # Retry up to 5 times with backoff for transient 5xx errors
+        last_err = None
+        for attempt in range(5):
+            try:
+                with urllib.request.urlopen(request, timeout=120) as response:
+                    archive = response.read()
+                break
+            except urllib.error.HTTPError as e:
+                last_err = e
+                if e.code >= 500 and attempt < 4:
+                    wait = 2 ** attempt
+                    print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                    time.sleep(wait)
+                    continue
+                raise
+            except Exception as e:
+                last_err = e
+                if attempt < 4:
+                    wait = 2 ** attempt
+                    print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                    time.sleep(wait)
+                    continue
+                raise
+        else:
+            raise last_err
+        with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
+            root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
+            for member in tar.getmembers():
+                name = member.name
+                if name == root_prefix[:-1]:
+                    continue
+                if name.startswith(root_prefix):
+                    member.name = name[len(root_prefix):]
+                if member.name:
+                    tar.extract(member, '.')
+        PY
 
     - name: Run production browser E2E
       shell: sh
@@ -617,7 +844,6 @@ jobs:
           -e E2E_BROWSER_CHANNEL=chromium \
           -e PLAYWRIGHT_HEADLESS=1 \
           -v "$PWD:/workspace" \
-          -v "$HOME/.npm:/root/.npm" \
           -w /workspace/apps/web \
           git.xiaoxiajianji.com/xiaoxia/base/playwright:v1.45.0-jammy \
           sh -lc 'npm ci && npx playwright test --reporter=line --project=chromium e2e/auth.spec.ts e2e/auth-guard.spec.ts e2e/core-upload.spec.ts e2e/core-generation.spec.ts e2e/core-titles.spec.ts'

.gitea/workflows/daily-check.yml

@@ -24,7 +24,46 @@ jobs:
           GITHUB_TOKEN: ${{ github.token }}
         run: |
           set -eu
-          bash scripts/ci_checkout.sh
+          python3 - <<'PY'
+          import io, os, tarfile, time, urllib.request, urllib.error
+          url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
+          request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
+          last_err = None
+          for attempt in range(5):
+              try:
+                  with urllib.request.urlopen(request, timeout=120) as response:
+                      archive = response.read()
+                  break
+              except urllib.error.HTTPError as e:
+                  last_err = e
+                  if e.code >= 500 and attempt < 4:
+                      wait = 2 ** attempt
+                      print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                      time.sleep(wait)
+                      continue
+                  raise
+              except Exception as e:
+                  last_err = e
+                  if attempt < 4:
+                      wait = 2 ** attempt
+                      print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                      time.sleep(wait)
+                      continue
+                  raise
+          else:
+              raise last_err
+          with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
+              root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
+              for member in tar.getmembers():
+                  name = member.name
+                  if name == root_prefix[:-1]:
+                      continue
+                  if name.startswith(root_prefix):
+                      member.name = name[len(root_prefix):]
+                  if member.name:
+                      tar.extract(member, '.')
+          PY
+
       - name: Production health check & smoke test
         id: smoke
         shell: sh
@@ -82,7 +121,46 @@ jobs:
           GITHUB_TOKEN: ${{ github.token }}
         run: |
           set -eu
-          bash scripts/ci_checkout.sh
+          python3 - <<'PY'
+          import io, os, tarfile, time, urllib.request, urllib.error
+          url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
+          request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
+          last_err = None
+          for attempt in range(5):
+              try:
+                  with urllib.request.urlopen(request, timeout=120) as response:
+                      archive = response.read()
+                  break
+              except urllib.error.HTTPError as e:
+                  last_err = e
+                  if e.code >= 500 and attempt < 4:
+                      wait = 2 ** attempt
+                      print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                      time.sleep(wait)
+                      continue
+                  raise
+              except Exception as e:
+                  last_err = e
+                  if attempt < 4:
+                      wait = 2 ** attempt
+                      print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                      time.sleep(wait)
+                      continue
+                  raise
+          else:
+              raise last_err
+          with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
+              root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
+              for member in tar.getmembers():
+                  name = member.name
+                  if name == root_prefix[:-1]:
+                      continue
+                  if name.startswith(root_prefix):
+                      member.name = name[len(root_prefix):]
+                  if member.name:
+                      tar.extract(member, '.')
+          PY
+
       - name: Run API smoke test on staging
         id: smoke
         shell: sh
@@ -180,14 +258,45 @@ jobs:
           GITHUB_TOKEN: ${{ github.token }}
         run: |
           set -eu
-          bash scripts/ci_checkout.sh
-      - name: Cache npm dependencies
-        uses: actions/cache@v3
-        with:
-          path: ~/.npm
-          key: ${{ runner.os }}-npm-${{ hashFiles('apps/web/package-lock.json') }}
-          restore-keys: |
-            ${{ runner.os }}-npm-
+          python3 - <<'PY'
+          import io, os, tarfile, time, urllib.request, urllib.error
+          url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
+          request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
+          last_err = None
+          for attempt in range(5):
+              try:
+                  with urllib.request.urlopen(request, timeout=120) as response:
+                      archive = response.read()
+                  break
+              except urllib.error.HTTPError as e:
+                  last_err = e
+                  if e.code >= 500 and attempt < 4:
+                      wait = 2 ** attempt
+                      print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                      time.sleep(wait)
+                      continue
+                  raise
+              except Exception as e:
+                  last_err = e
+                  if attempt < 4:
+                      wait = 2 ** attempt
+                      print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                      time.sleep(wait)
+                      continue
+                  raise
+          else:
+              raise last_err
+          with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
+              root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
+              for member in tar.getmembers():
+                  name = member.name
+                  if name == root_prefix[:-1]:
+                      continue
+                  if name.startswith(root_prefix):
+                      member.name = name[len(root_prefix):]
+                  if member.name:
+                      tar.extract(member, '.')
+          PY
 
       - name: Run Playwright E2E on staging
         id: e2e
@@ -201,7 +310,6 @@ jobs:
             -e E2E_BROWSER_CHANNEL=chromium \
             -e PLAYWRIGHT_HEADLESS=1 \
             -v "$PWD:/workspace" \
-            -v "$HOME/.npm:/root/.npm" \
             -w /workspace/apps/web \
             git.xiaoxiajianji.com/xiaoxia/base/playwright:v1.45.0-jammy \
             sh -lc 'npm ci && npx playwright test --reporter=line --project=chromium e2e/auth.spec.ts e2e/auth-guard.spec.ts e2e/core-upload.spec.ts e2e/core-generation.spec.ts e2e/core-titles.spec.ts' 2>&1 | tee /tmp/staging-e2e.log

apps/api/app/api/routes/chunked_upload.py

@@ -493,3 +493,5 @@ async def upload_chunk(
         "uploaded_chunks": len(meta["uploaded_chunks"]),
         "total_chunks": meta["total_chunks"],
     }
+
+

scripts/ci_checkout.sh

@@ -1,97 +0,0 @@
-#!/bin/sh
-# CI Checkout script - 从 Gitea API 下载源码 tar 包并解压
-# 用法: ci_checkout.sh [repo_api_base] [ref] [target_dir] [token]
-#   repo_api_base: 仓库 API 基础 URL，如 https://git.xiaoxiajianji.com/api/v1/repos/xiaoxia/xiaoxia-saas
-#   ref:           commit SHA 或分支名
-#   target_dir:    目标目录（默认当前目录）
-#   token:         API token
-# 所有参数均可省略，将从 Gitea Actions 环境变量中读取
-
-set -eu
-
-# ── 参数解析 ──────────────────────────────────────────────────
-REPO_API_BASE="${1:-}"
-REF="${2:-}"
-TARGET_DIR="${3:-.}"
-TOKEN="${4:-}"
-
-# 从环境变量补全默认值（兼容 Gitea Actions）
-if [ -z "$REPO_API_BASE" ]; then
-  REPO_API_BASE="${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}"
-fi
-if [ -z "$REF" ]; then
-  REF="${GITHUB_SHA}"
-fi
-if [ -z "$TOKEN" ]; then
-  TOKEN="${GITHUB_TOKEN:-}"
-fi
-
-if [ -z "$REPO_API_BASE" ] || [ -z "$REF" ]; then
-  echo "ERROR: repo API base and ref are required" >&2
-  echo "Usage: $0 [repo_api_base] [ref] [target_dir] [token]" >&2
-  exit 1
-fi
-
-# ── 下载并解压 ────────────────────────────────────────────────
-ARCHIVE_URL="${REPO_API_BASE}/archive/${REF}.tar.gz"
-
-echo "Checkout: ${ARCHIVE_URL}"
-echo "Target dir: ${TARGET_DIR}"
-
-mkdir -p "${TARGET_DIR}"
-
-# 通过环境变量传递给 Python
-_CHECKOUT_URL="${ARCHIVE_URL}" \
-_CHECKOUT_TOKEN="${TOKEN}" \
-_CHECKOUT_TARGET_DIR="${TARGET_DIR}" \
-python3 - <<'PY'
-import io, os, tarfile, time, urllib.request, urllib.error
-
-url = os.environ['_CHECKOUT_URL']
-token = os.environ.get('_CHECKOUT_TOKEN', '')
-target_dir = os.environ['_CHECKOUT_TARGET_DIR']
-
-headers = {}
-if token:
-    headers["Authorization"] = f"token {token}"
-
-request = urllib.request.Request(url, headers=headers)
-
-last_err = None
-for attempt in range(5):
-    try:
-        with urllib.request.urlopen(request, timeout=120) as response:
-            archive = response.read()
-        break
-    except urllib.error.HTTPError as e:
-        last_err = e
-        if e.code >= 500 and attempt < 4:
-            wait = 2 ** attempt
-            print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
-            time.sleep(wait)
-            continue
-        raise
-    except Exception as e:
-        last_err = e
-        if attempt < 4:
-            wait = 2 ** attempt
-            print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
-            time.sleep(wait)
-            continue
-        raise
-else:
-    raise last_err
-
-with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
-    root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
-    for member in tar.getmembers():
-        name = member.name
-        if name == root_prefix[:-1]:
-            continue
-        if name.startswith(root_prefix):
-            member.name = name[len(root_prefix):]
-        if member.name:
-            tar.extract(member, target_dir)
-
-print("Checkout complete.")
-PY

tests/conftest.py

@@ -11,47 +11,3 @@ if str(ROOT) not in sys.path:
 # 必须在任何 app 模块导入之前设置，否则 pydantic Settings 验证失败
 os.environ.setdefault("JWT_SECRET_KEY", "test-secret-key-for-all-tests")
 os.environ.setdefault("USE_IN_MEMORY_DB", "True")
-
-
-# ── Celery 全局 mock ──────────────────────────────────────────────────────
-# CI 环境没有 Redis，所有 Celery 异步任务都 mock 掉，避免连接超时报错
-# 集成测试只测 API 层逻辑（参数校验、权限、DB 操作），异步任务由 worker 单测覆盖
-
-from unittest.mock import MagicMock, patch
-
-
-def _mock_celery_task():
-    """全局 mock Celery 任务的 delay/apply_async/send_task 方法。"""
-    from celery import Celery, Task
-
-    # 保存原始方法
-    _orig_delay = Task.delay
-    _orig_apply_async = Task.apply_async
-    _orig_send_task = Celery.send_task
-
-    def _mock_delay(self, *args, **kwargs):
-        mock_result = MagicMock()
-        mock_result.id = "mock-task-id"
-        mock_result.state = "PENDING"
-        mock_result.ready.return_value = False
-        mock_result.get.return_value = None
-        return mock_result
-
-    def _mock_apply_async(self, *args, **kwargs):
-        return _mock_delay(self, *args, **kwargs)
-
-    def _mock_send_task(self, name, *args, **kwargs):
-        mock_result = MagicMock()
-        mock_result.id = f"mock-{name}"
-        mock_result.state = "PENDING"
-        mock_result.ready.return_value = False
-        mock_result.get.return_value = None
-        return mock_result
-
-    Task.delay = _mock_delay
-    Task.apply_async = _mock_apply_async
-    Celery.send_task = _mock_send_task
-
-
-# 在任何 app 模块导入之前就 patch 掉
-_mock_celery_task()

tests/integration/conftest.py

@@ -16,8 +16,8 @@ import pytest
 
 # ── 性能阈值配置 ──────────────────────────────────────────────────────────
 PERF_THRESHOLDS: Dict[str, int] = {
-    "core": 500,  # 核心接口：500ms
-    "normal": 1000,  # 普通接口：1000ms
+    "core": 500,    # 核心接口：500ms
+    "normal": 1000, # 普通接口：1000ms
     "heavy": 3000,  # 重操作：3000ms（涉及外部调用或复杂计算）
 }
 
@@ -32,11 +32,9 @@ PERF_TOLERANCE_RATIO = float(os.environ.get("PERF_TOLERANCE_RATIO", "0.34"))
 
 # ── 数据类 ────────────────────────────────────────────────────────────────
 
-
 @dataclass
 class PerfResult:
     """单次性能测试结果"""
-
     name: str
     threshold_ms: int
     times_ms: List[float] = field(default_factory=list)
@@ -82,7 +80,6 @@ class PerfResult:
 
 # ── 性能断言上下文管理器 ──────────────────────────────────────────────────
 
-
 class PerfAssert:
     """
     性能断言工具。
@@ -110,7 +107,9 @@ class PerfAssert:
             samples: 采样次数，默认使用全局配置
         """
         if threshold_level not in PERF_THRESHOLDS:
-            raise ValueError(f"未知的阈值级别: {threshold_level}，可选: {list(PERF_THRESHOLDS.keys())}")
+            raise ValueError(
+                f"未知的阈值级别: {threshold_level}，可选: {list(PERF_THRESHOLDS.keys())}"
+            )
 
         threshold_ms = PERF_THRESHOLDS[threshold_level]
         num_samples = samples or self.sample_count
@@ -120,7 +119,8 @@ class PerfAssert:
         yield result
         # 第一次调用已经记录在 result.times_ms 中（由调用方通过 measure 方法）
 
-    def measure(self, threshold_level: str = "core", name: str = "", samples: Optional[int] = None) -> Callable:
+    def measure(self, threshold_level: str = "core", name: str = "",
+                samples: Optional[int] = None) -> Callable:
         """
         返回一个装饰器/包装器，用于测量函数执行时间。
 
@@ -129,10 +129,11 @@ class PerfAssert:
                 lambda: client.post("/api/v1/auth/login", json={...})
             )
         """
-
         def wrapper(func):
             if threshold_level not in PERF_THRESHOLDS:
-                raise ValueError(f"未知的阈值级别: {threshold_level}，可选: {list(PERF_THRESHOLDS.keys())}")
+                raise ValueError(
+                    f"未知的阈值级别: {threshold_level}，可选: {list(PERF_THRESHOLDS.keys())}"
+                )
             threshold_ms = PERF_THRESHOLDS[threshold_level]
             num_samples = samples or self.sample_count
             result = PerfResult(name=name or threshold_level, threshold_ms=threshold_ms)
@@ -164,14 +165,20 @@ class PerfAssert:
                     f"[min={r.min_ms:.1f}, max={r.max_ms:.1f}, "
                     f"mean={r.mean_ms:.1f}, samples={len(r.times_ms)}]"
                 )
-            raise AssertionError(f"性能测试失败 ({len(failed)}/{len(self.results)}):\n" + "\n".join(lines))
+            raise AssertionError(
+                f"性能测试失败 ({len(failed)}/{len(self.results)}):\n"
+                + "\n".join(lines)
+            )
 
     def report(self) -> str:
         """生成性能报告文本"""
         lines = ["=" * 60, " 性能测试报告", "=" * 60]
         for r in self.results:
             status = "✅" if r.passed else "❌"
-            lines.append(f" {status} {r.name:<40s} " f"median={r.median_ms:>7.1f}ms / {r.threshold_ms:>5d}ms")
+            lines.append(
+                f" {status} {r.name:<40s} "
+                f"median={r.median_ms:>7.1f}ms / {r.threshold_ms:>5d}ms"
+            )
             lines.append(
                 f"     min={r.min_ms:.1f}ms  max={r.max_ms:.1f}ms  "
                 f"mean={r.mean_ms:.1f}ms  samples={len(r.times_ms)}"
@@ -186,13 +193,24 @@ class PerfAssert:
 
 # ── pytest fixtures ──────────────────────────────────────────────────────
 
-
 def pytest_configure(config):
     """注册自定义 marker"""
-    config.addinivalue_line("markers", "performance: 标记为性能测试（可通过 -m 'not performance' 跳过）")
-    config.addinivalue_line("markers", "perf_core: 核心接口性能测试（阈值 500ms）")
-    config.addinivalue_line("markers", "perf_normal: 普通接口性能测试（阈值 1000ms）")
-    config.addinivalue_line("markers", "perf_heavy: 重操作接口性能测试（阈值 3000ms）")
+    config.addinivalue_line(
+        "markers",
+        "performance: 标记为性能测试（可通过 -m 'not performance' 跳过）"
+    )
+    config.addinivalue_line(
+        "markers",
+        "perf_core: 核心接口性能测试（阈值 500ms）"
+    )
+    config.addinivalue_line(
+        "markers",
+        "perf_normal: 普通接口性能测试（阈值 1000ms）"
+    )
+    config.addinivalue_line(
+        "markers",
+        "perf_heavy: 重操作接口性能测试（阈值 3000ms）"
+    )
 
 
 def pytest_collection_modifyitems(config, items):
@@ -236,7 +254,6 @@ def perf_thresholds():
 
 # ── 辅助函数 ──────────────────────────────────────────────────────────────
 
-
 def run_perf_test(
     name: str,
     threshold_level: str,
@@ -256,7 +273,9 @@ def run_perf_test(
         PerfResult 对象
     """
     if threshold_level not in PERF_THRESHOLDS:
-        raise ValueError(f"未知的阈值级别: {threshold_level}，可选: {list(PERF_THRESHOLDS.keys())}")
+        raise ValueError(
+            f"未知的阈值级别: {threshold_level}，可选: {list(PERF_THRESHOLDS.keys())}"
+        )
 
     threshold_ms = PERF_THRESHOLDS[threshold_level]
     result = PerfResult(name=name, threshold_ms=threshold_ms)

tests/integration/test_api_performance.py

@@ -158,7 +158,9 @@ class TestCoreApiPerformance:
             )
         )
 
-        assert result.status_code == 200, f"登录接口返回状态码 {result.status_code}，预期 200"
+        assert result.status_code == 200, (
+            f"登录接口返回状态码 {result.status_code}，预期 200"
+        )
         assert result.passed, (
             f"登录接口性能不达标: 中位数 {result.median_ms:.1f}ms > 阈值 {result.threshold_ms}ms\n"
             f"  详情: min={result.min_ms:.1f}ms, max={result.max_ms:.1f}ms, "
@@ -169,9 +171,13 @@ class TestCoreApiPerformance:
         """GET /auth/me 获取当前用户信息性能"""
         headers = perf_test_user["headers"]
 
-        result = perf_assert.measure("core", "GET /auth/me")(lambda: client.get("/api/v1/auth/me", headers=headers))
+        result = perf_assert.measure("core", "GET /auth/me")(
+            lambda: client.get("/api/v1/auth/me", headers=headers)
+        )
 
-        assert result.status_code == 200, f"获取当前用户返回状态码 {result.status_code}，预期 200"
+        assert result.status_code == 200, (
+            f"获取当前用户返回状态码 {result.status_code}，预期 200"
+        )
         assert result.passed, (
             f"获取当前用户性能不达标: 中位数 {result.median_ms:.1f}ms > 阈值 {result.threshold_ms}ms\n"
             f"  详情: min={result.min_ms:.1f}ms, max={result.max_ms:.1f}ms, "
@@ -182,9 +188,13 @@ class TestCoreApiPerformance:
         """GET /projects 项目列表性能"""
         headers = perf_test_user["headers"]
 
-        result = perf_assert.measure("core", "GET /projects")(lambda: client.get("/api/v1/projects", headers=headers))
+        result = perf_assert.measure("core", "GET /projects")(
+            lambda: client.get("/api/v1/projects", headers=headers)
+        )
 
-        assert result.status_code == 200, f"项目列表返回状态码 {result.status_code}，预期 200"
+        assert result.status_code == 200, (
+            f"项目列表返回状态码 {result.status_code}，预期 200"
+        )
         assert result.passed, (
             f"项目列表性能不达标: 中位数 {result.median_ms:.1f}ms > 阈值 {result.threshold_ms}ms\n"
             f"  详情: min={result.min_ms:.1f}ms, max={result.max_ms:.1f}ms, "
@@ -195,9 +205,13 @@ class TestCoreApiPerformance:
         """GET /assets 素材列表性能"""
         headers = perf_test_user["headers"]
 
-        result = perf_assert.measure("core", "GET /assets")(lambda: client.get("/api/v1/assets", headers=headers))
+        result = perf_assert.measure("core", "GET /assets")(
+            lambda: client.get("/api/v1/assets", headers=headers)
+        )
 
-        assert result.status_code == 200, f"素材列表返回状态码 {result.status_code}，预期 200"
+        assert result.status_code == 200, (
+            f"素材列表返回状态码 {result.status_code}，预期 200"
+        )
         assert result.passed, (
             f"素材列表性能不达标: 中位数 {result.median_ms:.1f}ms > 阈值 {result.threshold_ms}ms\n"
             f"  详情: min={result.min_ms:.1f}ms, max={result.max_ms:.1f}ms, "
@@ -212,7 +226,9 @@ class TestCoreApiPerformance:
             lambda: client.get("/api/v1/generation/tasks", headers=headers)
         )
 
-        assert result.status_code == 200, f"生成任务列表返回状态码 {result.status_code}，预期 200"
+        assert result.status_code == 200, (
+            f"生成任务列表返回状态码 {result.status_code}，预期 200"
+        )
         assert result.passed, (
             f"生成任务列表性能不达标: 中位数 {result.median_ms:.1f}ms > 阈值 {result.threshold_ms}ms\n"
             f"  详情: min={result.min_ms:.1f}ms, max={result.max_ms:.1f}ms, "
@@ -227,7 +243,9 @@ class TestCoreApiPerformance:
             lambda: client.get("/api/v1/subscription/current", headers=headers)
         )
 
-        assert result.status_code == 200, f"订阅信息返回状态码 {result.status_code}，预期 200"
+        assert result.status_code == 200, (
+            f"订阅信息返回状态码 {result.status_code}，预期 200"
+        )
         assert result.passed, (
             f"订阅信息性能不达标: 中位数 {result.median_ms:.1f}ms > 阈值 {result.threshold_ms}ms\n"
             f"  详情: min={result.min_ms:.1f}ms, max={result.max_ms:.1f}ms, "
@@ -266,7 +284,9 @@ class TestNormalApiPerformance:
 
         result = perf_assert.measure("normal", "POST /projects")(_create)
 
-        assert result.status_code in (200, 201), f"创建项目返回状态码 {result.status_code}，预期 200/201"
+        assert result.status_code in (200, 201), (
+            f"创建项目返回状态码 {result.status_code}，预期 200/201"
+        )
         assert result.passed, (
             f"创建项目性能不达标: 中位数 {result.median_ms:.1f}ms > 阈值 {result.threshold_ms}ms\n"
             f"  详情: min={result.min_ms:.1f}ms, max={result.max_ms:.1f}ms, "
@@ -282,7 +302,9 @@ class TestNormalApiPerformance:
         )
 
         # 模板列表可能返回 200 或空列表，只要不是错误即可
-        assert result.status_code == 200, f"模板列表返回状态码 {result.status_code}，预期 200"
+        assert result.status_code == 200, (
+            f"模板列表返回状态码 {result.status_code}，预期 200"
+        )
         assert result.passed, (
             f"模板列表性能不达标: 中位数 {result.median_ms:.1f}ms > 阈值 {result.threshold_ms}ms\n"
             f"  详情: min={result.min_ms:.1f}ms, max={result.max_ms:.1f}ms, "
@@ -297,7 +319,9 @@ class TestNormalApiPerformance:
             lambda: client.get("/api/v1/edit-plans", headers=headers)
         )
 
-        assert result.status_code == 200, f"剪辑计划列表返回状态码 {result.status_code}，预期 200"
+        assert result.status_code == 200, (
+            f"剪辑计划列表返回状态码 {result.status_code}，预期 200"
+        )
         assert result.passed, (
             f"剪辑计划列表性能不达标: 中位数 {result.median_ms:.1f}ms > 阈值 {result.threshold_ms}ms\n"
             f"  详情: min={result.min_ms:.1f}ms, max={result.max_ms:.1f}ms, "
@@ -322,12 +346,6 @@ class TestHeavyApiPerformance:
     """
 
     def test_upload_direct_prepare_performance(self, perf_test_user, perf_assert):
-        # OSS 未配置时跳过此测试
-        from app.config import settings
-
-        if not settings.OSS_ACCESS_KEY_ID or not settings.OSS_ACCESS_KEY_SECRET:
-            pytest.skip("OSS credentials not configured, skipping upload signature test")
-
         """POST /upload/direct/prepare 获取上传签名性能"""
         headers = perf_test_user["headers"]
         project_id = perf_test_user["project_id"]
@@ -353,16 +371,15 @@ class TestHeavyApiPerformance:
                 },
             )
 
-        result = perf_assert.measure("heavy", "POST /upload/direct/prepare")(_prepare_upload)
+        result = perf_assert.measure("heavy", "POST /upload/direct/prepare")(
+            _prepare_upload
+        )
 
         # 上传签名接口可能因为 OSS 配置问题返回 503，这是预期的
         # 只要不超时、不返回 500 即可
-        assert result.status_code in (
-            200,
-            201,
-            400,
-            503,
-        ), f"获取上传签名返回状态码 {result.status_code}，预期 200/201/400/503"
+        assert result.status_code in (200, 201, 400, 503), (
+            f"获取上传签名返回状态码 {result.status_code}，预期 200/201/400/503"
+        )
         assert result.passed, (
             f"获取上传签名性能不达标: 中位数 {result.median_ms:.1f}ms > 阈值 {result.threshold_ms}ms\n"
             f"  详情: min={result.min_ms:.1f}ms, max={result.max_ms:.1f}ms, "
@@ -397,16 +414,15 @@ class TestHeavyApiPerformance:
                 },
             )
 
-        result = perf_assert.measure("heavy", "POST /generation/tasks")(_create_task)
+        result = perf_assert.measure("heavy", "POST /generation/tasks")(
+            _create_task
+        )
 
         # 创建生成任务可能因为缺少素材等返回 400，这是预期的
         # 性能测试关注响应时间，不关注业务成功与否
-        assert result.status_code in (
-            200,
-            201,
-            400,
-            404,
-        ), f"创建生成任务返回状态码 {result.status_code}，预期 200/201/400/404"
+        assert result.status_code in (200, 201, 400, 404), (
+            f"创建生成任务返回状态码 {result.status_code}，预期 200/201/400/404"
+        )
         assert result.passed, (
             f"创建生成任务性能不达标: 中位数 {result.median_ms:.1f}ms > 阈值 {result.threshold_ms}ms\n"
             f"  详情: min={result.min_ms:.1f}ms, max={result.max_ms:.1f}ms, "
@@ -433,15 +449,14 @@ class TestHeavyApiPerformance:
                 },
             )
 
-        result = perf_assert.measure("heavy", "POST /duplication/upload")(_upload)
+        result = perf_assert.measure("heavy", "POST /duplication/upload")(
+            _upload
+        )
 
         # 去重上传可能因为 OSS 配置问题返回 503，这是预期的
-        assert result.status_code in (
-            200,
-            201,
-            400,
-            503,
-        ), f"去重上传返回状态码 {result.status_code}，预期 200/201/400/503"
+        assert result.status_code in (200, 201, 400, 503), (
+            f"去重上传返回状态码 {result.status_code}，预期 200/201/400/503"
+        )
         assert result.passed, (
             f"去重上传性能不达标: 中位数 {result.median_ms:.1f}ms > 阈值 {result.threshold_ms}ms\n"
             f"  详情: min={result.min_ms:.1f}ms, max={result.max_ms:.1f}ms, "
@@ -495,11 +510,17 @@ def test_performance_summary(perf_test_user, perf_assert, capsys):
         )
     )
 
-    perf_assert.measure("core", "GET /auth/me")(lambda: client.get("/api/v1/auth/me", headers=headers))
+    perf_assert.measure("core", "GET /auth/me")(
+        lambda: client.get("/api/v1/auth/me", headers=headers)
+    )
 
-    perf_assert.measure("core", "GET /projects")(lambda: client.get("/api/v1/projects", headers=headers))
+    perf_assert.measure("core", "GET /projects")(
+        lambda: client.get("/api/v1/projects", headers=headers)
+    )
 
-    perf_assert.measure("core", "GET /assets")(lambda: client.get("/api/v1/assets", headers=headers))
+    perf_assert.measure("core", "GET /assets")(
+        lambda: client.get("/api/v1/assets", headers=headers)
+    )
 
     perf_assert.measure("core", "GET /generation/tasks")(
         lambda: client.get("/api/v1/generation/tasks", headers=headers)
@@ -518,9 +539,13 @@ def test_performance_summary(perf_test_user, perf_assert, capsys):
         )
     )
 
-    perf_assert.measure("normal", "GET /templates")(lambda: client.get("/api/v1/templates", headers=headers))
+    perf_assert.measure("normal", "GET /templates")(
+        lambda: client.get("/api/v1/templates", headers=headers)
+    )
 
-    perf_assert.measure("normal", "GET /edit-plans")(lambda: client.get("/api/v1/edit-plans", headers=headers))
+    perf_assert.measure("normal", "GET /edit-plans")(
+        lambda: client.get("/api/v1/edit-plans", headers=headers)
+    )
 
     # ── 重操作接口 ──
     perf_assert.measure("heavy", "POST /upload/direct/prepare")(
@@ -580,20 +605,20 @@ def test_performance_summary(perf_test_user, perf_assert, capsys):
 
     # 输出统计信息，方便 CI 解析
     with capsys.disabled():
-        print(f"\nPERF_STATS: total={total_count}, passed={passed_count}, " f"failed={total_count - passed_count}")
+        print(f"\nPERF_STATS: total={total_count}, passed={passed_count}, "
+              f"failed={total_count - passed_count}")
         for r in perf_assert.results:
             status = "PASS" if r.passed else "FAIL"
-            print(
-                f"PERF_RESULT: {status} | {r.name} | "
-                f"median={r.median_ms:.1f}ms | threshold={r.threshold_ms}ms | "
-                f"min={r.min_ms:.1f}ms | max={r.max_ms:.1f}ms | "
-                f"mean={r.mean_ms:.1f}ms | status_code={r.status_code}"
-            )
+            print(f"PERF_RESULT: {status} | {r.name} | "
+                  f"median={r.median_ms:.1f}ms | threshold={r.threshold_ms}ms | "
+                  f"min={r.min_ms:.1f}ms | max={r.max_ms:.1f}ms | "
+                  f"mean={r.mean_ms:.1f}ms | status_code={r.status_code}")
 
     # 这里使用宽松断言：只要超过一半通过就不报错
     # 具体的 CI 阻塞策略由 CI 配置控制（continue-on-error）
     assert passed_count >= total_count // 2, (
-        f"性能测试通过率过低: {passed_count}/{total_count} " f"({passed_count/total_count*100:.0f}%)，至少需要 50% 通过"
+        f"性能测试通过率过低: {passed_count}/{total_count} "
+        f"({passed_count/total_count*100:.0f}%)，至少需要 50% 通过"
     )
 
 

tests/integration/test_assets_api.py

@@ -53,6 +53,7 @@ from packages.domain import (
     User,
 )
 
+
 # ---------------------------------------------------------------------------
 # 1. Stub Repository 实现
 # ---------------------------------------------------------------------------

tests/integration/test_chunked_upload_api.py

@@ -47,6 +47,7 @@ from app.dependencies import (
 
 from packages.domain import AssetLibrary, AssetLibraryKind, Project, User
 
+
 # ---------------------------------------------------------------------------
 # 1. Stub Repository 实现
 # ---------------------------------------------------------------------------
@@ -209,14 +210,12 @@ def client(project, library, mock_storage):
     # 临时修改 CHUNK_STORAGE_ROOT 到测试临时目录
     test_temp_dir = tempfile.mkdtemp(prefix="test_chunked_upload_")
     import app.api.routes.chunked_upload as chunk_mod
-
     chunk_mod.CHUNK_STORAGE_ROOT = Path(test_temp_dir)
 
     yield TestClient(test_app)
 
     # 清理
     import shutil
-
     chunk_mod.CHUNK_STORAGE_ROOT = CHUNK_STORAGE_ROOT
     if Path(test_temp_dir).exists():
         shutil.rmtree(test_temp_dir)

tests/integration/test_classification_api.py

@@ -1,301 +0,0 @@
-"""
-分类任务 API 集成测试。
-
-覆盖端点：
-- POST /classification-jobs — 提交分类任务
-- GET  /classification-jobs/{job_id} — 获取分类任务详情
-
-使用 FastAPI TestClient + dependency_overrides 模式，
-导入真实路由模块，mock Celery 和 repository。
-"""
-
-from __future__ import annotations
-
-import os
-import sys
-from unittest.mock import MagicMock
-
-# ── 环境变量 & sys.path（必须在导入 app.* 之前设置） ──────────────────────────
-os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
-os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
-
-import pytest
-from fastapi import FastAPI
-from fastapi.testclient import TestClient
-
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api"))
-
-# mock celery_app 以避免实际发送任务
-import app.api.routes.classification_jobs as classification_routes
-from app.api.routes.classification_jobs import router
-from app.dependencies import get_classification_job_repository
-
-from packages.adapters.in_memory import InMemoryClassificationJobRepository
-from packages.domain import ClassificationJob, ClassificationJobStatus
-
-classification_routes.celery_app = MagicMock()
-
-
-# ---------------------------------------------------------------------------
-# 1. 辅助函数
-# ---------------------------------------------------------------------------
-
-
-def _make_job(
-    project_id: str = "proj-1",
-    asset_id: str = "asset-1",
-    status: ClassificationJobStatus = ClassificationJobStatus.PENDING,
-) -> ClassificationJob:
-    job = ClassificationJob.create(project_id=project_id, asset_id=asset_id)
-    if status == ClassificationJobStatus.PROCESSING:
-        job.status = ClassificationJobStatus.PROCESSING
-    elif status == ClassificationJobStatus.COMPLETED:
-        job.status = ClassificationJobStatus.COMPLETED
-        job.classification = "scenic"
-        job.confidence = 0.92
-    elif status == ClassificationJobStatus.FAILED:
-        job.status = ClassificationJobStatus.FAILED
-        job.error_message = "AI 服务不可用"
-    return job
-
-
-# ---------------------------------------------------------------------------
-# 2. Fixtures
-# ---------------------------------------------------------------------------
-
-
-@pytest.fixture
-def repo():
-    return InMemoryClassificationJobRepository()
-
-
-@pytest.fixture
-def client(repo):
-    """创建带有依赖覆盖的 TestClient。"""
-    test_app = FastAPI()
-    test_app.include_router(router, prefix="/classification-jobs")
-
-    def _override_repo():
-        return repo
-
-    test_app.dependency_overrides[get_classification_job_repository] = _override_repo
-
-    yield TestClient(test_app)
-
-    test_app.dependency_overrides.clear()
-
-
-# ---------------------------------------------------------------------------
-# 3. POST / — 提交分类任务
-# ---------------------------------------------------------------------------
-
-
-class TestSubmitClassificationJob:
-    """提交分类任务端点测试。"""
-
-    def test_submit_with_valid_data(self, client):
-        """使用有效数据提交分类任务应成功。"""
-        resp = client.post(
-            "/classification-jobs",
-            json={
-                "project_id": "proj-123",
-                "asset_id": "asset-456",
-            },
-        )
-        assert resp.status_code == 200
-        data = resp.json()
-        assert data["project_id"] == "proj-123"
-        assert data["asset_id"] == "asset-456"
-        assert data["status"] == "pending"
-        assert data["classification"] == ""
-        assert data["confidence"] == 0.0
-        assert data["error_message"] == ""
-        assert "id" in data
-        assert len(data["id"]) > 0
-
-    def test_submit_generates_unique_id(self, client):
-        """每次提交应生成不同的任务 ID。"""
-        resp1 = client.post("/classification-jobs", json={"project_id": "p1", "asset_id": "a1"})
-        resp2 = client.post("/classification-jobs", json={"project_id": "p1", "asset_id": "a2"})
-        assert resp1.json()["id"] != resp2.json()["id"]
-
-    def test_submit_missing_project_id_returns_422(self, client):
-        """缺少 project_id 应返回 422。"""
-        resp = client.post("/classification-jobs", json={"asset_id": "asset-1"})
-        assert resp.status_code == 422
-
-    def test_submit_missing_asset_id_returns_422(self, client):
-        """缺少 asset_id 应返回 422。"""
-        resp = client.post("/classification-jobs", json={"project_id": "proj-1"})
-        assert resp.status_code == 422
-
-    def test_submit_empty_project_id_returns_422(self, client):
-        """空 project_id 应返回 422。"""
-        resp = client.post("/classification-jobs", json={"project_id": "", "asset_id": "asset-1"})
-        assert resp.status_code == 422
-
-    def test_submit_empty_asset_id_returns_422(self, client):
-        """空 asset_id 应返回 422。"""
-        resp = client.post("/classification-jobs", json={"project_id": "proj-1", "asset_id": ""})
-        assert resp.status_code == 422
-
-    def test_submit_sends_celery_task(self, client):
-        """提交任务后应触发 Celery 异步任务。"""
-        classification_routes.celery_app.send_task.reset_mock()
-
-        resp = client.post("/classification-jobs", json={"project_id": "p1", "asset_id": "a1"})
-        assert resp.status_code == 200
-
-        job_id = resp.json()["id"]
-        classification_routes.celery_app.send_task.assert_called_once_with(
-            "worker.classify_asset",
-            args=[job_id],
-        )
-
-    def test_submit_persists_to_repository(self, client, repo):
-        """提交后任务应保存到 repository。"""
-        resp = client.post("/classification-jobs", json={"project_id": "p1", "asset_id": "a1"})
-        job_id = resp.json()["id"]
-
-        saved = repo.get(job_id)
-        assert saved is not None
-        assert saved.project_id == "p1"
-        assert saved.asset_id == "a1"
-        assert saved.status == ClassificationJobStatus.PENDING
-
-
-# ---------------------------------------------------------------------------
-# 4. GET /{job_id} — 获取分类任务详情
-# ---------------------------------------------------------------------------
-
-
-class TestGetClassificationJob:
-    """获取分类任务详情端点测试。"""
-
-    def test_get_pending_job(self, client, repo):
-        """获取 pending 状态的任务。"""
-        job = _make_job(status=ClassificationJobStatus.PENDING)
-        repo.create(job)
-
-        resp = client.get(f"/classification-jobs/{job.id}")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert data["id"] == job.id
-        assert data["status"] == "pending"
-        assert data["classification"] == ""
-        assert data["confidence"] == 0.0
-
-    def test_get_processing_job(self, client, repo):
-        """获取 processing 状态的任务。"""
-        job = _make_job(status=ClassificationJobStatus.PROCESSING)
-        repo.create(job)
-
-        resp = client.get(f"/classification-jobs/{job.id}")
-        assert resp.status_code == 200
-        assert resp.json()["status"] == "processing"
-
-    def test_get_completed_job(self, client, repo):
-        """获取已完成的任务应包含分类结果和置信度。"""
-        job = _make_job(status=ClassificationJobStatus.COMPLETED)
-        repo.create(job)
-
-        resp = client.get(f"/classification-jobs/{job.id}")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert data["status"] == "completed"
-        assert data["classification"] == "scenic"
-        assert data["confidence"] == 0.92
-        assert data["error_message"] == ""
-
-    def test_get_failed_job(self, client, repo):
-        """获取失败的任务应包含错误信息。"""
-        job = _make_job(status=ClassificationJobStatus.FAILED)
-        repo.create(job)
-
-        resp = client.get(f"/classification-jobs/{job.id}")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert data["status"] == "failed"
-        assert "AI 服务不可用" in data["error_message"]
-
-    def test_get_nonexistent_job_returns_404(self, client):
-        """获取不存在的任务应返回 404。"""
-        resp = client.get("/classification-jobs/nonexistent-job-id")
-        assert resp.status_code == 404
-
-    def test_response_contains_all_required_fields(self, client, repo):
-        """响应应包含所有必需字段。"""
-        job = _make_job()
-        repo.create(job)
-
-        resp = client.get(f"/classification-jobs/{job.id}")
-        data = resp.json()
-        for field in ["id", "project_id", "asset_id", "status", "classification", "confidence", "error_message"]:
-            assert field in data, f"缺少字段: {field}"
-
-
-# ---------------------------------------------------------------------------
-# 5. 跨端点场景
-# ---------------------------------------------------------------------------
-
-
-class TestClassificationApiScenarios:
-    """分类任务 API 跨端点集成场景。"""
-
-    def test_submit_then_get_pending(self, client, repo):
-        """提交任务后立即查询应为 pending 状态。"""
-        submit_resp = client.post(
-            "/classification-jobs",
-            json={"project_id": "proj-scenario", "asset_id": "asset-scenario"},
-        )
-        assert submit_resp.status_code == 200
-        job_id = submit_resp.json()["id"]
-
-        get_resp = client.get(f"/classification-jobs/{job_id}")
-        assert get_resp.status_code == 200
-        assert get_resp.json()["status"] == "pending"
-        assert get_resp.json()["project_id"] == "proj-scenario"
-        assert get_resp.json()["asset_id"] == "asset-scenario"
-
-    def test_submit_simulate_complete_then_get(self, client, repo):
-        """模拟 worker 完成任务后查询应返回结果。"""
-        submit_resp = client.post("/classification-jobs", json={"project_id": "p1", "asset_id": "a1"})
-        job_id = submit_resp.json()["id"]
-
-        # 模拟 worker 处理完成
-        job = repo.get(job_id)
-        assert job is not None
-        job.status = ClassificationJobStatus.COMPLETED
-        job.classification = "product"
-        job.confidence = 0.88
-        repo.update(job)
-
-        # 查询结果
-        get_resp = client.get(f"/classification-jobs/{job_id}")
-        assert get_resp.status_code == 200
-        data = get_resp.json()
-        assert data["status"] == "completed"
-        assert data["classification"] == "product"
-        assert data["confidence"] == 0.88
-
-    def test_submit_simulate_failure_then_get(self, client, repo):
-        """模拟 worker 失败后查询应返回错误信息。"""
-        submit_resp = client.post("/classification-jobs", json={"project_id": "p1", "asset_id": "a1"})
-        job_id = submit_resp.json()["id"]
-
-        # 模拟处理失败
-        job = repo.get(job_id)
-        assert job is not None
-        job.status = ClassificationJobStatus.FAILED
-        job.error_message = "网络超时"
-        repo.update(job)
-
-        get_resp = client.get(f"/classification-jobs/{job_id}")
-        assert get_resp.status_code == 200
-        data = get_resp.json()
-        assert data["status"] == "failed"
-        assert "网络超时" in data["error_message"]
-
-
-if __name__ == "__main__":
-    pytest.main([__file__, "-v"])

tests/integration/test_dashboard_api.py

@@ -1,496 +0,0 @@
-"""
-仪表盘 API 集成测试。
-
-覆盖端点：
-- GET /dashboard/overview — 仪表盘概览
-
-验证返回数据结构、空数据场景、数据汇总正确性。
-"""
-
-from __future__ import annotations
-
-import os
-import sys
-from datetime import datetime, timezone
-
-# ── 环境变量 & sys.path（必须在导入 app.* 之前设置） ──────────────────────────
-os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
-os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
-
-import pytest
-from fastapi import FastAPI
-from fastapi.testclient import TestClient
-
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api"))
-
-from app.api.routes.dashboard import router
-from app.auth import AuthenticatedUser, get_current_user
-from app.dependencies import (
-    get_asset_repository,
-    get_generation_task_repository,
-    get_project_repository,
-    get_title_library_repository,
-    get_voice_library_repository,
-)
-
-from packages.domain.entities import Project, User
-from packages.domain.generation_task import GenerationTask, GenerationTaskStatus
-
-# ---------------------------------------------------------------------------
-# 1. 内存 Repository
-# ---------------------------------------------------------------------------
-
-
-class InMemoryProjectRepository:
-    def __init__(self):
-        self._projects: dict[str, Project] = {}
-
-    def save(self, project: Project) -> None:
-        self._projects[project.id] = project
-
-    def find_by_id(self, project_id: str):
-        return self._projects.get(project_id)
-
-    def find_by_owner_user_id(self, owner_user_id: str):
-        return [p for p in self._projects.values() if p.owner_user_id == owner_user_id]
-
-    def find_accessible_projects(self, user_id: str):
-        return [p for p in self._projects.values() if p.owner_user_id == user_id]
-
-    def count_by_owner(self, owner_user_id: str) -> int:
-        return len(self.find_by_owner_user_id(owner_user_id))
-
-    def delete(self, project_id: str) -> bool:
-        if project_id in self._projects:
-            del self._projects[project_id]
-            return True
-        return False
-
-
-class InMemoryAssetRepository:
-    def __init__(self):
-        self._assets = []
-
-    def add_asset(self, project_id: str, storage_size: int = 0):
-        self._assets.append({"project_id": project_id, "storage_size": storage_size})
-
-    def count_by_project_ids(self, project_ids: list[str]) -> int:
-        return sum(1 for a in self._assets if a["project_id"] in project_ids)
-
-    def sum_storage_by_project_ids(self, project_ids: list[str]) -> int:
-        return sum(a["storage_size"] for a in self._assets if a["project_id"] in project_ids)
-
-    # 其他方法占位
-    def create(self, asset):
-        return asset
-
-    def find_by_id(self, asset_id):
-        return None
-
-    def find_by_project(self, project_id, **kwargs):
-        return []
-
-    def find_by_library(self, library_id, **kwargs):
-        return []
-
-    def update(self, asset):
-        return asset
-
-    def delete(self, asset_id):
-        return False
-
-    def batch_delete(self, asset_ids):
-        return 0
-
-    def search_candidates(self, **kwargs):
-        return []
-
-    def find_by_tag_ids(self, tag_ids):
-        return []
-
-    def count_by_project(self, project_id):
-        return 0
-
-    def find_by_library_and_file_type(self, library_id, file_type):
-        return []
-
-    def find_by_library_and_file_hash(self, library_id, file_hash):
-        return None
-
-
-class InMemoryGenerationTaskRepository:
-    def __init__(self):
-        self._tasks = {}
-
-    def add_task(self, task: GenerationTask):
-        self._tasks[task.id] = task
-
-    def count_by_user(self, user_id: str) -> int:
-        return len([t for t in self._tasks.values() if t.created_by_user_id == user_id])
-
-    def list_recent_by_user(self, user_id: str, limit: int = 5) -> list:
-        user_tasks = [t for t in self._tasks.values() if t.created_by_user_id == user_id]
-        # 按 created_at 倒序
-        user_tasks.sort(key=lambda t: t.created_at, reverse=True)
-        return user_tasks[:limit]
-
-    # 其他方法占位
-    def create(self, task):
-        return task
-
-    def get(self, task_id):
-        return None
-
-    def list_by_project(self, project_id):
-        return []
-
-    def list_by_user(self, user_id):
-        return []
-
-    def list_by_source_edit_plan(self, plan_id):
-        return []
-
-    def update(self, task):
-        return task
-
-
-class InMemoryTitleLibraryRepository:
-    def __init__(self):
-        self._items = {}
-
-    def add_item(self, user_id: str):
-        from uuid import uuid4
-
-        item_id = uuid4().hex
-        self._items[item_id] = {"id": item_id, "user_id": user_id}
-        return item_id
-
-    def count_by_user(self, user_id: str, is_active: bool = True) -> int:
-        return len([i for i in self._items.values() if i["user_id"] == user_id])
-
-    # 其他方法占位
-    def list_by_user(self, user_id, **kwargs):
-        return []
-
-    def get(self, title_id, user_id):
-        return None
-
-    def create(self, item):
-        return item
-
-    def update(self, item):
-        return item
-
-    def delete(self, title_id, user_id):
-        return False
-
-
-class InMemoryVoiceLibraryRepository:
-    def __init__(self):
-        self._items = {}
-
-    def add_item(self, user_id: str):
-        from uuid import uuid4
-
-        item_id = uuid4().hex
-        self._items[item_id] = {"id": item_id, "user_id": user_id}
-        return item_id
-
-    def count_by_user(self, user_id: str) -> int:
-        return len([i for i in self._items.values() if i["user_id"] == user_id])
-
-    # 其他方法占位
-    def list_by_user(self, user_id, **kwargs):
-        return []
-
-    def get(self, voice_id, user_id):
-        return None
-
-    def create(self, item):
-        return item
-
-    def update(self, item):
-        return item
-
-    def delete(self, voice_id, user_id):
-        return False
-
-
-# ---------------------------------------------------------------------------
-# 2. 辅助函数
-# ---------------------------------------------------------------------------
-
-
-def _make_user(**overrides) -> User:
-    defaults = dict(
-        id="user-test-001",
-        email="test@example.com",
-        display_name="Test User",
-        username="testuser",
-        subscription_plan="free",
-        subscription_status="active",
-        max_projects=3,
-        max_storage_gb=10,
-        created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
-    )
-    defaults.update(overrides)
-    return User(**defaults)
-
-
-def _make_project(project_id: str, owner_user_id: str = "user-test-001") -> Project:
-    return Project(
-        id=project_id,
-        name=f"Project {project_id}",
-        owner_user_id=owner_user_id,
-        description="",
-        created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
-    )
-
-
-def _make_generation_task(
-    task_id: str,
-    user_id: str = "user-test-001",
-    status: GenerationTaskStatus = GenerationTaskStatus.COMPLETED,
-    created_at: datetime | None = None,
-) -> GenerationTask:
-    return GenerationTask(
-        id=task_id,
-        project_id="proj-1",
-        asset_library_id="lib-1",
-        created_by_user_id=user_id,
-        status=status,
-        error_message="",
-        created_at=created_at or datetime.now(timezone.utc),
-        started_at=datetime.now(timezone.utc) if status != GenerationTaskStatus.PENDING else None,
-        completed_at=datetime.now(timezone.utc) if status == GenerationTaskStatus.COMPLETED else None,
-    )
-
-
-# ---------------------------------------------------------------------------
-# 3. Fixtures
-# ---------------------------------------------------------------------------
-
-
-@pytest.fixture
-def project_repo():
-    repo = InMemoryProjectRepository()
-    repo.save(_make_project("proj-1", "user-test-001"))
-    repo.save(_make_project("proj-2", "user-test-001"))
-    repo.save(_make_project("proj-other", "other-user"))
-    return repo
-
-
-@pytest.fixture
-def asset_repo():
-    return InMemoryAssetRepository()
-
-
-@pytest.fixture
-def generation_task_repo():
-    return InMemoryGenerationTaskRepository()
-
-
-@pytest.fixture
-def title_library_repo():
-    return InMemoryTitleLibraryRepository()
-
-
-@pytest.fixture
-def voice_library_repo():
-    return InMemoryVoiceLibraryRepository()
-
-
-@pytest.fixture
-def client(project_repo, asset_repo, generation_task_repo, title_library_repo, voice_library_repo):
-    """创建带有依赖覆盖的 TestClient。"""
-    test_app = FastAPI()
-    test_app.include_router(router, prefix="/dashboard")
-
-    def _override_current_user():
-        return AuthenticatedUser(user=_make_user())
-
-    test_app.dependency_overrides[get_current_user] = _override_current_user
-    test_app.dependency_overrides[get_project_repository] = lambda: project_repo
-    test_app.dependency_overrides[get_asset_repository] = lambda: asset_repo
-    test_app.dependency_overrides[get_generation_task_repository] = lambda: generation_task_repo
-    test_app.dependency_overrides[get_title_library_repository] = lambda: title_library_repo
-    test_app.dependency_overrides[get_voice_library_repository] = lambda: voice_library_repo
-
-    yield TestClient(test_app)
-
-    test_app.dependency_overrides.clear()
-
-
-# ---------------------------------------------------------------------------
-# 4. GET /overview — 仪表盘概览
-# ---------------------------------------------------------------------------
-
-
-class TestDashboardOverview:
-    """仪表盘概览端点测试。"""
-
-    def test_empty_data_returns_zeros(self, client):
-        """空数据时所有计数为 0。"""
-        resp = client.get("/dashboard/overview")
-        assert resp.status_code == 200
-        data = resp.json()
-
-        assert data["total_assets"] == 0
-        assert data["used_storage_bytes"] == 0
-        assert data["total_titles"] == 0
-        assert data["total_voices"] == 0
-        assert data["total_tasks"] == 0
-        assert data["total_products"] == 2  # fixture 中有 2 个项目
-        assert data["recent_tasks"] == []
-
-    def test_assets_count_and_storage(self, client, asset_repo):
-        """素材统计正确。"""
-        asset_repo.add_asset("proj-1", 1024)
-        asset_repo.add_asset("proj-1", 2048)
-        asset_repo.add_asset("proj-2", 4096)
-        # 其他用户的不计入
-        asset_repo.add_asset("proj-other", 9999)
-
-        resp = client.get("/dashboard/overview")
-        data = resp.json()
-
-        assert data["total_assets"] == 3
-        assert data["used_storage_bytes"] == 1024 + 2048 + 4096
-
-    def test_title_library_count(self, client, title_library_repo):
-        """标题库统计正确。"""
-        title_library_repo.add_item("user-test-001")
-        title_library_repo.add_item("user-test-001")
-        title_library_repo.add_item("user-test-001")
-        title_library_repo.add_item("other-user")
-
-        resp = client.get("/dashboard/overview")
-        data = resp.json()
-
-        assert data["total_titles"] == 3
-
-    def test_voice_library_count(self, client, voice_library_repo):
-        """配音库统计正确。"""
-        voice_library_repo.add_item("user-test-001")
-        voice_library_repo.add_item("other-user")
-
-        resp = client.get("/dashboard/overview")
-        data = resp.json()
-
-        assert data["total_voices"] == 1
-
-    def test_generation_tasks_count(self, client, generation_task_repo):
-        """生成任务统计正确。"""
-        generation_task_repo.add_task(_make_generation_task("task-1"))
-        generation_task_repo.add_task(_make_generation_task("task-2"))
-        generation_task_repo.add_task(_make_generation_task("task-other", user_id="other-user"))
-
-        resp = client.get("/dashboard/overview")
-        data = resp.json()
-
-        assert data["total_tasks"] == 2
-
-    def test_recent_tasks_limited_to_5(self, client, generation_task_repo):
-        """最近任务最多返回 5 个。"""
-        for i in range(10):
-            task = _make_generation_task(f"task-{i}")
-            generation_task_repo.add_task(task)
-
-        resp = client.get("/dashboard/overview")
-        data = resp.json()
-
-        assert len(data["recent_tasks"]) <= 5
-
-    def test_recent_tasks_have_correct_fields(self, client, generation_task_repo):
-        """最近任务包含正确字段。"""
-        task = _make_generation_task("task-1", status=GenerationTaskStatus.COMPLETED)
-        generation_task_repo.add_task(task)
-
-        resp = client.get("/dashboard/overview")
-        data = resp.json()
-
-        assert len(data["recent_tasks"]) == 1
-        item = data["recent_tasks"][0]
-        for field in ["id", "task_type", "status", "current_step", "error_message", "updated_at"]:
-            assert field in item, f"缺少字段: {field}"
-        assert item["task_type"] == "generation"
-
-    def test_subscription_info(self, client):
-        """订阅信息正确。"""
-        resp = client.get("/dashboard/overview")
-        data = resp.json()
-
-        assert "subscription" in data
-        sub = data["subscription"]
-        assert "plan" in sub
-        assert "is_active" in sub
-        assert sub["plan"] == "free"
-        assert sub["is_active"] is True
-
-    def test_pro_user_subscription(
-        self, project_repo, asset_repo, generation_task_repo, title_library_repo, voice_library_repo
-    ):
-        """Pro 用户订阅信息正确。"""
-        test_app = FastAPI()
-        test_app.include_router(router, prefix="/dashboard")
-
-        test_app.dependency_overrides[get_current_user] = lambda: AuthenticatedUser(
-            user=_make_user(subscription_plan="pro", subscription_status="active")
-        )
-        test_app.dependency_overrides[get_project_repository] = lambda: project_repo
-        test_app.dependency_overrides[get_asset_repository] = lambda: asset_repo
-        test_app.dependency_overrides[get_generation_task_repository] = lambda: generation_task_repo
-        test_app.dependency_overrides[get_title_library_repository] = lambda: title_library_repo
-        test_app.dependency_overrides[get_voice_library_repository] = lambda: voice_library_repo
-
-        c = TestClient(test_app)
-        resp = c.get("/dashboard/overview")
-        assert resp.status_code == 200
-        assert resp.json()["subscription"]["plan"] == "pro"
-        assert resp.json()["subscription"]["is_active"] is True
-        test_app.dependency_overrides.clear()
-
-    def test_total_products_count(self, client, project_repo):
-        """项目（产品）数量正确。"""
-        resp = client.get("/dashboard/overview")
-        data = resp.json()
-        assert data["total_products"] == 2
-
-        # 新增一个项目后
-        project_repo.save(_make_project("proj-3", "user-test-001"))
-        resp2 = client.get("/dashboard/overview")
-        assert resp2.json()["total_products"] == 3
-
-    def test_unauthorized_returns_401(
-        self, project_repo, asset_repo, generation_task_repo, title_library_repo, voice_library_repo
-    ):
-        """未授权访问返回 401/403。"""
-        test_app = FastAPI()
-        test_app.include_router(router, prefix="/dashboard")
-
-        test_app.dependency_overrides[get_project_repository] = lambda: project_repo
-        test_app.dependency_overrides[get_asset_repository] = lambda: asset_repo
-        test_app.dependency_overrides[get_generation_task_repository] = lambda: generation_task_repo
-        test_app.dependency_overrides[get_title_library_repository] = lambda: title_library_repo
-        test_app.dependency_overrides[get_voice_library_repository] = lambda: voice_library_repo
-
-        c = TestClient(test_app)
-        resp = c.get("/dashboard/overview")
-        assert resp.status_code in (401, 403)
-        test_app.dependency_overrides.clear()
-
-    def test_recent_tasks_status_mapping(self, client, generation_task_repo):
-        """不同状态的任务显示正确的当前步骤。"""
-        # 已完成任务
-        completed_task = _make_generation_task("task-completed", status=GenerationTaskStatus.COMPLETED)
-        generation_task_repo.add_task(completed_task)
-
-        resp = client.get("/dashboard/overview")
-        tasks = resp.json()["recent_tasks"]
-        completed = [t for t in tasks if t["id"] == "task-completed"][0]
-        assert completed["status"] == "completed"
-        assert "完成" in completed["current_step"] or "completed" in completed["current_step"].lower()
-
-
-if __name__ == "__main__":
-    pytest.main([__file__, "-v"])

tests/integration/test_generated_videos_api.py

@@ -1,554 +0,0 @@
-"""
-生成视频管理 API 集成测试。
-
-覆盖端点：
-- GET   /generated-videos — 列出生成视频
-- GET   /generated-videos/{video_id} — 获取生成视频详情
-- PATCH /generated-videos/{video_id}/review — 更新审核状态
-- GET   /generated-videos/{video_id}/download-url — 获取下载地址
-
-使用 FastAPI TestClient + dependency_overrides 模式，
-导入真实路由模块，mock 所有外部依赖。
-"""
-
-from __future__ import annotations
-
-import os
-import sys
-from dataclasses import replace
-from datetime import datetime, timezone
-from unittest.mock import MagicMock
-
-# ── 环境变量 & sys.path（必须在导入 app.* 之前设置） ──────────────────────────
-os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
-os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
-
-import pytest
-from fastapi import FastAPI
-from fastapi.testclient import TestClient
-
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api"))
-
-from app.api.routes.generated_videos import router
-from app.auth import AuthenticatedUser, get_current_user
-from app.core.storage import get_storage_service
-from app.dependencies import get_generated_video_repository, get_project_repository
-
-from packages.domain.entities import Project, User
-from packages.domain.generated_video import GeneratedVideo
-
-# ---------------------------------------------------------------------------
-# 1. 内存 Repository + 辅助函数
-# ---------------------------------------------------------------------------
-
-
-class InMemoryGeneratedVideoRepository:
-    """内存中的生成视频 Repository。"""
-
-    def __init__(self):
-        self._items: dict[str, GeneratedVideo] = {}
-
-    def create(self, video: GeneratedVideo) -> GeneratedVideo:
-        self._items[video.id] = video
-        return video
-
-    def get(self, video_id: str) -> GeneratedVideo | None:
-        return self._items.get(video_id)
-
-    def update(self, video: GeneratedVideo) -> GeneratedVideo:
-        self._items[video.id] = video
-        return video
-
-    def list_by_project(self, project_id: str) -> list[GeneratedVideo]:
-        return [v for v in self._items.values() if v.project_id == project_id]
-
-    def list_by_generation_task(self, generation_task_id: str) -> list[GeneratedVideo]:
-        return [v for v in self._items.values() if v.generation_task_id == generation_task_id]
-
-    def list_by_batch(self, batch_id: str) -> list[GeneratedVideo]:
-        return []
-
-
-class InMemoryProjectRepository:
-    """内存中的项目 Repository。"""
-
-    def __init__(self):
-        self._projects: dict[str, Project] = {}
-
-    def save(self, project: Project) -> None:
-        self._projects[project.id] = project
-
-    def find_by_id(self, project_id: str) -> Project | None:
-        return self._projects.get(project_id)
-
-    def find_by_owner_user_id(self, owner_user_id: str) -> list[Project]:
-        return [p for p in self._projects.values() if p.owner_user_id == owner_user_id]
-
-    def find_accessible_projects(self, user_id: str) -> list[Project]:
-        return [p for p in self._projects.values() if p.owner_user_id == user_id]
-
-    def count_by_owner(self, owner_user_id: str) -> int:
-        return len(self.find_by_owner_user_id(owner_user_id))
-
-    def delete(self, project_id: str) -> bool:
-        if project_id in self._projects:
-            del self._projects[project_id]
-            return True
-        return False
-
-
-class MockStorageService:
-    """Mock OSS 存储服务。"""
-
-    def get_download_url(self, file_url: str) -> str:
-        return f"https://cdn.example.com/download/{file_url}?token=abc123"
-
-
-def _make_user(**overrides) -> User:
-    defaults = dict(
-        id="user-test-001",
-        email="test@example.com",
-        display_name="Test User",
-        username="testuser",
-        subscription_plan="free",
-        subscription_status="active",
-        max_projects=3,
-        max_storage_gb=10,
-        created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
-    )
-    defaults.update(overrides)
-    return User(**defaults)
-
-
-def _make_project(project_id: str = "proj-1", owner_user_id: str = "user-test-001") -> Project:
-    return Project(
-        id=project_id,
-        name=f"Project {project_id}",
-        owner_user_id=owner_user_id,
-        description="",
-        created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
-    )
-
-
-def _make_video(
-    project_id: str = "proj-1",
-    name: str = "output.mp4",
-    status: str = "completed",
-    review_status: str = "pending_review",
-    **kwargs,
-) -> GeneratedVideo:
-    return GeneratedVideo.create(
-        project_id=project_id,
-        generation_task_id=kwargs.pop("generation_task_id", "task-1"),
-        name=name,
-        file_url=kwargs.pop("file_url", f"generated/{name}"),
-        file_size=kwargs.pop("file_size", 1024000),
-        duration=kwargs.pop("duration", 30.5),
-        width=kwargs.pop("width", 1920),
-        height=kwargs.pop("height", 1080),
-        fps=kwargs.pop("fps", 30.0),
-        thumbnail_url=kwargs.pop("thumbnail_url", None),
-        generation_params=kwargs.pop("generation_params", {"resolution": "1080p"}),
-    )
-
-
-# ---------------------------------------------------------------------------
-# 2. Fixtures
-# ---------------------------------------------------------------------------
-
-
-@pytest.fixture
-def video_repo():
-    return InMemoryGeneratedVideoRepository()
-
-
-@pytest.fixture
-def project_repo():
-    repo = InMemoryProjectRepository()
-    # 默认创建一个项目
-    repo.save(_make_project("proj-1", "user-test-001"))
-    repo.save(_make_project("proj-2", "user-test-001"))
-    repo.save(_make_project("proj-other", "other-user"))
-    return repo
-
-
-@pytest.fixture
-def storage_service():
-    return MockStorageService()
-
-
-@pytest.fixture
-def client(video_repo, project_repo, storage_service):
-    """创建带有依赖覆盖的 TestClient。"""
-    test_app = FastAPI()
-    test_app.include_router(router, prefix="/generated-videos")
-
-    def _override_current_user():
-        return AuthenticatedUser(user=_make_user())
-
-    def _override_video_repo():
-        return video_repo
-
-    def _override_project_repo():
-        return project_repo
-
-    def _override_storage():
-        return storage_service
-
-    test_app.dependency_overrides[get_current_user] = _override_current_user
-    test_app.dependency_overrides[get_generated_video_repository] = _override_video_repo
-    test_app.dependency_overrides[get_project_repository] = _override_project_repo
-    test_app.dependency_overrides[get_storage_service] = _override_storage
-
-    yield TestClient(test_app)
-
-    test_app.dependency_overrides.clear()
-
-
-# ---------------------------------------------------------------------------
-# 3. GET / — 列出生成视频
-# ---------------------------------------------------------------------------
-
-
-class TestListGeneratedVideos:
-    """列出生成视频端点测试。"""
-
-    def test_empty_list(self, client):
-        """无视频时返回空列表。"""
-        resp = client.get("/generated-videos")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert data["items"] == []
-
-    def test_list_all_user_videos(self, client, video_repo, project_repo):
-        """列出当前用户所有项目的视频。"""
-        v1 = _make_video(project_id="proj-1", name="video1.mp4")
-        v2 = _make_video(project_id="proj-2", name="video2.mp4")
-        v3 = _make_video(project_id="proj-other", name="other.mp4")  # 其他用户
-        video_repo.create(v1)
-        video_repo.create(v2)
-        video_repo.create(v3)
-
-        resp = client.get("/generated-videos")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert len(data["items"]) == 2
-        names = {item["name"] for item in data["items"]}
-        assert names == {"video1.mp4", "video2.mp4"}
-
-    def test_filter_by_project_id(self, client, video_repo):
-        """按 project_id 筛选视频。"""
-        v1 = _make_video(project_id="proj-1", name="a.mp4")
-        v2 = _make_video(project_id="proj-2", name="b.mp4")
-        video_repo.create(v1)
-        video_repo.create(v2)
-
-        resp = client.get("/generated-videos?project_id=proj-1")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert len(data["items"]) == 1
-        assert data["items"][0]["name"] == "a.mp4"
-
-    def test_filter_by_nonexistent_project_returns_404(self, client):
-        """筛选不存在的项目返回 404。"""
-        resp = client.get("/generated-videos?project_id=nonexistent")
-        assert resp.status_code == 404
-
-    def test_list_includes_download_url(self, client, video_repo):
-        """列表响应应包含下载地址。"""
-        v = _make_video(file_url="generated/test.mp4")
-        video_repo.create(v)
-
-        resp = client.get("/generated-videos")
-        assert resp.status_code == 200
-        item = resp.json()["items"][0]
-        assert "download_url" in item
-        assert item["download_url"] is not None
-        assert "cdn.example.com" in item["download_url"]
-
-    def test_list_response_fields(self, client, video_repo):
-        """列表响应包含所有必需字段。"""
-        v = _make_video()
-        video_repo.create(v)
-
-        resp = client.get("/generated-videos")
-        item = resp.json()["items"][0]
-        for field in [
-            "id",
-            "project_id",
-            "generation_task_id",
-            "name",
-            "file_url",
-            "file_size",
-            "duration",
-            "width",
-            "height",
-            "fps",
-            "status",
-            "review_status",
-            "generation_params",
-            "download_url",
-        ]:
-            assert field in item, f"缺少字段: {field}"
-
-    def test_unauthorized_returns_401(self, video_repo, project_repo, storage_service):
-        """未授权访问返回 401/403。"""
-        test_app = FastAPI()
-        test_app.include_router(router, prefix="/generated-videos")
-
-        # 不覆盖 get_current_user，使用默认（会拒绝无 token 请求）
-        test_app.dependency_overrides[get_generated_video_repository] = lambda: video_repo
-        test_app.dependency_overrides[get_project_repository] = lambda: project_repo
-        test_app.dependency_overrides[get_storage_service] = lambda: storage_service
-
-        c = TestClient(test_app)
-        resp = c.get("/generated-videos")
-        # 无 token 时 fastapi HTTPBearer auto_error=False 会返回 None，
-        # get_current_user 会抛 401
-        assert resp.status_code in (401, 403)
-        test_app.dependency_overrides.clear()
-
-
-# ---------------------------------------------------------------------------
-# 4. GET /{video_id} — 获取生成视频详情
-# ---------------------------------------------------------------------------
-
-
-class TestGetGeneratedVideo:
-    """获取生成视频详情端点测试。"""
-
-    def test_get_existing_video(self, client, video_repo):
-        """获取存在的视频返回详情。"""
-        v = _make_video(name="detail.mp4", duration=45.0)
-        video_repo.create(v)
-
-        resp = client.get(f"/generated-videos/{v.id}")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert data["id"] == v.id
-        assert data["name"] == "detail.mp4"
-        assert data["duration"] == 45.0
-        assert data["status"] == "completed"
-
-    def test_get_includes_download_url(self, client, video_repo):
-        """详情响应包含下载地址。"""
-        v = _make_video(file_url="generated/detail.mp4")
-        video_repo.create(v)
-
-        resp = client.get(f"/generated-videos/{v.id}")
-        data = resp.json()
-        assert "download_url" in data
-        assert "cdn.example.com" in data["download_url"]
-
-    def test_get_nonexistent_returns_404(self, client):
-        """获取不存在的视频返回 404。"""
-        resp = client.get("/generated-videos/nonexistent-video-id")
-        assert resp.status_code == 404
-        assert "not found" in resp.json()["detail"].lower()
-
-    def test_get_thumbnail_url(self, client, video_repo):
-        """有缩略图时返回缩略图 URL。"""
-        v = _make_video(thumbnail_url="thumbs/test.jpg")
-        video_repo.create(v)
-
-        resp = client.get(f"/generated-videos/{v.id}")
-        data = resp.json()
-        assert data["thumbnail_url"] == "thumbs/test.jpg"
-
-    def test_get_generation_params(self, client, video_repo):
-        """返回生成参数。"""
-        params = {"resolution": "4k", "style": "cinematic"}
-        v = _make_video(generation_params=params)
-        video_repo.create(v)
-
-        resp = client.get(f"/generated-videos/{v.id}")
-        data = resp.json()
-        assert data["generation_params"]["resolution"] == "4k"
-        assert data["generation_params"]["style"] == "cinematic"
-
-
-# ---------------------------------------------------------------------------
-# 5. PATCH /{video_id}/review — 更新审核状态
-# ---------------------------------------------------------------------------
-
-
-class TestUpdateReviewStatus:
-    """更新审核状态端点测试。"""
-
-    def test_approve_video(self, client, video_repo):
-        """审核通过。"""
-        v = _make_video(review_status="pending_review")
-        video_repo.create(v)
-
-        resp = client.patch(
-            f"/generated-videos/{v.id}/review",
-            json={"review_status": "approved"},
-        )
-        assert resp.status_code == 200
-        data = resp.json()
-        assert data["review_status"] == "approved"
-
-        # 验证 repository 已更新
-        updated = video_repo.get(v.id)
-        assert updated.review_status == "approved"
-
-    def test_reject_video(self, client, video_repo):
-        """审核拒绝。"""
-        v = _make_video(review_status="pending_review")
-        video_repo.create(v)
-
-        resp = client.patch(
-            f"/generated-videos/{v.id}/review",
-            json={"review_status": "rejected"},
-        )
-        assert resp.status_code == 200
-        assert resp.json()["review_status"] == "rejected"
-
-    def test_set_pending_review(self, client, video_repo):
-        """设置为待审核。"""
-        v = _make_video(review_status="approved")
-        video_repo.create(v)
-
-        resp = client.patch(
-            f"/generated-videos/{v.id}/review",
-            json={"review_status": "pending_review"},
-        )
-        assert resp.status_code == 200
-        assert resp.json()["review_status"] == "pending_review"
-
-    def test_nonexistent_video_returns_404(self, client):
-        """更新不存在的视频返回 404。"""
-        resp = client.patch(
-            "/nonexistent-id/review",
-            json={"review_status": "approved"},
-        )
-        assert resp.status_code == 404
-
-    def test_invalid_status_returns_422(self, client, video_repo):
-        """无效审核状态返回 422。"""
-        v = _make_video()
-        video_repo.create(v)
-
-        resp = client.patch(
-            f"/generated-videos/{v.id}/review",
-            json={"review_status": "invalid_status"},
-        )
-        assert resp.status_code == 422
-
-    def test_missing_status_returns_422(self, client, video_repo):
-        """缺少 review_status 字段返回 422。"""
-        v = _make_video()
-        video_repo.create(v)
-
-        resp = client.patch(f"/generated-videos/{v.id}/review", json={})
-        assert resp.status_code == 422
-
-    def test_update_returns_updated_fields(self, client, video_repo):
-        """更新后返回完整的视频信息。"""
-        v = _make_video(name="review_test.mp4")
-        video_repo.create(v)
-
-        resp = client.patch(
-            f"/generated-videos/{v.id}/review",
-            json={"review_status": "approved"},
-        )
-        data = resp.json()
-        assert data["name"] == "review_test.mp4"
-        assert "id" in data
-        assert "download_url" in data
-
-
-# ---------------------------------------------------------------------------
-# 6. GET /{video_id}/download-url — 获取下载地址
-# ---------------------------------------------------------------------------
-
-
-class TestGetDownloadUrl:
-    """获取下载地址端点测试。"""
-
-    def test_get_download_url_success(self, client, video_repo):
-        """获取下载地址成功。"""
-        v = _make_video(file_url="generated/video.mp4")
-        video_repo.create(v)
-
-        resp = client.get(f"/generated-videos/{v.id}/download-url")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert data["video_id"] == v.id
-        assert "download_url" in data
-        assert "cdn.example.com" in data["download_url"]
-
-    def test_nonexistent_video_returns_404(self, client):
-        """获取不存在视频的下载地址返回 404。"""
-        resp = client.get("/generated-videos/nonexistent-id/download-url")
-        assert resp.status_code == 404
-
-    def test_download_url_format(self, client, video_repo):
-        """下载地址格式正确。"""
-        v = _make_video(file_url="my-video.mp4")
-        video_repo.create(v)
-
-        resp = client.get(f"/generated-videos/{v.id}/download-url")
-        url = resp.json()["download_url"]
-        assert url.startswith("https://")
-        assert "token=" in url
-
-
-# ---------------------------------------------------------------------------
-# 7. 跨端点场景
-# ---------------------------------------------------------------------------
-
-
-class TestCrossEndpointScenarios:
-    """跨端点集成场景。"""
-
-    def test_create_list_detail_review_flow(self, client, video_repo):
-        """列表 → 详情 → 审核 完整流程。"""
-        # 准备数据
-        v = _make_video(name="flow.mp4", review_status="pending_review")
-        video_repo.create(v)
-
-        # 1. 列表
-        list_resp = client.get("/generated-videos")
-        assert list_resp.status_code == 200
-        assert len(list_resp.json()["items"]) == 1
-
-        # 2. 详情
-        detail_resp = client.get(f"/generated-videos/{v.id}")
-        assert detail_resp.status_code == 200
-        assert detail_resp.json()["name"] == "flow.mp4"
-        assert detail_resp.json()["review_status"] == "pending_review"
-
-        # 3. 审核通过
-        review_resp = client.patch(
-            f"/generated-videos/{v.id}/review",
-            json={"review_status": "approved"},
-        )
-        assert review_resp.status_code == 200
-        assert review_resp.json()["review_status"] == "approved"
-
-        # 4. 再次查看详情确认
-        detail_resp2 = client.get(f"/generated-videos/{v.id}")
-        assert detail_resp2.json()["review_status"] == "approved"
-
-        # 5. 获取下载地址
-        dl_resp = client.get(f"/generated-videos/{v.id}/download-url")
-        assert dl_resp.status_code == 200
-        assert dl_resp.json()["video_id"] == v.id
-
-    def test_multiple_videos_pagination_simulation(self, client, video_repo):
-        """多个视频时列表正确返回所有视频。"""
-        for i in range(5):
-            v = _make_video(project_id="proj-1", name=f"video_{i}.mp4")
-            video_repo.create(v)
-
-        resp = client.get("/generated-videos")
-        assert resp.status_code == 200
-        items = resp.json()["items"]
-        assert len(items) == 5
-        names = {item["name"] for item in items}
-        assert len(names) == 5  # 全部不同
-
-
-if __name__ == "__main__":
-    pytest.main([__file__, "-v"])

tests/integration/test_generation_api.py

@@ -53,6 +53,7 @@ from packages.domain import (
     User,
 )
 
+
 # ---------------------------------------------------------------------------
 # 1. Stub Repository 实现
 # ---------------------------------------------------------------------------

tests/integration/test_ingest_api.py

@@ -1,369 +0,0 @@
-"""
-摄入任务 API 集成测试。
-
-覆盖端点：
-- POST /ingest-jobs — 提交摄入任务
-- GET  /ingest-jobs/{job_id} — 获取摄入任务详情
-
-使用 FastAPI TestClient + dependency_overrides 模式，
-导入真实路由模块，mock Celery 和 repository。
-"""
-
-from __future__ import annotations
-
-import os
-import sys
-from unittest.mock import MagicMock
-
-# ── 环境变量 & sys.path（必须在导入 app.* 之前设置） ──────────────────────────
-os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
-os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
-
-import pytest
-from fastapi import FastAPI
-from fastapi.testclient import TestClient
-
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api"))
-
-# mock celery_app 以避免实际发送任务
-import app.api.routes.ingest_jobs as ingest_routes
-from app.api.routes.ingest_jobs import router
-from app.dependencies import get_ingest_job_repository
-
-from packages.adapters.in_memory import InMemoryIngestJobRepository
-from packages.domain import IngestJob, IngestJobStatus
-
-ingest_routes.celery_app = MagicMock()
-
-
-# ---------------------------------------------------------------------------
-# 1. 辅助函数
-# ---------------------------------------------------------------------------
-
-
-def _make_job(
-    project_id: str = "proj-1",
-    library_id: str = "lib-1",
-    storage_key: str = "uploads/test.mp4",
-    status: IngestJobStatus = IngestJobStatus.PENDING,
-) -> IngestJob:
-    job = IngestJob.create(
-        project_id=project_id,
-        library_id=library_id,
-        storage_key=storage_key,
-    )
-    if status == IngestJobStatus.PROCESSING:
-        job.status = IngestJobStatus.PROCESSING
-    elif status == IngestJobStatus.COMPLETED:
-        job.status = IngestJobStatus.COMPLETED
-        job.result_asset_id = "asset-completed-001"
-    elif status == IngestJobStatus.FAILED:
-        job.status = IngestJobStatus.FAILED
-        job.error_message = "文件解析失败"
-    return job
-
-
-# ---------------------------------------------------------------------------
-# 2. Fixtures
-# ---------------------------------------------------------------------------
-
-
-@pytest.fixture
-def repo():
-    return InMemoryIngestJobRepository()
-
-
-@pytest.fixture
-def client(repo):
-    """创建带有依赖覆盖的 TestClient。"""
-    test_app = FastAPI()
-    test_app.include_router(router, prefix="/ingest-jobs")
-
-    def _override_repo():
-        return repo
-
-    test_app.dependency_overrides[get_ingest_job_repository] = _override_repo
-
-    yield TestClient(test_app)
-
-    test_app.dependency_overrides.clear()
-
-
-# ---------------------------------------------------------------------------
-# 3. POST / — 提交摄入任务
-# ---------------------------------------------------------------------------
-
-
-class TestSubmitIngestJob:
-    """提交摄入任务端点测试。"""
-
-    def test_submit_with_valid_data(self, client):
-        """使用有效数据提交摄入任务应成功。"""
-        resp = client.post(
-            "/ingest-jobs",
-            json={
-                "project_id": "proj-123",
-                "library_id": "lib-456",
-                "storage_key": "uploads/video.mp4",
-            },
-        )
-        assert resp.status_code == 200
-        data = resp.json()
-        assert data["project_id"] == "proj-123"
-        assert data["library_id"] == "lib-456"
-        assert data["storage_key"] == "uploads/video.mp4"
-        assert data["status"] == "pending"
-        assert data["error_message"] == ""
-        assert data["result_asset_id"] == "" or data["result_asset_id"] is None
-        assert "id" in data
-        assert len(data["id"]) > 0
-
-    def test_submit_generates_unique_id(self, client):
-        """每次提交应生成不同的任务 ID。"""
-        resp1 = client.post(
-            "/ingest-jobs",
-            json={"project_id": "p1", "library_id": "l1", "storage_key": "a.mp4"},
-        )
-        resp2 = client.post(
-            "/ingest-jobs",
-            json={"project_id": "p1", "library_id": "l1", "storage_key": "b.mp4"},
-        )
-        assert resp1.json()["id"] != resp2.json()["id"]
-
-    def test_submit_missing_project_id_returns_422(self, client):
-        """缺少 project_id 应返回 422。"""
-        resp = client.post(
-            "/ingest-jobs",
-            json={"library_id": "lib-1", "storage_key": "uploads/test.mp4"},
-        )
-        assert resp.status_code == 422
-
-    def test_submit_missing_library_id_returns_422(self, client):
-        """缺少 library_id 应返回 422。"""
-        resp = client.post(
-            "/ingest-jobs",
-            json={"project_id": "proj-1", "storage_key": "uploads/test.mp4"},
-        )
-        assert resp.status_code == 422
-
-    def test_submit_missing_storage_key_returns_422(self, client):
-        """缺少 storage_key 应返回 422。"""
-        resp = client.post(
-            "/ingest-jobs",
-            json={"project_id": "proj-1", "library_id": "lib-1"},
-        )
-        assert resp.status_code == 422
-
-    def test_submit_empty_project_id_returns_422(self, client):
-        """空 project_id 应返回 422。"""
-        resp = client.post(
-            "/ingest-jobs",
-            json={"project_id": "", "library_id": "lib-1", "storage_key": "x.mp4"},
-        )
-        assert resp.status_code == 422
-
-    def test_submit_empty_library_id_returns_422(self, client):
-        """空 library_id 应返回 422。"""
-        resp = client.post(
-            "/ingest-jobs",
-            json={"project_id": "p1", "library_id": "", "storage_key": "x.mp4"},
-        )
-        assert resp.status_code == 422
-
-    def test_submit_empty_storage_key_returns_422(self, client):
-        """空 storage_key 应返回 422。"""
-        resp = client.post(
-            "/ingest-jobs",
-            json={"project_id": "p1", "library_id": "l1", "storage_key": ""},
-        )
-        assert resp.status_code == 422
-
-    def test_submit_sends_celery_task(self, client):
-        """提交任务后应触发 Celery 异步任务。"""
-        ingest_routes.celery_app.send_task.reset_mock()
-
-        resp = client.post(
-            "/ingest-jobs",
-            json={"project_id": "p1", "library_id": "l1", "storage_key": "x.mp4"},
-        )
-        assert resp.status_code == 200
-
-        job_id = resp.json()["id"]
-        ingest_routes.celery_app.send_task.assert_called_once_with(
-            "worker.ingest_asset",
-            args=[job_id],
-        )
-
-    def test_submit_persists_to_repository(self, client, repo):
-        """提交后任务应保存到 repository。"""
-        resp = client.post(
-            "/ingest-jobs",
-            json={"project_id": "p1", "library_id": "l1", "storage_key": "test.mp4"},
-        )
-        job_id = resp.json()["id"]
-
-        saved = repo.get(job_id)
-        assert saved is not None
-        assert saved.project_id == "p1"
-        assert saved.library_id == "l1"
-        assert saved.storage_key == "test.mp4"
-        assert saved.status == IngestJobStatus.PENDING
-
-    def test_submit_with_different_file_types(self, client):
-        """支持不同文件类型的 storage_key。"""
-        for key in ["uploads/image.jpg", "videos/clip.mov", "audio/sound.mp3"]:
-            resp = client.post(
-                "/ingest-jobs",
-                json={"project_id": "p1", "library_id": "l1", "storage_key": key},
-            )
-            assert resp.status_code == 200
-            assert resp.json()["storage_key"] == key
-
-
-# ---------------------------------------------------------------------------
-# 4. GET /{job_id} — 获取摄入任务详情
-# ---------------------------------------------------------------------------
-
-
-class TestGetIngestJob:
-    """获取摄入任务详情端点测试。"""
-
-    def test_get_pending_job(self, client, repo):
-        """获取 pending 状态的任务。"""
-        job = _make_job(status=IngestJobStatus.PENDING)
-        repo.create(job)
-
-        resp = client.get(f"/ingest-jobs/{job.id}")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert data["id"] == job.id
-        assert data["status"] == "pending"
-        assert data["result_asset_id"] == "" or data["result_asset_id"] is None
-
-    def test_get_processing_job(self, client, repo):
-        """获取 processing 状态的任务。"""
-        job = _make_job(status=IngestJobStatus.PROCESSING)
-        repo.create(job)
-
-        resp = client.get(f"/ingest-jobs/{job.id}")
-        assert resp.status_code == 200
-        assert resp.json()["status"] == "processing"
-
-    def test_get_completed_job(self, client, repo):
-        """获取已完成的任务应包含 result_asset_id。"""
-        job = _make_job(status=IngestJobStatus.COMPLETED)
-        repo.create(job)
-
-        resp = client.get(f"/ingest-jobs/{job.id}")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert data["status"] == "completed"
-        assert data["result_asset_id"] == "asset-completed-001"
-        assert data["error_message"] == ""
-
-    def test_get_failed_job(self, client, repo):
-        """获取失败的任务应包含错误信息。"""
-        job = _make_job(status=IngestJobStatus.FAILED)
-        repo.create(job)
-
-        resp = client.get(f"/ingest-jobs/{job.id}")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert data["status"] == "failed"
-        assert "文件解析失败" in data["error_message"]
-
-    def test_get_nonexistent_job_raises_error(self, client):
-        """获取不存在的任务会抛出 ValueError（当前实现未使用 HTTPException）。"""
-        # 注：路由中使用 raise ValueError 而非 HTTPException，
-        # 在 TestClient 中会以异常形式抛出。生产环境会返回 500。
-        # 此处验证当前行为：当 job 不存在时会报错。
-        try:
-            resp = client.get("/ingest-jobs/nonexistent-job-id")
-            # 如果 FastAPI 捕获了异常，会返回 500
-            assert resp.status_code == 500
-        except (ValueError, Exception):
-            # TestClient 中 ValueError 可能直接抛出
-            pass  # 符合预期：不存在的任务会报错
-
-    def test_response_contains_all_required_fields(self, client, repo):
-        """响应应包含所有必需字段。"""
-        job = _make_job()
-        repo.create(job)
-
-        resp = client.get(f"/ingest-jobs/{job.id}")
-        data = resp.json()
-        for field in ["id", "project_id", "library_id", "storage_key", "status", "error_message"]:
-            assert field in data, f"缺少字段: {field}"
-
-
-# ---------------------------------------------------------------------------
-# 5. 跨端点场景
-# ---------------------------------------------------------------------------
-
-
-class TestIngestApiScenarios:
-    """摄入任务 API 跨端点集成场景。"""
-
-    def test_submit_then_get_pending(self, client, repo):
-        """提交任务后立即查询应为 pending 状态。"""
-        submit_resp = client.post(
-            "/ingest-jobs",
-            json={
-                "project_id": "proj-scenario",
-                "library_id": "lib-scenario",
-                "storage_key": "uploads/scenario.mp4",
-            },
-        )
-        assert submit_resp.status_code == 200
-        job_id = submit_resp.json()["id"]
-
-        get_resp = client.get(f"/ingest-jobs/{job_id}")
-        assert get_resp.status_code == 200
-        assert get_resp.json()["status"] == "pending"
-        assert get_resp.json()["storage_key"] == "uploads/scenario.mp4"
-
-    def test_submit_simulate_complete_then_get(self, client, repo):
-        """模拟 worker 完成任务后查询应返回 asset_id。"""
-        submit_resp = client.post(
-            "/ingest-jobs",
-            json={"project_id": "p1", "library_id": "l1", "storage_key": "video.mp4"},
-        )
-        job_id = submit_resp.json()["id"]
-
-        # 模拟 worker 处理完成
-        job = repo.get(job_id)
-        assert job is not None
-        job.status = IngestJobStatus.COMPLETED
-        job.result_asset_id = "asset-new-001"
-        repo.update(job)
-
-        get_resp = client.get(f"/ingest-jobs/{job_id}")
-        assert get_resp.status_code == 200
-        data = get_resp.json()
-        assert data["status"] == "completed"
-        assert data["result_asset_id"] == "asset-new-001"
-
-    def test_submit_simulate_failure_then_get(self, client, repo):
-        """模拟 worker 失败后查询应返回错误信息。"""
-        submit_resp = client.post(
-            "/ingest-jobs",
-            json={"project_id": "p1", "library_id": "l1", "storage_key": "bad.mp4"},
-        )
-        job_id = submit_resp.json()["id"]
-
-        # 模拟处理失败
-        job = repo.get(job_id)
-        assert job is not None
-        job.status = IngestJobStatus.FAILED
-        job.error_message = "文件格式不支持"
-        repo.update(job)
-
-        get_resp = client.get(f"/ingest-jobs/{job_id}")
-        assert get_resp.status_code == 200
-        data = get_resp.json()
-        assert data["status"] == "failed"
-        assert "文件格式不支持" in data["error_message"]
-
-
-if __name__ == "__main__":
-    pytest.main([__file__, "-v"])

tests/integration/test_task_center_api.py

@@ -46,6 +46,7 @@ from packages.domain import (
     User,
 )
 
+
 # ---------------------------------------------------------------------------
 # 1. Stub Repository 实现
 # ---------------------------------------------------------------------------
@@ -113,7 +114,9 @@ class StubIngestJobRepository:
             job.status = status
 
     def list_by_project(self, project_id: str, skip: int = 0, limit: int = 50) -> list[IngestJob]:
-        return [j for j in self._jobs.values() if j.project_id == project_id][skip : skip + limit]
+        return [
+            j for j in self._jobs.values() if j.project_id == project_id
+        ][skip : skip + limit]
 
     def list_by_library(self, library_id: str, skip: int = 0, limit: int = 50) -> list[IngestJob]:
         return [j for j in self._jobs.values() if j.library_id == library_id][skip : skip + limit]

tests/integration/test_template_categories_api.py

@@ -1,349 +0,0 @@
-"""
-模板分类 CRUD API 集成测试。
-
-覆盖端点：
-- GET    /templates/categories/list — 列出分类
-- POST   /templates/categories — 创建分类
-- DELETE /templates/categories/{category_id} — 删除分类
-
-使用 FastAPI TestClient + dependency_overrides 模式，
-mock template repository，验证分类 CRUD 行为。
-"""
-
-from __future__ import annotations
-
-import os
-import sys
-from datetime import datetime, timezone
-from uuid import uuid4
-
-# ── 环境变量 & sys.path（必须在导入 app.* 之前设置） ──────────────────────────
-os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
-os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
-
-import pytest
-from fastapi import FastAPI
-from fastapi.testclient import TestClient
-
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api"))
-
-from app.api.routes import templates as templates_module
-from app.auth import AuthenticatedUser, get_current_user
-
-from packages.domain.entities import User
-from packages.domain.template import TemplateCategory
-
-# ---------------------------------------------------------------------------
-# 1. 内存 Repository
-# ---------------------------------------------------------------------------
-
-
-class InMemoryTemplateRepository:
-    """内存中的模板 Repository，仅实现分类相关方法。"""
-
-    def __init__(self):
-        self._categories: dict[str, TemplateCategory] = {}
-        self._templates = {}
-        self._segments = {}
-
-    # ── 分类相关 ──
-
-    def list_categories(self, user_id: str) -> list[TemplateCategory]:
-        return [c for c in self._categories.values() if c.user_id == user_id]
-
-    def create_category(self, category: TemplateCategory) -> TemplateCategory:
-        # 检查重复名称
-        existing = [c for c in self._categories.values() if c.user_id == category.user_id and c.name == category.name]
-        if existing:
-            raise ValueError(f"分类名称已存在: {category.name}")
-        self._categories[category.id] = category
-        return category
-
-    def get_category(self, category_id: str, user_id: str) -> TemplateCategory | None:
-        cat = self._categories.get(category_id)
-        if cat and cat.user_id == user_id:
-            return cat
-        return None
-
-    def delete_category(self, category_id: str, user_id: str) -> bool:
-        cat = self.get_category(category_id, user_id)
-        if cat:
-            del self._categories[category_id]
-            return True
-        return False
-
-    # ── 模板相关（路由可能调用，提供占位实现） ──
-
-    def list_by_user(self, user_id: str, *, skip: int = 0, limit: int = 50):
-        return []
-
-    def get(self, template_id: str, user_id: str):
-        return None
-
-    def create(self, template):
-        return template
-
-    def update(self, template):
-        return template
-
-    def delete(self, template_id: str, user_id: str) -> bool:
-        return False
-
-    def count_by_user(self, user_id: str) -> int:
-        return 0
-
-    def list_segments(self, template_id: str):
-        return []
-
-    def create_segments(self, segments):
-        return segments
-
-    def delete_segments_by_template(self, template_id: str) -> int:
-        return 0
-
-    def validate_template(self, *args, **kwargs):
-        return None
-
-
-# ---------------------------------------------------------------------------
-# 2. 辅助函数
-# ---------------------------------------------------------------------------
-
-
-def _make_user(**overrides) -> User:
-    defaults = dict(
-        id="user-test-001",
-        email="test@example.com",
-        display_name="Test User",
-        username="testuser",
-        subscription_plan="free",
-        subscription_status="active",
-        max_projects=3,
-        max_storage_gb=10,
-        created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
-    )
-    defaults.update(overrides)
-    return User(**defaults)
-
-
-def _make_category(
-    name: str,
-    user_id: str = "user-test-001",
-) -> TemplateCategory:
-    return TemplateCategory(
-        id=uuid4().hex,
-        user_id=user_id,
-        name=name,
-        created_at=datetime.now(timezone.utc),
-    )
-
-
-# ---------------------------------------------------------------------------
-# 3. Fixtures
-# ---------------------------------------------------------------------------
-
-
-@pytest.fixture
-def template_repo():
-    return InMemoryTemplateRepository()
-
-
-@pytest.fixture
-def client(template_repo):
-    """创建带有依赖覆盖的 TestClient。"""
-    test_app = FastAPI()
-    test_app.include_router(templates_module.router, prefix="/templates")
-
-    def _override_current_user():
-        return AuthenticatedUser(user=_make_user())
-
-    def _override_template_repo():
-        return template_repo
-
-    test_app.dependency_overrides[get_current_user] = _override_current_user
-    # 覆盖路由模块内的 _get_template_repository 依赖
-    test_app.dependency_overrides[templates_module._get_template_repository] = _override_template_repo
-
-    yield TestClient(test_app)
-
-    test_app.dependency_overrides.clear()
-
-
-# ---------------------------------------------------------------------------
-# 4. GET /categories/list — 列出分类
-# ---------------------------------------------------------------------------
-
-
-class TestListCategories:
-    """列出分类端点测试。"""
-
-    def test_empty_list(self, client):
-        """无分类时返回空列表。"""
-        resp = client.get("/templates/categories/list")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert data["items"] == []
-
-    def test_returns_user_categories(self, client, template_repo):
-        """只返回当前用户的分类。"""
-        c1 = _make_category("美食", "user-test-001")
-        c2 = _make_category("旅行", "user-test-001")
-        c3 = _make_category("科技", "other-user")
-        template_repo.create_category(c1)
-        template_repo.create_category(c2)
-        template_repo.create_category(c3)
-
-        resp = client.get("/templates/categories/list")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert len(data["items"]) == 2
-        names = {item["name"] for item in data["items"]}
-        assert names == {"美食", "旅行"}
-
-    def test_response_fields(self, client, template_repo):
-        """响应包含所有必需字段。"""
-        c = _make_category("测试分类")
-        template_repo.create_category(c)
-
-        resp = client.get("/templates/categories/list")
-        item = resp.json()["items"][0]
-        assert "id" in item
-        assert "user_id" in item
-        assert "name" in item
-        assert "created_at" in item
-
-
-# ---------------------------------------------------------------------------
-# 5. POST /categories — 创建分类
-# ---------------------------------------------------------------------------
-
-
-class TestCreateCategory:
-    """创建分类端点测试。"""
-
-    def test_create_valid_category(self, client):
-        """使用有效名称创建分类应成功。"""
-        resp = client.post("/templates/categories", json={"name": "vlog"})
-        assert resp.status_code == 201
-        data = resp.json()
-        assert data["name"] == "vlog"
-        assert "id" in data
-        assert data["user_id"] == "user-test-001"
-        assert "created_at" in data
-
-    def test_create_with_chinese_name(self, client):
-        """支持中文分类名称。"""
-        resp = client.post("/templates/categories", json={"name": "美食探店"})
-        assert resp.status_code == 201
-        assert resp.json()["name"] == "美食探店"
-
-    def test_create_persists_to_repo(self, client, template_repo):
-        """创建后分类保存到 repository。"""
-        resp = client.post("/templates/categories", json={"name": "新知识"})
-        cat_id = resp.json()["id"]
-
-        saved = template_repo.get_category(cat_id, "user-test-001")
-        assert saved is not None
-        assert saved.name == "新知识"
-
-    def test_create_missing_name_returns_422(self, client):
-        """缺少 name 字段返回 422。"""
-        resp = client.post("/templates/categories", json={})
-        assert resp.status_code == 422
-
-    def test_create_empty_name_returns_422(self, client):
-        """空名称返回 422（Pydantic min_length 校验）。"""
-        resp = client.post("/templates/categories", json={"name": ""})
-        # CreateCategoryRequest 没有 min_length 限制，此处验证实际行为
-        assert resp.status_code in (201, 422)
-
-    def test_create_multiple_categories(self, client, template_repo):
-        """可创建多个不同名称的分类。"""
-        names = ["美食", "旅行", "科技", "教育", "娱乐"]
-        for name in names:
-            resp = client.post("/templates/categories", json={"name": name})
-            assert resp.status_code == 201
-
-        all_cats = template_repo.list_categories("user-test-001")
-        assert len(all_cats) == 5
-
-
-# ---------------------------------------------------------------------------
-# 6. DELETE /categories/{category_id} — 删除分类
-# ---------------------------------------------------------------------------
-
-
-class TestDeleteCategory:
-    """删除分类端点测试。"""
-
-    def test_delete_existing_category(self, client, template_repo):
-        """删除存在的分类返回 204。"""
-        c = _make_category("待删除")
-        template_repo.create_category(c)
-
-        resp = client.delete(f"/templates/categories/{c.id}")
-        assert resp.status_code == 204
-
-        # 验证已删除
-        assert template_repo.get_category(c.id, "user-test-001") is None
-
-    def test_delete_nonexistent_returns_404(self, client):
-        """删除不存在的分类返回 404。"""
-        resp = client.delete("/templates/categories/nonexistent-id")
-        assert resp.status_code == 404
-        assert "not found" in resp.json()["detail"].lower() or "Category" in resp.json()["detail"]
-
-    def test_delete_other_user_category_returns_404(self, client, template_repo):
-        """删除其他用户的分类返回 404（安全隔离）。"""
-        c = _make_category("他人分类", user_id="other-user")
-        template_repo.create_category(c)
-
-        resp = client.delete(f"/templates/categories/{c.id}")
-        assert resp.status_code == 404
-        # 验证未被删除
-        assert template_repo.get_category(c.id, "other-user") is not None
-
-    def test_delete_idempotent(self, client, template_repo):
-        """删除后再次删除返回 404。"""
-        c = _make_category("幂等测试")
-        template_repo.create_category(c)
-
-        resp1 = client.delete(f"/templates/categories/{c.id}")
-        assert resp1.status_code == 204
-
-        resp2 = client.delete(f"/templates/categories/{c.id}")
-        assert resp2.status_code == 404
-
-
-# ---------------------------------------------------------------------------
-# 7. 跨端点场景
-# ---------------------------------------------------------------------------
-
-
-class TestCategoryCrudFlow:
-    """分类 CRUD 完整流程。"""
-
-    def test_create_list_delete_flow(self, client, template_repo):
-        """创建 → 列表 → 删除 完整流程。"""
-        # 1. 创建
-        create_resp = client.post("/templates/categories", json={"name": "流程测试"})
-        assert create_resp.status_code == 201
-        cat_id = create_resp.json()["id"]
-
-        # 2. 列表验证
-        list_resp = client.get("/templates/categories/list")
-        assert list_resp.status_code == 200
-        assert len(list_resp.json()["items"]) == 1
-        assert list_resp.json()["items"][0]["name"] == "流程测试"
-
-        # 3. 删除
-        del_resp = client.delete(f"/templates/categories/{cat_id}")
-        assert del_resp.status_code == 204
-
-        # 4. 再次列表验证已删除
-        list_resp2 = client.get("/templates/categories/list")
-        assert list_resp2.json()["items"] == []
-
-
-if __name__ == "__main__":
-    pytest.main([__file__, "-v"])

tests/integration/test_tts_api.py

@@ -1,918 +0,0 @@
-"""
-TTS 合成 API 集成测试。
-
-覆盖端点：
-- POST   /tts/synthesize — 创建 TTS 合成任务
-- GET    /tts/jobs — 列出 TTS 任务
-- GET    /tts/jobs/{job_id} — 获取 TTS 任务详情
-- GET    /tts/jobs/{job_id}/status — 获取 TTS 任务状态
-- DELETE /tts/jobs/{job_id} — 删除 TTS 任务
-- POST   /tts/jobs/{job_id}/save-to-library — 保存到音色库
-
-使用 FastAPI TestClient + dependency_overrides 模式，
-mock repository 和 CosyVoice 服务。
-"""
-
-from __future__ import annotations
-
-import os
-import sys
-from datetime import datetime, timezone
-from unittest.mock import MagicMock
-
-# ── 环境变量 & sys.path（必须在导入 app.* 之前设置） ──────────────────────────
-os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
-os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
-
-import pytest
-from fastapi import FastAPI
-from fastapi.testclient import TestClient
-
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api"))
-
-from app.api.routes.tts import router
-from app.auth import AuthenticatedUser, get_current_user
-from app.dependencies import (
-    get_cosyvoice_service,
-    get_user_repository,
-    get_voice_clone_profile_repository,
-    get_voice_library_repository,
-)
-
-from packages.domain.entities import User
-from packages.domain.tts_job import TTSJob, TTSJobStatus
-from packages.domain.voice_clone_profile import VoiceCloneProfile, VoiceCloneStatus
-
-# ---------------------------------------------------------------------------
-# 1. 内存 Repository
-# ---------------------------------------------------------------------------
-
-
-class InMemoryTTSJobRepository:
-    """内存中的 TTS 任务 Repository。"""
-
-    def __init__(self):
-        self._items: dict[str, TTSJob] = {}
-
-    def create(self, job: TTSJob) -> TTSJob:
-        self._items[job.id] = job
-        return job
-
-    def get(self, job_id: str) -> TTSJob | None:
-        return self._items.get(job_id)
-
-    def update(self, job: TTSJob) -> TTSJob:
-        self._items[job.id] = job
-        return job
-
-    def delete(self, job_id: str) -> bool:
-        if job_id in self._items:
-            del self._items[job_id]
-            return True
-        return False
-
-    def list_by_user(
-        self,
-        user_id: str,
-        *,
-        status=None,
-        limit: int = 50,
-        offset: int = 0,
-    ) -> list[TTSJob]:
-        items = [j for j in self._items.values() if j.user_id == user_id]
-        if status:
-            status_str = status.value if hasattr(status, "value") else str(status)
-            items = [j for j in items if j.status.value == status_str]
-        items.sort(key=lambda j: j.created_at, reverse=True)
-        return items[offset : offset + limit]
-
-    def count_by_user(
-        self,
-        user_id: str,
-        *,
-        status=None,
-    ) -> int:
-        items = [j for j in self._items.values() if j.user_id == user_id]
-        if status:
-            status_str = status.value if hasattr(status, "value") else str(status)
-            items = [j for j in items if j.status.value == status_str]
-        return len(items)
-
-    def list_by_profile(
-        self,
-        voice_clone_profile_id: str,
-        *,
-        status=None,
-        limit: int = 50,
-        offset: int = 0,
-    ) -> list[TTSJob]:
-        items = [j for j in self._items.values() if j.voice_clone_profile_id == voice_clone_profile_id]
-        if status:
-            status_str = status.value if hasattr(status, "value") else str(status)
-            items = [j for j in items if j.status.value == status_str]
-        return items[offset : offset + limit]
-
-
-class InMemoryVoiceCloneProfileRepository:
-    """内存中的音色克隆档案 Repository（用于 TTS 测试）。"""
-
-    def __init__(self):
-        self._items: dict[str, VoiceCloneProfile] = {}
-
-    def create(self, profile):
-        self._items[profile.id] = profile
-        return profile
-
-    def get(self, profile_id: str):
-        return self._items.get(profile_id)
-
-    def update(self, profile):
-        self._items[profile.id] = profile
-        return profile
-
-    def delete(self, profile_id):
-        if profile_id in self._items:
-            del self._items[profile_id]
-            return True
-        return False
-
-    def list_by_user(self, user_id, **kwargs):
-        return [p for p in self._items.values() if p.user_id == user_id]
-
-    def count_by_user(self, user_id, **kwargs):
-        return len([p for p in self._items.values() if p.user_id == user_id])
-
-    def find_by_voice_id(self, voice_id):
-        return None
-
-    def find_profile_ids_by_voice_ids(self, voice_ids):
-        return {}
-
-
-class InMemoryVoiceLibraryRepository:
-    """内存中的配音库 Repository。"""
-
-    def __init__(self):
-        self._items = {}
-
-    def create(self, item):
-        self._items[item.id] = item
-        return item
-
-    def get(self, voice_id: str, user_id: str):
-        item = self._items.get(voice_id)
-        if item and item.user_id == user_id:
-            return item
-        return None
-
-    def update(self, item):
-        self._items[item.id] = item
-        return item
-
-    def delete(self, voice_id: str, user_id: str) -> bool:
-        item = self.get(voice_id, user_id)
-        if item:
-            del self._items[voice_id]
-            return True
-        return False
-
-    def list_by_user(self, user_id, **kwargs):
-        return [i for i in self._items.values() if i.user_id == user_id]
-
-    def count_by_user(self, user_id: str) -> int:
-        return len([i for i in self._items.values() if i.user_id == user_id])
-
-
-class InMemoryUserRepository:
-    """内存中的用户 Repository。"""
-
-    def __init__(self):
-        self._users = {}
-
-    def save(self, user):
-        self._users[user.id] = user
-
-    def find_by_id(self, user_id: str):
-        return self._users.get(user_id)
-
-    def find_by_email(self, email: str):
-        for u in self._users.values():
-            if u.email == email:
-                return u
-        return None
-
-
-# ---------------------------------------------------------------------------
-# 2. Mock CosyVoice 服务
-# ---------------------------------------------------------------------------
-
-
-class MockCosyVoiceService:
-    """Mock CosyVoice 服务。"""
-
-    def __init__(self, *, fail_submit: bool = False):
-        self.fail_submit = fail_submit
-        self.submit_called = False
-
-    def submit_synthesize_task(self, *, text: str, voice_id: str = "", **kwargs) -> dict:
-        self.submit_called = True
-        if self.fail_submit:
-            from packages.application.cosyvoice_service import CosyVoiceError
-
-            raise CosyVoiceError("模拟 CosyVoice 合成失败")
-
-        return {
-            "task_id": "mock-tts-task-123",
-            "status": "processing",
-        }
-
-    def poll_synthesize_task(self, task_id: str, timeout: float = 120.0) -> dict:
-        return {
-            "status": "completed",
-            "audio_url": "https://cdn.example.com/tts/output.mp3",
-            "duration": 5.5,
-            "file_size": 88000,
-            "sample_rate": 22050,
-            "format": "mp3",
-        }
-
-    def synthesize_speech(self, *, text: str, voice_id: str = "", **kwargs) -> dict:
-        return {
-            "audio_url": "https://cdn.example.com/tts/output.mp3",
-            "duration": 5.5,
-            "file_size": 88000,
-        }
-
-    def submit_clone_task(self, **kwargs) -> dict:
-        return {"task_id": "clone-1", "status": "processing"}
-
-    def list_preset_voices(self) -> list:
-        return []
-
-
-# ---------------------------------------------------------------------------
-# 3. 辅助函数
-# ---------------------------------------------------------------------------
-
-
-def _make_user(**overrides) -> User:
-    defaults = dict(
-        id="user-test-001",
-        email="test@example.com",
-        display_name="Test User",
-        username="testuser",
-        subscription_plan="free",
-        subscription_status="active",
-        max_projects=3,
-        max_storage_gb=10,
-        created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
-    )
-    defaults.update(overrides)
-    return User(**defaults)
-
-
-def _make_tts_job(
-    text: str = "你好，这是一段测试文本。",
-    user_id: str = "user-test-001",
-    status: TTSJobStatus = TTSJobStatus.PENDING,
-    **kwargs,
-) -> TTSJob:
-    job = TTSJob.create(
-        user_id=user_id,
-        input_text=text,
-        voice_id=kwargs.get("voice_id", "voice-1"),
-        voice_model=kwargs.get("voice_model", "cosyvoice-v2"),
-        project_id=kwargs.get("project_id", ""),
-        voice_clone_profile_id=kwargs.get("voice_clone_profile_id", ""),
-        format=kwargs.get("format", "mp3"),
-        sample_rate=kwargs.get("sample_rate", 22050),
-        max_retries=kwargs.get("max_retries", 3),
-        metadata=kwargs.get("metadata", None),
-    )
-    # 设置状态
-    if status == TTSJobStatus.PROCESSING:
-        job.mark_processing()
-    elif status == TTSJobStatus.COMPLETED:
-        job.mark_processing()
-        job.mark_completed(
-            output_audio_url=kwargs.get("output_audio_url", "https://cdn.example.com/tts/out.mp3"),
-            output_audio_key=kwargs.get("output_audio_key", "tts/out.mp3"),
-            duration=kwargs.get("duration", 5.5),
-            file_size=kwargs.get("file_size", 88000),
-        )
-    elif status == TTSJobStatus.FAILED:
-        job.mark_processing()
-        job.mark_failed("合成失败")
-    elif status == TTSJobStatus.CANCELLED:
-        job.mark_cancelled()
-    return job
-
-
-def _make_voice_clone_profile(
-    user_id: str = "user-test-001",
-    status: VoiceCloneStatus = VoiceCloneStatus.READY,
-) -> VoiceCloneProfile:
-    profile = VoiceCloneProfile.create(
-        user_id=user_id,
-        name="测试克隆音色",
-        voice_model="cosyvoice-v2",
-    )
-    if status == VoiceCloneStatus.READY:
-        profile.mark_processing()
-        profile.mark_ready("clone-voice-001")
-    return profile
-
-
-# ---------------------------------------------------------------------------
-# 4. Fixtures
-# ---------------------------------------------------------------------------
-
-
-@pytest.fixture
-def tts_repo():
-    return InMemoryTTSJobRepository()
-
-
-@pytest.fixture
-def voice_clone_repo():
-    return InMemoryVoiceCloneProfileRepository()
-
-
-@pytest.fixture
-def voice_library_repo():
-    return InMemoryVoiceLibraryRepository()
-
-
-@pytest.fixture
-def user_repo():
-    repo = InMemoryUserRepository()
-    repo.save(_make_user())
-    return repo
-
-
-@pytest.fixture
-def cosyvoice_service():
-    return MockCosyVoiceService()
-
-
-@pytest.fixture
-def client(tts_repo, voice_clone_repo, voice_library_repo, user_repo, cosyvoice_service):
-    """创建带有依赖覆盖的 TestClient。"""
-    test_app = FastAPI()
-    test_app.include_router(router, prefix="/tts")
-
-    def _override_current_user():
-        return AuthenticatedUser(user=_make_user())
-
-    def _override_tts_repo():
-        return tts_repo
-
-    test_app.dependency_overrides[get_current_user] = _override_current_user
-    test_app.dependency_overrides[get_cosyvoice_service] = lambda: cosyvoice_service
-    test_app.dependency_overrides[get_voice_clone_profile_repository] = lambda: voice_clone_repo
-    test_app.dependency_overrides[get_voice_library_repository] = lambda: voice_library_repo
-    test_app.dependency_overrides[get_user_repository] = lambda: user_repo
-
-    # 使用 FastAPI dependency_overrides 覆盖 TTS repository
-    from app.api.routes import tts as tts_module
-
-    test_app.dependency_overrides[tts_module._get_repository] = lambda: tts_repo
-
-    yield TestClient(test_app)
-
-    test_app.dependency_overrides.clear()
-
-
-# ---------------------------------------------------------------------------
-# 5. POST /synthesize — 创建 TTS 合成任务
-# ---------------------------------------------------------------------------
-
-
-class TestCreateTTSJob:
-    """创建 TTS 合成任务端点测试。"""
-
-    def test_create_with_valid_text(self, client, cosyvoice_service):
-        """使用有效文本创建 TTS 任务。"""
-        resp = client.post(
-            "/tts/synthesize",
-            json={
-                "text": "你好，世界！",
-                "voice_id": "voice-1",
-                "voice_model": "cosyvoice-v2",
-            },
-        )
-        assert resp.status_code == 201
-        data = resp.json()
-        assert "job_id" in data
-        assert data["message"] == "合成任务已创建"
-        assert "status" in data
-
-    def test_create_persists_to_repository(self, client, tts_repo):
-        """创建后任务保存到 repository。"""
-        resp = client.post("/tts/synthesize", json={"text": "持久化测试"})
-        job_id = resp.json()["job_id"]
-
-        saved = tts_repo.get(job_id)
-        assert saved is not None
-        assert saved.input_text == "持久化测试"
-        assert saved.user_id == "user-test-001"
-
-    def test_create_missing_text_returns_422(self, client):
-        """缺少 text 返回 422。"""
-        resp = client.post("/tts/synthesize", json={})
-        assert resp.status_code == 422
-
-    def test_create_empty_text_returns_422(self, client):
-        """空 text 返回 422。"""
-        resp = client.post("/tts/synthesize", json={"text": ""})
-        assert resp.status_code == 422
-
-    def test_create_with_custom_format(self, client):
-        """支持指定输出格式。"""
-        for fmt in ["mp3", "wav", "pcm"]:
-            resp = client.post("/tts/synthesize", json={"text": "测试", "format": fmt})
-            assert resp.status_code == 201
-
-    def test_create_with_invalid_format_returns_422(self, client):
-        """无效格式在 Pydantic 层校验返回 422。"""
-        # format 参数不在 TTSSynthesizeRequest schema 中，
-        # 或者有默认值/枚举校验。此处测试额外字段会被忽略或校验失败。
-        # 实际：schema 中 format 是可选的，有默认值，无效值会在领域层被捕获
-        # 但 API 仍返回 201，任务标记为 failed（与音色克隆行为一致）
-        resp = client.post("/tts/synthesize", json={"text": "测试", "format": "flac"})
-        # 格式不在请求 schema 中时，FastAPI 会忽略额外字段，任务正常创建
-        assert resp.status_code == 201
-
-    def test_create_with_metadata(self, client):
-        """支持自定义 metadata。"""
-        resp = client.post(
-            "/tts/synthesize",
-            json={
-                "text": "元数据测试",
-                "metadata": {"source": "api", "version": "1.0"},
-            },
-        )
-        assert resp.status_code == 201
-
-    def test_create_with_voice_clone_profile_id(self, client, voice_clone_repo):
-        """使用音色克隆档案创建 TTS。"""
-        # 准备一个克隆档案
-        profile = _make_voice_clone_profile()
-        voice_clone_repo.create(profile)
-
-        resp = client.post(
-            "/tts/synthesize",
-            json={
-                "text": "使用克隆音色",
-                "voice_clone_profile_id": profile.id,
-            },
-        )
-        assert resp.status_code == 201
-
-    def test_create_with_nonexistent_clone_profile_returns_404(self, client):
-        """使用不存在的克隆档案返回 404。"""
-        resp = client.post(
-            "/tts/synthesize",
-            json={
-                "text": "测试",
-                "voice_clone_profile_id": "nonexistent-profile",
-            },
-        )
-        assert resp.status_code == 404
-
-    def test_create_with_other_user_clone_profile_returns_403(self, client, voice_clone_repo):
-        """使用其他用户的克隆档案返回 403。"""
-        profile = _make_voice_clone_profile(user_id="other-user")
-        voice_clone_repo.create(profile)
-
-        resp = client.post(
-            "/tts/synthesize",
-            json={
-                "text": "越权测试",
-                "voice_clone_profile_id": profile.id,
-            },
-        )
-        assert resp.status_code == 403
-
-
-# ---------------------------------------------------------------------------
-# 6. GET /jobs — 列出 TTS 任务
-# ---------------------------------------------------------------------------
-
-
-class TestListTTSJobs:
-    """列出 TTS 任务端点测试。"""
-
-    def test_empty_list(self, client):
-        """无任务时返回空列表。"""
-        resp = client.get("/tts/jobs")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert data["items"] == []
-        assert data["total"] == 0
-        assert data["page"] == 1
-        assert data["page_size"] == 20
-
-    def test_list_user_jobs(self, client, tts_repo):
-        """只返回当前用户的任务。"""
-        j1 = _make_tts_job("任务1", "user-test-001")
-        j2 = _make_tts_job("任务2", "user-test-001")
-        j3 = _make_tts_job("他人任务", "other-user")
-        tts_repo.create(j1)
-        tts_repo.create(j2)
-        tts_repo.create(j3)
-
-        resp = client.get("/tts/jobs")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert data["total"] == 2
-        assert len(data["items"]) == 2
-
-    def test_filter_by_status(self, client, tts_repo):
-        """按状态筛选。"""
-        completed = _make_tts_job("已完成", status=TTSJobStatus.COMPLETED)
-        failed = _make_tts_job("已失败", status=TTSJobStatus.FAILED)
-        tts_repo.create(completed)
-        tts_repo.create(failed)
-
-        resp = client.get("/tts/jobs?status=completed")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert data["total"] == 1
-        assert data["items"][0]["status"] == "completed"
-
-    def test_pagination(self, client, tts_repo):
-        """分页功能。"""
-        for i in range(5):
-            job = _make_tts_job(f"任务{i}")
-            tts_repo.create(job)
-
-        resp = client.get("/tts/jobs?page=1&page_size=2")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert data["total"] == 5
-        assert data["page"] == 1
-        assert data["page_size"] == 2
-        assert len(data["items"]) == 2
-
-        resp2 = client.get("/tts/jobs?page=2&page_size=2")
-        assert resp2.json()["page"] == 2
-        assert len(resp2.json()["items"]) == 2
-
-        resp3 = client.get("/tts/jobs?page=3&page_size=2")
-        assert len(resp3.json()["items"]) == 1
-
-    def test_list_response_fields(self, client, tts_repo):
-        """列表响应包含所有必需字段。"""
-        job = _make_tts_job("字段测试", status=TTSJobStatus.COMPLETED)
-        tts_repo.create(job)
-
-        resp = client.get("/tts/jobs")
-        item = resp.json()["items"][0]
-        for field in [
-            "id",
-            "user_id",
-            "input_text",
-            "voice_id",
-            "voice_model",
-            "status",
-            "output_audio_url",
-            "duration",
-            "format",
-            "error_message",
-            "retry_count",
-            "max_retries",
-            "created_at",
-            "updated_at",
-        ]:
-            assert field in item, f"缺少字段: {field}"
-
-
-# ---------------------------------------------------------------------------
-# 7. GET /jobs/{job_id} — 获取 TTS 任务详情
-# ---------------------------------------------------------------------------
-
-
-class TestGetTTSJob:
-    """获取 TTS 任务详情端点测试。"""
-
-    def test_get_existing_job(self, client, tts_repo):
-        """获取存在的任务返回详情。"""
-        job = _make_tts_job("详情测试", voice_model="cosyvoice-v2")
-        tts_repo.create(job)
-
-        resp = client.get(f"/tts/jobs/{job.id}")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert data["id"] == job.id
-        assert data["input_text"] == "详情测试"
-        assert data["voice_model"] == "cosyvoice-v2"
-
-    def test_get_nonexistent_returns_404(self, client):
-        """获取不存在的任务返回 404。"""
-        resp = client.get("/tts/jobs/nonexistent-job-id")
-        assert resp.status_code == 404
-        assert "not found" in resp.json()["detail"].lower()
-
-    def test_get_other_user_job_returns_404(self, client, tts_repo):
-        """获取其他用户的任务返回 404（安全隔离）。"""
-        job = _make_tts_job("他人任务", user_id="other-user")
-        tts_repo.create(job)
-
-        resp = client.get(f"/tts/jobs/{job.id}")
-        assert resp.status_code == 404
-
-    def test_get_completed_job(self, client, tts_repo):
-        """获取已完成任务包含音频 URL 和时长。"""
-        job = _make_tts_job("已完成", status=TTSJobStatus.COMPLETED, duration=10.5)
-        tts_repo.create(job)
-
-        resp = client.get(f"/tts/jobs/{job.id}")
-        data = resp.json()
-        assert data["status"] == "completed"
-        assert data["output_audio_url"] != ""
-        assert data["duration"] == 10.5
-        assert data["file_size"] > 0
-
-    def test_get_failed_job(self, client, tts_repo):
-        """获取失败任务包含错误信息。"""
-        job = _make_tts_job("失败任务", status=TTSJobStatus.FAILED)
-        tts_repo.create(job)
-
-        resp = client.get(f"/tts/jobs/{job.id}")
-        data = resp.json()
-        assert data["status"] == "failed"
-        assert data["error_message"] != ""
-
-
-# ---------------------------------------------------------------------------
-# 8. GET /jobs/{job_id}/status — 获取 TTS 任务状态
-# ---------------------------------------------------------------------------
-
-
-class TestGetTTSJobStatus:
-    """获取 TTS 任务状态端点测试。"""
-
-    def test_status_pending(self, client, tts_repo):
-        """pending 状态。"""
-        job = _make_tts_job("pending", status=TTSJobStatus.PENDING)
-        tts_repo.create(job)
-
-        resp = client.get(f"/tts/jobs/{job.id}/status")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert data["id"] == job.id
-        assert data["status"] == "pending"
-
-    def test_status_completed(self, client, tts_repo):
-        """completed 状态包含音频 URL。"""
-        job = _make_tts_job("completed", status=TTSJobStatus.COMPLETED)
-        tts_repo.create(job)
-
-        resp = client.get(f"/tts/jobs/{job.id}/status")
-        data = resp.json()
-        assert data["status"] == "completed"
-        assert data["output_audio_url"] != ""
-        assert data["duration"] > 0
-
-    def test_status_failed(self, client, tts_repo):
-        """failed 状态包含错误信息。"""
-        job = _make_tts_job("failed", status=TTSJobStatus.FAILED)
-        tts_repo.create(job)
-
-        resp = client.get(f"/tts/jobs/{job.id}/status")
-        data = resp.json()
-        assert data["status"] == "failed"
-        assert data["error_message"] != ""
-
-    def test_status_nonexistent_returns_404(self, client):
-        """获取不存在任务的状态返回 404。"""
-        resp = client.get("/tts/jobs/nonexistent/status")
-        assert resp.status_code == 404
-
-
-# ---------------------------------------------------------------------------
-# 9. DELETE /jobs/{job_id} — 删除 TTS 任务
-# ---------------------------------------------------------------------------
-
-
-class TestDeleteTTSJob:
-    """删除 TTS 任务端点测试。"""
-
-    def test_delete_existing_job(self, client, tts_repo):
-        """删除存在的任务返回 204。"""
-        job = _make_tts_job("待删除")
-        tts_repo.create(job)
-
-        resp = client.delete(f"/tts/jobs/{job.id}")
-        assert resp.status_code == 204
-
-        # 验证已删除
-        assert tts_repo.get(job.id) is None
-
-    def test_delete_nonexistent_returns_404(self, client):
-        """删除不存在的任务返回 404。"""
-        resp = client.delete("/tts/jobs/nonexistent-job-id")
-        assert resp.status_code == 404
-
-    def test_delete_other_user_job_returns_404(self, client, tts_repo):
-        """删除其他用户的任务返回 404（安全隔离）。"""
-        job = _make_tts_job("他人任务", user_id="other-user")
-        tts_repo.create(job)
-
-        resp = client.delete(f"/tts/jobs/{job.id}")
-        assert resp.status_code == 404
-        # 验证未被删除
-        assert tts_repo.get(job.id) is not None
-
-    def test_delete_idempotent(self, client, tts_repo):
-        """删除后再次删除返回 404。"""
-        job = _make_tts_job("幂等测试")
-        tts_repo.create(job)
-
-        resp1 = client.delete(f"/tts/jobs/{job.id}")
-        assert resp1.status_code == 204
-
-        resp2 = client.delete(f"/tts/jobs/{job.id}")
-        assert resp2.status_code == 404
-
-
-# ---------------------------------------------------------------------------
-# 10. POST /jobs/{job_id}/save-to-library — 保存到配音库
-# ---------------------------------------------------------------------------
-
-
-class TestSaveToLibrary:
-    """保存到配音库端点测试。"""
-
-    def test_save_completed_job(self, client, tts_repo):
-        """保存已完成的 TTS 任务到配音库。"""
-        job = _make_tts_job("保存测试", status=TTSJobStatus.COMPLETED, duration=5.5)
-        tts_repo.create(job)
-
-        resp = client.post(
-            f"/tts/jobs/{job.id}/save-to-library",
-            json={"name": "我的配音"},
-        )
-        assert resp.status_code == 201
-        data = resp.json()
-        assert data["name"] == "我的配音"
-        assert data["duration"] == 5.5
-        assert data["status"] == "completed"
-        assert "id" in data
-        assert "audio_url" in data
-        assert "voice_id" in data
-        assert "voice_name" in data
-
-    def test_save_pending_job_returns_400(self, client, tts_repo):
-        """保存未完成的任务返回 400。"""
-        job = _make_tts_job("未完成", status=TTSJobStatus.PENDING)
-        tts_repo.create(job)
-
-        resp = client.post(f"/tts/jobs/{job.id}/save-to-library")
-        assert resp.status_code == 400
-        assert "not completed" in resp.json()["detail"].lower() or "完成" in resp.json()["detail"]
-
-    def test_save_failed_job_returns_400(self, client, tts_repo):
-        """保存失败的任务返回 400。"""
-        job = _make_tts_job("失败", status=TTSJobStatus.FAILED)
-        tts_repo.create(job)
-
-        resp = client.post(f"/tts/jobs/{job.id}/save-to-library")
-        assert resp.status_code == 400
-
-    def test_save_nonexistent_job_returns_404(self, client):
-        """保存不存在的任务返回 404。"""
-        resp = client.post("/tts/jobs/nonexistent/save-to-library")
-        assert resp.status_code == 404
-
-    def test_save_other_user_job_returns_404(self, client, tts_repo):
-        """保存其他用户的任务返回 404。"""
-        job = _make_tts_job("他人任务", user_id="other-user", status=TTSJobStatus.COMPLETED)
-        tts_repo.create(job)
-
-        resp = client.post(f"/tts/jobs/{job.id}/save-to-library")
-        assert resp.status_code == 404
-
-    def test_save_auto_generates_name(self, client, tts_repo):
-        """不指定名称时自动生成。"""
-        job = _make_tts_job("自动命名", status=TTSJobStatus.COMPLETED)
-        tts_repo.create(job)
-
-        resp = client.post(f"/tts/jobs/{job.id}/save-to-library", json={})
-        assert resp.status_code == 201
-        data = resp.json()
-        assert data["name"] != ""
-        # 自动生成的名称应该以 TTS- 开头
-        assert data["name"].startswith("TTS-")
-
-    def test_save_creates_library_item(self, client, tts_repo, voice_library_repo):
-        """保存后配音库中新增一条记录。"""
-        before_count = voice_library_repo.count_by_user("user-test-001")
-
-        job = _make_tts_job("入库测试", status=TTSJobStatus.COMPLETED)
-        tts_repo.create(job)
-
-        resp = client.post(f"/tts/jobs/{job.id}/save-to-library", json={"name": "入库"})
-        assert resp.status_code == 201
-
-        after_count = voice_library_repo.count_by_user("user-test-001")
-        assert after_count == before_count + 1
-
-
-# ---------------------------------------------------------------------------
-# 11. 跨端点场景
-# ---------------------------------------------------------------------------
-
-
-class TestTTSLifecycle:
-    """TTS 完整生命周期测试。"""
-
-    def test_create_list_get_delete_flow(self, client, tts_repo):
-        """创建 → 列表 → 详情 → 删除 完整流程。"""
-        # 1. 创建
-        create_resp = client.post("/tts/synthesize", json={"text": "完整流程测试"})
-        assert create_resp.status_code == 201
-        job_id = create_resp.json()["job_id"]
-
-        # 2. 列表
-        list_resp = client.get("/tts/jobs")
-        assert list_resp.json()["total"] == 1
-
-        # 3. 详情
-        detail_resp = client.get(f"/tts/jobs/{job_id}")
-        assert detail_resp.status_code == 200
-        assert detail_resp.json()["input_text"] == "完整流程测试"
-
-        # 4. 状态
-        status_resp = client.get(f"/tts/jobs/{job_id}/status")
-        assert status_resp.status_code == 200
-
-        # 5. 删除
-        del_resp = client.delete(f"/tts/jobs/{job_id}")
-        assert del_resp.status_code == 204
-
-        # 6. 删除后列表为空
-        list_resp2 = client.get("/tts/jobs")
-        assert list_resp2.json()["total"] == 0
-
-    def test_create_simulate_complete_save_to_library(self, client, tts_repo):
-        """创建 → 模拟完成 → 保存到配音库 流程。"""
-        # 创建任务
-        create_resp = client.post("/tts/synthesize", json={"text": "入库流程"})
-        job_id = create_resp.json()["job_id"]
-
-        # 模拟 worker 完成
-        job = tts_repo.get(job_id)
-        assert job is not None
-        # 根据当前状态决定下一步：failed 先重置，pending 则转 processing，已是 processing 则跳过
-        if job.status == TTSJobStatus.FAILED:
-            job.prepare_retry()
-            job.mark_processing()
-        elif job.status == TTSJobStatus.PENDING:
-            job.mark_processing()
-        job.mark_completed(
-            output_audio_url="https://cdn.example.com/tts/final.mp3",
-            duration=8.0,
-            file_size=128000,
-        )
-        tts_repo.update(job)
-
-        # 确认完成
-        status_resp = client.get(f"/tts/jobs/{job_id}/status")
-        assert status_resp.json()["status"] == "completed"
-
-        # 保存到配音库
-        save_resp = client.post(
-            f"/tts/jobs/{job_id}/save-to-library",
-            json={"name": "最终配音"},
-        )
-        assert save_resp.status_code == 201
-        assert save_resp.json()["name"] == "最终配音"
-        assert save_resp.json()["duration"] == 8.0
-
-    def test_multiple_jobs_status_filter(self, client, tts_repo):
-        """多个任务时按状态筛选正确。"""
-        # 创建不同状态的任务
-        for text, status in [
-            ("任务A-完成", TTSJobStatus.COMPLETED),
-            ("任务B-完成", TTSJobStatus.COMPLETED),
-            ("任务C-失败", TTSJobStatus.FAILED),
-            ("任务D-处理中", TTSJobStatus.PROCESSING),
-        ]:
-            job = _make_tts_job(text, status=status)
-            tts_repo.create(job)
-
-        # 按状态筛选
-        completed_resp = client.get("/tts/jobs?status=completed")
-        assert completed_resp.json()["total"] == 2
-
-        failed_resp = client.get("/tts/jobs?status=failed")
-        assert failed_resp.json()["total"] == 1
-
-        processing_resp = client.get("/tts/jobs?status=processing")
-        assert processing_resp.json()["total"] == 1
-
-
-if __name__ == "__main__":
-    pytest.main([__file__, "-v"])

tests/integration/test_voice_clone_api.py

@@ -1,716 +0,0 @@
-"""
-声音克隆 API 集成测试。
-
-覆盖端点：
-- POST   /voice-clones — 创建声音克隆
-- GET    /voice-clones — 列出声音克隆
-- GET    /voice-clones/{clone_id} — 获取克隆详情
-- GET    /voice-clones/{clone_id}/status — 获取克隆状态
-- POST   /voice-clones/{clone_id}/retry — 重试克隆
-- DELETE /voice-clones/{clone_id} — 删除克隆
-
-使用 FastAPI TestClient + dependency_overrides 模式，
-mock repository 和 CosyVoice 服务。
-"""
-
-from __future__ import annotations
-
-import os
-import sys
-from datetime import datetime, timezone
-from unittest.mock import MagicMock
-
-# ── 环境变量 & sys.path（必须在导入 app.* 之前设置） ──────────────────────────
-os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
-os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
-
-import pytest
-from fastapi import FastAPI
-from fastapi.testclient import TestClient
-
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api"))
-
-from app.api.routes.voice_clones import router
-from app.auth import AuthenticatedUser, get_current_user
-from app.dependencies import get_cosyvoice_service, get_voice_clone_profile_repository
-
-from packages.domain.entities import User
-from packages.domain.voice_clone_profile import (
-    VoiceCloneProfile,
-    VoiceCloneStatus,
-)
-
-# ---------------------------------------------------------------------------
-# 1. 内存 Repository
-# ---------------------------------------------------------------------------
-
-
-class InMemoryVoiceCloneProfileRepository:
-    """内存中的音色克隆档案 Repository。"""
-
-    def __init__(self):
-        self._items: dict[str, VoiceCloneProfile] = {}
-
-    def create(self, profile: VoiceCloneProfile) -> VoiceCloneProfile:
-        self._items[profile.id] = profile
-        return profile
-
-    def get(self, profile_id: str) -> VoiceCloneProfile | None:
-        return self._items.get(profile_id)
-
-    def update(self, profile: VoiceCloneProfile) -> VoiceCloneProfile:
-        self._items[profile.id] = profile
-        return profile
-
-    def delete(self, profile_id: str) -> bool:
-        if profile_id in self._items:
-            del self._items[profile_id]
-            return True
-        return False
-
-    def list_by_user(
-        self,
-        user_id: str,
-        *,
-        status=None,
-        limit: int = 50,
-        offset: int = 0,
-    ) -> list[VoiceCloneProfile]:
-        items = [p for p in self._items.values() if p.user_id == user_id]
-        if status:
-            status_str = status.value if hasattr(status, "value") else str(status)
-            items = [p for p in items if p.status.value == status_str]
-        # 按 created_at 倒序
-        items.sort(key=lambda p: p.created_at, reverse=True)
-        return items[offset : offset + limit]
-
-    def count_by_user(
-        self,
-        user_id: str,
-        *,
-        status=None,
-    ) -> int:
-        items = [p for p in self._items.values() if p.user_id == user_id]
-        if status:
-            status_str = status.value if hasattr(status, "value") else str(status)
-            items = [p for p in items if p.status.value == status_str]
-        return len(items)
-
-    def find_by_voice_id(self, voice_id: str) -> VoiceCloneProfile | None:
-        for p in self._items.values():
-            if p.voice_id == voice_id:
-                return p
-        return None
-
-    def find_profile_ids_by_voice_ids(self, voice_ids: list[str]) -> dict[str, str]:
-        result = {}
-        for p in self._items.values():
-            if p.voice_id in voice_ids:
-                result[p.voice_id] = p.id
-        return result
-
-
-# ---------------------------------------------------------------------------
-# 2. Mock CosyVoice 服务
-# ---------------------------------------------------------------------------
-
-
-class MockCosyVoiceService:
-    """Mock CosyVoice 服务，模拟克隆任务提交和状态查询。"""
-
-    def __init__(self, *, fail_submit: bool = False, async_mode: bool = True):
-        self.fail_submit = fail_submit
-        self.async_mode = async_mode
-        self.submit_called = False
-        self.submit_args = None
-
-    def submit_clone_task(self, *, audio_url: str, voice_name: str, language: str = "zh-CN") -> dict:
-        self.submit_called = True
-        self.submit_args = {"audio_url": audio_url, "voice_name": voice_name, "language": language}
-
-        if self.fail_submit:
-            from packages.application.cosyvoice_service import CosyVoiceError
-
-            raise CosyVoiceError("模拟 CosyVoice 提交失败")
-
-        if self.async_mode:
-            # 异步模式：返回 task_id，需要轮询
-            return {"task_id": "mock-task-123", "request_id": "req-456", "status": "processing"}
-        else:
-            # 同步模式：直接返回 voice_id
-            return {"voice_id": "mock-voice-789", "status": "success"}
-
-    def check_task_status(self, task_id: str) -> dict:
-        return {"status": "completed", "voice_id": "mock-voice-789"}
-
-    def list_preset_voices(self) -> list:
-        return []
-
-    def submit_synthesize_task(self, **kwargs) -> dict:
-        return {"task_id": "synth-1", "status": "processing"}
-
-    def synthesize_speech(self, **kwargs) -> dict:
-        return {"audio_url": "https://example.com/audio.mp3", "duration": 5.0}
-
-    def poll_synthesize_task(self, task_id: str, timeout: float = 120.0) -> dict:
-        return {
-            "status": "completed",
-            "audio_url": "https://example.com/audio.mp3",
-            "duration": 5.0,
-        }
-
-
-# ---------------------------------------------------------------------------
-# 3. 辅助函数
-# ---------------------------------------------------------------------------
-
-
-def _make_user(**overrides) -> User:
-    defaults = dict(
-        id="user-test-001",
-        email="test@example.com",
-        display_name="Test User",
-        username="testuser",
-        subscription_plan="free",
-        subscription_status="active",
-        max_projects=3,
-        max_storage_gb=10,
-        created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
-    )
-    defaults.update(overrides)
-    return User(**defaults)
-
-
-def _make_clone_profile(
-    name: str = "我的音色",
-    user_id: str = "user-test-001",
-    status: VoiceCloneStatus = VoiceCloneStatus.PENDING,
-    source_audio_url: str = "https://example.com/source.wav",
-    **kwargs,
-) -> VoiceCloneProfile:
-    profile = VoiceCloneProfile.create(
-        user_id=user_id,
-        name=name,
-        source_audio_url=source_audio_url,
-        voice_model=kwargs.get("voice_model", "cosyvoice-v2"),
-        language=kwargs.get("language", "zh-CN"),
-        gender=kwargs.get("gender", "female"),
-        max_retries=kwargs.get("max_retries", 3),
-        metadata=kwargs.get("metadata", None),
-        description=kwargs.get("description", ""),
-    )
-    # 设置状态
-    if status == VoiceCloneStatus.PROCESSING:
-        profile.mark_processing()
-        profile.metadata = {"cosyvoice_task_id": "task-123"}
-    elif status == VoiceCloneStatus.READY:
-        profile.mark_processing()
-        profile.mark_ready("voice-ready-001")
-    elif status == VoiceCloneStatus.FAILED:
-        profile.mark_processing()
-        profile.mark_failed("模拟失败")
-    elif status == VoiceCloneStatus.DISABLED:
-        profile.mark_disabled()
-    return profile
-
-
-# ---------------------------------------------------------------------------
-# 4. Fixtures
-# ---------------------------------------------------------------------------
-
-
-@pytest.fixture
-def clone_repo():
-    return InMemoryVoiceCloneProfileRepository()
-
-
-@pytest.fixture
-def cosyvoice_service():
-    return MockCosyVoiceService(async_mode=False)  # 同步模式，简化测试
-
-
-@pytest.fixture
-def client(clone_repo, cosyvoice_service):
-    """创建带有依赖覆盖的 TestClient。"""
-    test_app = FastAPI()
-    test_app.include_router(router, prefix="/voice-clones")
-
-    def _override_current_user():
-        return AuthenticatedUser(user=_make_user())
-
-    test_app.dependency_overrides[get_current_user] = _override_current_user
-    test_app.dependency_overrides[get_voice_clone_profile_repository] = lambda: clone_repo
-    test_app.dependency_overrides[get_cosyvoice_service] = lambda: cosyvoice_service
-
-    yield TestClient(test_app)
-
-    test_app.dependency_overrides.clear()
-
-
-# ---------------------------------------------------------------------------
-# 5. POST / — 创建声音克隆
-# ---------------------------------------------------------------------------
-
-
-class TestCreateVoiceClone:
-    """创建声音克隆端点测试。"""
-
-    def test_create_with_source_audio(self, client, cosyvoice_service):
-        """提供源音频时创建克隆，同步模式下直接 ready。"""
-        resp = client.post(
-            "/voice-clones",
-            json={
-                "name": "我的专属音色",
-                "source_audio_url": "https://example.com/voice.wav",
-                "voice_model": "cosyvoice-v2",
-                "language": "zh-CN",
-                "gender": "female",
-            },
-        )
-        assert resp.status_code == 201
-        data = resp.json()
-        assert data["name"] == "我的专属音色"
-        assert data["source_audio_url"] == "https://example.com/voice.wav"
-        assert data["voice_model"] == "cosyvoice-v2"
-        assert data["language"] == "zh-CN"
-        assert data["gender"] == "female"
-        assert "id" in data
-        assert len(data["id"]) > 0
-
-        # 同步模式下应直接 ready
-        assert data["status"] == "ready"
-        assert data["voice_id"] == "mock-voice-789"
-        assert data["error_message"] == ""
-
-    def test_create_without_source_audio(self, client):
-        """不提供源音频时创建，状态为 pending。"""
-        resp = client.post(
-            "/voice-clones",
-            json={
-                "name": "待上传音色",
-                "description": "等待上传音频",
-            },
-        )
-        assert resp.status_code == 201
-        data = resp.json()
-        assert data["name"] == "待上传音色"
-        assert data["status"] == "pending"
-        assert data["source_audio_url"] == ""
-        assert data["voice_id"] == ""
-
-    def test_create_persists_to_repository(self, client, clone_repo):
-        """创建后档案保存到 repository。"""
-        resp = client.post("/voice-clones", json={"name": "持久化测试"})
-        profile_id = resp.json()["id"]
-
-        saved = clone_repo.get(profile_id)
-        assert saved is not None
-        assert saved.name == "持久化测试"
-        assert saved.user_id == "user-test-001"
-
-    def test_create_missing_name_returns_422(self, client):
-        """缺少 name 返回 422。"""
-        resp = client.post("/voice-clones", json={})
-        assert resp.status_code == 422
-
-    def test_create_empty_name_returns_422(self, client):
-        """空 name 返回 422。"""
-        resp = client.post("/voice-clones", json={"name": ""})
-        assert resp.status_code == 422
-
-    def test_create_name_too_long_returns_422(self, client):
-        """名称超长返回 422。"""
-        long_name = "a" * 101
-        resp = client.post("/voice-clones", json={"name": long_name})
-        assert resp.status_code == 422
-
-    def test_create_with_metadata(self, client, cosyvoice_service):
-        """支持自定义 metadata。"""
-        resp = client.post(
-            "/voice-clones",
-            json={
-                "name": "带元数据的克隆",
-                "source_audio_url": "https://example.com/v.wav",
-                "metadata": {"source": "mobile_app", "version": "1.0"},
-            },
-        )
-        assert resp.status_code == 201
-        data = resp.json()
-        assert data["metadata"]["source"] == "mobile_app"
-        assert data["metadata"]["version"] == "1.0"
-
-    def test_create_cosyvoice_failure_returns_failed(self, client, clone_repo, cosyvoice_service):
-        """CosyVoice 提交失败时返回 201 + failed 状态（不抛 500）。"""
-        cosyvoice_service.fail_submit = True
-        cosyvoice_service.async_mode = True  # 异步模式才会调用 submit_clone_task
-
-        resp = client.post(
-            "/voice-clones",
-            json={
-                "name": "会失败的克隆",
-                "source_audio_url": "https://example.com/bad.wav",
-            },
-        )
-        # 不抛 500，返回 201 + failed 状态
-        assert resp.status_code == 201
-        data = resp.json()
-        assert data["status"] == "failed"
-        assert data["error_message"] != ""
-
-
-# ---------------------------------------------------------------------------
-# 6. GET / — 列出声音克隆
-# ---------------------------------------------------------------------------
-
-
-class TestListVoiceClones:
-    """列出声音克隆端点测试。"""
-
-    def test_empty_list(self, client):
-        """无克隆时返回空列表。"""
-        resp = client.get("/voice-clones")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert data["items"] == []
-        assert data["total"] == 0
-
-    def test_list_user_clones(self, client, clone_repo):
-        """只返回当前用户的克隆。"""
-        p1 = _make_clone_profile("音色1", "user-test-001")
-        p2 = _make_clone_profile("音色2", "user-test-001")
-        p3 = _make_clone_profile("他人音色", "other-user")
-        clone_repo.create(p1)
-        clone_repo.create(p2)
-        clone_repo.create(p3)
-
-        resp = client.get("/voice-clones")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert data["total"] == 2
-        assert len(data["items"]) == 2
-        names = {item["name"] for item in data["items"]}
-        assert names == {"音色1", "音色2"}
-
-    def test_filter_by_status(self, client, clone_repo):
-        """按状态筛选。"""
-        ready = _make_clone_profile("已就绪", status=VoiceCloneStatus.READY)
-        failed = _make_clone_profile("已失败", status=VoiceCloneStatus.FAILED)
-        clone_repo.create(ready)
-        clone_repo.create(failed)
-
-        resp = client.get("/voice-clones?status=ready")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert data["total"] == 1
-        assert data["items"][0]["name"] == "已就绪"
-
-    def test_filter_by_failed_status(self, client, clone_repo):
-        """筛选失败状态。"""
-        failed = _make_clone_profile("失败的", status=VoiceCloneStatus.FAILED)
-        ready = _make_clone_profile("成功的", status=VoiceCloneStatus.READY)
-        clone_repo.create(failed)
-        clone_repo.create(ready)
-
-        resp = client.get("/voice-clones?status=failed")
-        assert resp.status_code == 200
-        assert resp.json()["total"] == 1
-        assert resp.json()["items"][0]["name"] == "失败的"
-
-    def test_list_response_fields(self, client, clone_repo):
-        """列表响应包含所有必需字段。"""
-        p = _make_clone_profile("字段测试")
-        clone_repo.create(p)
-
-        resp = client.get("/voice-clones")
-        item = resp.json()["items"][0]
-        for field in [
-            "id",
-            "user_id",
-            "name",
-            "description",
-            "source_audio_url",
-            "voice_id",
-            "voice_model",
-            "language",
-            "gender",
-            "status",
-            "error_message",
-            "retry_count",
-            "max_retries",
-            "created_at",
-            "updated_at",
-        ]:
-            assert field in item, f"缺少字段: {field}"
-
-
-# ---------------------------------------------------------------------------
-# 7. GET /{clone_id} — 获取克隆详情
-# ---------------------------------------------------------------------------
-
-
-class TestGetVoiceClone:
-    """获取克隆详情端点测试。"""
-
-    def test_get_existing_clone(self, client, clone_repo):
-        """获取存在的克隆返回详情。"""
-        p = _make_clone_profile("详情测试", description="这是一段描述")
-        clone_repo.create(p)
-
-        resp = client.get(f"/voice-clones/{p.id}")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert data["id"] == p.id
-        assert data["name"] == "详情测试"
-        assert data["description"] == "这是一段描述"
-
-    def test_get_nonexistent_returns_404(self, client):
-        """获取不存在的克隆返回 404。"""
-        resp = client.get("/voice-clones/nonexistent-clone-id")
-        assert resp.status_code == 404
-        assert "not found" in resp.json()["detail"].lower()
-
-    def test_get_other_user_clone_returns_404(self, client, clone_repo):
-        """获取其他用户的克隆返回 404（安全隔离）。"""
-        p = _make_clone_profile("他人音色", user_id="other-user")
-        clone_repo.create(p)
-
-        resp = client.get(f"/voice-clones/{p.id}")
-        assert resp.status_code == 404
-
-    def test_get_ready_clone_has_voice_id(self, client, clone_repo):
-        """就绪状态的克隆有 voice_id。"""
-        p = _make_clone_profile("就绪音色", status=VoiceCloneStatus.READY)
-        clone_repo.create(p)
-
-        resp = client.get(f"/voice-clones/{p.id}")
-        data = resp.json()
-        assert data["status"] == "ready"
-        assert data["voice_id"] == "voice-ready-001"
-
-    def test_get_failed_clone_has_error_message(self, client, clone_repo):
-        """失败状态的克隆有错误信息。"""
-        p = _make_clone_profile("失败音色", status=VoiceCloneStatus.FAILED)
-        clone_repo.create(p)
-
-        resp = client.get(f"/voice-clones/{p.id}")
-        data = resp.json()
-        assert data["status"] == "failed"
-        assert "模拟失败" in data["error_message"]
-
-
-# ---------------------------------------------------------------------------
-# 8. GET /{clone_id}/status — 获取克隆状态
-# ---------------------------------------------------------------------------
-
-
-class TestGetVoiceCloneStatus:
-    """获取克隆状态端点测试。"""
-
-    def test_status_pending(self, client, clone_repo):
-        """pending 状态。"""
-        p = _make_clone_profile("pending", status=VoiceCloneStatus.PENDING)
-        clone_repo.create(p)
-
-        resp = client.get(f"/voice-clones/{p.id}/status")
-        assert resp.status_code == 200
-        data = resp.json()
-        assert data["id"] == p.id
-        assert data["status"] == "pending"
-        assert data["retry_count"] == 0
-
-    def test_status_ready(self, client, clone_repo):
-        """ready 状态包含 voice_id。"""
-        p = _make_clone_profile("ready", status=VoiceCloneStatus.READY)
-        clone_repo.create(p)
-
-        resp = client.get(f"/voice-clones/{p.id}/status")
-        data = resp.json()
-        assert data["status"] == "ready"
-        assert data["voice_id"] == "voice-ready-001"
-
-    def test_status_failed(self, client, clone_repo):
-        """failed 状态包含错误信息。"""
-        p = _make_clone_profile("failed", status=VoiceCloneStatus.FAILED)
-        clone_repo.create(p)
-
-        resp = client.get(f"/voice-clones/{p.id}/status")
-        data = resp.json()
-        assert data["status"] == "failed"
-        assert data["error_message"] != ""
-        assert data["retry_count"] == 0  # mark_failed 不增加 retry_count，只有重试时才增加
-
-    def test_status_nonexistent_returns_404(self, client):
-        """获取不存在克隆的状态返回 404。"""
-        resp = client.get("/voice-clones/nonexistent/status")
-        assert resp.status_code == 404
-
-
-# ---------------------------------------------------------------------------
-# 9. POST /{clone_id}/retry — 重试克隆
-# ---------------------------------------------------------------------------
-
-
-class TestRetryVoiceClone:
-    """重试克隆端点测试。"""
-
-    def test_retry_failed_clone(self, client, clone_repo, cosyvoice_service):
-        """重试失败的克隆应成功。"""
-        cosyvoice_service.async_mode = False
-        p = _make_clone_profile("重试测试", status=VoiceCloneStatus.FAILED)
-        clone_repo.create(p)
-
-        resp = client.post(f"/voice-clones/{p.id}/retry")
-        assert resp.status_code == 200
-        data = resp.json()
-        # 同步模式下重试后应变为 ready
-        assert data["status"] == "ready"
-        assert data["retry_count"] >= 1
-
-    def test_retry_nonexistent_returns_404(self, client):
-        """重试不存在的克隆返回 404。"""
-        resp = client.post("/voice-clones/nonexistent/retry")
-        assert resp.status_code == 404
-
-    def test_retry_ready_clone_returns_400(self, client, clone_repo):
-        """重试已就绪的克隆返回 400（不可重试）。"""
-        p = _make_clone_profile("已就绪", status=VoiceCloneStatus.READY)
-        clone_repo.create(p)
-
-        resp = client.post(f"/voice-clones/{p.id}/retry")
-        assert resp.status_code == 400
-        assert "retryable" in resp.json()["detail"].lower() or "not" in resp.json()["detail"].lower()
-
-    def test_retry_processing_clone_returns_400(self, client, clone_repo):
-        """重试处理中的克隆返回 400。"""
-        p = _make_clone_profile("处理中", status=VoiceCloneStatus.PROCESSING)
-        clone_repo.create(p)
-
-        resp = client.post(f"/voice-clones/{p.id}/retry")
-        assert resp.status_code == 400
-
-    def test_retry_increments_retry_count(self, client, clone_repo, cosyvoice_service):
-        """重试后重试次数增加。"""
-        cosyvoice_service.async_mode = False
-        p = _make_clone_profile("重试计数", status=VoiceCloneStatus.FAILED)
-        clone_repo.create(p)
-
-        before_count = p.retry_count
-        resp = client.post(f"/voice-clones/{p.id}/retry")
-        after_count = resp.json()["retry_count"]
-
-        assert after_count > before_count
-
-
-# ---------------------------------------------------------------------------
-# 10. DELETE /{clone_id} — 删除克隆
-# ---------------------------------------------------------------------------
-
-
-class TestDeleteVoiceClone:
-    """删除克隆端点测试。"""
-
-    def test_delete_existing_clone(self, client, clone_repo):
-        """删除存在的克隆返回 204。"""
-        p = _make_clone_profile("待删除")
-        clone_repo.create(p)
-
-        resp = client.delete(f"/voice-clones/{p.id}")
-        assert resp.status_code == 204
-
-        # 验证已删除
-        assert clone_repo.get(p.id) is None
-
-    def test_delete_nonexistent_returns_404(self, client):
-        """删除不存在的克隆返回 404。"""
-        resp = client.delete("/voice-clones/nonexistent-clone-id")
-        assert resp.status_code == 404
-
-    def test_delete_other_user_clone_returns_404(self, client, clone_repo):
-        """删除其他用户的克隆返回 404（安全隔离）。"""
-        p = _make_clone_profile("他人音色", user_id="other-user")
-        clone_repo.create(p)
-
-        resp = client.delete(f"/voice-clones/{p.id}")
-        assert resp.status_code == 404
-        # 验证未被删除
-        assert clone_repo.get(p.id) is not None
-
-    def test_delete_idempotent(self, client, clone_repo):
-        """删除后再次删除返回 404。"""
-        p = _make_clone_profile("幂等测试")
-        clone_repo.create(p)
-
-        resp1 = client.delete(f"/voice-clones/{p.id}")
-        assert resp1.status_code == 204
-
-        resp2 = client.delete(f"/voice-clones/{p.id}")
-        assert resp2.status_code == 404
-
-
-# ---------------------------------------------------------------------------
-# 11. 跨端点场景
-# ---------------------------------------------------------------------------
-
-
-class TestVoiceCloneLifecycle:
-    """音色克隆完整生命周期测试。"""
-
-    def test_full_lifecycle_create_list_get_delete(self, client, clone_repo, cosyvoice_service):
-        """创建 → 列表 → 详情 → 删除 完整流程。"""
-        # 1. 创建
-        create_resp = client.post(
-            "/voice-clones",
-            json={
-                "name": "生命周期测试",
-                "source_audio_url": "https://example.com/voice.wav",
-            },
-        )
-        assert create_resp.status_code == 201
-        clone_id = create_resp.json()["id"]
-
-        # 2. 列表
-        list_resp = client.get("/voice-clones")
-        assert list_resp.json()["total"] == 1
-
-        # 3. 详情
-        detail_resp = client.get(f"/voice-clones/{clone_id}")
-        assert detail_resp.status_code == 200
-        assert detail_resp.json()["name"] == "生命周期测试"
-
-        # 4. 状态
-        status_resp = client.get(f"/voice-clones/{clone_id}/status")
-        assert status_resp.status_code == 200
-        assert status_resp.json()["status"] == "ready"
-
-        # 5. 删除
-        del_resp = client.delete(f"/voice-clones/{clone_id}")
-        assert del_resp.status_code == 204
-
-        # 6. 删除后列表为空
-        list_resp2 = client.get("/voice-clones")
-        assert list_resp2.json()["total"] == 0
-
-    def test_failed_retry_flow(self, client, clone_repo, cosyvoice_service):
-        """失败 → 重试 → 成功 流程。"""
-        # 创建一个失败的克隆
-        p = _make_clone_profile("失败重试", status=VoiceCloneStatus.FAILED)
-        clone_repo.create(p)
-
-        # 确认状态
-        status_resp = client.get(f"/voice-clones/{p.id}/status")
-        assert status_resp.json()["status"] == "failed"
-
-        # 重试
-        cosyvoice_service.async_mode = False
-        retry_resp = client.post(f"/voice-clones/{p.id}/retry")
-        assert retry_resp.status_code == 200
-        assert retry_resp.json()["status"] == "ready"
-
-        # 再次确认状态
-        status_resp2 = client.get(f"/voice-clones/{p.id}/status")
-        assert status_resp2.json()["status"] == "ready"
-        assert status_resp2.json()["voice_id"] != ""
-
-
-if __name__ == "__main__":
-    pytest.main([__file__, "-v"])

tests/unit/test_payment_callback.py

@@ -31,7 +31,8 @@ from fastapi.testclient import TestClient
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
 
-from app.api.routes.subscription import _get_plan_name, _get_plan_price, router
+from app.api.routes.subscription import router, _get_plan_name, _get_plan_price
+
 
 # ---------------------------------------------------------------------------
 # 1. Mock Billing Repository
@@ -308,16 +309,11 @@ class TestPaymentCallbackIdempotency:
         """mark_paid 方法对已支付账单返回 False（幂等）。"""
         repo = MockBillingRepository()
 
-        repo.create(
-            dict(
-                id="bill-001",
-                user_id="u1",
-                plan_name="Pro 专业版",
-                amount=299.0,
-                billing_cycle="monthly",
-                status="pending",
-            )
-        )
+        repo.create(dict(
+            id="bill-001", user_id="u1",
+            plan_name="Pro 专业版", amount=299.0,
+            billing_cycle="monthly", status="pending",
+        ))
 
         # 第一次标记为已支付
         result1 = repo.mark_paid("bill-001", "alipay", "pay-001")
@@ -404,9 +400,7 @@ class TestPaymentCallbackValidation:
         resp = client.post(
             "/subscription/payment-callback",
             params={
-                "user_id": "u1",
-                "plan": "pro",
-                "billing_cycle": "monthly",
+                "user_id": "u1", "plan": "pro", "billing_cycle": "monthly",
                 "amount": -100.0,
             },
         )
@@ -462,16 +456,11 @@ class TestMockBillingRepository:
     def test_create_record(self):
         """创建账单记录。"""
         repo = MockBillingRepository()
-        record = repo.create(
-            dict(
-                id="bill-001",
-                user_id="user-001",
-                plan_name="Pro 专业版",
-                amount=299.0,
-                billing_cycle="monthly",
-                status="pending",
-            )
-        )
+        record = repo.create(dict(
+            id="bill-001", user_id="user-001",
+            plan_name="Pro 专业版", amount=299.0,
+            billing_cycle="monthly", status="pending",
+        ))
         assert record.id == "bill-001"
         assert record.status == "pending"
         assert repo.created_count == 1
@@ -479,16 +468,10 @@ class TestMockBillingRepository:
     def test_find_by_id(self):
         """按 ID 查询账单。"""
         repo = MockBillingRepository()
-        repo.create(
-            dict(
-                id="bill-001",
-                user_id="user-1",
-                plan_name="Pro",
-                amount=299,
-                billing_cycle="monthly",
-                status="pending",
-            )
-        )
+        repo.create(dict(
+            id="bill-001", user_id="user-1",
+            plan_name="Pro", amount=299, billing_cycle="monthly", status="pending",
+        ))
 
         found = repo.find_by_id("bill-001")
         assert found is not None
@@ -501,9 +484,7 @@ class TestMockBillingRepository:
         """按用户查询账单。"""
         repo = MockBillingRepository()
         repo.create(dict(id="b1", user_id="u1", plan_name="Pro", amount=299, billing_cycle="monthly", status="pending"))
-        repo.create(
-            dict(id="b2", user_id="u1", plan_name="Standard", amount=99, billing_cycle="monthly", status="pending")
-        )
+        repo.create(dict(id="b2", user_id="u1", plan_name="Standard", amount=99, billing_cycle="monthly", status="pending"))
         repo.create(dict(id="b3", user_id="u2", plan_name="Pro", amount=299, billing_cycle="monthly", status="pending"))
 
         user1_records = repo.find_by_user("u1")
@@ -515,16 +496,10 @@ class TestMockBillingRepository:
     def test_mark_paid_transitions_status(self):
         """mark_paid 正确转换状态。"""
         repo = MockBillingRepository()
-        repo.create(
-            dict(
-                id="bill-001",
-                user_id="u1",
-                plan_name="Pro",
-                amount=299,
-                billing_cycle="monthly",
-                status="pending",
-            )
-        )
+        repo.create(dict(
+            id="bill-001", user_id="u1",
+            plan_name="Pro", amount=299, billing_cycle="monthly", status="pending",
+        ))
 
         result = repo.mark_paid("bill-001", "alipay", "pay-001")
         assert result is True
@@ -538,16 +513,10 @@ class TestMockBillingRepository:
     def test_mark_paid_idempotent(self):
         """mark_paid 对已支付账单幂等。"""
         repo = MockBillingRepository()
-        repo.create(
-            dict(
-                id="bill-001",
-                user_id="u1",
-                plan_name="Pro",
-                amount=299,
-                billing_cycle="monthly",
-                status="pending",
-            )
-        )
+        repo.create(dict(
+            id="bill-001", user_id="u1",
+            plan_name="Pro", amount=299, billing_cycle="monthly", status="pending",
+        ))
 
         repo.mark_paid("bill-001", "alipay", "pay-001")
         paid_at_first = repo.find_by_id("bill-001").paid_at