test: add unit and integration test suite for the new SDK
- 268 unit tests covering all sync and async client methods - 196 integration tests against a real sandbox API - Mock-based tests using respx for sync and async HTTP mocking - Paging, error handling, and edge case coverage (400/401/403/404/500/418) - Integration test utilities and test data files - .env.example with sandbox environment variables
This commit is contained in:
Stevan Freeborn
@@ -0,0 +1,188 @@
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from dotenv import load_dotenv
|
||||
|
||||
from onspring_api_sdk import AsyncOnspringClient, OnspringClient
|
||||
|
||||
load_dotenv()
|
||||
|
||||
|
||||
def _required_env(key: str) -> str:
|
||||
value = os.environ.get(key)
|
||||
if value is None:
|
||||
pytest.fail(f"{key} is not defined")
|
||||
return value
|
||||
|
||||
|
||||
def _optional_env(key: str) -> str | None:
|
||||
return os.environ.get(key)
|
||||
|
||||
|
||||
def pytest_configure(config):
|
||||
config.addinivalue_line("markers", "integration: marks tests as integration tests (real API calls)")
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def base_url():
|
||||
return _required_env("API_BASE_URL")
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def api_key():
|
||||
return _required_env("SANDBOX_API_KEY")
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_app_id():
|
||||
return int(_required_env("TEST_APP_ID"))
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_survey_id():
|
||||
return int(_required_env("TEST_SURVEY_ID"))
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_app_id_no_access():
|
||||
return int(_required_env("TEST_APP_ID_NO_ACCESS"))
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_app_ids():
|
||||
raw = _required_env("TEST_APP_IDS")
|
||||
return [int(x) for x in raw.split(",")]
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_app_ids_no_access():
|
||||
raw = _required_env("TEST_APP_IDS_NO_ACCESS")
|
||||
return [int(x) for x in raw.split(",")]
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_field_id():
|
||||
return int(_required_env("TEST_FIELD_ID"))
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_field_id_no_access():
|
||||
return int(_required_env("TEST_FIELD_ID_NO_ACCESS"))
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_field_ids():
|
||||
raw = _required_env("TEST_FIELD_IDS")
|
||||
return [int(x) for x in raw.split(",")]
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_field_ids_no_access():
|
||||
raw = _required_env("TEST_FIELD_IDS_NO_ACCESS")
|
||||
return [int(x) for x in raw.split(",")]
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_record():
|
||||
return int(_required_env("TEST_RECORD"))
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_survey_record_id():
|
||||
return int(_required_env("TEST_SURVEY_RECORD_ID"))
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_text_field():
|
||||
return int(_required_env("TEST_TEXT_FIELD"))
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_attachment_field():
|
||||
return int(_required_env("TEST_ATTACHMENT_FIELD"))
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_attachment_field_no_access_field():
|
||||
return int(_required_env("TEST_ATTACHMENT_FIELD_NO_ACCESS_FIELD"))
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_attachment_field_no_access_app():
|
||||
return int(_required_env("TEST_ATTACHMENT_FIELD_NO_ACCESS_APP"))
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_attachment():
|
||||
return int(_required_env("TEST_ATTACHMENT"))
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_image_field():
|
||||
return int(_required_env("TEST_IMAGE_FIELD"))
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_image():
|
||||
return int(_required_env("TEST_IMAGE"))
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_list_field():
|
||||
return int(_required_env("TEST_LIST_FIELD"))
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_list_field_no_access():
|
||||
return int(_required_env("TEST_LIST_FIELD_NO_ACCESS"))
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_list_id():
|
||||
return int(_required_env("TEST_LIST_ID"))
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_list_id_no_access():
|
||||
return int(_required_env("TEST_LIST_ID_NO_ACCESS"))
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_list_item_id_no_access():
|
||||
return _required_env("TEST_LIST_ITEM_ID_NO_ACCESS")
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_report():
|
||||
return int(_required_env("TEST_REPORT"))
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_report_no_access():
|
||||
return int(_required_env("TEST_REPORT_NO_ACCESS"))
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_report_with_chart_data():
|
||||
return int(_required_env("TEST_REPORT_WITH_CHART_DATA"))
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_survey_auto_number_field():
|
||||
return _optional_env("TEST_SURVEY_AUTO_NUMBER_FIELD")
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def testdata_dir():
|
||||
return Path(__file__).parent / "testdata"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(base_url, api_key):
|
||||
return OnspringClient(base_url, api_key)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def async_client(base_url, api_key):
|
||||
return AsyncOnspringClient(base_url, api_key)
|
||||
@@ -0,0 +1,243 @@
|
||||
import pytest
|
||||
|
||||
from onspring_api_sdk.models import PagingRequest
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
class TestGetApps:
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_sync(self, client):
|
||||
response = client.get_apps()
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.message is None
|
||||
assert response.data is not None
|
||||
assert response.data.page_number is not None
|
||||
assert response.data.page_size is not None
|
||||
assert response.data.total_pages is not None
|
||||
assert response.data.total_records is not None
|
||||
assert response.data.apps is not None
|
||||
assert len(response.data.apps) > 0
|
||||
|
||||
for app in response.data.apps:
|
||||
assert app.id is not None
|
||||
assert app.name is not None
|
||||
assert app.href is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_async(self, async_client):
|
||||
response = await async_client.get_apps()
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert len(response.data.apps) > 0
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_with_paging_request(self, client):
|
||||
response = client.get_apps(paging_request=PagingRequest(page_number=1, page_size=1))
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert response.data.page_number == 1
|
||||
assert response.data.page_size == 1
|
||||
assert len(response.data.apps) <= 1
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_with_paging_request_async(self, async_client):
|
||||
response = await async_client.get_apps(paging_request=PagingRequest(page_number=1, page_size=1))
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.data is not None
|
||||
assert response.data.page_number == 1
|
||||
assert response.data.page_size == 1
|
||||
assert len(response.data.apps) <= 1
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_page_size_returns_400(self, client):
|
||||
response = client.get_apps(paging_request=PagingRequest(page_number=1, page_size=-1))
|
||||
|
||||
assert response.status_code == 400
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_invalid_page_size_returns_400_async(self, async_client):
|
||||
response = await async_client.get_apps(paging_request=PagingRequest(page_number=1, page_size=-1))
|
||||
|
||||
assert response.status_code == 400
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_api_key_returns_401(self, base_url):
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
bad = OnspringClient(base_url, "invalid")
|
||||
response = bad.get_apps()
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_invalid_api_key_returns_401_async(self, base_url):
|
||||
from onspring_api_sdk import AsyncOnspringClient
|
||||
|
||||
bad = AsyncOnspringClient(base_url, "invalid")
|
||||
response = await bad.get_apps()
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
|
||||
class TestGetAppById:
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_sync(self, client, test_app_id):
|
||||
response = client.get_app_by_id(app_id=test_app_id)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.message is None
|
||||
assert response.data is not None
|
||||
assert response.data.app.id == test_app_id
|
||||
assert response.data.app.name is not None
|
||||
assert response.data.app.href is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_async(self, async_client, test_app_id):
|
||||
response = await async_client.get_app_by_id(app_id=test_app_id)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert response.data.app.id == test_app_id
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_api_key_returns_401(self, base_url):
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
bad = OnspringClient(base_url, "invalid")
|
||||
response = bad.get_app_by_id(app_id=1)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_invalid_api_key_returns_401_async(self, base_url):
|
||||
from onspring_api_sdk import AsyncOnspringClient
|
||||
|
||||
bad = AsyncOnspringClient(base_url, "invalid")
|
||||
response = await bad.get_app_by_id(app_id=1)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_no_access_returns_403(self, client, test_app_id_no_access):
|
||||
response = client.get_app_by_id(app_id=test_app_id_no_access)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_no_access_returns_403_async(self, async_client, test_app_id_no_access):
|
||||
response = await async_client.get_app_by_id(app_id=test_app_id_no_access)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_not_found_returns_404(self, client):
|
||||
response = client.get_app_by_id(app_id=0)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_not_found_returns_404_async(self, async_client):
|
||||
response = await async_client.get_app_by_id(app_id=0)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
|
||||
class TestGetAppsByIds:
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_sync(self, client, test_app_ids):
|
||||
response = client.get_apps_by_ids(app_ids=test_app_ids)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.message is None
|
||||
assert response.data is not None
|
||||
assert response.data.apps is not None
|
||||
assert len(response.data.apps) == len(test_app_ids)
|
||||
|
||||
for app in response.data.apps:
|
||||
assert app.id is not None
|
||||
assert app.name is not None
|
||||
assert app.href is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_async(self, async_client, test_app_ids):
|
||||
response = await async_client.get_apps_by_ids(app_ids=test_app_ids)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert len(response.data.apps) == len(test_app_ids)
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_api_key_returns_401(self, base_url):
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
bad = OnspringClient(base_url, "invalid")
|
||||
response = bad.get_apps_by_ids(app_ids=[1])
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_invalid_api_key_returns_401_async(self, base_url):
|
||||
from onspring_api_sdk import AsyncOnspringClient
|
||||
|
||||
bad = AsyncOnspringClient(base_url, "invalid")
|
||||
response = await bad.get_apps_by_ids(app_ids=[1])
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_no_access_returns_403(self, client, test_app_ids_no_access):
|
||||
response = client.get_apps_by_ids(app_ids=test_app_ids_no_access)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_no_access_returns_403_async(self, async_client, test_app_ids_no_access):
|
||||
response = await async_client.get_apps_by_ids(app_ids=test_app_ids_no_access)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
@@ -0,0 +1,284 @@
|
||||
import pytest
|
||||
|
||||
from onspring_api_sdk.models import PagingRequest
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
class TestGetFieldById:
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_sync(self, client, test_field_id):
|
||||
response = client.get_field_by_id(field_id=test_field_id)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.message is None
|
||||
assert response.data is not None
|
||||
assert response.data.field.id == test_field_id
|
||||
assert response.data.field.name is not None
|
||||
assert response.data.field.app_id is not None
|
||||
assert response.data.field.type is not None
|
||||
assert response.data.field.status is not None
|
||||
assert response.data.field.is_required is not None
|
||||
assert response.data.field.is_unique is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_async(self, async_client, test_field_id):
|
||||
response = await async_client.get_field_by_id(field_id=test_field_id)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert response.data.field.id == test_field_id
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_api_key_returns_401(self, base_url):
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
bad = OnspringClient(base_url, "invalid")
|
||||
response = bad.get_field_by_id(field_id=1)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_invalid_api_key_returns_401_async(self, base_url):
|
||||
from onspring_api_sdk import AsyncOnspringClient
|
||||
|
||||
bad = AsyncOnspringClient(base_url, "invalid")
|
||||
response = await bad.get_field_by_id(field_id=1)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_no_access_returns_403(self, client, test_field_id_no_access):
|
||||
response = client.get_field_by_id(field_id=test_field_id_no_access)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_no_access_returns_403_async(self, async_client, test_field_id_no_access):
|
||||
response = await async_client.get_field_by_id(field_id=test_field_id_no_access)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_not_found_returns_404(self, client):
|
||||
response = client.get_field_by_id(field_id=0)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_not_found_returns_404_async(self, async_client):
|
||||
response = await async_client.get_field_by_id(field_id=0)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
|
||||
class TestGetFieldsByAppId:
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_sync(self, client, test_survey_id):
|
||||
response = client.get_fields_by_app_id(app_id=test_survey_id)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.message is None
|
||||
assert response.data is not None
|
||||
assert response.data.page_number is not None
|
||||
assert response.data.page_size is not None
|
||||
assert response.data.total_pages is not None
|
||||
assert response.data.total_records is not None
|
||||
assert response.data.fields is not None
|
||||
assert len(response.data.fields) > 0
|
||||
|
||||
for field in response.data.fields:
|
||||
assert field.id is not None
|
||||
assert field.name is not None
|
||||
assert field.app_id is not None
|
||||
assert field.type is not None
|
||||
assert field.status is not None
|
||||
assert field.is_required is not None
|
||||
assert field.is_unique is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_async(self, async_client, test_survey_id):
|
||||
response = await async_client.get_fields_by_app_id(app_id=test_survey_id)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert len(response.data.fields) > 0
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_with_paging_request(self, client, test_survey_id):
|
||||
response = client.get_fields_by_app_id(
|
||||
app_id=test_survey_id,
|
||||
paging_request=PagingRequest(page_number=1, page_size=1),
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert response.data.page_number == 1
|
||||
assert response.data.page_size == 1
|
||||
assert len(response.data.fields) <= 1
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_with_paging_request_async(self, async_client, test_survey_id):
|
||||
response = await async_client.get_fields_by_app_id(
|
||||
app_id=test_survey_id,
|
||||
paging_request=PagingRequest(page_number=1, page_size=1),
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.data is not None
|
||||
assert response.data.page_number == 1
|
||||
assert response.data.page_size == 1
|
||||
assert len(response.data.fields) <= 1
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_page_size_returns_400(self, client, test_survey_id):
|
||||
response = client.get_fields_by_app_id(
|
||||
app_id=test_survey_id,
|
||||
paging_request=PagingRequest(page_number=1, page_size=-1),
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_invalid_page_size_returns_400_async(self, async_client, test_survey_id):
|
||||
response = await async_client.get_fields_by_app_id(
|
||||
app_id=test_survey_id,
|
||||
paging_request=PagingRequest(page_number=1, page_size=-1),
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_api_key_returns_401(self, base_url):
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
bad = OnspringClient(base_url, "invalid")
|
||||
response = bad.get_fields_by_app_id(app_id=1)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_invalid_api_key_returns_401_async(self, base_url):
|
||||
from onspring_api_sdk import AsyncOnspringClient
|
||||
|
||||
bad = AsyncOnspringClient(base_url, "invalid")
|
||||
response = await bad.get_fields_by_app_id(app_id=1)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_no_access_returns_403(self, client, test_app_id_no_access):
|
||||
response = client.get_fields_by_app_id(app_id=test_app_id_no_access)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_no_access_returns_403_async(self, async_client, test_app_id_no_access):
|
||||
response = await async_client.get_fields_by_app_id(app_id=test_app_id_no_access)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
|
||||
class TestGetFieldsByIds:
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_sync(self, client, test_field_ids):
|
||||
response = client.get_fields_by_ids(field_ids=test_field_ids)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.message is None
|
||||
assert response.data is not None
|
||||
assert response.data.count == len(test_field_ids)
|
||||
assert len(response.data.fields) == len(test_field_ids)
|
||||
|
||||
for field in response.data.fields:
|
||||
assert field.id is not None
|
||||
assert field.name is not None
|
||||
assert field.app_id is not None
|
||||
assert field.type is not None
|
||||
assert field.status is not None
|
||||
assert field.is_required is not None
|
||||
assert field.is_unique is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_async(self, async_client, test_field_ids):
|
||||
response = await async_client.get_fields_by_ids(field_ids=test_field_ids)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert response.data.count == len(test_field_ids)
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_api_key_returns_401(self, base_url):
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
bad = OnspringClient(base_url, "invalid")
|
||||
response = bad.get_fields_by_ids(field_ids=[1, 2, 3])
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_invalid_api_key_returns_401_async(self, base_url):
|
||||
from onspring_api_sdk import AsyncOnspringClient
|
||||
|
||||
bad = AsyncOnspringClient(base_url, "invalid")
|
||||
response = await bad.get_fields_by_ids(field_ids=[1, 2, 3])
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_no_access_returns_403(self, client, test_field_ids_no_access):
|
||||
response = client.get_fields_by_ids(field_ids=test_field_ids_no_access)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_no_access_returns_403_async(self, async_client, test_field_ids_no_access):
|
||||
response = await async_client.get_fields_by_ids(field_ids=test_field_ids_no_access)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
@@ -0,0 +1,679 @@
|
||||
import pytest
|
||||
|
||||
from onspring_api_sdk.models import SaveFileRequest
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
def _attachment_path(testdata_dir):
|
||||
return str(testdata_dir / "test-attachment.txt")
|
||||
|
||||
|
||||
def _image_path(testdata_dir):
|
||||
return str(testdata_dir / "test-image.jpeg")
|
||||
|
||||
|
||||
class TestGetFileInfoById:
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_attachment_field(self, client, test_record, test_attachment_field, test_attachment):
|
||||
response = client.get_file_info_by_id(
|
||||
record_id=test_record,
|
||||
field_id=test_attachment_field,
|
||||
file_id=test_attachment,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.message is None
|
||||
assert response.data is not None
|
||||
assert response.data.file_info.created_date is not None
|
||||
assert response.data.file_info.content_type is not None
|
||||
assert response.data.file_info.file_href is not None
|
||||
assert response.data.file_info.name is not None
|
||||
assert response.data.file_info.modified_date is not None
|
||||
assert response.data.file_info.type is not None
|
||||
assert response.data.file_info.owner is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_attachment_field_async(self, async_client, test_record, test_attachment_field, test_attachment):
|
||||
response = await async_client.get_file_info_by_id(
|
||||
record_id=test_record,
|
||||
field_id=test_attachment_field,
|
||||
file_id=test_attachment,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_image_field(self, client, test_record, test_image_field, test_image):
|
||||
response = client.get_file_info_by_id(
|
||||
record_id=test_record,
|
||||
field_id=test_image_field,
|
||||
file_id=test_image,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.message is None
|
||||
assert response.data is not None
|
||||
assert response.data.file_info.created_date is not None
|
||||
assert response.data.file_info.content_type is not None
|
||||
assert response.data.file_info.file_href is not None
|
||||
assert response.data.file_info.name is not None
|
||||
assert response.data.file_info.modified_date is not None
|
||||
assert response.data.file_info.type is not None
|
||||
assert response.data.file_info.owner is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_image_field_async(self, async_client, test_record, test_image_field, test_image):
|
||||
response = await async_client.get_file_info_by_id(
|
||||
record_id=test_record,
|
||||
field_id=test_image_field,
|
||||
file_id=test_image,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_non_file_field_returns_400(self, client, test_record, test_text_field, test_attachment):
|
||||
response = client.get_file_info_by_id(
|
||||
record_id=test_record,
|
||||
field_id=test_text_field,
|
||||
file_id=test_attachment,
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_non_file_field_returns_400_async(self, async_client, test_record, test_text_field, test_attachment):
|
||||
response = await async_client.get_file_info_by_id(
|
||||
record_id=test_record,
|
||||
field_id=test_text_field,
|
||||
file_id=test_attachment,
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_api_key_returns_401(self, base_url, test_record, test_attachment_field, test_attachment):
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
bad = OnspringClient(base_url, "invalid")
|
||||
response = bad.get_file_info_by_id(
|
||||
record_id=test_record,
|
||||
field_id=test_attachment_field,
|
||||
file_id=test_attachment,
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_invalid_api_key_returns_401_async(
|
||||
self, base_url, test_record, test_attachment_field, test_attachment
|
||||
):
|
||||
from onspring_api_sdk import AsyncOnspringClient
|
||||
|
||||
bad = AsyncOnspringClient(base_url, "invalid")
|
||||
response = await bad.get_file_info_by_id(
|
||||
record_id=test_record,
|
||||
field_id=test_attachment_field,
|
||||
file_id=test_attachment,
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_no_field_access_returns_403(self, client, test_attachment_field_no_access_field):
|
||||
response = client.get_file_info_by_id(record_id=1, field_id=test_attachment_field_no_access_field, file_id=1)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_no_field_access_returns_403_async(self, async_client, test_attachment_field_no_access_field):
|
||||
response = await async_client.get_file_info_by_id(
|
||||
record_id=1, field_id=test_attachment_field_no_access_field, file_id=1
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_no_app_access_returns_403(self, client, test_attachment_field_no_access_app):
|
||||
response = client.get_file_info_by_id(record_id=1, field_id=test_attachment_field_no_access_app, file_id=1)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_no_app_access_returns_403_async(self, async_client, test_attachment_field_no_access_app):
|
||||
response = await async_client.get_file_info_by_id(
|
||||
record_id=1, field_id=test_attachment_field_no_access_app, file_id=1
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_field_not_found_returns_404(self, client):
|
||||
response = client.get_file_info_by_id(record_id=1, field_id=0, file_id=1)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_field_not_found_returns_404_async(self, async_client):
|
||||
response = await async_client.get_file_info_by_id(record_id=1, field_id=0, file_id=1)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_record_not_found_returns_404(self, client, test_attachment_field):
|
||||
response = client.get_file_info_by_id(record_id=0, field_id=test_attachment_field, file_id=1)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_record_not_found_returns_404_async(self, async_client, test_attachment_field):
|
||||
response = await async_client.get_file_info_by_id(record_id=0, field_id=test_attachment_field, file_id=1)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
|
||||
class TestGetFileById:
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_attachment_field(self, client, test_record, test_attachment_field, test_attachment):
|
||||
response = client.get_file_by_id(record_id=test_record, field_id=test_attachment_field, file_id=test_attachment)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.message is None
|
||||
assert response.data is not None
|
||||
assert response.data.file.content_length is not None
|
||||
assert response.data.file.content_type is not None
|
||||
assert response.data.file.name is not None
|
||||
assert response.data.file.content is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_attachment_field_async(self, async_client, test_record, test_attachment_field, test_attachment):
|
||||
response = await async_client.get_file_by_id(
|
||||
record_id=test_record, field_id=test_attachment_field, file_id=test_attachment
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_image_field(self, client, test_record, test_image_field, test_image):
|
||||
response = client.get_file_by_id(record_id=test_record, field_id=test_image_field, file_id=test_image)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.message is None
|
||||
assert response.data is not None
|
||||
assert response.data.file.content_length is not None
|
||||
assert response.data.file.content_type is not None
|
||||
assert response.data.file.name is not None
|
||||
assert response.data.file.content is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_image_field_async(self, async_client, test_record, test_image_field, test_image):
|
||||
response = await async_client.get_file_by_id(
|
||||
record_id=test_record, field_id=test_image_field, file_id=test_image
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_non_file_field_returns_400(self, client, test_record, test_text_field, test_attachment):
|
||||
response = client.get_file_by_id(record_id=test_record, field_id=test_text_field, file_id=test_attachment)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_non_file_field_returns_400_async(self, async_client, test_record, test_text_field, test_attachment):
|
||||
response = await async_client.get_file_by_id(
|
||||
record_id=test_record, field_id=test_text_field, file_id=test_attachment
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_api_key_returns_401(self, base_url, test_record, test_attachment_field, test_attachment):
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
bad = OnspringClient(base_url, "invalid")
|
||||
response = bad.get_file_by_id(record_id=test_record, field_id=test_attachment_field, file_id=test_attachment)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_invalid_api_key_returns_401_async(
|
||||
self, base_url, test_record, test_attachment_field, test_attachment
|
||||
):
|
||||
from onspring_api_sdk import AsyncOnspringClient
|
||||
|
||||
bad = AsyncOnspringClient(base_url, "invalid")
|
||||
response = await bad.get_file_by_id(
|
||||
record_id=test_record, field_id=test_attachment_field, file_id=test_attachment
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_no_field_access_returns_403(self, client, test_attachment_field_no_access_field):
|
||||
response = client.get_file_by_id(record_id=1, field_id=test_attachment_field_no_access_field, file_id=1)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_no_field_access_returns_403_async(self, async_client, test_attachment_field_no_access_field):
|
||||
response = await async_client.get_file_by_id(
|
||||
record_id=1, field_id=test_attachment_field_no_access_field, file_id=1
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_no_app_access_returns_403(self, client, test_attachment_field_no_access_app):
|
||||
response = client.get_file_by_id(record_id=1, field_id=test_attachment_field_no_access_app, file_id=1)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_no_app_access_returns_403_async(self, async_client, test_attachment_field_no_access_app):
|
||||
response = await async_client.get_file_by_id(
|
||||
record_id=1, field_id=test_attachment_field_no_access_app, file_id=1
|
||||
)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_field_not_found_returns_404(self, client):
|
||||
response = client.get_file_by_id(record_id=1, field_id=0, file_id=1)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_field_not_found_returns_404_async(self, async_client):
|
||||
response = await async_client.get_file_by_id(record_id=1, field_id=0, file_id=1)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_record_not_found_returns_404(self, client, test_attachment_field):
|
||||
response = client.get_file_by_id(record_id=0, field_id=test_attachment_field, file_id=1)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_record_not_found_returns_404_async(self, async_client, test_attachment_field):
|
||||
response = await async_client.get_file_by_id(record_id=0, field_id=test_attachment_field, file_id=1)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
|
||||
class TestSaveFile:
|
||||
_attachment_file_ids: list[int] = []
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def cleanup_attachments(self, client, test_record, test_attachment_field):
|
||||
yield
|
||||
|
||||
for file_id in self._attachment_file_ids:
|
||||
client.delete_file_by_id(record_id=test_record, field_id=test_attachment_field, file_id=file_id)
|
||||
|
||||
self._attachment_file_ids.clear()
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_attachment_field(self, client, test_record, test_attachment_field, testdata_dir):
|
||||
request = SaveFileRequest(
|
||||
record_id=test_record,
|
||||
field_id=test_attachment_field,
|
||||
file_name="test-attachment.txt",
|
||||
file_path=_attachment_path(testdata_dir),
|
||||
content_type="text/plain",
|
||||
notes="integration test",
|
||||
)
|
||||
|
||||
response = client.save_file(request)
|
||||
|
||||
assert response.status_code == 201
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert response.data.id is not None
|
||||
|
||||
self._attachment_file_ids.append(response.data.id)
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_attachment_field_async(self, async_client, test_record, test_attachment_field, testdata_dir):
|
||||
request = SaveFileRequest(
|
||||
record_id=test_record,
|
||||
field_id=test_attachment_field,
|
||||
file_name="test-attachment.txt",
|
||||
file_path=_attachment_path(testdata_dir),
|
||||
content_type="text/plain",
|
||||
notes="integration test",
|
||||
)
|
||||
|
||||
response = await async_client.save_file(request)
|
||||
|
||||
assert response.status_code == 201
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert response.data.id is not None
|
||||
|
||||
self._attachment_file_ids.append(response.data.id)
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_image_field(self, client, test_record, test_image_field, testdata_dir):
|
||||
request = SaveFileRequest(
|
||||
record_id=test_record,
|
||||
field_id=test_image_field,
|
||||
file_name="test-image.jpeg",
|
||||
file_path=_image_path(testdata_dir),
|
||||
content_type="image/jpeg",
|
||||
notes="integration test",
|
||||
)
|
||||
|
||||
response = client.save_file(request)
|
||||
|
||||
assert response.status_code == 201
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert response.data.id is not None
|
||||
|
||||
self._attachment_file_ids.append(response.data.id)
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_non_file_field_returns_400(self, client, test_record, test_text_field, testdata_dir):
|
||||
request = SaveFileRequest(
|
||||
record_id=test_record,
|
||||
field_id=test_text_field,
|
||||
file_name="test-attachment.txt",
|
||||
file_path=_attachment_path(testdata_dir),
|
||||
content_type="text/plain",
|
||||
notes="integration test",
|
||||
)
|
||||
|
||||
response = client.save_file(request)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_api_key_returns_401(self, base_url, testdata_dir):
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
bad = OnspringClient(base_url, "invalid")
|
||||
request = SaveFileRequest(
|
||||
record_id=1,
|
||||
field_id=1,
|
||||
file_name="test-attachment.txt",
|
||||
file_path=_attachment_path(testdata_dir),
|
||||
content_type="text/plain",
|
||||
)
|
||||
|
||||
response = bad.save_file(request)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_no_field_access_returns_403(self, client, test_attachment_field_no_access_field, testdata_dir):
|
||||
request = SaveFileRequest(
|
||||
record_id=1,
|
||||
field_id=test_attachment_field_no_access_field,
|
||||
file_name="test-attachment.txt",
|
||||
file_path=_attachment_path(testdata_dir),
|
||||
content_type="text/plain",
|
||||
)
|
||||
|
||||
response = client.save_file(request)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_no_app_access_returns_403(self, client, test_attachment_field_no_access_app, testdata_dir):
|
||||
request = SaveFileRequest(
|
||||
record_id=1,
|
||||
field_id=test_attachment_field_no_access_app,
|
||||
file_name="test-attachment.txt",
|
||||
file_path=_attachment_path(testdata_dir),
|
||||
content_type="text/plain",
|
||||
)
|
||||
|
||||
response = client.save_file(request)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_field_not_found_returns_404(self, client, testdata_dir):
|
||||
request = SaveFileRequest(
|
||||
record_id=1,
|
||||
field_id=0,
|
||||
file_name="test-attachment.txt",
|
||||
file_path=_attachment_path(testdata_dir),
|
||||
content_type="text/plain",
|
||||
)
|
||||
|
||||
response = client.save_file(request)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_record_not_found_returns_404(self, client, test_attachment_field, testdata_dir):
|
||||
request = SaveFileRequest(
|
||||
record_id=0,
|
||||
field_id=test_attachment_field,
|
||||
file_name="test-attachment.txt",
|
||||
file_path=_attachment_path(testdata_dir),
|
||||
content_type="text/plain",
|
||||
)
|
||||
|
||||
response = client.save_file(request)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
|
||||
class TestDeleteFileById:
|
||||
_attachment_file_id: int | None = None
|
||||
_image_file_id: int | None = None
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def setup_files(self, client, test_record, test_attachment_field, test_image_field, testdata_dir):
|
||||
if self._attachment_file_id is None:
|
||||
req = SaveFileRequest(
|
||||
record_id=test_record,
|
||||
field_id=test_attachment_field,
|
||||
file_name="test-attachment.txt",
|
||||
file_path=_attachment_path(testdata_dir),
|
||||
content_type="text/plain",
|
||||
notes="delete test",
|
||||
)
|
||||
|
||||
resp = client.save_file(req)
|
||||
|
||||
if resp.data is not None:
|
||||
self._attachment_file_id = resp.data.id
|
||||
|
||||
if self._image_file_id is None:
|
||||
req = SaveFileRequest(
|
||||
record_id=test_record,
|
||||
field_id=test_image_field,
|
||||
file_name="test-image.jpeg",
|
||||
file_path=_image_path(testdata_dir),
|
||||
content_type="image/jpeg",
|
||||
notes="delete test",
|
||||
)
|
||||
|
||||
resp = client.save_file(req)
|
||||
|
||||
if resp.data is not None:
|
||||
self._image_file_id = resp.data.id
|
||||
|
||||
yield
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_attachment_field(self, client, test_record, test_attachment_field):
|
||||
assert self._attachment_file_id is not None
|
||||
|
||||
response = client.delete_file_by_id(
|
||||
record_id=test_record,
|
||||
field_id=test_attachment_field,
|
||||
file_id=self._attachment_file_id,
|
||||
)
|
||||
|
||||
assert response.status_code == 204
|
||||
assert response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_image_field(self, client, test_record, test_image_field):
|
||||
assert self._image_file_id is not None
|
||||
|
||||
response = client.delete_file_by_id(
|
||||
record_id=test_record,
|
||||
field_id=test_image_field,
|
||||
file_id=self._image_file_id,
|
||||
)
|
||||
|
||||
assert response.status_code == 204
|
||||
assert response.is_successful
|
||||
assert response.message is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_non_file_field_returns_400(self, client, test_record, test_text_field):
|
||||
response = client.delete_file_by_id(record_id=test_record, field_id=test_text_field, file_id=1)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_api_key_returns_401(self, base_url):
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
bad = OnspringClient(base_url, "invalid")
|
||||
response = bad.delete_file_by_id(record_id=1, field_id=1, file_id=1)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_no_field_access_returns_403(self, client, test_attachment_field_no_access_field):
|
||||
response = client.delete_file_by_id(record_id=1, field_id=test_attachment_field_no_access_field, file_id=1)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_no_app_access_returns_403(self, client, test_attachment_field_no_access_app):
|
||||
response = client.delete_file_by_id(record_id=1, field_id=test_attachment_field_no_access_app, file_id=1)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_field_not_found_returns_404(self, client):
|
||||
response = client.delete_file_by_id(record_id=1, field_id=0, file_id=1)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_record_not_found_returns_404(self, client, test_attachment_field):
|
||||
response = client.delete_file_by_id(record_id=0, field_id=test_attachment_field, file_id=1)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
@@ -0,0 +1,277 @@
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
|
||||
from onspring_api_sdk.models import ListItemRequest
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
def _unique_name():
|
||||
return f"test_list_value_{uuid.uuid4().hex[:8]}"
|
||||
|
||||
|
||||
class TestAddOrUpdateListItem:
|
||||
_new_item_ids: list[uuid.UUID] = []
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def cleanup(self, client, test_list_id):
|
||||
yield
|
||||
|
||||
for item_id in self._new_item_ids:
|
||||
client.delete_list_item(list_id=test_list_id, item_id=str(item_id))
|
||||
|
||||
self._new_item_ids.clear()
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_add(self, client, test_list_id):
|
||||
request = ListItemRequest(
|
||||
list_id=test_list_id,
|
||||
name=_unique_name(),
|
||||
numeric_value=1,
|
||||
color="#000000",
|
||||
)
|
||||
|
||||
response = client.add_or_update_list_item(request)
|
||||
|
||||
assert response.status_code == 201
|
||||
assert response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is not None
|
||||
assert response.data.id is not None
|
||||
|
||||
self._new_item_ids.append(response.data.id)
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_add_async(self, async_client, test_list_id):
|
||||
request = ListItemRequest(
|
||||
list_id=test_list_id,
|
||||
name=_unique_name(),
|
||||
numeric_value=1,
|
||||
color="#000000",
|
||||
)
|
||||
|
||||
response = await async_client.add_or_update_list_item(request)
|
||||
|
||||
assert response.status_code == 201
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert response.data.id is not None
|
||||
|
||||
if response.data.id:
|
||||
import os
|
||||
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
sync = OnspringClient(os.environ["API_BASE_URL"], os.environ["SANDBOX_API_KEY"])
|
||||
sync.delete_list_item(list_id=test_list_id, item_id=str(response.data.id))
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_update(self, client, test_list_id):
|
||||
add_request = ListItemRequest(
|
||||
list_id=test_list_id,
|
||||
name=_unique_name(),
|
||||
numeric_value=1,
|
||||
color="#000000",
|
||||
)
|
||||
|
||||
add_response = client.add_or_update_list_item(add_request)
|
||||
|
||||
assert add_response.data is not None
|
||||
|
||||
item_id = add_response.data.id
|
||||
self._new_item_ids.append(item_id)
|
||||
|
||||
update_request = ListItemRequest(
|
||||
list_id=test_list_id,
|
||||
name=_unique_name(),
|
||||
id=item_id,
|
||||
numeric_value=1,
|
||||
color="#000000",
|
||||
)
|
||||
|
||||
update_response = client.add_or_update_list_item(update_request)
|
||||
|
||||
assert update_response.status_code == 200
|
||||
assert update_response.is_successful
|
||||
assert update_response.message is not None
|
||||
assert update_response.data is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_update_async(self, async_client, test_list_id):
|
||||
add_request = ListItemRequest(
|
||||
list_id=test_list_id,
|
||||
name=_unique_name(),
|
||||
numeric_value=1,
|
||||
color="#000000",
|
||||
)
|
||||
|
||||
add_response = await async_client.add_or_update_list_item(add_request)
|
||||
|
||||
assert add_response.data is not None
|
||||
|
||||
item_id = add_response.data.id
|
||||
|
||||
import os
|
||||
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
sync = OnspringClient(os.environ["API_BASE_URL"], os.environ["SANDBOX_API_KEY"])
|
||||
|
||||
update_request = ListItemRequest(
|
||||
list_id=test_list_id,
|
||||
name=_unique_name(),
|
||||
id=item_id,
|
||||
numeric_value=1,
|
||||
color="#000000",
|
||||
)
|
||||
|
||||
update_response = await async_client.add_or_update_list_item(update_request)
|
||||
|
||||
assert update_response.status_code == 200
|
||||
assert update_response.is_successful
|
||||
assert update_response.data is not None
|
||||
|
||||
sync.delete_list_item(list_id=test_list_id, item_id=str(item_id))
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_api_key_returns_401(self, base_url):
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
bad = OnspringClient(base_url, "invalid")
|
||||
request = ListItemRequest(list_id=1, name="test", numeric_value=1, color="#000000")
|
||||
|
||||
response = bad.add_or_update_list_item(request)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_no_access_returns_403(self, client):
|
||||
request = ListItemRequest(list_id=1, name="test", numeric_value=1, color="#000000")
|
||||
|
||||
response = client.add_or_update_list_item(request)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_list_not_found_returns_404(self, client):
|
||||
request = ListItemRequest(list_id=0, name="test", numeric_value=1, color="#000000")
|
||||
|
||||
response = client.add_or_update_list_item(request)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_item_not_found_returns_404(self, client, test_list_id):
|
||||
request = ListItemRequest(
|
||||
list_id=test_list_id,
|
||||
id=uuid.UUID("3fa85f64-5717-4562-b3fc-2c963f66afa6"),
|
||||
name="test",
|
||||
)
|
||||
|
||||
response = client.add_or_update_list_item(request)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
|
||||
class TestDeleteListItem:
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_sync(self, client, test_list_id):
|
||||
add_request = ListItemRequest(
|
||||
list_id=test_list_id,
|
||||
name=_unique_name(),
|
||||
numeric_value=1,
|
||||
color="#000000",
|
||||
)
|
||||
|
||||
add_response = client.add_or_update_list_item(add_request)
|
||||
|
||||
assert add_response.data is not None
|
||||
|
||||
item_id = str(add_response.data.id)
|
||||
|
||||
response = client.delete_list_item(list_id=test_list_id, item_id=item_id)
|
||||
|
||||
assert response.status_code == 204
|
||||
assert response.is_successful
|
||||
assert response.message is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_async(self, async_client, test_list_id):
|
||||
import os
|
||||
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
sync = OnspringClient(os.environ["API_BASE_URL"], os.environ["SANDBOX_API_KEY"])
|
||||
|
||||
add_request = ListItemRequest(
|
||||
list_id=test_list_id,
|
||||
name=_unique_name(),
|
||||
numeric_value=1,
|
||||
color="#000000",
|
||||
)
|
||||
|
||||
add_response = sync.add_or_update_list_item(add_request)
|
||||
|
||||
assert add_response.data is not None
|
||||
|
||||
item_id = str(add_response.data.id)
|
||||
|
||||
response = await async_client.delete_list_item(list_id=test_list_id, item_id=item_id)
|
||||
|
||||
assert response.status_code == 204
|
||||
assert response.is_successful
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_api_key_returns_401(self, base_url):
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
bad = OnspringClient(base_url, "invalid")
|
||||
response = bad.delete_list_item(list_id=1, item_id="1")
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_no_access_returns_403(self, client, test_list_id_no_access, test_list_item_id_no_access):
|
||||
response = client.delete_list_item(list_id=test_list_id_no_access, item_id=test_list_item_id_no_access)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_list_not_found_returns_404(self, client):
|
||||
response = client.delete_list_item(list_id=0, item_id="3fa85f64-5717-4562-b3fc-2c963f66afa6")
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_item_not_found_returns_404(self, client, test_list_id):
|
||||
response = client.delete_list_item(
|
||||
list_id=test_list_id,
|
||||
item_id="3fa85f64-5717-4562-b3fc-2c963f66afa6",
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
@@ -0,0 +1,15 @@
|
||||
import pytest
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
class TestCanConnect:
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_sync(self, client):
|
||||
result = client.can_connect()
|
||||
assert result is True
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_async(self, async_client):
|
||||
result = await async_client.can_connect()
|
||||
assert result is True
|
||||
@@ -0,0 +1,804 @@
|
||||
import pytest
|
||||
|
||||
from onspring_api_sdk.enums import DataFormat
|
||||
from onspring_api_sdk.models import (
|
||||
DeleteBatchRecordsRequest,
|
||||
GetBatchRecordsRequest,
|
||||
GetRecordByIdRequest,
|
||||
GetRecordsByAppRequest,
|
||||
QueryRecordsRequest,
|
||||
Record,
|
||||
StringFieldValue,
|
||||
)
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
class TestGetRecordsByAppId:
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_sync(self, client, test_survey_id):
|
||||
request = GetRecordsByAppRequest(app_id=test_survey_id)
|
||||
|
||||
response = client.get_records_by_app_id(request)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.message is None
|
||||
assert response.data is not None
|
||||
assert response.data.page_number is not None
|
||||
assert response.data.page_size is not None
|
||||
assert response.data.total_pages is not None
|
||||
assert response.data.total_records is not None
|
||||
assert response.data.records is not None
|
||||
|
||||
for record in response.data.records:
|
||||
assert record.app_id == test_survey_id
|
||||
assert record.record_id is not None
|
||||
assert record.fields is not None
|
||||
assert len(record.fields) > 0
|
||||
|
||||
for field in record.fields:
|
||||
assert field.field_id is not None
|
||||
assert field.type is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_async(self, async_client, test_survey_id):
|
||||
request = GetRecordsByAppRequest(app_id=test_survey_id)
|
||||
|
||||
response = await async_client.get_records_by_app_id(request)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert len(response.data.records) > 0
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_with_params(self, client, test_survey_id, test_text_field):
|
||||
request = GetRecordsByAppRequest(
|
||||
app_id=test_survey_id,
|
||||
field_ids=[test_text_field],
|
||||
data_format=DataFormat.Formatted.name,
|
||||
page_number=1,
|
||||
page_size=1,
|
||||
)
|
||||
|
||||
response = client.get_records_by_app_id(request)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert response.data.page_number == 1
|
||||
assert response.data.page_size == 1
|
||||
assert len(response.data.records) == 1
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_with_params_async(self, async_client, test_survey_id, test_text_field):
|
||||
request = GetRecordsByAppRequest(
|
||||
app_id=test_survey_id,
|
||||
field_ids=[test_text_field],
|
||||
data_format=DataFormat.Formatted.name,
|
||||
page_number=1,
|
||||
page_size=1,
|
||||
)
|
||||
|
||||
response = await async_client.get_records_by_app_id(request)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert response.data.page_number == 1
|
||||
assert response.data.page_size == 1
|
||||
assert len(response.data.records) == 1
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_api_key_returns_401(self, base_url):
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
bad = OnspringClient(base_url, "invalid")
|
||||
request = GetRecordsByAppRequest(app_id=0)
|
||||
|
||||
response = bad.get_records_by_app_id(request)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_invalid_api_key_returns_401_async(self, base_url):
|
||||
from onspring_api_sdk import AsyncOnspringClient
|
||||
|
||||
bad = AsyncOnspringClient(base_url, "invalid")
|
||||
request = GetRecordsByAppRequest(app_id=0)
|
||||
|
||||
response = await bad.get_records_by_app_id(request)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_no_access_returns_403(self, client, test_app_id_no_access):
|
||||
request = GetRecordsByAppRequest(app_id=test_app_id_no_access)
|
||||
|
||||
response = client.get_records_by_app_id(request)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_no_access_returns_403_async(self, async_client, test_app_id_no_access):
|
||||
request = GetRecordsByAppRequest(app_id=test_app_id_no_access)
|
||||
|
||||
response = await async_client.get_records_by_app_id(request)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
|
||||
class TestGetRecordById:
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_sync(self, client, test_survey_id, test_survey_record_id):
|
||||
request = GetRecordByIdRequest(app_id=test_survey_id, record_id=test_survey_record_id)
|
||||
|
||||
response = client.get_record_by_id(request)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.message is None
|
||||
assert response.data is not None
|
||||
assert response.data.app_id == test_survey_id
|
||||
assert response.data.record_id == test_survey_record_id
|
||||
assert response.data.fields is not None
|
||||
assert len(response.data.fields) > 0
|
||||
|
||||
for field in response.data.fields:
|
||||
assert field.field_id is not None
|
||||
assert field.type is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_async(self, async_client, test_survey_id, test_survey_record_id):
|
||||
request = GetRecordByIdRequest(app_id=test_survey_id, record_id=test_survey_record_id)
|
||||
|
||||
response = await async_client.get_record_by_id(request)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert response.data.app_id == test_survey_id
|
||||
assert response.data.record_id == test_survey_record_id
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_with_params(self, client, test_survey_id, test_survey_record_id, test_text_field):
|
||||
request = GetRecordByIdRequest(
|
||||
app_id=test_survey_id,
|
||||
record_id=test_survey_record_id,
|
||||
field_ids=[test_text_field],
|
||||
data_format=DataFormat.Formatted.name,
|
||||
)
|
||||
|
||||
response = client.get_record_by_id(request)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert response.data.app_id == test_survey_id
|
||||
assert response.data.record_id == test_survey_record_id
|
||||
assert len(response.data.fields) > 0
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_with_params_async(self, async_client, test_survey_id, test_survey_record_id, test_text_field):
|
||||
request = GetRecordByIdRequest(
|
||||
app_id=test_survey_id,
|
||||
record_id=test_survey_record_id,
|
||||
field_ids=[test_text_field],
|
||||
data_format=DataFormat.Formatted.name,
|
||||
)
|
||||
|
||||
response = await async_client.get_record_by_id(request)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_api_key_returns_401(self, base_url):
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
bad = OnspringClient(base_url, "invalid")
|
||||
request = GetRecordByIdRequest(app_id=1, record_id=1)
|
||||
|
||||
response = bad.get_record_by_id(request)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_invalid_api_key_returns_401_async(self, base_url):
|
||||
from onspring_api_sdk import AsyncOnspringClient
|
||||
|
||||
bad = AsyncOnspringClient(base_url, "invalid")
|
||||
request = GetRecordByIdRequest(app_id=1, record_id=1)
|
||||
|
||||
response = await bad.get_record_by_id(request)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_record_not_found_returns_404(self, client, test_survey_id):
|
||||
request = GetRecordByIdRequest(app_id=test_survey_id, record_id=0)
|
||||
|
||||
response = client.get_record_by_id(request)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_record_not_found_returns_404_async(self, async_client, test_survey_id):
|
||||
request = GetRecordByIdRequest(app_id=test_survey_id, record_id=0)
|
||||
|
||||
response = await async_client.get_record_by_id(request)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
|
||||
class TestSaveRecord:
|
||||
_new_records: list[dict] = []
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def cleanup(self, client):
|
||||
yield
|
||||
|
||||
for rec in self._new_records:
|
||||
client.delete_record_by_id(app_id=rec["app_id"], record_id=rec["record_id"])
|
||||
|
||||
self._new_records.clear()
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_add(self, client, test_survey_id, test_text_field):
|
||||
record = Record(
|
||||
app_id=test_survey_id,
|
||||
fields=[StringFieldValue(field_id=test_text_field, value="Test")],
|
||||
)
|
||||
|
||||
response = client.add_or_update_record(record)
|
||||
|
||||
assert response.status_code == 201
|
||||
assert response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is not None
|
||||
assert response.data.id is not None
|
||||
assert response.data.warnings is not None
|
||||
|
||||
self._new_records.append({"app_id": test_survey_id, "record_id": response.data.id})
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_add_async(self, async_client, test_survey_id, test_text_field):
|
||||
record = Record(
|
||||
app_id=test_survey_id,
|
||||
fields=[StringFieldValue(field_id=test_text_field, value="Test")],
|
||||
)
|
||||
|
||||
response = await async_client.add_or_update_record(record)
|
||||
|
||||
assert response.status_code == 201
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert response.data.id is not None
|
||||
self._new_records.append({"app_id": test_survey_id, "record_id": response.data.id})
|
||||
|
||||
if response.data.id:
|
||||
import os
|
||||
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
sync = OnspringClient(os.environ["API_BASE_URL"], os.environ["SANDBOX_API_KEY"])
|
||||
sync.delete_record_by_id(app_id=test_survey_id, record_id=response.data.id)
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_update(self, client, test_survey_id, test_text_field):
|
||||
new_record = Record(
|
||||
app_id=test_survey_id,
|
||||
fields=[StringFieldValue(field_id=test_text_field, value="Test")],
|
||||
)
|
||||
|
||||
new_response = client.add_or_update_record(new_record)
|
||||
|
||||
assert new_response.data is not None
|
||||
|
||||
new_id = new_response.data.id
|
||||
|
||||
self._new_records.append({"app_id": test_survey_id, "record_id": new_id})
|
||||
|
||||
update_record = Record(
|
||||
app_id=test_survey_id,
|
||||
record_id=new_id,
|
||||
fields=[StringFieldValue(field_id=test_text_field, value="updated")],
|
||||
)
|
||||
|
||||
response = client.add_or_update_record(update_record)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert response.data.id == new_id
|
||||
assert response.data.warnings is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_empty_fields_returns_400(self, client, test_survey_id):
|
||||
record = Record(app_id=test_survey_id, fields=[])
|
||||
|
||||
response = client.add_or_update_record(record)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_api_key_returns_401(self, base_url):
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
bad = OnspringClient(base_url, "invalid")
|
||||
record = Record(app_id=0, fields=[])
|
||||
|
||||
response = bad.add_or_update_record(record)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_no_access_returns_403(self, client, test_app_id_no_access):
|
||||
record = Record(app_id=test_app_id_no_access, fields=[])
|
||||
|
||||
response = client.add_or_update_record(record)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_record_not_found_returns_404(self, client, test_survey_id):
|
||||
record = Record(app_id=test_survey_id, record_id=0, fields=[])
|
||||
|
||||
response = client.add_or_update_record(record)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
|
||||
class TestDeleteRecordById:
|
||||
def _create_temp_record(self, client, test_survey_id, test_text_field):
|
||||
record = Record(
|
||||
app_id=test_survey_id,
|
||||
fields=[StringFieldValue(field_id=test_text_field, value="to_delete")],
|
||||
)
|
||||
|
||||
response = client.add_or_update_record(record)
|
||||
|
||||
assert response.data is not None
|
||||
return response.data.id
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_sync(self, client, test_survey_id, test_text_field):
|
||||
record_id = self._create_temp_record(client, test_survey_id, test_text_field)
|
||||
|
||||
response = client.delete_record_by_id(app_id=test_survey_id, record_id=record_id)
|
||||
|
||||
assert response.status_code == 204
|
||||
assert response.is_successful
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_async(self, async_client, test_survey_id, test_text_field):
|
||||
import os
|
||||
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
sync = OnspringClient(os.environ["API_BASE_URL"], os.environ["SANDBOX_API_KEY"])
|
||||
|
||||
record_id = self._create_temp_record(sync, test_survey_id, test_text_field)
|
||||
|
||||
response = await async_client.delete_record_by_id(app_id=test_survey_id, record_id=record_id)
|
||||
|
||||
assert response.status_code == 204
|
||||
assert response.is_successful
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_api_key_returns_401(self, base_url):
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
bad = OnspringClient(base_url, "invalid")
|
||||
response = bad.delete_record_by_id(app_id=1, record_id=1)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_no_access_returns_403(self, client, test_app_id_no_access):
|
||||
response = client.delete_record_by_id(app_id=test_app_id_no_access, record_id=1)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_record_not_found_returns_404(self, client, test_app_id):
|
||||
response = client.delete_record_by_id(app_id=test_app_id, record_id=0)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
|
||||
class TestDeleteRecordsByIds:
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_sync(self, client, test_survey_id, test_text_field):
|
||||
r1 = client.add_or_update_record(
|
||||
Record(app_id=test_survey_id, fields=[StringFieldValue(field_id=test_text_field, value="del1")])
|
||||
)
|
||||
|
||||
r2 = client.add_or_update_record(
|
||||
Record(app_id=test_survey_id, fields=[StringFieldValue(field_id=test_text_field, value="del2")])
|
||||
)
|
||||
|
||||
assert r1.data is not None and r2.data is not None
|
||||
|
||||
request = DeleteBatchRecordsRequest(app_id=test_survey_id, record_ids=[r1.data.id, r2.data.id])
|
||||
|
||||
response = client.delete_records_by_ids(request)
|
||||
|
||||
assert response.status_code == 204
|
||||
assert response.is_successful
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_empty_ids_returns_400(self, client, test_survey_id):
|
||||
request = DeleteBatchRecordsRequest(app_id=test_survey_id, record_ids=[])
|
||||
|
||||
response = client.delete_records_by_ids(request)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_api_key_returns_401(self, base_url):
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
bad = OnspringClient(base_url, "invalid")
|
||||
request = DeleteBatchRecordsRequest(app_id=1, record_ids=[1])
|
||||
|
||||
response = bad.delete_records_by_ids(request)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_no_access_returns_403(self, client, test_app_id_no_access):
|
||||
request = DeleteBatchRecordsRequest(app_id=test_app_id_no_access, record_ids=[1])
|
||||
|
||||
response = client.delete_records_by_ids(request)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
|
||||
class TestGetRecordsByIds:
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_sync(self, client, test_survey_id, test_survey_record_id):
|
||||
request = GetBatchRecordsRequest(app_id=test_survey_id, record_ids=[test_survey_record_id])
|
||||
|
||||
response = client.get_records_by_ids(request)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.message is None
|
||||
assert response.data is not None
|
||||
assert response.data.count is not None
|
||||
assert len(response.data.records) > 0
|
||||
|
||||
for record in response.data.records:
|
||||
assert record.app_id == test_survey_id
|
||||
assert record.record_id is not None
|
||||
assert record.fields is not None
|
||||
assert len(record.fields) > 0
|
||||
|
||||
for field in record.fields:
|
||||
assert field.field_id is not None
|
||||
assert field.type is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_async(self, async_client, test_survey_id, test_survey_record_id):
|
||||
request = GetBatchRecordsRequest(app_id=test_survey_id, record_ids=[test_survey_record_id])
|
||||
|
||||
response = await async_client.get_records_by_ids(request)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert len(response.data.records) > 0
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_with_params(self, client, test_survey_id, test_survey_record_id, test_text_field):
|
||||
request = GetBatchRecordsRequest(
|
||||
app_id=test_survey_id,
|
||||
record_ids=[test_survey_record_id],
|
||||
field_ids=[test_text_field],
|
||||
data_format=DataFormat.Formatted.name,
|
||||
)
|
||||
|
||||
response = client.get_records_by_ids(request)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert len(response.data.records) > 0
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_with_params_async(self, async_client, test_survey_id, test_survey_record_id, test_text_field):
|
||||
request = GetBatchRecordsRequest(
|
||||
app_id=test_survey_id,
|
||||
record_ids=[test_survey_record_id],
|
||||
field_ids=[test_text_field],
|
||||
data_format=DataFormat.Formatted.name,
|
||||
)
|
||||
|
||||
response = await async_client.get_records_by_ids(request)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_too_many_ids_returns_400(self, client):
|
||||
record_ids = list(range(1, 102))
|
||||
request = GetBatchRecordsRequest(app_id=1, record_ids=record_ids)
|
||||
|
||||
response = client.get_records_by_ids(request)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_too_many_ids_returns_400_async(self, async_client):
|
||||
record_ids = list(range(1, 102))
|
||||
request = GetBatchRecordsRequest(app_id=1, record_ids=record_ids)
|
||||
|
||||
response = await async_client.get_records_by_ids(request)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_api_key_returns_401(self, base_url):
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
bad = OnspringClient(base_url, "invalid")
|
||||
request = GetBatchRecordsRequest(app_id=1, record_ids=[1])
|
||||
|
||||
response = bad.get_records_by_ids(request)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_invalid_api_key_returns_401_async(self, base_url):
|
||||
from onspring_api_sdk import AsyncOnspringClient
|
||||
|
||||
bad = AsyncOnspringClient(base_url, "invalid")
|
||||
request = GetBatchRecordsRequest(app_id=1, record_ids=[1])
|
||||
|
||||
response = await bad.get_records_by_ids(request)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_no_access_returns_403(self, client, test_app_id_no_access):
|
||||
request = GetBatchRecordsRequest(app_id=test_app_id_no_access, record_ids=[1])
|
||||
|
||||
response = client.get_records_by_ids(request)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_no_access_returns_403_async(self, async_client, test_app_id_no_access):
|
||||
request = GetBatchRecordsRequest(app_id=test_app_id_no_access, record_ids=[1])
|
||||
|
||||
response = await async_client.get_records_by_ids(request)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
|
||||
class TestQueryRecords:
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_sync(self, client, test_survey_id, test_survey_auto_number_field):
|
||||
filter_str = f"{test_survey_auto_number_field} gt 0"
|
||||
request = QueryRecordsRequest(app_id=test_survey_id, filter=filter_str)
|
||||
|
||||
response = client.query_records(request)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.message is None
|
||||
assert response.data is not None
|
||||
assert response.data.page_number is not None
|
||||
assert response.data.page_size is not None
|
||||
assert response.data.total_pages is not None
|
||||
assert response.data.total_records is not None
|
||||
|
||||
for record in response.data.records:
|
||||
assert record.app_id == test_survey_id
|
||||
assert record.record_id is not None
|
||||
assert record.fields is not None
|
||||
|
||||
if record.fields:
|
||||
for field in record.fields:
|
||||
assert field.field_id is not None
|
||||
assert field.type is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_async(self, async_client, test_survey_id, test_survey_auto_number_field):
|
||||
filter_str = f"{test_survey_auto_number_field} gt 0"
|
||||
request = QueryRecordsRequest(app_id=test_survey_id, filter=filter_str)
|
||||
|
||||
response = await async_client.query_records(request)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert response.data.total_records is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_with_params(self, client, test_survey_id, test_survey_auto_number_field):
|
||||
filter_str = f"{test_survey_auto_number_field} gt 0"
|
||||
request = QueryRecordsRequest(
|
||||
app_id=test_survey_id,
|
||||
filter=filter_str,
|
||||
field_ids=[test_survey_auto_number_field],
|
||||
data_format=DataFormat.Formatted.name,
|
||||
page_number=1,
|
||||
page_size=1,
|
||||
)
|
||||
|
||||
response = client.query_records(request)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert response.data.page_number == 1
|
||||
assert response.data.page_size == 1
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_with_params_async(self, async_client, test_survey_id, test_survey_auto_number_field):
|
||||
filter_str = f"{test_survey_auto_number_field} gt 0"
|
||||
request = QueryRecordsRequest(
|
||||
app_id=test_survey_id,
|
||||
filter=filter_str,
|
||||
field_ids=[test_survey_auto_number_field],
|
||||
data_format=DataFormat.Formatted.name,
|
||||
page_number=1,
|
||||
page_size=1,
|
||||
)
|
||||
|
||||
response = await async_client.query_records(request)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert response.data.page_number == 1
|
||||
assert response.data.page_size == 1
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_page_size_returns_400(self, client, test_survey_auto_number_field):
|
||||
filter_str = f"{test_survey_auto_number_field} gt 0"
|
||||
request = QueryRecordsRequest(
|
||||
app_id=1,
|
||||
filter=filter_str,
|
||||
page_size=-1,
|
||||
)
|
||||
|
||||
response = client.query_records(request)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_invalid_page_size_returns_400_async(self, async_client, test_survey_auto_number_field):
|
||||
filter_str = f"{test_survey_auto_number_field} gt 0"
|
||||
request = QueryRecordsRequest(
|
||||
app_id=1,
|
||||
filter=filter_str,
|
||||
page_size=-1,
|
||||
)
|
||||
|
||||
response = await async_client.query_records(request)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_api_key_returns_401(self, base_url):
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
bad = OnspringClient(base_url, "invalid")
|
||||
request = QueryRecordsRequest(app_id=1, filter="")
|
||||
|
||||
response = bad.query_records(request)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_invalid_api_key_returns_401_async(self, base_url):
|
||||
from onspring_api_sdk import AsyncOnspringClient
|
||||
|
||||
bad = AsyncOnspringClient(base_url, "invalid")
|
||||
request = QueryRecordsRequest(app_id=1, filter="")
|
||||
|
||||
response = await bad.query_records(request)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_no_access_returns_403(self, client, test_app_id_no_access, test_text_field):
|
||||
filter_str = f"{test_text_field} gt ''"
|
||||
request = QueryRecordsRequest(app_id=test_app_id_no_access, filter=filter_str)
|
||||
|
||||
response = client.query_records(request)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_no_access_returns_403_async(self, async_client, test_app_id_no_access, test_text_field):
|
||||
filter_str = f"{test_text_field} gt ''"
|
||||
request = QueryRecordsRequest(app_id=test_app_id_no_access, filter=filter_str)
|
||||
|
||||
response = await async_client.query_records(request)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
@@ -0,0 +1,278 @@
|
||||
import pytest
|
||||
|
||||
from onspring_api_sdk.enums import ReportDataType
|
||||
from onspring_api_sdk.models import GetReportByIdRequest, PagingRequest
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
class TestGetReportById:
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_sync(self, client, test_report):
|
||||
request = GetReportByIdRequest(report_id=test_report)
|
||||
|
||||
response = client.get_report_by_id(request)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.message is None
|
||||
assert response.data is not None
|
||||
assert response.data.columns is not None
|
||||
assert len(response.data.columns) > 0
|
||||
assert response.data.rows is not None
|
||||
assert len(response.data.rows) > 0
|
||||
|
||||
for row in response.data.rows:
|
||||
assert row.record_id is not None
|
||||
assert row.cells is not None
|
||||
assert len(row.cells) > 0
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_async(self, async_client, test_report):
|
||||
request = GetReportByIdRequest(report_id=test_report)
|
||||
|
||||
response = await async_client.get_report_by_id(request)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert len(response.data.rows) > 0
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_with_chart_data_report(self, client, test_report_with_chart_data):
|
||||
request = GetReportByIdRequest(report_id=test_report_with_chart_data)
|
||||
|
||||
response = client.get_report_by_id(request)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert response.data.columns is not None
|
||||
assert response.data.rows is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_with_chart_data_report_async(self, async_client, test_report_with_chart_data):
|
||||
request = GetReportByIdRequest(report_id=test_report_with_chart_data)
|
||||
|
||||
response = await async_client.get_report_by_id(request)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_request_chart_data(self, client, test_report_with_chart_data):
|
||||
request = GetReportByIdRequest(
|
||||
report_id=test_report_with_chart_data,
|
||||
data_type=ReportDataType.ChartData.name,
|
||||
)
|
||||
|
||||
response = client.get_report_by_id(request)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert response.data.columns is not None
|
||||
assert response.data.rows is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_request_chart_data_async(self, async_client, test_report_with_chart_data):
|
||||
request = GetReportByIdRequest(
|
||||
report_id=test_report_with_chart_data,
|
||||
data_type=ReportDataType.ChartData.name,
|
||||
)
|
||||
|
||||
response = await async_client.get_report_by_id(request)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_chart_data_on_report_without_returns_400(self, client, test_report):
|
||||
request = GetReportByIdRequest(
|
||||
report_id=test_report,
|
||||
data_type=ReportDataType.ChartData.name,
|
||||
)
|
||||
|
||||
response = client.get_report_by_id(request)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_chart_data_on_report_without_returns_400_async(self, async_client, test_report):
|
||||
request = GetReportByIdRequest(
|
||||
report_id=test_report,
|
||||
data_type=ReportDataType.ChartData.name,
|
||||
)
|
||||
|
||||
response = await async_client.get_report_by_id(request)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_api_key_returns_401(self, base_url):
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
bad = OnspringClient(base_url, "invalid")
|
||||
request = GetReportByIdRequest(report_id=1)
|
||||
|
||||
response = bad.get_report_by_id(request)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_invalid_api_key_returns_401_async(self, base_url):
|
||||
from onspring_api_sdk import AsyncOnspringClient
|
||||
|
||||
bad = AsyncOnspringClient(base_url, "invalid")
|
||||
request = GetReportByIdRequest(report_id=1)
|
||||
|
||||
response = await bad.get_report_by_id(request)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_no_access_returns_403(self, client):
|
||||
request = GetReportByIdRequest(report_id=1)
|
||||
|
||||
response = client.get_report_by_id(request)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_no_access_returns_403_async(self, async_client):
|
||||
request = GetReportByIdRequest(report_id=1)
|
||||
|
||||
response = await async_client.get_report_by_id(request)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_not_found_returns_404(self, client):
|
||||
request = GetReportByIdRequest(report_id=0)
|
||||
|
||||
response = client.get_report_by_id(request)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_not_found_returns_404_async(self, async_client):
|
||||
request = GetReportByIdRequest(report_id=0)
|
||||
|
||||
response = await async_client.get_report_by_id(request)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
|
||||
class TestGetReportsByAppId:
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_sync(self, client, test_survey_id):
|
||||
response = client.get_reports_by_app_id(app_id=test_survey_id)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.message is None
|
||||
assert response.data is not None
|
||||
assert response.data.page_number is not None
|
||||
assert response.data.page_size is not None
|
||||
assert response.data.total_pages is not None
|
||||
assert response.data.total_records is not None
|
||||
assert response.data.reports is not None
|
||||
assert len(response.data.reports) > 0
|
||||
|
||||
for report in response.data.reports:
|
||||
assert report.app_id is not None
|
||||
assert report.id is not None
|
||||
assert report.name is not None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_async(self, async_client, test_survey_id):
|
||||
response = await async_client.get_reports_by_app_id(app_id=test_survey_id)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.is_successful
|
||||
assert response.data is not None
|
||||
assert len(response.data.reports) > 0
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_page_size_returns_400(self, client, test_survey_id):
|
||||
response = client.get_reports_by_app_id(
|
||||
app_id=test_survey_id,
|
||||
paging_request=PagingRequest(page_number=1, page_size=-1),
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_invalid_page_size_returns_400_async(self, async_client, test_survey_id):
|
||||
response = await async_client.get_reports_by_app_id(
|
||||
app_id=test_survey_id,
|
||||
paging_request=PagingRequest(page_number=1, page_size=-1),
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_invalid_api_key_returns_401(self, base_url):
|
||||
from onspring_api_sdk import OnspringClient
|
||||
|
||||
bad = OnspringClient(base_url, "invalid")
|
||||
response = bad.get_reports_by_app_id(app_id=1)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_invalid_api_key_returns_401_async(self, base_url):
|
||||
from onspring_api_sdk import AsyncOnspringClient
|
||||
|
||||
bad = AsyncOnspringClient(base_url, "invalid")
|
||||
response = await bad.get_reports_by_app_id(app_id=1)
|
||||
|
||||
assert response.status_code == 401
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
def test_no_access_returns_403(self, client, test_app_id_no_access):
|
||||
response = client.get_reports_by_app_id(app_id=test_app_id_no_access)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.message is not None
|
||||
assert response.data is None
|
||||
|
||||
@pytest.mark.flaky(reruns=3)
|
||||
async def test_no_access_returns_403_async(self, async_client, test_app_id_no_access):
|
||||
response = await async_client.get_reports_by_app_id(app_id=test_app_id_no_access)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert not response.is_successful
|
||||
assert response.data is None
|
||||
@@ -0,0 +1 @@
|
||||
This is a test attachment.
|
||||
BIN
Binary file not shown.
|
After Width: | Height: | Size: 96 KiB |
@@ -0,0 +1,33 @@
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
from onspring_api_sdk import OnspringClient
|
||||
from onspring_api_sdk.models import Record, StringFieldValue
|
||||
|
||||
|
||||
def add_record(base_url: str, api_key: str) -> int:
|
||||
client = OnspringClient(base_url, api_key)
|
||||
|
||||
survey_id = os.environ.get("TEST_SURVEY_ID")
|
||||
text_field = os.environ.get("TEST_TEXT_FIELD")
|
||||
|
||||
if survey_id is None:
|
||||
pytest.fail("TEST_SURVEY_ID is not defined")
|
||||
if text_field is None:
|
||||
pytest.fail("TEST_TEXT_FIELD is not defined")
|
||||
|
||||
app_id = int(survey_id)
|
||||
field_id = int(text_field)
|
||||
|
||||
record = Record(
|
||||
app_id=app_id,
|
||||
fields=[StringFieldValue(field_id=field_id, value="test")],
|
||||
)
|
||||
|
||||
response = client.add_or_update_record(record)
|
||||
|
||||
if response.data is None or response.data.id is None:
|
||||
pytest.fail("Record ID is not defined")
|
||||
|
||||
return response.data.id
|
||||
Reference in New Issue
Block a user