Spaces:
Sleeping
Sleeping
| """ | |
| Test suite for Response Inspector | |
| Tests response analysis logic for determining credit actions: | |
| - Confirm credits (successful operations) | |
| - Refund credits (failed operations) | |
| - Keep reserved (pending async operations) | |
| """ | |
| import pytest | |
| import json | |
| from fastapi import Response | |
| from services.credit_service.response_inspector import ResponseInspector | |
| # ============================================================================= | |
| # Synchronous Endpoint Tests | |
| # ============================================================================= | |
| def test_should_confirm_sync_success(): | |
| """Test confirmation for successful sync operation (200).""" | |
| response = Response(content=json.dumps({"result": "success"}), status_code=200) | |
| inspector = ResponseInspector() | |
| assert inspector.should_confirm(response, "sync", {"result": "success"}) is True | |
| assert inspector.should_refund(response, "sync", {"result": "success"}) is False | |
| def test_should_confirm_sync_created(): | |
| """Test confirmation for sync operation (201).""" | |
| response = Response(content=json.dumps({"id": "123"}), status_code=201) | |
| inspector = ResponseInspector() | |
| assert inspector.should_confirm(response, "sync", {"id": "123"}) is True | |
| def test_should_refund_sync_client_error(): | |
| """Test refund for sync client error (400).""" | |
| response = Response( | |
| content=json.dumps({"detail": "Invalid request"}), | |
| status_code=400 | |
| ) | |
| inspector = ResponseInspector() | |
| assert inspector.should_confirm(response, "sync", {"detail": "Invalid request"}) is False | |
| assert inspector.should_refund(response, "sync", {"detail": "Invalid request"}) is True | |
| def test_should_refund_sync_server_error(): | |
| """Test refund for sync server error (500).""" | |
| response = Response( | |
| content=json.dumps({"detail": "Internal error"}), | |
| status_code=500 | |
| ) | |
| inspector = ResponseInspector() | |
| assert inspector.should_refund(response, "sync", {"detail": "Internal error"}) is True | |
| # ============================================================================= | |
| # Asynchronous Endpoint Tests - Job Creation | |
| # ============================================================================= | |
| def test_async_job_creation_success(): | |
| """Test async job creation - should keep reserved.""" | |
| response = Response( | |
| content=json.dumps({"job_id": "job_123", "status": "queued"}), | |
| status_code=200 | |
| ) | |
| inspector = ResponseInspector() | |
| response_data = {"job_id": "job_123", "status": "queued"} | |
| # Job created successfully, but not complete yet | |
| assert inspector.should_confirm(response, "async", response_data) is False | |
| assert inspector.should_refund(response, "async", response_data) is False | |
| def test_async_job_creation_failure(): | |
| """Test async job creation failure - should refund.""" | |
| response = Response( | |
| content=json.dumps({"detail": "Validation failed"}), | |
| status_code=400 | |
| ) | |
| inspector = ResponseInspector() | |
| response_data = {"detail": "Validation failed"} | |
| # Job creation failed, refund credits | |
| assert inspector.should_confirm(response, "async", response_data) is False | |
| assert inspector.should_refund(response, "async", response_data) is True | |
| # ============================================================================= | |
| # Asynchronous Endpoint Tests - Status Checks | |
| # ============================================================================= | |
| def test_async_status_completed(): | |
| """Test async job status check - completed.""" | |
| response = Response( | |
| content=json.dumps({"job_id": "job_123", "status": "completed", "result": "..."}), | |
| status_code=200 | |
| ) | |
| inspector = ResponseInspector() | |
| response_data = {"job_id": "job_123", "status": "completed", "result": "..."} | |
| # Job completed, confirm credits | |
| assert inspector.should_confirm(response, "async", response_data) is True | |
| assert inspector.should_refund(response, "async", response_data) is False | |
| def test_async_status_processing(): | |
| """Test async job status check - still processing.""" | |
| response = Response( | |
| content=json.dumps({"job_id": "job_123", "status": "processing"}), | |
| status_code=200 | |
| ) | |
| inspector = ResponseInspector() | |
| response_data = {"job_id": "job_123", "status": "processing"} | |
| # Job still processing, keep reserved | |
| assert inspector.should_confirm(response, "async", response_data) is False | |
| assert inspector.should_refund(response, "async", response_data) is False | |
| def test_async_status_queued(): | |
| """Test async job status check - still queued.""" | |
| response = Response( | |
| content=json.dumps({"job_id": "job_123", "status": "queued", "position": 5}), | |
| status_code=200 | |
| ) | |
| inspector = ResponseInspector() | |
| response_data = {"job_id": "job_123", "status": "queued", "position": 5} | |
| # Job still queued, keep reserved | |
| assert inspector.should_confirm(response, "async", response_data) is False | |
| assert inspector.should_refund(response, "async", response_data) is False | |
| def test_async_status_failed_refundable(): | |
| """Test async job status check - failed with refundable error.""" | |
| response = Response( | |
| content=json.dumps({ | |
| "job_id": "job_123", | |
| "status": "failed", | |
| "error_message": "API_KEY_INVALID - The API key is invalid" | |
| }), | |
| status_code=200 | |
| ) | |
| inspector = ResponseInspector() | |
| response_data = { | |
| "job_id": "job_123", | |
| "status": "failed", | |
| "error_message": "API_KEY_INVALID - The API key is invalid" | |
| } | |
| # Job failed with refundable error, refund credits | |
| assert inspector.should_confirm(response, "async", response_data) is False | |
| assert inspector.should_refund(response, "async", response_data) is True | |
| def test_async_status_failed_non_refundable(): | |
| """Test async job status check - failed with non-refundable error.""" | |
| response = Response( | |
| content=json.dumps({ | |
| "job_id": "job_123", | |
| "status": "failed", | |
| "error_message": "SAFETY: Content blocked by safety filters" | |
| }), | |
| status_code=200 | |
| ) | |
| inspector = ResponseInspector() | |
| response_data = { | |
| "job_id": "job_123", | |
| "status": "failed", | |
| "error_message": "SAFETY: Content blocked by safety filters" | |
| } | |
| # Job failed with non-refundable error, confirm credits (keep deducted) | |
| assert inspector.should_confirm(response, "async", response_data) is False | |
| assert inspector.should_refund(response, "async", response_data) is False | |
| # ============================================================================= | |
| # Refund Reason Tests | |
| # ============================================================================= | |
| def test_get_refund_reason_server_error(): | |
| """Test refund reason for server error.""" | |
| response = Response(content="", status_code=500) | |
| inspector = ResponseInspector() | |
| reason = inspector.get_refund_reason(response, None) | |
| assert "Server error: 500" == reason | |
| def test_get_refund_reason_client_error(): | |
| """Test refund reason for client error.""" | |
| response = Response( | |
| content=json.dumps({"detail": "Invalid input"}), | |
| status_code=400 | |
| ) | |
| inspector = ResponseInspector() | |
| response_data = {"detail": "Invalid input"} | |
| reason = inspector.get_refund_reason(response, response_data) | |
| assert "Request error: Invalid input" == reason | |
| def test_get_refund_reason_job_failure(): | |
| """Test refund reason for job failure.""" | |
| response = Response(content="", status_code=200) | |
| inspector = ResponseInspector() | |
| response_data = {"error_message": "API timeout after 60 seconds"} | |
| reason = inspector.get_refund_reason(response, response_data) | |
| assert "Job failed" in reason | |
| assert "API timeout" in reason | |
| def test_get_refund_reason_unknown(): | |
| """Test refund reason when unknown.""" | |
| response = Response(content="", status_code=200) | |
| inspector = ResponseInspector() | |
| reason = inspector.get_refund_reason(response, None) | |
| assert reason == "Unknown error" | |
| # ============================================================================= | |
| # Response Body Parsing Tests | |
| # ============================================================================= | |
| def test_parse_response_body_valid_json(): | |
| """Test parsing valid JSON response body.""" | |
| body = b'{"key": "value", "number": 123}' | |
| inspector = ResponseInspector() | |
| parsed = inspector.parse_response_body(body) | |
| assert parsed == {"key": "value", "number": 123} | |
| def test_parse_response_body_invalid_json(): | |
| """Test parsing invalid JSON response body.""" | |
| body = b'not json' | |
| inspector = ResponseInspector() | |
| parsed = inspector.parse_response_body(body) | |
| assert parsed is None | |
| def test_parse_response_body_empty(): | |
| """Test parsing empty response body.""" | |
| body = b'' | |
| inspector = ResponseInspector() | |
| parsed = inspector.parse_response_body(body) | |
| assert parsed is None | |
| # ============================================================================= | |
| # Edge Cases | |
| # ============================================================================= | |
| def test_free_endpoint(): | |
| """Test free endpoint (no credit cost).""" | |
| response = Response(content=json.dumps({"result": "ok"}), status_code=200) | |
| inspector = ResponseInspector() | |
| # Free endpoints shouldn't trigger credit actions | |
| # (handled by middleware checking cost=0, but inspector should be safe) | |
| assert inspector.should_confirm(response, "free", {"result": "ok"}) is False | |
| def test_async_missing_status_field(): | |
| """Test async response missing status field.""" | |
| response = Response( | |
| content=json.dumps({"job_id": "abc", "message": "Processing"}), | |
| status_code=200 | |
| ) | |
| inspector = ResponseInspector() | |
| response_data = {"job_id": "abc", "message": "Processing"} | |
| # No status field, should keep reserved | |
| assert inspector.should_confirm(response, "async", response_data) is False | |
| assert inspector.should_refund(response, "async", response_data) is False | |
| def test_async_none_response_data(): | |
| """Test async with None response data.""" | |
| response = Response(content=b"", status_code=500) | |
| inspector = ResponseInspector() | |
| # Server error with no parseable response | |
| assert inspector.should_confirm(response, "async", None) is False | |
| assert inspector.should_refund(response, "async", None) is True | |