proposa-ai/scripts/tests/test_api_callback.py

import json
import sys
import unittest
from io import StringIO
from unittest.mock import patch

from scripts import test_api_callback


class ApiCallbackScriptTests(unittest.TestCase):
    def test_completed_payload_matches_current_api_contract(self):
        payload = test_api_callback.build_callback_payload("completed")

        self.assertEqual(payload["status"], "completed")
        self.assertIn("final_review_url", payload)
        self.assertNotIn("error", payload)
        self.assertEqual(
            set(payload),
            {
                "request_id",
                "txbId",
                "status",
                "resultPath",
                "files",
                "final_review_url",
            },
        )

    def test_failed_payload_matches_current_api_contract(self):
        payload = test_api_callback.build_callback_payload("failed")

        self.assertEqual(payload["status"], "failed")
        self.assertTrue(payload["resultPath"].startswith("error: "))
        self.assertIn("error", payload)
        self.assertNotIn("final_review_url", payload)

    def test_send_callback_posts_utf8_json_and_returns_response(self):
        class FakeResponse:
            status = 200

            def __enter__(self):
                return self

            def __exit__(self, *_args):
                return False

            def read(self):
                return '{"code":0,"message":"成功"}'.encode("utf-8")

        payload = test_api_callback.build_callback_payload("completed")
        with patch.object(
            test_api_callback, "urlopen", return_value=FakeResponse()
        ) as mocked_urlopen:
            response = test_api_callback.send_callback(
                test_api_callback.CALLBACK_URL, payload
            )

        request = mocked_urlopen.call_args.args[0]
        self.assertEqual(request.method, "POST")
        self.assertEqual(
            request.headers["Content-type"], "application/json; charset=utf-8"
        )
        self.assertEqual(json.loads(request.data.decode("utf-8")), payload)
        self.assertEqual(response.status, 200)
        self.assertIn("成功", response.body)

    def test_main_treats_http_200_with_business_400_as_failure(self):
        response = test_api_callback.CallbackResponse(
            status=200,
            body='{"code":400,"message":"txbId不存在"}',
        )
        with patch.object(sys, "argv", ["test_api_callback.py", "--send"]), patch.object(
            test_api_callback, "send_callback", return_value=response
        ), patch("sys.stdout", new_callable=StringIO) as output:
            exit_code = test_api_callback.main()

        self.assertEqual(exit_code, 1)
        self.assertIn("业务码: 400（失败）", output.getvalue())


if __name__ == "__main__":
    unittest.main()