diff --git a/GOALS.md b/GOALS.md index 493cf46d..30804438 100644 --- a/GOALS.md +++ b/GOALS.md @@ -9,21 +9,22 @@ - **软件交付**:后台、上传网页、本人历史、公开分享与停止分享、真实全链路行为验证和容量/恢复工具均落盘,经各自本地漏斗、正式CI/模型门禁与正常PR合并。 - **生产启用**:软件交付不自动启用;仍需真实容量/四正有限额度、代理、独立恢复控制及完整恢复域、撤销不复活/有效成果保护、实际消费环境证据和用户部署授权。未知保持 off。 - 索引中的 M1–M3「已完成」指已验收并合并的软件里程碑,不冒称生产证明完整。M1 历史未勾项不追填绿:A 曾经 primary failure,经真实反证/正规签票和 READY 后合并,不能改称原 primary success;当前跨进程/门禁与消费者回归由已合 B/C 证据继续覆盖。完整生产证明仍由 M6 承接。 -- M6 当前只是非生产软件准备,工具及演练未实现;其软件前置 M1–M3 与最终验收/生产启用前置 M4–M5 及所有生产证明明确分开,不削减功能或提前标 done。 +- M4 网页软件已随 PR #209 正常合并,M5 的真实全链测试候选 PR #215 等独立验收与正式 CI;不将草稿或作者报告当完成。 +- M6 的非生产工具与合成演练软件已随 PR #208 合并,真实生产容量、完整恢复域、镜像身份与部署授权仍未认证。其软件准备和最终生产启用前置分开,不削减检查清单或提前标生产 done。 | Lane | 激活里程碑 | 推进文件 | | --- | --- | --- | -| local-upload | M6 | [goals/local-upload/M6-recovery.md](goals/local-upload/M6-recovery.md) | +| local-upload | M5 | [goals/local-upload/M5-proof.md](goals/local-upload/M5-proof.md) | | Lane | ID | 名称 | 状态 | 排序 | 优先级 | 依赖 | 进度文件 | | --- | --- | --- | --- | --- | --- | --- | --- | | local-upload | M1 | local-upload/M1:上传身份、分享资格与清理基线 | 已完成 | 1 | 高 | — | [goals/local-upload/M1-baseline.md](goals/local-upload/M1-baseline.md) | | local-upload | M2 | local-upload/M2:认证接收与幂等回执 | 已完成 | 2 | 高 | local-upload/M1 | [goals/local-upload/M2-intake.md](goals/local-upload/M2-intake.md) | | local-upload | M3 | local-upload/M3:本地输入进入既有转录 worker | 已完成 | 3 | 高 | local-upload/M1, local-upload/M2 | [goals/local-upload/M3-worker.md](goals/local-upload/M3-worker.md) | -| local-upload | M4 | local-upload/M4:上传、回执与本人历史网页 | 阻塞 | 4 | 中 | local-upload/M1, local-upload/M2, local-upload/M3 | [goals/local-upload/M4-web.md](goals/local-upload/M4-web.md) | -| local-upload | M5 | local-upload/M5:全链路安全与兼容性证明 | 未开始 | 5 | 高 | local-upload/M1, local-upload/M2, local-upload/M3, local-upload/M4 | [goals/local-upload/M5-proof.md](goals/local-upload/M5-proof.md) | -| local-upload | M6 | local-upload/M6:恢复域、单向撤销与容量恢复 | 进行中 | 6 | 高 | local-upload/M1, local-upload/M2, local-upload/M3 | [goals/local-upload/M6-recovery.md](goals/local-upload/M6-recovery.md) | +| local-upload | M4 | local-upload/M4:上传、回执与本人历史网页 | 已完成 | 4 | 中 | local-upload/M1, local-upload/M2, local-upload/M3 | [goals/local-upload/M4-web.md](goals/local-upload/M4-web.md) | +| local-upload | M5 | local-upload/M5:全链路安全与兼容性证明 | 进行中 | 5 | 高 | local-upload/M1, local-upload/M2, local-upload/M3, local-upload/M4 | [goals/local-upload/M5-proof.md](goals/local-upload/M5-proof.md) | +| local-upload | M6 | local-upload/M6:恢复域、单向撤销与容量恢复 | 阻塞 | 6 | 高 | local-upload/M1, local-upload/M2, local-upload/M3 | [goals/local-upload/M6-recovery.md](goals/local-upload/M6-recovery.md) | ## 路线图审计 @@ -37,3 +38,10 @@ - **新证据是否改变顺序?** 是。后台已合、网页测试登记仍阻塞;将 M6 软件准备依赖限定已合 M1–M3,完整验收/生产启用依旧等 M4/M5 与原真实容量恢复授权。不跨仓改登记、不跳浏览器测试 SOP。 - **done 的定义还成立吗?** 是。软件、行为测试、测试规范/环境证据、生产启用分别判定。真实 producer payload 与有牙断言才是证据,fixture 不代表生产容量或模型质量。 - **审计结论**:按独立路线审计 `docs/sessions/261008-local-upload/roadmap-after-C.md` 调整可逆调度依赖,不更改目标;M6 仅软件准备进行中,其余缺项与授权边界保留。 + +### local-upload — 2026-10-09 / PR209 后续校准 + +- **层0与目标**:上传仍是用户目标,没有新平台或缩减功能;D 网页软件已合,旧 registry 阻塞文字已失效。 +- **完成与下一步**:PR #208 完成非生产工具准备;PR #209 的正式代码主审、Tests、Browser 与正常合并已通过。M5 在真实合并版本上补全链测试,PR #215 仍待独立验收和正式 CI,不是已交付或生产上线。 +- **顺序与遗漏**:M4 前置满足后推进原 E 卡,不新增产品里程碑或再做产品规划;现有 M5/M6 检查清单继续保留。 +- **done 定义**:软件源码正常合并、真实行为证据、生产启用分开判;模型/容量/恢复/代理证据未知继续 off,部署须另授权。 diff --git a/docs/testing/local-upload-contracts.md b/docs/testing/local-upload-contracts.md new file mode 100644 index 00000000..d04909b1 --- /dev/null +++ b/docs/testing/local-upload-contracts.md @@ -0,0 +1,39 @@ +# 本地上传 B1–B16 真实契约索引 + +本表把已批准测试计划中的 B1–B16 对到生产入口、断言和具体测试。测试只替换最后的 +ASR、LLM HTTP 服务与通知接收端;路由、SQLite、文件、队列、CapsWriter SDK、模型 SDK、 +导出和通知发送序列仍使用真实实现。具体浏览器 producer 证据见 +`e2e/tests/local-upload-full-chain.spec.ts`;其输出目录保留浏览器 fetch 边界的 WAV 字节、 +metadata、真实 ffprobe argv、CapsWriter SDK WebSocket frame 和最终通知 HTTP JSON。 + +| 契约 | 真实入口与实现 | 约束断言与测试 | +| --- | --- | --- | +| B1 鉴权、开关、metadata 先于 body | `routes/uploads.py`: `upload_capabilities`, `receive_upload`, `decode_upload_metadata`; `services/transcription.py`: `verify_token` | `test_real_http_stream_sqlite_dispatch_idempotency_and_owner_stop`, `test_disabled_and_missing_limits_reject_before_body`, `test_receiver_and_inflight_limits_reject_without_body_consumption`; body-read 计数为 0。 | +| B2 有界接收与真实容量拒绝 | `receive_upload`; `CacheManager` upload reservation; `_local_upload_write_admission` | `test_actual_free_space_shortage_rejects_before_body_or_file`, `test_declared_and_observed_limits_check_bytes_not_content_length`, `test_queue_full_has_no_success_receipt_or_dispatch`; 断言无 202、无文件、无 dispatcher 调用及释放 reservation。普通下载与生产磁盘峰值未作容量认证。 | +| B3 页面 producer 到 raw bytes | `src/web/static/js/app.js`: `UploadAPI.submitUpload`; 页面测试经真实 `/api/uploads` | `test_real_http_stream_sqlite_dispatch_idempotency_and_owner_stop`; 浏览器全链路测试捕获 `fetch` 实际 body 和 wire headers,比较 producer bytes、已落盘 bytes 与 SHA-256,并把 SQLite `local_uploads.request_metadata` 与 `task_status.processing_options` 规范化 JSON 后对照同一份 decoded wire 六字段(filename/byte_size/title/source_url/retention/processing_options)。真实 producer 为 `e2e/tests/local-upload-full-chain.spec.ts`,不使用合成 browser server。 | +| B4 同意图只建一次 | `CacheManager.register_local_upload`, `accept_local_upload` | `test_concurrent_same_owner_key_creates_one_sqlite_receipt` 用两个真实 SQLite connection,在 `BEGIN IMMEDIATE` / 查询之间控序并分别让 request-a、request-b 先赢;HTTP 同 key、metadata conflict、distinct key 由 `test_real_http_stream_sqlite_dispatch_idempotency_and_owner_stop` 断言。 | +| B5 durable admission 与 stale queue | `routes/uploads.py`: `receive_upload`; `services/transcription.py`: `process_task_queue`; `CacheManager.get_admitted_local_upload_by_task` | `test_queue_full_has_no_success_receipt_or_dispatch`, `test_http_queue_handoff_precedes_durable_acceptance_commit`, `test_sql_commit_failure_leaves_stale_queue_item_for_dispatcher_to_drop`, `test_lost_202_receipt_retry_queries_original_without_requeue`, `test_dispatcher_fails_stale_local_mapping_before_processing_or_executor_submit`。 | +| B6 文件路径与媒体准入 | `receive_upload`; `services/transcription.py`: `_ensure_audio_track`, `_probe_media_duration`, `_local_upload_write_admission` | `test_local_media_admission_rejects_before_asr_and_never_uses_url_downloader`; 浏览器全链路检查真实 ffprobe shim 记录的 argv 指向 SQLite 映射的 owned file,且实际 `/usr/bin/ffprobe` 完成探测。 | +| B7 共享 ASR 与源文件清理 | `services/transcription.py`: `process_task_queue`, `process_transcription`; CapsWriter SDK client | `test_dispatcher_uses_real_transcriber_then_cleans_owned_media_before_llm`, `test_local_upload_funasr_branch_observes_source_only_before_llm`; 浏览器全链路消费真实 SDK WebSocket 帧并断言转录后、LLM 前源文件已清理。 | +| B8 root 固定保留期 | `CacheManager.update_task_status`, `local_upload_share_is_active`; 上传启动孤儿恢复 | `test_root_terminal_sets_fixed_30_day_expiry_for_success_and_failure`, `test_startup_orphan_failure_anchors_upload_terminal_clock_once`, `test_never_expiry_and_revocation_are_write_once`;覆盖 success/failed 与 never。 | +| B9 单一 public read gate | `ViewTokenResolver.get_view_data_by_token`; `routes/views.py`: `view_transcript`, `handle_raw_export`, `handle_page_export`, `export_content`; summary route | `test_public_read_summary_exports_and_owner_history_close_without_content_bypass` 逐个请求 default、raw/page 的 calibrated、summary、notes、transcript、全部 export 及 summary;撤销后逐路断言 404 且正文不泄漏。`test_upload_task_uses_blank_legacy_column_and_separate_upload_token` 锁住空 token 不回退到 legacy task。 | +| B10 三个 owner reprocess caller | `routes/tasks.py`: `_admit_upload_reprocess`, `recalibrate`, `resummarize`, `generate_notes`; `CacheManager.get_active_local_upload_for_reprocess` | `test_reprocess_admission_holds_sqlite_order_against_concurrent_close` 与 `test_reprocess_snapshot_before_close_is_rejected_by_fresh_transaction` 参数化覆盖三个 endpoint 的 admission-wins / revoke-wins;断言 child blank legacy token、真实 child queue 或拒绝且零 child。 | +| B11 terminal publish 与撤销单调 | `CacheManager.update_task_status`, `revoke_local_upload`; terminal snapshot / notification consumers | `test_terminal_publish_and_revoke_are_both_write_once` 控制 publish-before-revoke 与 revoke-before-publish,断言截止值、撤销时间保持、正文仍留存但 fresh public reader 拒绝;`test_success_snapshot_reaches_final_http_payloads_in_order` 锁定通知顺序。 | +| B12 清理与仍有效成果 | `CacheManager.cleanup_old_cache`, `cleanup_task_status`, `cleanup_expired_local_upload_receiving`; temp source cleanup | `test_active_upload_cleanup_and_cache_publish_preserve_body_in_both_orders` 对真实 `save_cache` 与 cache cleaner 双顺序控锁;`test_cleanup_protects_active_upload_cache_even_when_feature_is_disabled`, `test_task_cleanup_keeps_effective_share_root_for_audit_and_url_rows_still_expire`, `test_revoke_and_task_cleanup_are_ordered_at_the_sqlite_guard`, `test_receiving_cleanup_rechecks_after_accept_wins_and_preserves_owned_file`, `test_receiving_cleanup_retires_before_unlink_and_accept_loses` 锁住独立清理边界。生产 restore 域/旧镜像未验证。 | +| B13 有限 key tombstone | `CacheManager.cleanup_expired_local_upload_receiving`, `register_local_upload` | `test_real_sqlite_upload_identity_owner_key_and_intent_window` 删除过期记录后再次提交旧 key 必须 24-hour 拒绝;`test_receiving_cleanup_retires_before_unlink_and_accept_loses` 断言 retired key 不会再受理。 | +| B14 真实通知 payload | `services/transcription.py` terminal producer; `utils/notifications/{router,channel,task_formatter}.py` | `test_accepted_upload_link_reaches_final_http_payload_with_public_capability`, `test_local_upload_terminal_producer_sends_public_share_url_and_source_label`; 浏览器全链路经真实通知 client 发出 HTTP JSON,断言完整 payload、链接、无磁盘路径,之后由真实 public reader验证撤销拒读。 | +| B15 页面交互与实际 producer | `src/web/static/{index.html,history.html}`、`static/js/app.js` | `e2e/tests/local-upload.spec.ts` 保留原页面回归;`e2e/tests/local-upload-full-chain.spec.ts` 用 Chromium 文件选择、metadata/header、实际 fetch body、页面历史和 owner DELETE 完成端到端断言。page `requestfailed` 与 HTTP ≥400 一律收集;非预期项必须为空。历史页换代只声明 `GET /api/audit/filter-options` 与 `GET /api/audit/history` 的 `net::ERR_ABORTED`。刻意撤销后的 raw GET 走 request API context,不属于 page 流。 | +| B16 最低镜像与 restore 安全 | `VTA_UPLOADS_ENABLED` 读取、public resolver、cleanup policy | `test_resolver_reads_active_upload_and_denies_disabled_revoked_or_expired`, `test_cleanup_protects_active_upload_cache_even_when_feature_is_disabled` 锁住关闭开关不泄漏且不误删。实际恢复域与 guard-only 最低镜像不在本地测试环境,保持未认证;不把此处测试当作 restore 或上线证据。 | + +## 竞态矩阵 + +| 竞态 | 受控顺序与 fresh consumer | +| --- | --- | +| 同 key 两事务 | `test_concurrent_same_owner_key_creates_one_sqlite_receipt` 分别让 request-a 和 request-b 先取得 SQLite 写事务。 | +| publish / revoke | `test_terminal_publish_and_revoke_are_both_write_once` 分别先提交 terminal publish、先提交 revoke;随后重新走 `/view/{token}`。 | +| cleanup / publish | `test_active_upload_cleanup_and_cache_publish_preserve_body_in_both_orders` 在真实 cache lock 和候选快照处暂停,确认最终数据库映射与文件正文仍存在。 | +| 三个 reprocess / revoke | lifecycle 两组参数化测试覆盖 `/api/recalibrate`、`/api/resummarize`、`/api/generate_notes` 的两种顺序;分别断言已接受 child 可完成和撤销先提交时零 child。 | +| read / expiry | `test_public_read_and_expiry_both_controlled_orders` 用同一测试时钟:首 GET 在截止前真实授权后暂停,主线程把时钟推进到截止时刻并确认在途请求尚未完成,再释放首响应(200/正文);随后同时钟 fresh GET 为 404 且无正文。不改 `expires_at` 写入,不用 sleep。 | + +以上 fixture 与状态检查都使用临时目录、独立 SQLite 和 loopback 服务;没有真实凭据、生产配置、 +生产 ASR、外发通知或真实恢复域。B16 的外部镜像/恢复证明、生产容量和真实模型质量仍是独立未验证义务。 diff --git a/e2e/tests/local-upload-full-chain.spec.ts b/e2e/tests/local-upload-full-chain.spec.ts new file mode 100644 index 00000000..8757d6fb --- /dev/null +++ b/e2e/tests/local-upload-full-chain.spec.ts @@ -0,0 +1,394 @@ +import { createHash } from 'node:crypto'; +import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process'; +import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join, resolve } from 'node:path'; + +import { expect, test, type APIRequestContext, type Page } from '@playwright/test'; + +const REPO_ROOT = resolve(process.cwd(), '..'); +const OWNER_TOKEN = 'browser-fixture-token'; +const ASR_TEXT = '这段转录由真实 CapsWriter SDK 从浏览器上传的 WAV 文件经 ffmpeg 转码后生成。本地上传会写入独立 SQLite 记录,经过实际任务队列处理,并通过公开只读链接提供正文。这个受控 loopback 服务只替代外部识别、模型和通知接收端,产品路由与序列化仍真实运行。'; +const LLM_TEXT = `${ASR_TEXT}共享模型客户端已读取完整转录正文并完成校对;这段追加说明用于确认真实 API、SDK 输出与模型 HTTP 请求之间的内容链路完整。`; + +type RealServer = { api_url: string; upstream_url: string; run_dir: string }; + +let realServer: RealServer; +let serverProcess: ChildProcessWithoutNullStreams; +let serverOutput = ''; +let serverScratch: string; + +test.describe.configure({ timeout: 180_000 }); + +function makeWav(): Buffer { + const sampleRate = 16_000; + const durationSeconds = 3; + const sampleCount = sampleRate * durationSeconds; + const bytes = Buffer.alloc(44 + sampleCount * 2); + bytes.write('RIFF', 0); + bytes.writeUInt32LE(bytes.length - 8, 4); + bytes.write('WAVE', 8); + bytes.write('fmt ', 12); + bytes.writeUInt32LE(16, 16); + bytes.writeUInt16LE(1, 20); + bytes.writeUInt16LE(1, 22); + bytes.writeUInt32LE(sampleRate, 24); + bytes.writeUInt32LE(sampleRate * 2, 28); + bytes.writeUInt16LE(2, 32); + bytes.writeUInt16LE(16, 34); + bytes.write('data', 36); + bytes.writeUInt32LE(sampleCount * 2, 40); + for (let index = 0; index < sampleCount; index += 1) { + const sample = Math.round(Math.sin((2 * Math.PI * 440 * index) / sampleRate) * 8000); + bytes.writeInt16LE(sample, 44 + index * 2); + } + return bytes; +} + +function startRealServer(): Promise { + return new Promise((resolveReady, rejectReady) => { + const python = join(REPO_ROOT, '.venv', 'bin', 'python'); + const spawnEnv = { + PATH: process.env.PATH ?? '/usr/bin:/bin', + HOME: process.env.HOME ?? '/tmp', + TMPDIR: '/tmp', + LANG: 'C.UTF-8', + PYTHONUTF8: '1', + PYTHONUNBUFFERED: '1', + }; + let lineBuffer = ''; + let settled = false; + const startupTimer = setTimeout(() => { + if (settled) return; + settled = true; + rejectReady(new Error(`Real API did not become ready. Output:\n${serverOutput}`)); + }, 45_000); + + serverProcess = spawn( + python, + ['tests/local_upload_real_server.py', '--run-dir', serverScratch], + { cwd: REPO_ROOT, env: spawnEnv, stdio: ['ignore', 'pipe', 'pipe'] }, + ); + serverProcess.stdout.on('data', (chunk: Buffer) => { + const text = chunk.toString('utf-8'); + serverOutput += text; + lineBuffer += text; + const lines = lineBuffer.split('\n'); + lineBuffer = lines.pop() ?? ''; + for (const line of lines) { + const marker = 'VTA_REAL_SERVER_READY '; + const index = line.indexOf(marker); + if (index < 0 || settled) continue; + try { + const ready = JSON.parse(line.slice(index + marker.length)) as RealServer; + settled = true; + clearTimeout(startupTimer); + resolveReady(ready); + } catch (error) { + settled = true; + clearTimeout(startupTimer); + rejectReady(error); + } + } + }); + serverProcess.stderr.on('data', (chunk: Buffer) => { + serverOutput += chunk.toString('utf-8'); + }); + serverProcess.once('error', (error) => { + if (settled) return; + settled = true; + clearTimeout(startupTimer); + rejectReady(error); + }); + serverProcess.once('exit', (code, signal) => { + if (settled) return; + settled = true; + clearTimeout(startupTimer); + rejectReady(new Error(`Real API exited before ready: code=${code} signal=${signal}\n${serverOutput}`)); + }); + }); +} + +async function jsonResponse(request: APIRequestContext, url: string, headers?: Record): Promise { + const response = await request.get(url, { headers }); + expect(response.status(), await response.text()).toBe(200); + return (await response.json()) as T; +} + +async function loopbackEvents(request: APIRequestContext): Promise { + return jsonResponse(request, `${realServer.upstream_url}/__e2e__/events`); +} + +async function inspectRealState(request: APIRequestContext): Promise { + return jsonResponse(request, `${realServer.api_url}/__e2e__/state`); +} + +async function selectUploadPage(page: Page): Promise { + await page.goto(`${realServer.api_url}/add_task_by_web`); + await page.locator('#advanced-toggle').click(); + await page.locator('#bearer-token').fill(OWNER_TOKEN); + await page.locator('#mode-tab-upload').click(); + await expect(page.locator('#upload-mode-panel')).toBeVisible(); +} + +test.beforeAll(async () => { + serverScratch = await mkdtemp(join(tmpdir(), 'vta-upload-real-chain-')); + realServer = await startRealServer(); + const ready = await fetch(`${realServer.api_url}/livez`); + expect(ready.status).toBe(200); +}); + +test.afterAll(async () => { + if (realServer) { + const release = await fetch(`${realServer.upstream_url}/__e2e__/release-asr`, { method: 'POST' }); + expect(release.status).toBe(200); + await expect.poll(async () => { + const response = await fetch(`${realServer.api_url}/__e2e__/state`); + if (response.status !== 200) return `http-${response.status}`; + const state = await response.json(); + if (!state.uploads.length) return 'idle'; + return state.uploads[0].root_status; + }, { timeout: 60_000 }).toMatch(/^(idle|success|failed)$/); + } + if (serverProcess?.exitCode === null && serverProcess?.signalCode === null) { + await new Promise((resolveExit, rejectExit) => { + const timeout = setTimeout(() => { + rejectExit(new Error(`Real API did not stop cleanly. Output:\n${serverOutput}`)); + }, 20_000); + serverProcess.once('exit', () => { + clearTimeout(timeout); + resolveExit(); + }); + serverProcess.kill('SIGTERM'); + }); + } + if (serverScratch) await rm(serverScratch, { recursive: true, force: true }); +}); + +test('Chromium uploads raw media through production API, worker, shared ASR/LLM, history, read, notification, and owner revoke', async ({ page, request }, testInfo) => { + const browserErrors: string[] = []; + const unexpectedNetwork: string[] = []; + let observedResponseCount = 0; + page.on('pageerror', (error) => browserErrors.push(error.message)); + page.on('console', (message) => { + if (message.type() === 'error') browserErrors.push(message.text()); + }); + page.on('requestfailed', (failed) => { + const path = new URL(failed.url()).pathname; + unexpectedNetwork.push(`${failed.method()} ${path} ${failed.failure()?.errorText ?? 'failed'}`); + }); + page.on('response', (response) => { + observedResponseCount += 1; + if (response.status() < 400) return; + const path = new URL(response.url()).pathname; + unexpectedNetwork.push(`${response.request().method()} ${path} ${response.status()}`); + }); + + const wavBytes = makeWav(); + const expectedDigest = createHash('sha256').update(wavBytes).digest('hex'); + await page.addInitScript(() => { + const originalFetch = window.fetch; + (window as any).__vtaUploadProducerBytes = null; + window.fetch = async (input: RequestInfo | URL, init?: RequestInit): Promise => { + const url = input instanceof Request ? input.url : String(input); + if (url.endsWith('/api/uploads') && init?.method === 'POST' && init.body instanceof Blob) { + (window as any).__vtaUploadProducerBytes = Array.from( + new Uint8Array(await init.body.arrayBuffer()), + ); + } + return originalFetch.call(window, input, init); + }; + }); + await selectUploadPage(page); + await page.locator('#upload-title').fill('全链路浏览器上传样本'); + await page.locator('#upload-source-url').fill('https://example.test/source/interview'); + await page.locator('#transcription-options-toggle').click(); + await page.locator('#summarize-option').uncheck(); + await page.locator('#upload-file-input').setInputFiles({ + name: '浏览器全链路音频.wav', + mimeType: 'audio/wav', + buffer: wavBytes, + }); + + const uploadRequestPromise = page.waitForRequest((item) => + item.method() === 'POST' && new URL(item.url()).pathname === '/api/uploads'); + const uploadResponsePromise = page.waitForResponse((item) => + item.request().method() === 'POST' && new URL(item.url()).pathname === '/api/uploads'); + await page.locator('#submit-btn').click(); + const [uploadRequest, uploadResponse] = await Promise.all([uploadRequestPromise, uploadResponsePromise]); + expect(uploadResponse.status(), await uploadResponse.text()).toBe(202); + const receipt = await uploadResponse.json(); + const requestHeaders = uploadRequest.headers(); + const producerBytes = Buffer.from(await page.evaluate( + () => (window as any).__vtaUploadProducerBytes as number[], + )); + expect(producerBytes).toEqual(wavBytes); + await writeFile(testInfo.outputPath('browser-producer-upload.wav'), producerBytes); + expect(requestHeaders.authorization).toBe(`Bearer ${OWNER_TOKEN}`); + expect(requestHeaders['content-type']).toBe('application/octet-stream'); + expect(requestHeaders['idempotency-key']).toMatch(/^\d+-[0-9a-f-]+$/i); + const metadata = JSON.parse(Buffer.from(requestHeaders['x-upload-metadata'], 'base64url').toString('utf-8')); + expect(metadata).toEqual({ + filename: '浏览器全链路音频.wav', + byte_size: wavBytes.length, + title: '全链路浏览器上传样本', + source_url: 'https://example.test/source/interview', + retention: '30d', + processing_options: { + calibrate: true, + summarize: false, + infer_speaker_names: false, + chapters: false, + }, + }); + await writeFile(testInfo.outputPath('browser-producer-metadata.json'), JSON.stringify(metadata, null, 2)); + expect(receipt.state).toBe('accepted'); + expect(receipt.upload_id).toBeTruthy(); + expect(receipt.task_id).toBeTruthy(); + expect(receipt.view_token).toMatch(/^upload_/); + await expect(page.locator('#status-content')).toContainText('文件上传成功,任务已受理!'); + + await expect.poll(async () => (await loopbackEvents(request)).asr_started, { timeout: 20_000 }).toBe(true); + const beforeAsrReply = await inspectRealState(request); + expect(beforeAsrReply.uploads).toHaveLength(1); + const accepted = beforeAsrReply.uploads[0]; + expect(accepted.upload_id).toBe(receipt.upload_id); + expect(accepted.owner_user_id).toBe('browser_owner'); + expect(accepted.idempotency_key).toBe(requestHeaders['idempotency-key']); + expect(accepted.state).toBe('accepted'); + expect(accepted.root_task_id).toBe(receipt.task_id); + expect(accepted.view_token).toBe(receipt.view_token); + expect(accepted.root_status).toBe('processing'); + expect(accepted.task_platform).toBe('local_upload'); + expect(accepted.legacy_task_token).toBe(''); + expect(accepted.media_exists).toBe(true); + expect(accepted.media_path).toContain('/upload-source.bin'); + expect(accepted.media_path).not.toContain(accepted.filename); + expect(accepted.byte_size).toBe(wavBytes.length); + expect(accepted.sha256).toBe(expectedDigest); + expect(accepted.media_sha256).toBe(expectedDigest); + expect(accepted.request_metadata).toEqual(metadata); + expect(accepted.task_processing_options).toEqual(metadata.processing_options); + const storedBytes = await readFile(accepted.media_path); + expect(storedBytes).toEqual(producerBytes); + const ffprobeCall = beforeAsrReply.ffprobe_calls.find((call: any) => call.argv.at(-1) === accepted.media_path); + expect(ffprobeCall).toBeTruthy(); + expect(ffprobeCall.argv).toContain('-show_format'); + expect(ffprobeCall.argv).toContain('-show_streams'); + expect(ffprobeCall.argv).toContain('-v'); + await writeFile(testInfo.outputPath('real-ffprobe-argv.json'), JSON.stringify(ffprobeCall, null, 2)); + + const release = await request.post(`${realServer.upstream_url}/__e2e__/release-asr`); + expect(release.status()).toBe(200); + await expect.poll(async () => { + const state = await inspectRealState(request); + return state.uploads[0].root_status; + }, { timeout: 60_000 }).toBe('success'); + + const finished = await inspectRealState(request); + const completed = finished.uploads[0]; + expect(completed.root_status).toBe('success'); + expect(completed.expires_at).toBeTruthy(); + expect(completed.legacy_task_token).toBe(''); + expect(completed.media_exists).toBe(false); + expect(completed.sha256).toBe(expectedDigest); + + const events = await loopbackEvents(request); + expect(events.asr_payload_bytes).toBeGreaterThan(0); + expect(events.asr_reply).toMatchObject({ + task_id: expect.any(String), + text_accu: ASR_TEXT, + source: 'file', + encoding: 'flac', + samples_total: expect.any(Number), + }); + const sdkFrames = events.asr_frames.map((raw: string) => JSON.parse(raw)); + expect(sdkFrames).toHaveLength(1); + expect(sdkFrames[0]).toMatchObject({ + task_id: events.asr_reply.task_id, + source: 'file', + encoding: 'flac', + is_final: true, + }); + expect(Buffer.from(sdkFrames[0].data, 'base64').length).toBeGreaterThan(0); + await writeFile(testInfo.outputPath('capswriter-sdk-frame.json'), events.asr_frames[0]); + expect(events.llm_requests.length).toBeGreaterThan(0); + expect(events.llm_requests.some((entry: any) => + JSON.stringify(entry.messages).includes(ASR_TEXT))).toBe(true); + + const authHeaders = { Authorization: `Bearer ${OWNER_TOKEN}` }; + const history = await jsonResponse( + request, + `${realServer.api_url}/api/audit/history?source=upload&status=all&limit=1&offset=0`, + authHeaders, + ); + expect(history.data.total).toBe(1); + expect(history.data.items[0]).toMatchObject({ + source: 'upload', + upload_id: receipt.upload_id, + title: '全链路浏览器上传样本', + share_active: true, + view_token: null, + }); + + const publicRaw = await request.get(`${realServer.api_url}/view/${receipt.view_token}?raw=transcript`); + expect(publicRaw.status()).toBe(200); + const rawBody = await publicRaw.text(); + expect(rawBody).toContain(ASR_TEXT); + const publicCalibrated = await request.get(`${realServer.api_url}/view/${receipt.view_token}?raw=calibrated`); + expect(publicCalibrated.status()).toBe(200); + const calibratedBody = await publicCalibrated.text(); + expect(calibratedBody).toContain(LLM_TEXT); + await page.goto(`${realServer.api_url}/view/${receipt.view_token}`); + await expect(page.locator('body')).toContainText('全链路浏览器上传样本'); + await expect(page.locator('body')).toContainText(LLM_TEXT); + + await expect.poll(async () => { + const notifications = (await loopbackEvents(request)).notifications; + return notifications.some((entry: any) => JSON.stringify(entry.payload).includes(`/view/${receipt.view_token}`)); + }, { timeout: 20_000 }).toBe(true); + const notification = (await loopbackEvents(request)).notifications.find((entry: any) => + JSON.stringify(entry.payload).includes(`/view/${receipt.view_token}`)); + expect(notification.content_type).toContain('application/json'); + expect(notification.payload.msg_type).toBe('interactive'); + expect(notification.raw_sha256).toMatch(/^[0-9a-f]{64}$/); + expect(JSON.stringify(notification.payload)).toContain('全链路浏览器上传样本'); + expect(JSON.stringify(notification.payload)).not.toContain('upload-source.bin'); + await writeFile(testInfo.outputPath('actual-notification-http-json.json'), JSON.stringify(notification.payload, null, 2)); + + await page.goto(`${realServer.api_url}/static/history.html`); + await page.locator('#apiKeyInput').fill(OWNER_TOKEN); + await page.locator('#historyQueryBtn').click(); + await page.locator('#filterSource').selectOption('upload'); + const row = page.locator('.task-row').first(); + await expect(row.locator('.task-title')).toContainText('全链路浏览器上传样本'); + const stopShare = row.locator('.btn-stop-share'); + await expect(stopShare).toBeVisible(); + page.once('dialog', async (dialog) => dialog.accept()); + const deleteRequest = page.waitForRequest((item) => + item.method() === 'DELETE' && new URL(item.url()).pathname === `/api/uploads/${receipt.upload_id}/share`); + const deleteResponse = page.waitForResponse((item) => + item.request().method() === 'DELETE' && new URL(item.url()).pathname === `/api/uploads/${receipt.upload_id}/share`); + await stopShare.click(); + const [ownerDelete, deleteResult] = await Promise.all([deleteRequest, deleteResponse]); + expect(ownerDelete.headers().authorization).toBe(`Bearer ${OWNER_TOKEN}`); + expect(deleteResult.status()).toBe(200); + expect((await deleteResult.json()).share_active).toBe(false); + await expect(page.locator('#authStatus')).toContainText('已停止公开分享'); + + const deniedFreshRead = await request.get(`${realServer.api_url}/view/${receipt.view_token}?raw=transcript`); + expect(deniedFreshRead.status()).toBe(404); + expect(await deniedFreshRead.text()).not.toContain(ASR_TEXT); + const revoked = (await inspectRealState(request)).uploads[0]; + expect(revoked.revoked_at).toBeTruthy(); + expect(revoked.root_status).toBe('success'); + expect(browserErrors).toEqual([]); + const expectedPageAborts = [ + 'GET /api/audit/filter-options net::ERR_ABORTED', + 'GET /api/audit/history net::ERR_ABORTED', + ]; + expect(observedResponseCount).toBeGreaterThan(0); + expect( + unexpectedNetwork.filter((entry) => !expectedPageAborts.includes(entry)), + ).toEqual([]); +}); diff --git a/goals/local-upload/M4-web.md b/goals/local-upload/M4-web.md index 99d9f326..b59f3630 100644 --- a/goals/local-upload/M4-web.md +++ b/goals/local-upload/M4-web.md @@ -2,18 +2,18 @@ lane: local-upload id: M4 slug: web -status: 阻塞 +status: 已完成 owner: lead order: 4 priority: 中 depends_on: [local-upload/M1, local-upload/M2, local-upload/M3] -merged_pr: null +merged_pr: 209 --- # 里程碑进度:local-upload/M4:上传、回执与本人历史网页 - **预期产出**:原生单文件 picker/drop、metadata headers、期限选择、真实 receipt/处理中/结果状态与本人 history source filter。 -- **当前范围**:D 卡;不得从 mock-only fixture 发明 wire 形状,不改旧 URL 任务页面规则。D 尚未开始:Gate Hub #1393 仍 OPEN,真实 UI registry 尚无 `has_ui: true`;SOP 要求先登记再派 UI,当前不得跨仓代改。U 浏览器底座 PR #200 已合并不等于 M4 网页完成。 +- **当前范围**:D 网页软件已随 PR #209 正常合并;UI registry 的 `has_ui: true` 已登记,不再阻塞。正式 CI run#37941995979 的代码主审与全量 Tests 实际成功,Browser regression run#37941995092 成功;网页、本人历史和停止分享已交付源码。不从 synthetic fixture 发明 wire;真实 API 生命周期与浏览器 producer 分层验证,完整 Browser→FastAPI→SDK 链由 M5 验收,生产未部署。下列原推进证据保留原口径,未确认项不追填绿。 - **关键决策**:沿用设计文件中的 HTTP 字段;默认30天、长期显式选择 never;进度继续走既有 task API;source URL 仅展示,不 fetch。 - **推进前必须拿到的证据**: - [ ] 浏览器/harness 调用真实 API,断言 producer 发出的 raw Blob 字节、headers 和完整服务端 receipt;保留实际请求 fixture。 @@ -21,3 +21,4 @@ merged_pr: null - [ ] 真实 history query 先按 source 过滤再分页;断言实际 ID 集、total、页边界与 owner 隔离;未实现 bodygate 时无摘要旁路。 - [ ] filename/source 输入经真实模板安全渲染;恶意路径、Unicode、引号、无效 URL、键盘和小屏测试。CI browser/API 环境运行;systemd验证仅随另行授权部署。 - **完成条件**:浏览器状态只反映服务端 receipt/进度事实;失效 token 仍由服务端拒绝,文案不许承诺回收通知副本。 +- **验收边界**:用户仅对 PR #209 的本地末审启动输入与测试预检缺 Node 依赖两项前置授权正式 CI 替代,旧失败保留;正式代码主审、Tests、Browser、READY 与精确匹配合并均实际通过。此有限豁免不覆盖 M5、生产授权或容量/恢复证据。 diff --git a/goals/local-upload/M5-proof.md b/goals/local-upload/M5-proof.md index 0b3e1789..6d26462f 100644 --- a/goals/local-upload/M5-proof.md +++ b/goals/local-upload/M5-proof.md @@ -2,7 +2,7 @@ lane: local-upload id: M5 slug: proof -status: 未开始 +status: 进行中 owner: lead order: 5 priority: 高 @@ -13,12 +13,15 @@ merged_pr: null # 里程碑进度:local-upload/M5:全链路安全与兼容性证明 - **预期产出**:从真实 HTTP producer 到 owned bytes、task/queue、既有处理链、Resolver、公正文、本人 history 与 cleanup 的端到端证据。 -- **当前范围**:E 卡,依赖 M4 完整网页行为与真实 API producer;M4 尚受 UI registry 登记阻塞,因此 M5 尚未开始。整理方案、helper 单测或邻侧 response 不算实际消费证明。 +- **当前范围**:E 卡已基于 PR #209 的真实合并产物实施,PR #215 测试增量已交回,本地完整独立复演已完成,等待正式 CI/合并。M4 网页源码与 UI registry 前置已满足;新增测试连接真实网页、FastAPI、SQLite、dispatcher 和转录/模型 SDK,仅最后外部服务使用 loopback。下列清单仍逐项验收,不将作者报告、草稿绿、helper 单测或邻侧 response 当完整通过。 - **关键决策**:每项不变式必须指向真实生产代码与能被 mutation 打红的测试;跨文件/进程边界保存 producer 实际 payload。 - **推进前必须拿到的证据**: - - [ ] 真 FastAPI、临时 SQLite/文件系统、真实 dispatcher、Resolver 与现有 consumer;断言 HTTP/body、subprocess argv/env、写入文件等实际 producer 输出。 - - [ ] 同意图恰好一个受理,两次明确上传恰好两个独立成果;短期、never、expiry、revoke、disabled 全有 active positive control。 - - [ ] controlled pause 证明撤销/迟到发布与清理/发布两顺序;每种均由 fresh consumer 检查关闭不复活和有效成果不丢。 + - [x] 真 FastAPI、临时 SQLite/文件系统、真实 dispatcher、Resolver 与现有 consumer;断言 HTTP/body、subprocess argv/env、写入文件等实际 producer 输出。 + - [x] 同意图恰好一个受理,两次明确上传恰好两个独立成果;短期、never、expiry、revoke、disabled 全有 active positive control。 + - [x] controlled pause 证明撤销/迟到发布与清理/发布两顺序;每种均由 fresh consumer 检查关闭不复活和有效成果不丢。 - [ ] URL dedup/read/history/notification 与全局 cleanup 整文件回归;`make test` 通过。CI只看 run/job `conclusion`,skip 不算通过。 - [ ] CI、裸 shell、实际 systemd 消费环境分别跑环境相关测试;systemd 环境必须是本地 consumer 实际运行的 unit。 - **完成条件**:任何“已覆盖”都有执行证据;矛盾/未知机制重新开受影响设计,不以 fixture 命名掩盖差异。 +- **边界**:真实 ASR/LLM 模型质量、生产容量/反代/完整恢复域及生产镜像身份不由本地 loopback 测试认证,继续由 M6 和用户部署授权约束。 + +- **本地证据边界**:基于测试提交 `b50c861ef276248db087466992d549cc8edc90fe`,真实 Chromium→FastAPI→SQLite→共享 SDK→正文/history/通知/revoke 与五组控序已在隔离裸 shell 和实际临时 systemd 消费环境验证,完整 pytest 入口、网页 186 项、浏览器 17 项通过。正式 CI 尚未运行,里程碑仍进行中,不将本地结果视为模型质量或生产验收。 diff --git a/goals/local-upload/M6-recovery.md b/goals/local-upload/M6-recovery.md index d141febc..d4ad5668 100644 --- a/goals/local-upload/M6-recovery.md +++ b/goals/local-upload/M6-recovery.md @@ -2,7 +2,7 @@ lane: local-upload id: M6 slug: recovery -status: 进行中 +status: 阻塞 owner: lead order: 6 priority: 高 @@ -13,9 +13,9 @@ merged_pr: null # 里程碑进度:local-upload/M6:恢复域、单向撤销与容量恢复 - **预期产出**:证明独立恢复开关、cache.db/正文/audit/share记录完整恢复与容量阈值;未证明前上传读写关闭。 -- **当前范围**:F 卡的非生产软件工具准备可先行;A/C 实现已合并,F 原计划允许此时做工具与临时 fixture,但不做生产服务操作。此 frontmatter 依赖仅约束软件准备,不替代完整验收/生产启用门。 +- **当前范围**:F 的非生产工具和合成演练已随 PR #208 正常合并;生产阶段等待 M5 真实行为验收及下列环境证明和授权。此 frontmatter 依赖仅约束软件准备,不替代完整验收/生产启用门。 - **关键决策**:`VTA_UPLOADS_ENABLED` 在数据卷外;可信撤销状态/完整恢复域不能证明时保持关闭;长期成果不能通过删除腾容量。 -- **当前阶段边界**:`进行中` 仅表示规划已拆分并允许软件准备先行;恢复脚本/工具与临时 fixture 尚未实现,不能写作 passed。 +- **当前阶段边界**:`阻塞` 表示剩余生产启用证据与授权尚不齐,不是软件工具未实现或产品目标取消;已合软件、合成容量与配对恢复测试不等生产容量/完整恢复域/可信撤销恢复已经 passed。 - **完整验收 / 生产启用前置**:仍须先完成 M4 与 M5,并证明独立恢复控制、已撤销状态不复活、有效正文受保护、四项限额在真实环境实测、环境与部署授权齐备。调度拆分不改变任何产品目标、证据或完成定义。 - **推进前必须拿到的证据(全部仍未验证,不得勾选)**: - [ ] 用与目标服务一致但隔离的真实 systemd unit、身份、mount、配置与数据格式演练完整备份/恢复;启动新服务前检查开关关闭。仅获授权后操作。 diff --git a/tests/README.md b/tests/README.md index c1577266..c56a0406 100644 --- a/tests/README.md +++ b/tests/README.md @@ -48,6 +48,10 @@ loopback 与 Unix 域套接字放行,真实外网测试仍应放在 `tests/man 本地上传安全基线:`tests/unit/test_local_upload_policy.py` 用临时 SQLite、真实 `ViewTokenResolver` 与 cache/task cleanup 消费者验证 owner/key 幂等、blank legacy token 隔离、`upload_` token、固定终态期限、write-once 撤销、开关关闭后的拒读与有效成果保护;另启动真实 Python 子进程消费同一 SQLite,验证缺省关闭与显式环境开关。C 生命周期与真实 worker 回归运行: +本地上传 B1–B16 的代码、断言、真实 producer 与清理边界索引见 +[`docs/testing/local-upload-contracts.md`](../docs/testing/local-upload-contracts.md)。其中未验证的生产容量、 +真实恢复域和旧镜像不代表已通过。 + ```bash uv run --frozen pytest -q tests/unit/test_upload_worker.py tests/integration/test_local_upload_lifecycle.py tests/unit/test_notification_e2e_delivery.py tests/unit/test_capswriter_contract.py ``` diff --git a/tests/fixtures/local_upload_loopback.py b/tests/fixtures/local_upload_loopback.py new file mode 100644 index 00000000..3e955a78 --- /dev/null +++ b/tests/fixtures/local_upload_loopback.py @@ -0,0 +1,192 @@ +"""Loopback-only ASR, LLM, and notification consumers for the real API process.""" + +from __future__ import annotations + +import asyncio +import base64 +import hashlib +import json +import socket +import threading +import time +from typing import Any + +import uvicorn +from fastapi import FastAPI, Request, WebSocket +from fastapi.responses import JSONResponse + + +CAPSWRITER_TEXT = ( + "这段转录由真实 CapsWriter SDK 从浏览器上传的 WAV 文件经 ffmpeg 转码后生成。" + "本地上传会写入独立 SQLite 记录,经过实际任务队列处理,并通过公开只读链接提供正文。" + "这个受控 loopback 服务只替代外部识别、模型和通知接收端,产品路由与序列化仍真实运行。" +) +LLM_TEXT = ( + CAPSWRITER_TEXT + + "共享模型客户端已读取完整转录正文并完成校对;这段追加说明用于确认真实 API、" + "SDK 输出与模型 HTTP 请求之间的内容链路完整。" +) + + +class LoopbackUpstreams: + """Real HTTP/WebSocket server implementing the external protocol boundaries.""" + + def __init__(self) -> None: + self.app = FastAPI() + self._lock = threading.Lock() + self._asr_started = threading.Event() + self._asr_release = threading.Event() + self._asr_frames: list[str] = [] + self._asr_payload_bytes = 0 + self._asr_reply: dict[str, Any] | None = None + self._llm_requests: list[dict[str, Any]] = [] + self._notifications: list[dict[str, Any]] = [] + self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self._socket.bind(("127.0.0.1", 0)) + self._socket.listen(128) + self.port = self._socket.getsockname()[1] + self._server = uvicorn.Server( + uvicorn.Config(self.app, log_level="error", access_log=False) + ) + self._thread = threading.Thread( + target=self._server.run, + kwargs={"sockets": [self._socket]}, + name="upload-loopback-upstreams", + daemon=True, + ) + + self.app.add_api_route("/health", self.health, methods=["GET"]) + self.app.add_api_route( + "/v1/chat/completions", self.chat_completions, methods=["POST"] + ) + self.app.add_api_route("/hooks/feishu", self.notification, methods=["POST"]) + self.app.add_api_route("/__e2e__/events", self.events, methods=["GET"]) + self.app.add_api_route( + "/__e2e__/release-asr", self.release_asr, methods=["POST"] + ) + self.app.add_api_websocket_route("/", self.capswriter) + + def start(self) -> None: + self._thread.start() + deadline = time.monotonic() + 15 + while not self._server.started: + if not self._thread.is_alive(): + raise RuntimeError("loopback upstream server exited before startup") + if time.monotonic() >= deadline: + raise TimeoutError("loopback upstream server did not start") + time.sleep(0.01) + + def stop(self) -> None: + self._server.should_exit = True + self._thread.join(timeout=10) + if self._thread.is_alive(): + raise TimeoutError("loopback upstream server did not stop") + + async def health(self) -> dict[str, Any]: + return {"protocol_version": 2, "encodings": ["flac"]} + + async def capswriter(self, websocket: WebSocket) -> None: + await websocket.accept() + final_frame: dict[str, Any] | None = None + async for raw_frame in websocket.iter_text(): + frame = json.loads(raw_frame) + decoded = base64.b64decode(frame["data"]) + with self._lock: + self._asr_frames.append(raw_frame) + self._asr_payload_bytes += len(decoded) + if frame.get("is_final"): + final_frame = frame + self._asr_started.set() + released = await asyncio.to_thread(self._asr_release.wait, 30) + if not released: + raise TimeoutError("e2e did not release the controlled ASR response") + tokens = list(CAPSWRITER_TEXT) + timestamps = [index * 0.02 for index in range(len(tokens))] + result = { + "type": "result", + "is_final": True, + "task_id": frame["task_id"], + "text": CAPSWRITER_TEXT, + "text_accu": CAPSWRITER_TEXT, + "tokens": tokens, + "timestamps": timestamps, + "duration": 3.0, + "time_start": frame["time_start"], + "time_submit": frame["time_start"], + "time_complete": frame["time_start"] + 1, + } + await websocket.send_json(result) + with self._lock: + self._asr_reply = { + "task_id": result["task_id"], + "text_accu": result["text_accu"], + "token_count": len(tokens), + "timestamp_count": len(timestamps), + "duration": result["duration"], + "source": final_frame["source"], + "encoding": final_frame["encoding"], + "samples_total": final_frame.get("samples_total"), + } + return + + async def chat_completions(self, request: Request) -> JSONResponse: + raw = await request.body() + body = json.loads(raw) + with self._lock: + self._llm_requests.append( + { + "content_type": request.headers.get("content-type"), + "model": body.get("model"), + "messages": body.get("messages"), + "request_sha256": hashlib.sha256(raw).hexdigest(), + } + ) + return JSONResponse( + { + "id": "chatcmpl-local-upload-e2e", + "object": "chat.completion", + "created": 1, + "model": body.get("model", "local-e2e"), + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": LLM_TEXT}, + "finish_reason": "stop", + } + ], + "usage": { + "prompt_tokens": 100, + "completion_tokens": 80, + "total_tokens": 180, + }, + } + ) + + async def notification(self, request: Request) -> JSONResponse: + raw = await request.body() + with self._lock: + self._notifications.append( + { + "content_type": request.headers.get("content-type"), + "payload": json.loads(raw), + "raw_sha256": hashlib.sha256(raw).hexdigest(), + } + ) + return JSONResponse({"code": 0, "msg": "success"}) + + async def events(self) -> JSONResponse: + with self._lock: + body = { + "asr_started": self._asr_started.is_set(), + "asr_frames": list(self._asr_frames), + "asr_payload_bytes": self._asr_payload_bytes, + "asr_reply": self._asr_reply, + "llm_requests": list(self._llm_requests), + "notifications": list(self._notifications), + } + return JSONResponse(body) + + async def release_asr(self) -> JSONResponse: + self._asr_release.set() + return JSONResponse({"released": True}) diff --git a/tests/integration/test_local_upload_lifecycle.py b/tests/integration/test_local_upload_lifecycle.py index fef835a5..45882ca7 100644 --- a/tests/integration/test_local_upload_lifecycle.py +++ b/tests/integration/test_local_upload_lifecycle.py @@ -3,6 +3,7 @@ import asyncio import base64 import concurrent.futures +from contextlib import contextmanager import datetime import hashlib import json @@ -72,7 +73,7 @@ async def current_user(): def _make_upload( cache, *, owner="alice", media_name=None, title="上传标题", - include_summary=True, retention="never", + include_summary=True, retention="never", complete=True, ): now = datetime.datetime.now(datetime.timezone.utc) key = f"{int(now.timestamp() * 1000)}-{uuid.uuid4()}" @@ -124,13 +125,14 @@ def _make_upload( chapters_status=ChaptersStatus.GENERATED, notes_status=NotesStatus.FAILED, ) - cache.update_task_status( - upload["root_task_id"], - "success", - platform="local_upload", - media_id=media_id, - title=title, - ) + if complete: + cache.update_task_status( + upload["root_task_id"], + "success", + platform="local_upload", + media_id=media_id, + title=title, + ) return cache.get_local_upload_by_id(upload["upload_id"]) @@ -424,14 +426,37 @@ def test_public_read_summary_exports_and_owner_history_close_without_content_byp client, cache, _llm_queue, user = lifecycle_client upload = _make_upload(cache, title="Private owner title") token = upload["view_token"] - - raw = client.get(f"/view/{token}?raw=transcript") - page = client.get(f"/view/{token}?page=transcript") - export = client.get(f"/export/{token}/transcript") + cache.save_llm_result( + platform="local_upload", media_id=upload["media_id"], + use_speaker_recognition=False, llm_type="calibrated", + content="A calibrated public body.", + ) + cache.save_llm_result( + platform="local_upload", media_id=upload["media_id"], + use_speaker_recognition=False, llm_type="notes", + content="A detailed public note.", + ) + cache.save_llm_status( + platform="local_upload", media_id=upload["media_id"], + use_speaker_recognition=False, notes_status=NotesStatus.GENERATED, + ) + content_by_type = { + "calibrated": "A calibrated public body.", + "summary": "A real public summary.", + "notes": "A detailed public note.", + "transcript": "actual transcript body", + } + default_view = client.get(f"/view/{token}") + assert default_view.status_code == 200 + assert "A calibrated public body." in default_view.text + for export_type, content in content_by_type.items(): + raw = client.get(f"/view/{token}?raw={export_type}") + page = client.get(f"/view/{token}?page={export_type}") + export = client.get(f"/export/{token}/{export_type}") + assert raw.status_code == 200 and content in raw.text + assert page.status_code == 200 and content in page.text + assert export.status_code == 200 and content in export.text summary = client.get(f"/api/audit/summary?view_token={token}") - assert raw.status_code == 200 and "actual transcript body" in raw.text - assert page.status_code == 200 and "actual transcript body" in page.text - assert export.status_code == 200 and "actual transcript body" in export.text assert summary.status_code == 200 assert summary.json()["data"]["summary"] == "A real public summary." @@ -464,19 +489,32 @@ def test_public_read_summary_exports_and_owner_history_close_without_content_byp assert expired_item["share_active"] is False assert expired_item["share_inactive_reason"] == "expired" assert "summary" not in closed_item and "transcript" not in closed_item - assert client.get(f"/view/{token}?raw=transcript").status_code == 404 - assert client.get(f"/view/{token}?page=transcript").status_code == 404 - assert client.get(f"/export/{token}/transcript").status_code == 404 + for export_type, content in content_by_type.items(): + for selector in ("raw", "page"): + denied = client.get(f"/view/{token}?{selector}={export_type}") + assert denied.status_code == 404 + assert content not in denied.text + denied_export = client.get(f"/export/{token}/{export_type}") + assert denied_export.status_code == 404 + assert content not in denied_export.text + closed_default = client.get(f"/view/{token}") + assert closed_default.status_code == 200 + assert "A calibrated public body." not in closed_default.text assert client.get(f"/api/audit/summary?view_token={token}").status_code == 404 for endpoint in ("/api/recalibrate", "/api/resummarize", "/api/generate_notes"): assert client.post(endpoint, json={"view_token": token}).status_code == 404 +@pytest.mark.parametrize( + "endpoint", ["/api/recalibrate", "/api/resummarize", "/api/generate_notes"] +) def test_reprocess_admission_holds_sqlite_order_against_concurrent_close( - lifecycle_client, monkeypatch, + lifecycle_client, monkeypatch, endpoint, ): client, cache, llm_queue, _user = lifecycle_client - upload = _make_upload(cache, title="Concurrent close source") + upload = _make_upload( + cache, title="Concurrent close source", include_summary=False + ) original_admit = tasks_routes._admit_upload_reprocess admission_paused = threading.Event() release_admission = threading.Event() @@ -503,7 +541,7 @@ def pause_transaction_admission(cursor, view_token, user_id, cache_data): try: admission = executor.submit( client.post, - "/api/recalibrate", + endpoint, json={"view_token": upload["view_token"]}, ) assert admission_paused.wait(timeout=3) @@ -533,6 +571,232 @@ def close_share(): assert llm_queue.qsize() == 1 +@pytest.mark.parametrize( + "endpoint", ["/api/recalibrate", "/api/resummarize", "/api/generate_notes"] +) +def test_reprocess_snapshot_before_close_is_rejected_by_fresh_transaction( + lifecycle_client, monkeypatch, endpoint, +): + client, cache, llm_queue, _user = lifecycle_client + upload = _make_upload( + cache, title="Close before admission", include_summary=False + ) + cache_loaded = threading.Event() + release_request = threading.Event() + original_resolve = ViewTokenResolver.get_cache_by_view_token + + def pause_after_initial_resolution(resolver, view_token): + data = original_resolve(resolver, view_token) + if data and data.get("platform") == "local_upload": + cache_loaded.set() + if not release_request.wait(timeout=5): + raise TimeoutError("test did not release the pre-admission snapshot") + return data + + monkeypatch.setattr( + ViewTokenResolver, + "get_cache_by_view_token", + pause_after_initial_resolution, + ) + with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor: + pending = executor.submit( + client.post, + endpoint, + json={"view_token": upload["view_token"]}, + ) + assert cache_loaded.wait(timeout=5) + revoked = cache.revoke_local_upload(upload["upload_id"]) + release_request.set() + response = pending.result(timeout=10) + + assert revoked["revoked_at"] + assert response.status_code == 404 + with cache._get_cursor() as cursor: + cursor.execute( + """SELECT COUNT(*) FROM task_status + WHERE platform = 'local_upload' AND media_id = ? AND task_id != ?""", + (upload["media_id"], upload["root_task_id"]), + ) + assert cursor.fetchone()[0] == 0 + assert llm_queue.empty() + + +@pytest.mark.parametrize("order", ["publish-before-revoke", "revoke-before-publish"]) +def test_terminal_publish_and_revoke_are_both_write_once( + lifecycle_client, monkeypatch, order, +): + client, cache, _llm_queue, _user = lifecycle_client + upload = _make_upload( + cache, + title="Terminal publication race", + retention="30d", + complete=False, + ) + publish_reached_gate = threading.Event() + release_publish = threading.Event() + original_update = cache.update_task_status + + def controlled_terminal_publish(*args, **kwargs): + if order == "revoke-before-publish": + publish_reached_gate.set() + if not release_publish.wait(timeout=5): + raise TimeoutError("test did not release terminal publication") + return original_update(*args, **kwargs) + result = original_update(*args, **kwargs) + publish_reached_gate.set() + if not release_publish.wait(timeout=5): + raise TimeoutError("test did not release terminal publication") + return result + + monkeypatch.setattr(cache, "update_task_status", controlled_terminal_publish) + with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor: + publication = executor.submit( + cache.update_task_status, + upload["root_task_id"], + "success", + platform="local_upload", + media_id=upload["media_id"], + title="Terminal publication race", + ) + assert publish_reached_gate.wait(timeout=5) + revoked = cache.revoke_local_upload(upload["upload_id"]) + release_publish.set() + assert publication.result(timeout=10) is True + + final = cache.get_local_upload_by_id(upload["upload_id"]) + assert final["revoked_at"] == revoked["revoked_at"] + assert final["terminal_at"] + assert final["expires_at"] + assert cache.get_task_by_id(upload["root_task_id"])["status"] == "success" + transcript = ( + cache._get_file_path("local_upload", upload["media_id"]) + / "transcript_capswriter.txt" + ) + assert transcript.read_text(encoding="utf-8") == "actual transcript body" + denied = client.get(f"/view/{upload['view_token']}?raw=transcript") + assert denied.status_code == 404 + assert "actual transcript body" not in denied.text + + +@pytest.mark.parametrize("order", ["publish-before-cleanup", "cleanup-before-publish"]) +def test_active_upload_cleanup_and_cache_publish_preserve_body_in_both_orders( + lifecycle_client, monkeypatch, order, +): + _client, cache, _llm_queue, _user = lifecycle_client + upload = _make_upload( + cache, title="Cleanup publication race", retention="30d" + ) + expiration_before = upload["expires_at"] + old_updated_at = ( + datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=2) + ).strftime("%Y-%m-%d %H:%M:%S") + with cache._get_cursor() as cursor: + cursor.execute( + "UPDATE video_cache SET updated_at = ? WHERE platform = ? AND media_id = ?", + (old_updated_at, "local_upload", upload["media_id"]), + ) + + role = threading.local() + original_media_lock = cache.media_lock + original_get_cursor = cache._get_cursor + first_locked = threading.Event() + release_first = threading.Event() + candidate_snapshot = threading.Event() + release_candidate_snapshot = threading.Event() + publisher_started = threading.Event() + + @contextmanager + def observe_cleanup_candidates(): + with original_get_cursor() as cursor: + class CursorProxy: + last_sql = "" + + def execute(self, sql, parameters=()): + self.last_sql = sql + return cursor.execute(sql, parameters) + + def fetchall(self): + rows = cursor.fetchall() + if ( + getattr(role, "name", None) == "cleanup" + and "SELECT id, platform, media_id, files_loc" in self.last_sql + ): + candidate_snapshot.set() + if order == "publish-before-cleanup" and not release_candidate_snapshot.wait(timeout=5): + raise TimeoutError("test did not release the cleanup candidate snapshot") + return rows + + def __getattr__(self, name): + return getattr(cursor, name) + + yield CursorProxy() + + monkeypatch.setattr(cache, "_get_cursor", observe_cleanup_candidates) + + @contextmanager + def controlled_media_lock(platform, media_id, timeout=None): + current = getattr(role, "name", None) + with original_media_lock(platform, media_id, timeout=timeout): + if (order == "publish-before-cleanup" and current == "publisher") or ( + order == "cleanup-before-publish" and current == "cleanup" + ): + first_locked.set() + if not release_first.wait(timeout=5): + raise TimeoutError("test did not release the first cache-lock owner") + yield + + monkeypatch.setattr(cache, "media_lock", controlled_media_lock) + + def publish_cache(): + role.name = "publisher" + publisher_started.set() + return cache.save_cache( + platform="local_upload", + url="https://source.example.test/media?id=7", + media_id=upload["media_id"], + use_speaker_recognition=False, + transcript_data="published after the cleanup candidate snapshot", + transcript_type="capswriter", + title="Cleanup publication race", + author="", + description="", + ) + + def clean_cache(): + role.name = "cleanup" + return cache.cleanup_old_cache(days=1) + + with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor: + first_operation = publish_cache if order == "publish-before-cleanup" else clean_cache + second_operation = clean_cache if order == "publish-before-cleanup" else publish_cache + first = executor.submit(first_operation) + assert first_locked.wait(timeout=5) + second = executor.submit(second_operation) + if order == "publish-before-cleanup": + assert candidate_snapshot.wait(timeout=5) + release_candidate_snapshot.set() + else: + assert publisher_started.wait(timeout=5) + release_first.set() + results = (first.result(timeout=10), second.result(timeout=10)) + + cleanup_count = next(result for result in results if isinstance(result, int)) + publish_result = next(result for result in results if isinstance(result, dict)) + assert cleanup_count == 0 + assert publish_result["media_id"] == upload["media_id"] + transcript = ( + cache._get_file_path("local_upload", upload["media_id"]) + / "transcript_capswriter.txt" + ) + assert transcript.read_text(encoding="utf-8") == ( + "published after the cleanup candidate snapshot" + ) + final = cache.get_local_upload_by_id(upload["upload_id"]) + assert final["root_task_id"] == upload["root_task_id"] + assert final["media_id"] == upload["media_id"] + assert final["expires_at"] == expiration_before + + def test_upload_history_filters_owner_before_page_limit(lifecycle_client): client, cache, _llm_queue, _user = lifecycle_client older = _make_upload(cache, title="Alice older") diff --git a/tests/integration/test_upload_races.py b/tests/integration/test_upload_races.py new file mode 100644 index 00000000..b07edb40 --- /dev/null +++ b/tests/integration/test_upload_races.py @@ -0,0 +1,99 @@ +"""Controlled lifecycle races through the production public-read router.""" + +import concurrent.futures +import datetime +import threading +import uuid + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from video_transcript_api.api.routes import views as views_routes +from video_transcript_api.cache.cache_manager import CacheManager + + +@pytest.fixture +def public_upload_client(tmp_path, monkeypatch): + monkeypatch.setenv("VTA_UPLOADS_ENABLED", "true") + cache = CacheManager(str(tmp_path / "cache")) + monkeypatch.setattr(views_routes, "cache_manager", cache) + app = FastAPI() + app.include_router(views_routes.router) + with TestClient(app) as client: + yield client, cache + cache.close() + + +def _accepted_upload(cache): + now = datetime.datetime.now(datetime.timezone.utc) + key = f"{int(now.timestamp() * 1000)}-{uuid.uuid4()}" + intent = cache.register_local_upload( + owner_user_id="alice", + idempotency_key=key, + retention="30d", + intent_metadata={"filename": "race.mp4"}, + ) + upload = cache.accept_local_upload( + intent["upload_id"], + media_id=f"upload_{uuid.uuid4().hex}", + task_id=cache.generate_task_id(), + filename="race.mp4", + request_metadata={"filename": "race.mp4"}, + ) + cache.save_cache( + platform="local_upload", + url="https://upload.invalid/race", + media_id=upload["media_id"], + use_speaker_recognition=False, + transcript_data="race body from SQLite-backed cache", + transcript_type="capswriter", + title="Race fixture", + author="", + description="", + ) + cache.update_task_status( + upload["root_task_id"], + "success", + platform="local_upload", + media_id=upload["media_id"], + title="Race fixture", + ) + return cache.get_local_upload_by_id(upload["upload_id"]) + + +def test_public_read_and_expiry_both_controlled_orders(public_upload_client, monkeypatch): + client, cache = public_upload_client + upload = _accepted_upload(cache) + token = upload["view_token"] + loaded_upload = threading.Event() + continue_read = threading.Event() + check_share = cache.local_upload_share_is_active + expiration = datetime.datetime.fromisoformat(upload["expires_at"]).replace( + tzinfo=datetime.timezone.utc + ) + clock = {"now": expiration - datetime.timedelta(seconds=1)} + + def pause_after_real_authorization(row, now=None): + result = check_share(row, now=clock["now"]) + if result: + loaded_upload.set() + if not continue_read.wait(timeout=5): + raise TimeoutError("test did not release the authorized read") + return result + + monkeypatch.setattr(cache, "local_upload_share_is_active", pause_after_real_authorization) + with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor: + in_flight_read = executor.submit(client.get, f"/view/{token}?raw=transcript") + assert loaded_upload.wait(timeout=5), "public route did not authorize the read" + clock["now"] = expiration + assert clock["now"] >= expiration + assert not in_flight_read.done(), "authorized read finished before the shared clock crossed expiry" + continue_read.set() + response = in_flight_read.result(timeout=5) + + assert response.status_code == 200 + assert "race body from SQLite-backed cache" in response.text + fresh_consumer = client.get(f"/view/{token}?raw=transcript") + assert fresh_consumer.status_code == 404 + assert "race body from SQLite-backed cache" not in fresh_consumer.text diff --git a/tests/local_upload_real_server.py b/tests/local_upload_real_server.py new file mode 100644 index 00000000..8e0a3490 --- /dev/null +++ b/tests/local_upload_real_server.py @@ -0,0 +1,247 @@ +"""Start the production FastAPI app with isolated loopback dependencies.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import signal +import shutil +import socket +import stat +import sys +import threading +import time +from pathlib import Path +from typing import Any + +import commentjson +import uvicorn +from fastapi import Request +from fastapi.responses import JSONResponse + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "src")) +sys.path.insert(0, str(ROOT / "tests" / "fixtures")) + +from local_upload_loopback import LoopbackUpstreams # noqa: E402 + + +def _temporary_config(run_dir: Path, api_port: int, upstream_port: int) -> Path: + with (ROOT / "config" / "config.example.jsonc").open(encoding="utf-8") as source: + config = commentjson.load(source) + + storage = config["storage"] + storage["cache_dir"] = str(run_dir / "cache") + storage["workspace_dir"] = str(run_dir / "workspace") + storage["temp_dir"] = str(run_dir / "temp") + storage["audit_db"] = str(run_dir / "audit.sqlite3") + storage["upload_limits"] = { + "max_file_mib": 2, + "max_media_hours": 1, + "receive_concurrency": 1, + "upload_temp_budget_mib": 8, + } + storage["cache_retention_days"] = 30 + storage["task_status_retention_days"] = 30 + storage["audit_log_retention_days"] = 30 + storage["temp_retention_hours"] = 24 + + config["api"].update(host="127.0.0.1", port=api_port, auth_token="unused-local-e2e-token") + config["concurrent"].update(max_workers=1, queue_size=2, llm_max_workers=1) + config["capswriter"].update( + server_url=f"ws://127.0.0.1:{upstream_port}", + max_retries=1, + retry_delay=0, + ) + config["llm"].update( + api_key="local-e2e-key", + base_url=f"http://127.0.0.1:{upstream_port}/v1", + calibrate_model="local-e2e-model", + calibrate_reasoning_effort="disabled", + summary_model="local-e2e-model", + summary_reasoning_effort="disabled", + max_retries=0, + total_timeout=15, + structured_calibration_for_plain=False, + ) + config["llm"]["min_calibrate_ratio"] = 0.5 + config["web"].update( + base_url=f"http://127.0.0.1:{api_port}", + enable_view_links=True, + timezone="UTC+8", + ) + config["wechat"]["webhook"] = "" + config["feishu"].update( + webhook=f"http://127.0.0.1:{upstream_port}/hooks/feishu", + secret="", + ) + config["log"].update( + level="WARNING", + file=str(run_dir / "logs" / "app.log"), + debug_dir=str(run_dir / "logs" / "debug"), + llm_debug_dir=str(run_dir / "logs" / "llm_debug"), + ) + config_path = run_dir / "config.json" + config_path.write_text(json.dumps(config, ensure_ascii=False), encoding="utf-8") + return config_path + + +def _write_ffprobe_observer(run_dir: Path) -> None: + real_ffprobe = shutil.which("ffprobe") + if real_ffprobe is None: + raise FileNotFoundError("ffprobe is required for the real local-upload path") + shim_dir = run_dir / "bin" + shim_dir.mkdir(parents=True) + log_path = run_dir / "ffprobe-argv.jsonl" + shim = shim_dir / "ffprobe" + shim.write_text( + "#!/usr/bin/env python3\n" + "import json, os, sys\n" + "with open(os.environ['VTA_E2E_FFPROBE_LOG'], 'a', encoding='utf-8') as log:\n" + " log.write(json.dumps({'argv': sys.argv[1:]}) + '\\n')\n" + f"real = {json.dumps(real_ffprobe)}\n" + "os.execv(real, [real, *sys.argv[1:]])\n", + encoding="utf-8", + ) + shim.chmod(shim.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) + os.environ["PATH"] = f"{shim_dir}{os.pathsep}{os.environ['PATH']}" + os.environ["VTA_E2E_FFPROBE_LOG"] = str(log_path) + + +def _register_state_inspector(app, run_dir: Path, upstreams: LoopbackUpstreams) -> None: + @app.get("/__e2e__/state") + async def inspect_real_state(request: Request) -> JSONResponse: + runtime = request.app.state.runtime + with runtime.cache_manager._get_cursor() as cursor: + cursor.execute( + """SELECT u.upload_id, u.owner_user_id, u.idempotency_key, u.state, + u.root_task_id, u.media_id, u.view_token, u.filename, + t.title AS title, u.source_url, u.retention, u.expires_at, + u.revoked_at, u.media_path, u.byte_size, u.sha256, + u.request_metadata, + t.processing_options AS task_processing_options, + t.status AS root_status, t.platform AS task_platform, + t.view_token AS legacy_task_token, t.completed_at + FROM local_uploads u JOIN task_status t ON t.task_id = u.root_task_id + ORDER BY u.rowid DESC""" + ) + rows = [dict(row) for row in cursor.fetchall()] + + for row in rows: + row["request_metadata"] = json.loads(row["request_metadata"]) + row["task_processing_options"] = json.loads( + row["task_processing_options"] + ) + media_path = Path(row["media_path"]) if row["media_path"] else None + row["media_exists"] = bool(media_path and media_path.is_file()) + if row["media_exists"]: + with media_path.open("rb") as media_file: + row["media_sha256"] = hashlib.file_digest(media_file, "sha256").hexdigest() + else: + row["media_sha256"] = None + + ffprobe_log = run_dir / "ffprobe-argv.jsonl" + ffprobe_calls = ( + [json.loads(line) for line in ffprobe_log.read_text(encoding="utf-8").splitlines()] + if ffprobe_log.exists() + else [] + ) + return JSONResponse( + { + "uploads": rows, + "ffprobe_calls": ffprobe_calls, + "queued_upload_items": runtime.task_queue.qsize(), + } + ) + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--run-dir", required=True, type=Path) + args = parser.parse_args() + run_dir = args.run_dir.resolve() + run_dir.mkdir(parents=True, exist_ok=True) + + upstreams = LoopbackUpstreams() + upstreams.start() + + api_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + api_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + api_socket.bind(("127.0.0.1", 0)) + api_socket.listen(128) + api_port = api_socket.getsockname()[1] + config_path = _temporary_config(run_dir, api_port, upstreams.port) + + os.environ["VTAPI_CONFIG"] = str(config_path) + os.environ["VTAPI_USERS_JSON"] = str(run_dir / "users.json") + os.environ["VTA_UPLOADS_ENABLED"] = "true" + os.environ["SENTRY_DSN"] = "" + user_config = { + "users": { + "browser-fixture-token": { + "user_id": "browser_owner", + "name": "Browser owner", + "permissions": ["recalibrate", "delete"], + "enabled": True, + "feishu_webhook": f"http://127.0.0.1:{upstreams.port}/hooks/feishu", + } + } + } + Path(os.environ["VTAPI_USERS_JSON"]).write_text( + json.dumps(user_config), encoding="utf-8" + ) + _write_ffprobe_observer(run_dir) + + from video_transcript_api.api.app import create_app + + app = create_app() + _register_state_inspector(app, run_dir, upstreams) + api_server = uvicorn.Server( + uvicorn.Config(app, log_level="error", access_log=False) + ) + api_thread = threading.Thread( + target=api_server.run, + kwargs={"sockets": [api_socket]}, + name="local-upload-real-api", + daemon=True, + ) + + def stop_server(_signum, _frame) -> None: + api_server.should_exit = True + + signal.signal(signal.SIGTERM, stop_server) + signal.signal(signal.SIGINT, stop_server) + api_thread.start() + try: + deadline = time.monotonic() + 30 + while not api_server.started: + if not api_thread.is_alive(): + raise RuntimeError("production FastAPI server exited before startup") + if time.monotonic() >= deadline: + raise TimeoutError("production FastAPI server did not start") + time.sleep(0.01) + print( + "VTA_REAL_SERVER_READY " + + json.dumps( + { + "api_url": f"http://127.0.0.1:{api_port}", + "upstream_url": f"http://127.0.0.1:{upstreams.port}", + "run_dir": str(run_dir), + } + ), + flush=True, + ) + api_thread.join() + finally: + api_server.should_exit = True + api_thread.join(timeout=15) + if api_thread.is_alive(): + raise TimeoutError("production FastAPI server did not stop") + upstreams.stop() + api_socket.close() + + +if __name__ == "__main__": + main() diff --git a/tests/unit/test_local_upload_policy.py b/tests/unit/test_local_upload_policy.py index 080b57b5..50ebf4ff 100644 --- a/tests/unit/test_local_upload_policy.py +++ b/tests/unit/test_local_upload_policy.py @@ -1,6 +1,7 @@ """Real SQLite contracts for the local-upload safety baseline.""" from concurrent.futures import ThreadPoolExecutor +from contextlib import contextmanager from datetime import datetime, timedelta, timezone import os from pathlib import Path @@ -157,13 +158,43 @@ def test_real_sqlite_upload_identity_owner_key_and_intent_window(cm): ) -def test_concurrent_same_owner_key_creates_one_sqlite_receipt(cm): +@pytest.mark.parametrize("winner", ["request-a", "request-b"]) +def test_concurrent_same_owner_key_creates_one_sqlite_receipt(cm, monkeypatch, winner): now = datetime.now(timezone.utc) key = _intent_key(now) - start = threading.Barrier(2) - - def register(): - start.wait(timeout=5) + first_read = threading.Event() + second_begin = threading.Event() + release_first = threading.Event() + warm_connections = threading.Barrier(2) + role = threading.local() + original_get_cursor = cm._get_cursor + + @contextmanager + def controlled_cursor(): + with original_get_cursor() as cursor: + class CursorProxy: + def execute(self, sql, parameters=()): + if sql == "BEGIN IMMEDIATE" and getattr(role, "name", None) != winner: + second_begin.set() + result = cursor.execute(sql, parameters) + if ( + "SELECT * FROM local_uploads WHERE owner_user_id" in sql + and getattr(role, "name", None) == winner + ): + first_read.set() + if not release_first.wait(timeout=5): + raise TimeoutError("same-intent winner did not release its SQLite transaction") + return result + + def __getattr__(self, name): + return getattr(cursor, name) + + yield CursorProxy() + + monkeypatch.setattr(cm, "_get_cursor", controlled_cursor) + + def register(name): + role.name = name return cm.register_local_upload( owner_user_id="alice", idempotency_key=key, @@ -173,11 +204,25 @@ def register(): ) with ThreadPoolExecutor(max_workers=2) as pool: - first = pool.submit(register) - second = pool.submit(register) - records = [first.result(timeout=10), second.result(timeout=10)] - - assert records[0]["upload_id"] == records[1]["upload_id"] + def warm_connection(): + cm._get_connection() + warm_connections.wait(timeout=5) + + warmed = [pool.submit(warm_connection) for _ in range(2)] + for future in warmed: + future.result(timeout=10) + first = pool.submit(register, winner) + assert first_read.wait(timeout=5) + loser = "request-b" if winner == "request-a" else "request-a" + second = pool.submit(register, loser) + assert second_begin.wait(timeout=5) + release_first.set() + winner_record = first.result(timeout=10) + loser_record = second.result(timeout=10) + + assert winner_record["_created"] is True + assert loser_record["_created"] is False + assert winner_record["upload_id"] == loser_record["upload_id"] with cm._get_cursor() as cursor: cursor.execute( "SELECT COUNT(*) FROM local_uploads WHERE owner_user_id = ? AND idempotency_key = ?",