onspring-api-sdk-python/tests/integration/test_records.py

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