"""Integration tests for the FastAPI + Temporal MVP.""" import asyncio import pytest async def wait_for_workflow_status(client, order_id: int, expected_status: str, attempts: int = 120): """Poll the workflow status endpoint until it reaches a target status.""" last_payload = None for _ in range(attempts): response = await client.get(f"/api/v1/workflows/{order_id}") if response.status_code == 200: last_payload = response.json() if last_payload["workflow_status"] == expected_status: return last_payload await asyncio.sleep(0.05) raise AssertionError(f"Workflow {order_id} never reached status {expected_status!r}: {last_payload}") async def wait_for_step_count(client, order_id: int, step_name: str, minimum_count: int, attempts: int = 120): """Poll until a workflow step has been recorded a minimum number of times.""" last_payload = None for _ in range(attempts): response = await client.get(f"/api/v1/workflows/{order_id}") if response.status_code == 200: last_payload = response.json() count = sum(1 for step in last_payload["steps"] if step["step_name"] == step_name) if count >= minimum_count: return last_payload await asyncio.sleep(0.05) raise AssertionError( f"Workflow {order_id} never recorded step {step_name!r} {minimum_count} times: {last_payload}" ) async def create_mid_end_order(client): """Create a standard semi-pro order for review-path tests.""" resources = await create_workflow_resources(client, include_scene=True) response = await client.post( "/api/v1/orders", json={ "customer_level": "mid", "service_mode": "semi_pro", "model_id": resources["model"]["id"], "garment_asset_id": resources["garment"]["id"], "scene_ref_asset_id": resources["scene"]["id"], }, ) assert response.status_code == 201 return response.json() async def create_library_resource(client, payload: dict) -> dict: """Create a resource-library item for workflow integration tests.""" response = await client.post("/api/v1/library/resources", json=payload) assert response.status_code == 201 return response.json() async def create_workflow_resources(client, *, include_scene: bool) -> dict[str, dict]: """Create a minimal set of real library resources for workflow tests.""" model_resource = await create_library_resource( client, { "resource_type": "model", "name": "Ava Studio", "description": "棚拍女模特", "tags": ["女装", "棚拍"], "gender": "female", "age_group": "adult", "files": [ { "file_role": "original", "storage_key": "library/models/ava/original.png", "public_url": "https://images.example.com/library/models/ava/original.png", "mime_type": "image/png", "size_bytes": 1024, "sort_order": 0, "width": 1200, "height": 1600, }, { "file_role": "thumbnail", "storage_key": "library/models/ava/thumb.png", "public_url": "https://images.example.com/library/models/ava/thumb.png", "mime_type": "image/png", "size_bytes": 256, "sort_order": 0, }, ], }, ) garment_resource = await create_library_resource( client, { "resource_type": "garment", "name": "Cream Dress", "description": "米白色连衣裙", "tags": ["女装"], "category": "dress", "files": [ { "file_role": "original", "storage_key": "library/garments/cream-dress/original.png", "public_url": "https://images.example.com/library/garments/cream-dress/original.png", "mime_type": "image/png", "size_bytes": 1024, "sort_order": 0, }, { "file_role": "thumbnail", "storage_key": "library/garments/cream-dress/thumb.png", "public_url": "https://images.example.com/library/garments/cream-dress/thumb.png", "mime_type": "image/png", "size_bytes": 256, "sort_order": 0, }, ], }, ) resources = { "model": model_resource, "garment": garment_resource, } if include_scene: resources["scene"] = await create_library_resource( client, { "resource_type": "scene", "name": "Loft Window", "description": "暖调室内场景", "tags": ["室内"], "environment": "indoor", "files": [ { "file_role": "original", "storage_key": "library/scenes/loft/original.png", "public_url": "https://images.example.com/library/scenes/loft/original.png", "mime_type": "image/png", "size_bytes": 1024, "sort_order": 0, }, { "file_role": "thumbnail", "storage_key": "library/scenes/loft/thumb.png", "public_url": "https://images.example.com/library/scenes/loft/thumb.png", "mime_type": "image/png", "size_bytes": 256, "sort_order": 0, }, ], }, ) return resources @pytest.mark.asyncio async def test_healthcheck(api_runtime): """The health endpoint should always respond successfully.""" client, _ = api_runtime response = await client.get("/healthz") assert response.status_code == 200 assert response.json() == {"status": "ok"} @pytest.mark.asyncio async def test_low_end_order_completes(api_runtime): """Low-end orders should run through the full automated pipeline.""" client, env = api_runtime resources = await create_workflow_resources(client, include_scene=True) response = await client.post( "/api/v1/orders", json={ "customer_level": "low", "service_mode": "auto_basic", "model_id": resources["model"]["id"], "garment_asset_id": resources["garment"]["id"], "scene_ref_asset_id": resources["scene"]["id"], }, ) assert response.status_code == 201 payload = response.json() assert payload["workflow_id"] == f"order-{payload['order_id']}" handle = env.client.get_workflow_handle(payload["workflow_id"]) result = await handle.result() assert result["status"] == "succeeded" order_response = await client.get(f"/api/v1/orders/{payload['order_id']}") assert order_response.status_code == 200 assert order_response.json()["status"] == "succeeded" assets_response = await client.get(f"/api/v1/orders/{payload['order_id']}/assets") assert assets_response.status_code == 200 assert any(asset["asset_type"] == "final" for asset in assets_response.json()) workflow_response = await client.get(f"/api/v1/workflows/{payload['order_id']}") assert workflow_response.status_code == 200 assert workflow_response.json()["workflow_status"] == "succeeded" @pytest.mark.asyncio async def test_prepare_model_step_uses_library_original_file(api_runtime): """prepare_model should use the model resource's original file rather than a mock URI.""" def expected_input_snapshot(resource: dict) -> dict: original_file = next(file for file in resource["files"] if file["file_role"] == "original") snapshot = { "resource_id": resource["id"], "resource_name": resource["name"], "original_file_id": original_file["id"], "original_url": resource["original_url"], "mime_type": original_file["mime_type"], "width": original_file["width"], "height": original_file["height"], } return {key: value for key, value in snapshot.items() if value is not None} client, env = api_runtime resources = await create_workflow_resources(client, include_scene=True) model_resource = resources["model"] garment_resource = resources["garment"] scene_resource = resources["scene"] response = await client.post( "/api/v1/orders", json={ "customer_level": "low", "service_mode": "auto_basic", "model_id": model_resource["id"], "garment_asset_id": garment_resource["id"], "scene_ref_asset_id": scene_resource["id"], }, ) assert response.status_code == 201 payload = response.json() handle = env.client.get_workflow_handle(payload["workflow_id"]) result = await handle.result() assert result["status"] == "succeeded" assets_response = await client.get(f"/api/v1/orders/{payload['order_id']}/assets") assert assets_response.status_code == 200 prepared_asset = next( asset for asset in assets_response.json() if asset["asset_type"] == "prepared_model" ) assert prepared_asset["uri"] == model_resource["original_url"] assert prepared_asset["metadata_json"]["library_resource_id"] == model_resource["id"] assert prepared_asset["metadata_json"]["library_original_file_id"] == next( file["id"] for file in model_resource["files"] if file["file_role"] == "original" ) assert prepared_asset["metadata_json"]["library_original_url"] == model_resource["original_url"] assert prepared_asset["metadata_json"]["model_input"] == expected_input_snapshot(model_resource) assert prepared_asset["metadata_json"]["garment_input"] == expected_input_snapshot(garment_resource) assert prepared_asset["metadata_json"]["scene_input"] == expected_input_snapshot(scene_resource) @pytest.mark.asyncio async def test_low_end_order_completes_without_scene(api_runtime): """Low-end orders should still complete when scene input is omitted.""" client, env = api_runtime resources = await create_workflow_resources(client, include_scene=False) response = await client.post( "/api/v1/orders", json={ "customer_level": "low", "service_mode": "auto_basic", "model_id": resources["model"]["id"], "garment_asset_id": resources["garment"]["id"], }, ) assert response.status_code == 201 payload = response.json() handle = env.client.get_workflow_handle(payload["workflow_id"]) result = await handle.result() assert result["status"] == "succeeded" order_response = await client.get(f"/api/v1/orders/{payload['order_id']}") assert order_response.status_code == 200 assert order_response.json()["scene_ref_asset_id"] is None assert order_response.json()["status"] == "succeeded" @pytest.mark.asyncio async def test_low_end_order_reuses_tryon_uri_for_qc_and_final_without_scene(api_runtime): """Without scene input, qc/export should keep the real try-on image URI.""" client, env = api_runtime resources = await create_workflow_resources(client, include_scene=False) response = await client.post( "/api/v1/orders", json={ "customer_level": "low", "service_mode": "auto_basic", "model_id": resources["model"]["id"], "garment_asset_id": resources["garment"]["id"], }, ) assert response.status_code == 201 payload = response.json() handle = env.client.get_workflow_handle(payload["workflow_id"]) result = await handle.result() assert result["status"] == "succeeded" assets_response = await client.get(f"/api/v1/orders/{payload['order_id']}/assets") assert assets_response.status_code == 200 assets = assets_response.json() tryon_asset = next(asset for asset in assets if asset["asset_type"] == "tryon") qc_asset = next(asset for asset in assets if asset["asset_type"] == "qc_candidate") final_asset = next(asset for asset in assets if asset["asset_type"] == "final") assert qc_asset["uri"] == tryon_asset["uri"] assert final_asset["uri"] == tryon_asset["uri"] assert final_asset["metadata_json"]["source_asset_id"] == qc_asset["id"] @pytest.mark.asyncio async def test_mid_end_order_waits_review_then_approves_without_scene(api_runtime): """Mid-end orders should still reach review and approve when scene input is omitted.""" client, env = api_runtime resources = await create_workflow_resources(client, include_scene=False) response = await client.post( "/api/v1/orders", json={ "customer_level": "mid", "service_mode": "semi_pro", "model_id": resources["model"]["id"], "garment_asset_id": resources["garment"]["id"], }, ) assert response.status_code == 201 payload = response.json() await wait_for_workflow_status(client, payload["order_id"], "waiting_review") review_response = await client.post( f"/api/v1/reviews/{payload['order_id']}/submit", json={"decision": "approve", "reviewer_id": 77, "comment": "通过"}, ) assert review_response.status_code == 200 handle = env.client.get_workflow_handle(payload["workflow_id"]) result = await handle.result() assert result["status"] == "succeeded" @pytest.mark.asyncio async def test_mid_end_order_waits_review_then_approves(api_runtime): """Mid-end orders should pause for review and continue after approval.""" client, env = api_runtime payload = await create_mid_end_order(client) await wait_for_workflow_status(client, payload["order_id"], "waiting_review") pending_response = await client.get("/api/v1/reviews/pending") assert pending_response.status_code == 200 assert any(item["order_id"] == payload["order_id"] for item in pending_response.json()) review_response = await client.post( f"/api/v1/reviews/{payload['order_id']}/submit", json={"decision": "approve", "reviewer_id": 77, "comment": "通过"}, ) assert review_response.status_code == 200 handle = env.client.get_workflow_handle(payload["workflow_id"]) result = await handle.result() assert result["status"] == "succeeded" order_response = await client.get(f"/api/v1/orders/{payload['order_id']}") assert order_response.status_code == 200 assert order_response.json()["status"] == "succeeded" @pytest.mark.asyncio async def test_library_upload_presign_returns_direct_upload_metadata(api_runtime): """Library upload presign should return S3 direct-upload metadata and a public URL.""" client, _ = api_runtime response = await client.post( "/api/v1/library/uploads/presign", json={ "resource_type": "model", "file_role": "original", "file_name": "ava.png", "content_type": "image/png", }, ) assert response.status_code == 200 payload = response.json() assert payload["method"] == "PUT" assert payload["storage_key"].startswith("library/models/") assert payload["public_url"].startswith("https://") assert "library/models/" in payload["upload_url"] @pytest.mark.asyncio async def test_library_resource_can_be_created_and_listed(api_runtime): """Creating a library resource should persist original, thumbnail, and gallery files.""" client, _ = api_runtime create_response = await client.post( "/api/v1/library/resources", json={ "resource_type": "model", "name": "Ava Studio", "description": "棚拍女模特", "tags": ["女装", "棚拍"], "gender": "female", "age_group": "adult", "files": [ { "file_role": "original", "storage_key": "library/models/ava/original.png", "public_url": "https://images.marcusd.me/library/models/ava/original.png", "mime_type": "image/png", "size_bytes": 1024, "sort_order": 0, }, { "file_role": "thumbnail", "storage_key": "library/models/ava/thumb.png", "public_url": "https://images.marcusd.me/library/models/ava/thumb.png", "mime_type": "image/png", "size_bytes": 256, "sort_order": 0, }, { "file_role": "gallery", "storage_key": "library/models/ava/gallery-1.png", "public_url": "https://images.marcusd.me/library/models/ava/gallery-1.png", "mime_type": "image/png", "size_bytes": 2048, "sort_order": 1, }, ], }, ) assert create_response.status_code == 201 created = create_response.json() assert created["resource_type"] == "model" assert created["pose_id"] is None assert created["cover_url"] == "https://images.marcusd.me/library/models/ava/thumb.png" assert created["original_url"] == "https://images.marcusd.me/library/models/ava/original.png" assert len(created["files"]) == 3 list_response = await client.get("/api/v1/library/resources", params={"resource_type": "model"}) assert list_response.status_code == 200 listing = list_response.json() assert listing["total"] == 1 assert listing["items"][0]["name"] == "Ava Studio" assert listing["items"][0]["gender"] == "female" assert listing["items"][0]["pose_id"] is None @pytest.mark.asyncio async def test_library_resource_list_supports_type_specific_filters(api_runtime): """Library listing should support resource-type-specific filter fields.""" client, _ = api_runtime payloads = [ { "resource_type": "scene", "name": "Loft Window", "description": "暖调室内场景", "tags": ["室内"], "environment": "indoor", "files": [ { "file_role": "original", "storage_key": "library/scenes/loft/original.png", "public_url": "https://images.marcusd.me/library/scenes/loft/original.png", "mime_type": "image/png", "size_bytes": 1024, "sort_order": 0, }, { "file_role": "thumbnail", "storage_key": "library/scenes/loft/thumb.png", "public_url": "https://images.marcusd.me/library/scenes/loft/thumb.png", "mime_type": "image/png", "size_bytes": 256, "sort_order": 0, }, ], }, { "resource_type": "scene", "name": "Garden Walk", "description": "自然光室外场景", "tags": ["室外"], "environment": "outdoor", "files": [ { "file_role": "original", "storage_key": "library/scenes/garden/original.png", "public_url": "https://images.marcusd.me/library/scenes/garden/original.png", "mime_type": "image/png", "size_bytes": 1024, "sort_order": 0, }, { "file_role": "thumbnail", "storage_key": "library/scenes/garden/thumb.png", "public_url": "https://images.marcusd.me/library/scenes/garden/thumb.png", "mime_type": "image/png", "size_bytes": 256, "sort_order": 0, }, ], }, ] for payload in payloads: response = await client.post("/api/v1/library/resources", json=payload) assert response.status_code == 201 indoor_response = await client.get( "/api/v1/library/resources", params={"resource_type": "scene", "environment": "indoor"}, ) assert indoor_response.status_code == 200 indoor_payload = indoor_response.json() assert indoor_payload["total"] == 1 assert indoor_payload["items"][0]["name"] == "Loft Window" @pytest.mark.asyncio @pytest.mark.parametrize( ("decision", "expected_step"), [ ("rerun_scene", "scene"), ("rerun_face", "face"), ("rerun_fusion", "fusion"), ], ) async def test_mid_end_rerun_paths_return_to_review(api_runtime, decision: str, expected_step: str): """Each rerun decision should branch back to the correct step and pause again for review.""" client, env = api_runtime payload = await create_mid_end_order(client) await wait_for_workflow_status(client, payload["order_id"], "waiting_review") review_response = await client.post( f"/api/v1/reviews/{payload['order_id']}/submit", json={"decision": decision, "reviewer_id": 77, "comment": f"trigger {decision}"}, ) assert review_response.status_code == 200 await wait_for_step_count(client, payload["order_id"], expected_step, 2) await wait_for_step_count(client, payload["order_id"], "review", 2) workflow_payload = await wait_for_workflow_status(client, payload["order_id"], "waiting_review") assert workflow_payload["workflow_status"] == "waiting_review" approve_response = await client.post( f"/api/v1/reviews/{payload['order_id']}/submit", json={"decision": "approve", "reviewer_id": 77, "comment": "批准最终结果"}, ) assert approve_response.status_code == 200 handle = env.client.get_workflow_handle(payload["workflow_id"]) result = await handle.result() assert result["status"] == "succeeded" @pytest.mark.asyncio async def test_mid_end_order_registers_manual_revision_and_updates_pending_queue(api_runtime): """Registering a manual revision should keep the workflow paused and mark the queue item accordingly.""" client, _ = api_runtime payload = await create_mid_end_order(client) workflow_payload = await wait_for_workflow_status(client, payload["order_id"], "waiting_review") parent_asset_id = workflow_payload["steps"][-2]["output_json"]["candidate_asset_ids"][0] register_response = await client.post( f"/api/v1/orders/{payload['order_id']}/revisions", json={ "parent_asset_id": parent_asset_id, "uploaded_uri": "mock://manual-revision-v1", "reviewer_id": 88, "comment": "人工修订第一版", }, ) assert register_response.status_code == 201 register_payload = register_response.json() assert register_payload["order_id"] == payload["order_id"] assert register_payload["parent_asset_id"] == parent_asset_id assert register_payload["root_asset_id"] == parent_asset_id assert register_payload["version_no"] == 1 assert register_payload["review_task_status"] == "revision_uploaded" pending_response = await client.get("/api/v1/reviews/pending") assert pending_response.status_code == 200 queue_item = next(item for item in pending_response.json() if item["order_id"] == payload["order_id"]) assert queue_item["review_task_status"] == "revision_uploaded" assert queue_item["latest_revision_asset_id"] == register_payload["asset_id"] assert queue_item["revision_count"] == 1 @pytest.mark.asyncio async def test_mid_end_order_lists_single_line_revision_chain(api_runtime): """Listing revisions should return the uploaded manual revision chain in version order.""" client, _ = api_runtime payload = await create_mid_end_order(client) workflow_payload = await wait_for_workflow_status(client, payload["order_id"], "waiting_review") root_asset_id = workflow_payload["steps"][-2]["output_json"]["candidate_asset_ids"][0] first_response = await client.post( f"/api/v1/orders/{payload['order_id']}/revisions", json={ "parent_asset_id": root_asset_id, "uploaded_uri": "mock://manual-revision-v1", "reviewer_id": 88, "comment": "人工修订第一版", }, ) assert first_response.status_code == 201 first_payload = first_response.json() second_response = await client.post( f"/api/v1/orders/{payload['order_id']}/revisions", json={ "parent_asset_id": first_payload["asset_id"], "uploaded_uri": "mock://manual-revision-v2", "reviewer_id": 88, "comment": "人工修订第二版", }, ) assert second_response.status_code == 201 second_payload = second_response.json() chain_response = await client.get(f"/api/v1/orders/{payload['order_id']}/revisions") assert chain_response.status_code == 200 chain_payload = chain_response.json() assert chain_payload["order_id"] == payload["order_id"] assert [item["asset_id"] for item in chain_payload["items"]] == [ first_payload["asset_id"], second_payload["asset_id"], ] assert [item["version_no"] for item in chain_payload["items"]] == [1, 2] assert chain_payload["items"][0]["parent_asset_id"] == root_asset_id assert chain_payload["items"][1]["parent_asset_id"] == first_payload["asset_id"] assert chain_payload["items"][-1]["is_current_version"] is True @pytest.mark.asyncio async def test_mid_end_order_confirms_manual_revision_and_exports_revision_asset(api_runtime): """Confirming a manual revision should resume the workflow and export the revision asset.""" client, env = api_runtime payload = await create_mid_end_order(client) workflow_payload = await wait_for_workflow_status(client, payload["order_id"], "waiting_review") parent_asset_id = workflow_payload["steps"][-2]["output_json"]["candidate_asset_ids"][0] register_response = await client.post( f"/api/v1/orders/{payload['order_id']}/revisions", json={ "parent_asset_id": parent_asset_id, "uploaded_uri": "mock://manual-revision-v1", "reviewer_id": 88, "comment": "人工修订第一版", }, ) assert register_response.status_code == 201 register_payload = register_response.json() confirm_response = await client.post( f"/api/v1/reviews/{payload['order_id']}/confirm-revision", json={ "reviewer_id": 88, "comment": "确认继续流水线", }, ) assert confirm_response.status_code == 200 confirm_payload = confirm_response.json() assert confirm_payload["revision_asset_id"] == register_payload["asset_id"] assert confirm_payload["decision"] == "approve" assert confirm_payload["status"] == "submitted" handle = env.client.get_workflow_handle(payload["workflow_id"]) result = await handle.result() assert result["status"] == "succeeded" assert result["final_asset_id"] is not None order_response = await client.get(f"/api/v1/orders/{payload['order_id']}") assert order_response.status_code == 200 order_payload = order_response.json() assert order_payload["status"] == "succeeded" assert order_payload["final_asset"]["asset_type"] == "final" assert order_payload["final_asset"]["metadata_json"]["source_asset_id"] == register_payload["asset_id"] @pytest.mark.asyncio async def test_orders_list_returns_recent_orders_with_revision_summary(api_runtime): """Orders list should expose pagination metadata, filtering, and revision summary fields.""" client, env = api_runtime low_resources = await create_workflow_resources(client, include_scene=True) low_order = await client.post( "/api/v1/orders", json={ "customer_level": "low", "service_mode": "auto_basic", "model_id": low_resources["model"]["id"], "pose_id": 11, "garment_asset_id": low_resources["garment"]["id"], "scene_ref_asset_id": low_resources["scene"]["id"], }, ) assert low_order.status_code == 201 low_payload = low_order.json() await env.client.get_workflow_handle(low_payload["workflow_id"]).result() mid_payload = await create_mid_end_order(client) workflow_payload = await wait_for_workflow_status(client, mid_payload["order_id"], "waiting_review") parent_asset_id = workflow_payload["steps"][-2]["output_json"]["candidate_asset_ids"][0] register_response = await client.post( f"/api/v1/orders/{mid_payload['order_id']}/revisions", json={ "parent_asset_id": parent_asset_id, "uploaded_uri": "mock://manual-revision-v1", "reviewer_id": 99, "comment": "人工修订第一版", }, ) assert register_response.status_code == 201 list_response = await client.get("/api/v1/orders", params={"page": 1, "limit": 1}) assert list_response.status_code == 200 first_page = list_response.json() assert first_page["page"] == 1 assert first_page["limit"] == 1 assert first_page["total"] == 2 assert first_page["total_pages"] == 2 assert [item["order_id"] for item in first_page["items"]] == [mid_payload["order_id"]] assert first_page["items"][0]["workflow_id"] == mid_payload["workflow_id"] assert first_page["items"][0]["review_task_status"] == "revision_uploaded" assert first_page["items"][0]["latest_revision_version"] == 1 assert first_page["items"][0]["revision_count"] == 1 assert first_page["items"][0]["pending_manual_confirm"] is True second_page_response = await client.get("/api/v1/orders", params={"page": 2, "limit": 1}) assert second_page_response.status_code == 200 second_page = second_page_response.json() assert second_page["page"] == 2 assert second_page["limit"] == 1 assert second_page["total"] == 2 assert second_page["total_pages"] == 2 assert [item["order_id"] for item in second_page["items"]] == [low_payload["order_id"]] assert second_page["items"][0]["status"] == "succeeded" filtered_response = await client.get( "/api/v1/orders", params={"page": 1, "limit": 10, "status": "waiting_review"} ) assert filtered_response.status_code == 200 filtered_payload = filtered_response.json() assert filtered_payload["page"] == 1 assert filtered_payload["limit"] == 10 assert filtered_payload["total"] == 1 assert filtered_payload["total_pages"] == 1 assert [item["order_id"] for item in filtered_payload["items"]] == [mid_payload["order_id"]] query_response = await client.get( "/api/v1/orders", params={"page": 1, "limit": 10, "query": mid_payload["workflow_id"]}, ) assert query_response.status_code == 200 query_payload = query_response.json() assert query_payload["total"] == 1 assert [item["order_id"] for item in query_payload["items"]] == [mid_payload["order_id"]] @pytest.mark.asyncio async def test_workflows_list_returns_recent_runs_with_failure_count(api_runtime): """Workflow list should expose pagination metadata, filtering, and revision summary.""" client, env = api_runtime low_resources = await create_workflow_resources(client, include_scene=True) low_order = await client.post( "/api/v1/orders", json={ "customer_level": "low", "service_mode": "auto_basic", "model_id": low_resources["model"]["id"], "pose_id": 21, "garment_asset_id": low_resources["garment"]["id"], "scene_ref_asset_id": low_resources["scene"]["id"], }, ) assert low_order.status_code == 201 low_payload = low_order.json() await env.client.get_workflow_handle(low_payload["workflow_id"]).result() mid_payload = await create_mid_end_order(client) workflow_payload = await wait_for_workflow_status(client, mid_payload["order_id"], "waiting_review") parent_asset_id = workflow_payload["steps"][-2]["output_json"]["candidate_asset_ids"][0] register_response = await client.post( f"/api/v1/orders/{mid_payload['order_id']}/revisions", json={ "parent_asset_id": parent_asset_id, "uploaded_uri": "mock://manual-revision-v1", "reviewer_id": 77, "comment": "人工修订第一版", }, ) assert register_response.status_code == 201 list_response = await client.get("/api/v1/workflows", params={"page": 1, "limit": 1}) assert list_response.status_code == 200 first_page = list_response.json() assert first_page["page"] == 1 assert first_page["limit"] == 1 assert first_page["total"] == 2 assert first_page["total_pages"] == 2 assert [item["order_id"] for item in first_page["items"]] == [mid_payload["order_id"]] assert first_page["items"][0]["workflow_id"] == mid_payload["workflow_id"] assert first_page["items"][0]["workflow_status"] == "waiting_review" assert first_page["items"][0]["review_task_status"] == "revision_uploaded" assert first_page["items"][0]["latest_revision_version"] == 1 assert first_page["items"][0]["revision_count"] == 1 assert first_page["items"][0]["pending_manual_confirm"] is True assert first_page["items"][0]["failure_count"] == 0 second_page_response = await client.get("/api/v1/workflows", params={"page": 2, "limit": 1}) assert second_page_response.status_code == 200 second_page = second_page_response.json() assert second_page["page"] == 2 assert second_page["limit"] == 1 assert second_page["total"] == 2 assert second_page["total_pages"] == 2 assert [item["order_id"] for item in second_page["items"]] == [low_payload["order_id"]] assert second_page["items"][0]["workflow_status"] == "succeeded" filtered_response = await client.get( "/api/v1/workflows", params={"page": 1, "limit": 10, "status": "waiting_review"} ) assert filtered_response.status_code == 200 filtered_payload = filtered_response.json() assert filtered_payload["page"] == 1 assert filtered_payload["limit"] == 10 assert filtered_payload["total"] == 1 assert filtered_payload["total_pages"] == 1 assert [item["order_id"] for item in filtered_payload["items"]] == [mid_payload["order_id"]] query_response = await client.get( "/api/v1/workflows", params={"page": 1, "limit": 10, "query": str(low_payload["order_id"])}, ) assert query_response.status_code == 200 query_payload = query_response.json() assert query_payload["total"] == 1 assert [item["order_id"] for item in query_payload["items"]] == [low_payload["order_id"]]