source/PaddleOCR/tests/api_client/test_http.py

# Copyright (c) 2026 PaddlePaddle Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import json
from http.server import BaseHTTPRequestHandler, HTTPServer
from threading import Thread

import pytest

from paddleocr._api_client._http import API_PATH, DEFAULT_BASE_URL, HTTPClient
from paddleocr._api_client.client import PaddleOCRClient
from paddleocr._api_client.errors import (
    APIError,
    AuthError,
    InvalidRequestError,
    NetworkError,
    RateLimitError,
    RequestTimeoutError,
    ServiceUnavailableError,
)
from paddleocr._api_client.models import Model


class _MockHandler(BaseHTTPRequestHandler):
    """Records requests and responds with canned JSON."""

    requests_log = []
    response_body = {"code": 0, "data": {"jobId": "job-1"}}
    response_status = 200

    def do_POST(self):
        content_length = int(self.headers.get("Content-Length", 0))
        body = self.rfile.read(content_length) if content_length else b""
        self.__class__.requests_log.append(
            {
                "method": "POST",
                "path": self.path,
                "headers": dict(self.headers),
                "body": body,
            }
        )
        self.send_response(self.__class__.response_status)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(json.dumps(self.__class__.response_body).encode())

    def do_GET(self):
        self.__class__.requests_log.append(
            {"method": "GET", "path": self.path, "headers": dict(self.headers)}
        )
        self.send_response(self.__class__.response_status)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(json.dumps(self.__class__.response_body).encode())

    def log_message(self, format, *args):
        pass


@pytest.fixture()
def mock_server():
    _MockHandler.requests_log = []
    _MockHandler.response_body = {"code": 0, "data": {"jobId": "job-1"}}
    _MockHandler.response_status = 200

    server = HTTPServer(("127.0.0.1", 0), _MockHandler)
    port = server.server_address[1]
    thread = Thread(target=server.serve_forever, daemon=True)
    thread.start()
    yield f"http://127.0.0.1:{port}", _MockHandler
    server.shutdown()


def test_base_url_appends_api_path(mock_server):
    base_url, handler = mock_server
    client = HTTPClient("test-token", base_url, timeout=5.0)

    client.submit_url("PP-OCRv5", "https://example.test/file.pdf", {})

    assert len(handler.requests_log) == 1
    assert handler.requests_log[0]["path"] == API_PATH


def test_get_job_status_url(mock_server):
    base_url, handler = mock_server
    handler.response_body = {"code": 0, "data": {"jobId": "job-1", "state": "running"}}
    client = HTTPClient("test-token", base_url, timeout=5.0)

    client.get_job_status("job-123")

    assert handler.requests_log[0]["path"] == f"{API_PATH}/job-123"


def test_get_batch_status_url(mock_server):
    base_url, handler = mock_server
    handler.response_body = {"code": 0, "data": {"batchId": "b-1", "jobs": []}}
    client = HTTPClient("test-token", base_url, timeout=5.0)

    client.get_batch_status("batch-abc")

    assert handler.requests_log[0]["path"] == f"{API_PATH}/batch/batch-abc"


def test_base_url_trailing_slashes_normalized(mock_server):
    base_url, handler = mock_server
    client = HTTPClient("test-token", base_url + "///", timeout=5.0)

    client.submit_url("PP-OCRv5", "https://example.test/file.pdf", {})

    assert handler.requests_log[0]["path"] == API_PATH


def test_authorization_header_sent(mock_server):
    base_url, handler = mock_server
    client = HTTPClient("my-secret-token", base_url, timeout=5.0)

    client.submit_url("PP-OCRv5", "https://example.test/file.pdf", {})

    assert (
        handler.requests_log[0]["headers"]["Authorization"] == "Bearer my-secret-token"
    )


def test_client_platform_header(mock_server):
    base_url, handler = mock_server
    client = HTTPClient("token", base_url, timeout=5.0, client_platform="sdk-test")

    client.submit_url("PP-OCRv5", "https://example.test/file.pdf", {})

    assert handler.requests_log[0]["headers"]["Client-Platform"] == "sdk-test"


def test_submit_url_posts_correct_body(mock_server):
    base_url, handler = mock_server
    client = HTTPClient("token", base_url, timeout=5.0)

    job_id = client.submit_url(
        "PP-OCRv5",
        "https://example.test/file.pdf",
        {"visualize": True},
        page_ranges="1-3",
        batch_id="batch-1",
    )

    assert job_id == "job-1"
    body = json.loads(handler.requests_log[0]["body"])
    assert body == {
        "fileUrl": "https://example.test/file.pdf",
        "model": "PP-OCRv5",
        "optionalPayload": {"visualize": True},
        "pageRanges": "1-3",
        "batchId": "batch-1",
    }


def test_error_mapping_401(mock_server):
    base_url, handler = mock_server
    handler.response_status = 401
    handler.response_body = {"code": 401, "msg": "Unauthorized"}
    client = HTTPClient("bad-token", base_url, timeout=5.0)

    with pytest.raises(AuthError):
        client.submit_url("PP-OCRv5", "https://example.test/file.pdf", {})


def test_error_mapping_400(mock_server):
    base_url, handler = mock_server
    handler.response_status = 400
    handler.response_body = {"code": 400, "msg": "Bad request"}
    client = HTTPClient("token", base_url, timeout=5.0)

    with pytest.raises(InvalidRequestError):
        client.submit_url("PP-OCRv5", "https://example.test/file.pdf", {})


def test_error_mapping_429(mock_server):
    base_url, handler = mock_server
    handler.response_status = 429
    handler.response_body = {"code": 429, "msg": "Rate limited"}
    client = HTTPClient("token", base_url, timeout=5.0)

    with pytest.raises(RateLimitError):
        client.submit_url("PP-OCRv5", "https://example.test/file.pdf", {})


def test_error_mapping_503(mock_server):
    base_url, handler = mock_server
    handler.response_status = 503
    handler.response_body = {"code": 503, "msg": "Service unavailable"}
    client = HTTPClient("token", base_url, timeout=5.0)

    with pytest.raises(ServiceUnavailableError):
        client.submit_url("PP-OCRv5", "https://example.test/file.pdf", {})


def test_error_mapping_500(mock_server):
    base_url, handler = mock_server
    handler.response_status = 500
    handler.response_body = {"code": 500, "msg": "Internal error"}
    client = HTTPClient("token", base_url, timeout=5.0)

    with pytest.raises(APIError):
        client.submit_url("PP-OCRv5", "https://example.test/file.pdf", {})


def test_paddleocr_client_requires_token(monkeypatch):
    monkeypatch.delenv("PADDLEOCR_ACCESS_TOKEN", raising=False)
    with pytest.raises(AuthError):
        PaddleOCRClient()


def test_paddleocr_client_reads_env_token(monkeypatch):
    monkeypatch.setenv("PADDLEOCR_ACCESS_TOKEN", "env-token")
    client = PaddleOCRClient()
    client.close()


def test_default_base_url_is_host_only():
    assert "/api/" not in DEFAULT_BASE_URL
    assert DEFAULT_BASE_URL == "https://paddleocr.aistudio-app.com"


def test_paddleocr_client_client_platform_header(monkeypatch):
    class FakeResponse:
        status_code = 200
        text = ""

        def json(self):
            return {"code": 0, "data": {"jobId": "job-1"}}

    class FakeSession:
        def __init__(self):
            self.headers = {}
            self.captured_headers = None

        def post(self, *args, **kwargs):
            self.captured_headers = dict(self.headers)
            return FakeResponse()

        def close(self):
            return None

    fake_session = FakeSession()
    monkeypatch.setattr(
        "paddleocr._api_client._http.requests.Session",
        lambda: fake_session,
    )

    client = PaddleOCRClient(token="token", client_platform="my-app")
    job = client.submit_ocr(file_url="https://example.test/input.pdf")

    assert job.job_id == "job-1"
    assert fake_session.captured_headers["Client-Platform"] == "my-app"