mindee-api-python/tests/v1/api/test_async_response.py at e16f8434815a5bbefe5660121395c5b6b00fa793 · mindee/mindee-api-python · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
import json

import pytest
import requests

from mindee.client import Client
from mindee.input.sources.path_input import PathInput
from mindee.mindee_http.response_validation import is_valid_async_response
from mindee.parsing.common.api_request import RequestStatus
from mindee.parsing.common.async_predict_response import AsyncPredictResponse
from mindee.product.invoice_splitter.invoice_splitter_v1 import InvoiceSplitterV1
from tests.utils import V1_DATA_DIR, V1_PRODUCT_DATA_DIR

ASYNC_DIR = V1_DATA_DIR / "async"

FILE_PATH_POST_SUCCESS = ASYNC_DIR / "post_success.json"
FILE_PATH_POST_FAIL = ASYNC_DIR / "post_fail_forbidden.json"
FILE_PATH_GET_PROCESSING = ASYNC_DIR / "get_processing.json"
FILE_PATH_GET_COMPLETED = ASYNC_DIR / "get_completed.json"
FILE_PATH_GET_FAILED_JOB = ASYNC_DIR / "get_failed_job_error.json"


class FakeResponse(requests.Response):
    def __init__(self, json_data, _status_code=200):
        super().__init__()
        self._json_data = json_data
        self.status_code = _status_code
        self._ok = True

    def set_ok_status(self, ok_status):
        self._ok = ok_status

    @property
    def ok(self):
        return self._ok

    @property
    def content(self) -> str:
        return json.dumps(self._json_data)


@pytest.fixture
def dummy_file_input() -> PathInput:
    file_input = PathInput(
        V1_PRODUCT_DATA_DIR / "invoice_splitter" / "default_sample.pdf"
    )
    return file_input


@pytest.fixture
def dummy_client() -> Client:
    return Client(api_key="dummy")


def test_async_response_post_success():
    response = json.load(open(FILE_PATH_POST_SUCCESS))
    parsed_response = AsyncPredictResponse(InvoiceSplitterV1, response)
    fake_response = FakeResponse(response)
    fake_response.set_ok_status(True)
    assert is_valid_async_response(fake_response) is True
    assert parsed_response.job is not None
    assert (
        parsed_response.job.issued_at.isoformat() == "2023-02-16T12:33:49.602947+00:00"
    )
    assert parsed_response.job.available_at is None
    assert parsed_response.job.status == "waiting"
    assert parsed_response.job.id == "76c90710-3a1b-4b91-8a39-31a6543e347c"
    assert not parsed_response.api_request.error


def test_async_response_post_fail():
    response = json.load(open(FILE_PATH_POST_FAIL))
    fake_response = FakeResponse(response)
    fake_response.set_ok_status(False)
    assert is_valid_async_response(fake_response) is False


def test_async_get_processing():
    response = json.load(open(FILE_PATH_GET_PROCESSING))
    parsed_response = AsyncPredictResponse(InvoiceSplitterV1, response)
    fake_response = FakeResponse(response)
    fake_response.set_ok_status(True)
    assert is_valid_async_response(fake_response) is True
    assert parsed_response.job is not None
    assert parsed_response.job.issued_at.isoformat() == "2023-03-16T12:33:49.602947"
    assert parsed_response.job.available_at is None
    assert parsed_response.job.status == "processing"
    assert parsed_response.job.id == "76c90710-3a1b-4b91-8a39-31a6543e347c"
    assert not parsed_response.api_request.error


def test_async_response_get_completed():
    response = json.load(open(FILE_PATH_GET_COMPLETED))
    parsed_response = AsyncPredictResponse(InvoiceSplitterV1, response)
    fake_response = FakeResponse(response)
    fake_response.set_ok_status(True)
    assert is_valid_async_response(fake_response) is True
    assert parsed_response.job is not None
    assert parsed_response.job.issued_at.isoformat() == "2023-03-21T13:52:56.326107"
    assert parsed_response.job.available_at.isoformat() == "2023-03-21T13:53:00.990339"
    assert parsed_response.job.status == "completed"
    assert parsed_response.api_request.error == {}


def test_async_get_failed_job():
    response = json.load(open(FILE_PATH_GET_FAILED_JOB))
    parsed_response = AsyncPredictResponse(InvoiceSplitterV1, response)
    fake_response = FakeResponse(response)
    fake_response.set_ok_status(False)
    assert is_valid_async_response(fake_response) is False
    assert parsed_response.api_request.status == RequestStatus.SUCCESS
    assert parsed_response.api_request.status_code == 200
    assert parsed_response.job.issued_at.isoformat() == "2024-02-20T10:31:06.878599"
    assert parsed_response.job.available_at.isoformat() == "2024-02-20T10:31:06.878599"
    assert parsed_response.job.status == "failed"
    assert parsed_response.job.error["code"] == "ServerError"