unstructured-python-client/_test_contract/conftest.py at b43d445bc0bfdd7114ca05b6797e40e35676be20 · Unstructured-IO/unstructured-python-client · 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
117
118
119
120
121
122
123
124
125
import os
from datetime import timedelta
from pathlib import Path

from freezegun import freeze_time
import pytest

from unstructured_client import UnstructuredClient, utils

FAKE_API_KEY = "91pmLBeETAbXCpNylRsLq11FdiZPTk"


@pytest.fixture(scope="module")
def platform_client(platform_api_url) -> UnstructuredClient:
    # settings the retry config to always try 3 times after a fail = 4 requests sent
    _client = UnstructuredClient(
        api_key_auth=FAKE_API_KEY,
        server_url=platform_api_url,
        retry_config=utils.RetryConfig(
            "backoff", utils.BackoffStrategy(
                initial_interval=3000,
                max_interval=3000,
                exponent=1.0,
                max_elapsed_time=8000
            ),
            retry_connection_errors=True
        )
    )
    yield _client


@pytest.fixture(scope="module")
def serverless_client(serverless_api_url) -> UnstructuredClient:
    # settings the retry config to always try 3 times after a fail = 4 requests sent
    _client = UnstructuredClient(
        api_key_auth=FAKE_API_KEY,
        server_url=serverless_api_url,
        retry_config = utils.RetryConfig(
        "backoff", utils.BackoffStrategy(
            initial_interval=3000,
            max_interval=3000,
            exponent=1.0,
            max_elapsed_time=8000
        ),
        retry_connection_errors=True
    )
    )
    yield _client


@pytest.fixture()
def freezer():
    ignore = ['_pytest.terminal', '_pytest.runner']
    freezer = freeze_time(ignore=ignore)
    frozen_time = freezer.start()
    yield frozen_time
    freezer.stop()

@pytest.fixture(autouse=True)
def mock_sleep(mocker, freezer):
    sleep_mock = mocker.patch("time.sleep")
    sleep_mock.side_effect = lambda seconds: freezer.tick(timedelta(seconds=seconds))
    yield sleep_mock


@pytest.fixture(scope="module")
def platform_api_url():
    return "https://platform.unstructuredapp.io"


@pytest.fixture(scope="module")
def serverless_api_url():
    return "https://api.unstructuredapp.io"


@pytest.fixture(scope="module")
def dummy_partitioned_text():
    return """[
  {
    "type": "NarrativeText",
    "element_id": "b7dca0363a83468b9e7326c0c1caf93e",
    "text": "March 17, 2022",
    "metadata": {
      "detection_class_prob": 0.35799261927604675,
      "coordinates": {
        "points": [
          [
            1447.871337890625,
            301.74810791015625
          ],
          [
            1447.871337890625,
            326.5603332519531
          ],
          [
            1616.6922607421875,
            326.5603332519531
          ],
          [
            1616.6922607421875,
            301.74810791015625
          ]
        ],
        "system": "PixelSpace",
        "layout_width": 1700,
        "layout_height": 2200
      },
      "last_modified": "2024-02-07T14:23:29",
      "filetype": "application/pdf",
      "languages": [
        "eng"
      ],
      "page_number": 1,
      "file_directory": "data",
      "filename": "MyDocument.pdf"
    }
  }
]"""


@pytest.fixture(scope="module")
def doc_path() -> Path:
    samples_path = Path(__file__).resolve().parents[1] / "_sample_docs"
    assert samples_path.exists()
    return samples_path