Repository: gtalarico/pyairtable
Pull request #330 (2024-01-21 11:58:32)

Support for iterating through the audit log
This branch allows enterprise users to iterate through [audit log events](https://airtable.com/developers/web/api/audit-log-events) with something like this:

```python
enterprise = api.enterprise("ent...")
for page in enterprise.audit_log(sort_asc=True):
    for event in page.events:
        handle_event(event)
```


codecov[bot]: ## [Codecov](https://app.codecov.io/gh/gtalarico/pyairtable/pull/330?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Gui+Talarico) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Comparison is base [(`54ef64f`)](https://app.codecov.io/gh/gtalarico/pyairtable/commit/54ef64fb6eb56df1d91cef6e9d8a7f092cbb0e1d?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Gui+Talarico) 99.61% compared to head [(`bc419d2`)](https://app.codecov.io/gh/gtalarico/pyairtable/pull/330?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Gui+Talarico) 99.62%.


<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main     #330      +/-   ##
==========================================
+ Coverage   99.61%   99.62%   +0.01%     
==========================================
  Files          24       25       +1     
  Lines        1801     1889      +88     
==========================================
+ Hits         1794     1882      +88     
  Misses          7        7              
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/gtalarico/pyairtable/pull/330?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Gui+Talarico).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Gui+Talarico).


Patch:
diff --git a/docs/source/api.rst b/docs/source/api.rst
index 8815c7d..2c0d9ba 100644
--- a/docs/source/api.rst
+++ b/docs/source/api.rst
@@ -49,6 +49,14 @@ API: pyairtable.models
     :inherited-members: AirtableModel
 
 
+API: pyairtable.models.audit
+********************************
+
+.. automodule:: pyairtable.models.audit
+    :members:
+    :inherited-members: AirtableModel
+
+
 API: pyairtable.models.schema
 ********************************
 
diff --git a/docs/source/changelog.rst b/docs/source/changelog.rst
index 004f249..0ed91d5 100644
--- a/docs/source/changelog.rst
+++ b/docs/source/changelog.rst
@@ -5,6 +5,9 @@ Changelog
 2.3.0 (TBD)
 ------------------------
 
+* Added :meth:`Enterprise.audit_log <pyairtable.Enterprise.audit_log>`
+  to iterate page-by-page through `audit log events <https://airtable.com/developers/web/api/audit-logs-overview>`__.
+  - `PR #330 <https://github.com/gtalarico/pyairtable/pull/330>`_.
 * :meth:`Api.base <pyairtable.Api.base>`,
   :meth:`Api.table <pyairtable.Api.table>`,
   and :meth:`Base.table <pyairtable.Base.table>`
diff --git a/docs/source/enterprise.rst b/docs/source/enterprise.rst
new file mode 100644
index 0000000..871c231
--- /dev/null
+++ b/docs/source/enterprise.rst
@@ -0,0 +1,29 @@
+.. include:: _substitutions.rst
+.. include:: _warn_latest.rst
+
+Enterprise Features
+==============================
+
+
+pyAirtable exposes a number of classes and methods for interacting with enterprise organizations.
+The following methods are only available on an `Enterprise plan <https://airtable.com/pricing>`__.
+If you call one of them against a base that is not part of an enterprise workspace, Airtable will
+return a 404 error, and pyAirtable will add a reminder to the exception to check your billing plan.
+
+.. automethod:: pyairtable.Api.enterprise
+    :noindex:
+
+.. automethod:: pyairtable.Base.collaborators
+    :noindex:
+
+.. automethod:: pyairtable.Base.shares
+    :noindex:
+
+.. automethod:: pyairtable.Workspace.collaborators
+    :noindex:
+
+.. automethod:: pyairtable.Enterprise.info
+    :noindex:
+
+.. automethod:: pyairtable.Enterprise.audit_log
+    :noindex:
diff --git a/docs/source/index.rst b/docs/source/index.rst
index 32e0964..ec4ca99 100644
--- a/docs/source/index.rst
+++ b/docs/source/index.rst
@@ -27,9 +27,9 @@ pyAirtable
    getting-started
    tables
    orm
-   webhooks
    metadata
-   migrations
+   webhooks
+   enterprise
    api
 
 
@@ -37,6 +37,7 @@ pyAirtable
    :caption: More
    :hidden:
 
+   migrations
    about
    changelog
    contributing
diff --git a/docs/source/metadata.rst b/docs/source/metadata.rst
index dfac6de..fd09fc9 100644
--- a/docs/source/metadata.rst
+++ b/docs/source/metadata.rst
@@ -33,30 +33,6 @@ You'll find more detail in the API reference for :mod:`pyairtable.models.schema`
     :noindex:
 
 
-Enterprise information
------------------------------
-
-pyAirtable exposes a number of classes and methods for interacting with enterprise organizations.
-The following methods are only available on an `Enterprise plan <https://airtable.com/pricing>`__.
-If you call one of them against a base that is not part of an enterprise workspace, Airtable will
-return a 404 error, and pyAirtable will add a reminder to the exception to check your billing plan.
-
-.. automethod:: pyairtable.Api.enterprise
-    :noindex:
-
-.. automethod:: pyairtable.Base.collaborators
-    :noindex:
-
-.. automethod:: pyairtable.Base.shares
-    :noindex:
-
-.. automethod:: pyairtable.Workspace.collaborators
-    :noindex:
-
-.. automethod:: pyairtable.Enterprise.info
-    :noindex:
-
-
 Modifying existing schema
 -----------------------------
 
diff --git a/docs/source/webhooks.rst b/docs/source/webhooks.rst
index 5a74d53..9f6a132 100644
--- a/docs/source/webhooks.rst
+++ b/docs/source/webhooks.rst
@@ -1,3 +1,6 @@
+.. include:: _substitutions.rst
+.. include:: _warn_latest.rst
+
 Webhooks
 ==============================
 
diff --git a/pyairtable/api/api.py b/pyairtable/api/api.py
index 33319c4..a2708af 100644
--- a/pyairtable/api/api.py
+++ b/pyairtable/api/api.py
@@ -293,6 +293,7 @@ class Api:
         url: str,
         fallback: Optional[Tuple[str, str]] = None,
         options: Optional[Dict[str, Any]] = None,
+        params: Optional[Dict[str, Any]] = None,
         offset_field: str = "offset",
     ) -> Iterator[Any]:
         """
@@ -311,18 +312,35 @@ class Api:
             fallback: The method and URL to use if we have to convert a GET to a POST.
             options: Airtable-specific query params to use while fetching records.
                 See :ref:`Parameters` for valid options.
+            params: Additional query params to append to the URL as-is.
             offset_field: The key to use in the API response to determine whether
                 there are additional pages to retrieve.
         """
         options = options or {}
+        params = params or {}
+
+        def _get_offset_field(response: Dict[str, Any]) -> Optional[str]:
+            value = response.get("pagination") or response  # see Enterprise.audit_log
+            field_names = offset_field.split(".")
+            while field_names:
+                if not (value := value.get(field_names.pop(0))):
+                    return None
+            return str(value)
+
         while True:
-            response = self.request(method, url, fallback=fallback, options=options)
+            response = self.request(
+                method=method,
+                url=url,
+                fallback=fallback,
+                options=options,
+                params=params,
+            )
             yield response
             if not isinstance(response, dict):
                 return
-            if not (offset := response.get(offset_field)):
+            if not (offset := _get_offset_field(response)):
                 return
-            options = {**options, offset_field: offset}
+            params = {**params, offset_field: offset}
 
     def chunked(self, iterable: Sequence[T]) -> Iterator[Sequence[T]]:
         """
diff --git a/pyairtable/api/enterprise.py b/pyairtable/api/enterprise.py
index 96d5912..10a1c24 100644
--- a/pyairtable/api/enterprise.py
+++ b/pyairtable/api/enterprise.py
@@ -1,7 +1,15 @@
-from typing import Iterable, List, Optional
+import datetime
+from typing import Iterable, Iterator, List, Optional, Union
 
+from pyairtable.models.audit import AuditLogResponse
 from pyairtable.models.schema import EnterpriseInfo, UserGroup, UserInfo
-from pyairtable.utils import cache_unless_forced, enterprise_only
+from pyairtable.utils import (
+    cache_unless_forced,
+    coerce_iso_str,
+    coerce_list_str,
+    enterprise_only,
+    is_user_id,
+)
 
 
 @enterprise_only
@@ -95,6 +103,129 @@ class Enterprise:
         }
         return list(users.values())
 
+    def audit_log(
+        self,
+        *,
+        page_size: Optional[int] = None,
+        page_limit: Optional[int] = None,
+        sort_asc: Optional[bool] = False,
+        previous: Optional[str] = None,
+        next: Optional[str] = None,
+        start_time: Optional[Union[str, datetime.date, datetime.datetime]] = None,
+        end_time: Optional[Union[str, datetime.date, datetime.datetime]] = None,
+        user_id: Optional[Union[str, Iterable[str]]] = None,
+        event_type: Optional[Union[str, Iterable[str]]] = None,
+        model_id: Optional[Union[str, Iterable[str]]] = None,
+        category: Optional[Union[str, Iterable[str]]] = None,
+    ) -> Iterator[AuditLogResponse]:
+        """
+        Retrieve and yield results from the `Audit Log <https://airtable.com/developers/web/api/audit-logs-integration-guide>`__,
+        one page of results at a time. Each result is an instance of :class:`~pyairtable.models.audit.AuditLogResponse`
+        and contains the pagination IDs returned from the API, as described in the linked documentation.
+
+        By default, the Airtable API will return up to 180 days of audit log events, going backwards from most recent.
+        Retrieving all records may take some time, but is as straightforward as:
+
+            >>> enterprise = Enterprise("entYourEnterpriseId")
+            >>> events = [
+            ...     event
+            ...     for page in enterprise.audit_log()
+            ...     for event in page.events
+            ... ]
+
+        If you are creating a record of all audit log events, you probably want to start with the earliest
+        events in the retention window and iterate chronologically. You'll likely have a job running
+        periodically in the background, so you'll need some way to persist the pagination IDs retrieved
+        from the API in case that job is interrupted and needs to be restarted.
+
+        The sample code below will use a local file to remember the next page's ID, so that if the job is
+        interrupted, it will resume where it left off (potentially processing some entries twice).
+
+        .. code-block:: python
+
+            import os
+            import shelve
+            import pyairtable
+
+            def handle_event(event):
+                print(event)
+
+            api = pyairtable.Api(os.environ["AIRTABLE_API_KEY"])
+            enterprise = api.enterprise(os.environ["AIRTABLE_ENTERPRISE_ID"])
+            persistence = shelve.open("audit_log.db")
+            first_page = persistence.get("next", None)
+
+            for page in enterprise.audit_log(sort_asc=True, next=first_page):
+                for event in page.events:
+                    handle_event(event)
+                persistence["next"] = page.pagination.next
+
+        For more information on any of the keyword parameters below, refer to the
+        `audit log events <https://airtable.com/developers/web/api/audit-log-events>`__
+        API documentation.
+
+        Args:
+            page_size: How many events per page to return (maximum 100).
+            page_limit: How many pages to return before stopping.
+            sort_asc: Whether to sort in ascending order (earliest to latest)
+                rather than descending order (latest to earliest).
+            previous: Requests the previous page of results from the given ID.
+                See the `audit log integration guide <https://airtable.com/developers/web/api/audit-logs-integration-guide>`__
+                for more information on pagination parameters.
+            next: Requests the next page of results according to the given ID.
+                See the `audit log integration guide <https://airtable.com/developers/web/api/audit-logs-integration-guide>`__
+                for more information on pagination parameters.
+            start_time: Earliest timestamp to retrieve (inclusive).
+            end_time: Latest timestamp to retrieve (inclusive).
+            originating_user_id: Retrieve audit log events originating
+                from the provided user ID or IDs (maximum 100).
+            event_type: Retrieve audit log events falling under the provided
+                `audit log event type <https://airtable.com/developers/web/api/audit-log-event-types>`__
+                or types (maximum 100).
+            model_id: Retrieve audit log events taking action on, or involving,
+                the provided model ID or IDs (maximum 100).
+            category: Retrieve audit log events belonging to the provided
+                audit log event category or categories.
+
+        Returns:
+            An object representing a single page of audit log results.
+        """
+
+        start_time = coerce_iso_str(start_time)
+        end_time = coerce_iso_str(end_time)
+        user_id = coerce_list_str(user_id)
+        event_type = coerce_list_str(event_type)
+        model_id = coerce_list_str(model_id)
+        category = coerce_list_str(category)
+        params = {
+            "startTime": start_time,
+            "endTime": end_time,
+            "originatingUserId": user_id,
+            "eventType": event_type,
+            "modelId": model_id,
+            "category": category,
+            "pageSize": page_size,
+            "sortOrder": ("ascending" if sort_asc else "descending"),
+            "previous": previous,
+            "next": next,
+        }
+        params = {k: v for (k, v) in params.items() if v}
+        offset_field = "next" if sort_asc else "previous"
+        url = self.api.build_url(f"meta/enterpriseAccounts/{self.id}/auditLogEvents")
+        iter_requests = self.api.iterate_requests(
+            method="GET",
+            url=url,
+            params=params,
+            offset_field=offset_field,
+        )
+        for count, response in enumerate(iter_requests, start=1):
+            parsed = AuditLogResponse.parse_obj(response)
+            yield parsed
+            if not parsed.events:
+                return
+            if page_limit is not None and count >= page_limit:
+                return
+
 
 # These are at the bottom of the module to avoid circular imports
 import pyairtable.api.api  # noqa
diff --git a/pyairtable/models/audit.py b/pyairtable/models/audit.py
new file mode 100644
index 0000000..00a63f2
--- /dev/null
+++ b/pyairtable/models/audit.py
@@ -0,0 +1,75 @@
+from typing import Any, Dict, List, Optional
+
+from typing_extensions import TypeAlias
+
+from pyairtable.models._base import AirtableModel, update_forward_refs
+
+
+class AuditLogResponse(AirtableModel):
+    """
+    Represents a page of audit log events.
+
+    See `Audit log events <https://airtable.com/developers/web/api/audit-log-events>`__
+    for more information on how to interpret this data structure.
+    """
+
+    events: List["AuditLogEvent"]
+    pagination: Optional["AuditLogResponse.Pagination"] = None
+
+    class Pagination(AirtableModel):
+        next: Optional[str]
+        previous: Optional[str]
+
+
+class AuditLogEvent(AirtableModel):
+    """
+    Represents a single audit log event.
+
+    See `Audit log events <https://airtable.com/developers/web/api/audit-log-events>`__
+    for more information on how to interpret this data structure.
+    """
+
+    id: str
+    timestamp: str
+    action: str
+    actor: "AuditLogActor"
+    model_id: str
+    model_type: str
+    payload: "AuditLogPayload"
+    payload_version: str
+    context: "AuditLogEvent.Context"
+    origin: "AuditLogEvent.Origin"
+
+    class Context(AirtableModel):
+        base_id: Optional[str] = None
+        action_id: str
+        enterprise_account_id: str
+        descendant_enterprise_account_id: Optional[str] = None
+        interface_id: Optional[str] = None
+        workspace_id: Optional[str] = None
+
+    class Origin(AirtableModel):
+        ip_address: str
+        user_agent: str
+        oauth_access_token_id: Optional[str] = None
+        personal_access_token_id: Optional[str] = None
+        session_id: Optional[str] = None
+
+
+class AuditLogActor(AirtableModel):
+    type: str
+    user: Optional["AuditLogActor.UserInfo"] = None
+    view_id: Optional[str] = None
+    automation_id: Optional[str] = None
+
+    class UserInfo(AirtableModel):
+        id: str
+        email: str
+        name: Optional[str] = None
+
+
+# Placeholder until we can parse https://airtable.com/developers/web/api/audit-log-event-types
+AuditLogPayload: TypeAlias = Dict[str, Any]
+
+
+update_forward_refs(vars())
diff --git a/pyairtable/utils.py b/pyairtable/utils.py
index 4c52b89..8588ea7 100644
--- a/pyairtable/utils.py
+++ b/pyairtable/utils.py
@@ -3,7 +3,19 @@ import re
 import textwrap
 from datetime import date, datetime
 from functools import partial, wraps
-from typing import Any, Callable, Generic, Iterator, Sequence, TypeVar, Union, cast
+from typing import (
+    Any,
+    Callable,
+    Generic,
+    Iterable,
+    Iterator,
+    List,
+    Optional,
+    Sequence,
+    TypeVar,
+    Union,
+    cast,
+)
 
 import requests
 from typing_extensions import ParamSpec, Protocol
@@ -208,3 +220,29 @@ def cache_unless_forced(func: Callable[P, R]) -> FetchMethod[R]:
     _append_docstring_text(_inner, "Args:\n\tforce: |kwarg_force_metadata|")
 
     return _inner
+
+
+def coerce_iso_str(value: Any) -> Optional[str]:
+    """
+    Given an input that might be a date or datetime, or an ISO 8601 formatted str,
+    convert the value into an ISO 8601 formatted str.
+    """
+    if value is None:
+        return value
+    if isinstance(value, str):
+        datetime.fromisoformat(value)  # validates type, nothing more
+        return value
+    if isinstance(value, (date, datetime)):
+        return value.isoformat()
+    raise TypeError(f"cannot coerce {type(value)} into ISO 8601 str")
+
+
+def coerce_list_str(value: Optional[Union[str, Iterable[str]]]) -> List[str]:
+    """
+    Given an input that is either a str or an iterable of str, return a list.
+    """
+    if value is None:
+        return []
+    if isinstance(value, str):
+        return [value]
+    return list(value)


Test patch:
diff --git a/tests/integration/test_integration_enterprise.py b/tests/integration/test_integration_enterprise.py
index 5b1aae6..4deb2d4 100644
--- a/tests/integration/test_integration_enterprise.py
+++ b/tests/integration/test_integration_enterprise.py
@@ -133,3 +133,16 @@ def test_update_field(blank_base: pyairtable.Base):
     field.description = "Renamed"
     field.save()
     assert reload_field().description == "Renamed"
+
+
+def test_audit_log(api):
+    """
+    Test that we can call the audit log endpoint.
+    """
+    if "AIRTABLE_ENTERPRISE_ID" not in os.environ:
+        return pytest.skip("test_audit_log requires AIRTABLE_ENTERPRISE_ID")
+
+    enterprise = api.enterprise(os.environ["AIRTABLE_ENTERPRISE_ID"])
+    for page in enterprise.audit_log(page_limit=1):
+        for event in page.events:
+            assert isinstance(event.action, str)
diff --git a/tests/test_api_enterprise.py b/tests/test_api_enterprise.py
index d74eaf3..61ea163 100644
--- a/tests/test_api_enterprise.py
+++ b/tests/test_api_enterprise.py
@@ -1,9 +1,11 @@
-from unittest.mock import Mock
+import datetime
+from unittest.mock import Mock, call, patch
 
 import pytest
 
 from pyairtable.api.enterprise import Enterprise
 from pyairtable.models.schema import EnterpriseInfo, UserGroup, UserInfo
+from pyairtable.testing import fake_id
 
 
 @pytest.fixture
@@ -11,7 +13,11 @@ def enterprise(api):
     return Enterprise(api, "entUBq2RGdihxl3vU")
 
 
-@pytest.fixture
+N_AUDIT_PAGES = 15
+N_AUDIT_PAGE_SIZE = 10
+
+
+@pytest.fixture(autouse=True)
 def enterprise_mocks(enterprise, requests_mock, sample_json):
     m = Mock()
     m.json_user = sample_json("UserInfo")
@@ -25,9 +31,48 @@ def enterprise_mocks(enterprise, requests_mock, sample_json):
         enterprise.api.build_url(f"meta/groups/{m.json_group['id']}"),
         json=m.json_group,
     )
+    m.get_audit_log = requests_mock.get(
+        enterprise.api.build_url(
+            f"meta/enterpriseAccounts/{enterprise.id}/auditLogEvents"
+        ),
+        response_list=[
+            {
+                "json": {
+                    "events": fake_audit_log_events(n),
+                    "pagination": (
+                        None if n == N_AUDIT_PAGES - 1 else {"previous": "dummy"}
+                    ),
+                }
+            }
+            for n in range(N_AUDIT_PAGES)
+        ],
+    )
     return m
 
 
+def fake_audit_log_events(counter, page_size=N_AUDIT_PAGE_SIZE):
+    return [
+        {
+            "id": str(counter * 1000 + n),
+            "timestamp": datetime.datetime.now().isoformat(),
+            "action": "viewBase",
+            "actor": {"type": "anonymousUser"},
+            "model_id": (base_id := fake_id("app")),
+            "model_type": "base",
+            "payload": {"name": "The Base Name"},
+            "payloadVersion": "1.0",
+            "context": {
+                "baseId": base_id,
+                "actionId": fake_id("act"),
+                "enterpriseAccountId": fake_id("ent"),
+                "workspaceId": fake_id("wsp"),
+            },
+            "origin": {"ipAddress": "8.8.8.8", "userAgent": "Internet Explorer"},
+        }
+        for n in range(page_size)
+    ]
+
+
 def test_info(enterprise, enterprise_mocks):
     assert isinstance(info := enterprise.info(), EnterpriseInfo)
     assert info.id == "entUBq2RGdihxl3vU"
@@ -70,7 +115,7 @@ def test_user__no_collaboration(enterprise, enterprise_mocks):
         ["usrL2PNC5o3H4lBEi", "foo@bar.com"],  # should not return duplicates
     ),
 )
-def test_users(enterprise, enterprise_mocks, search_for):
+def test_users(enterprise, search_for):
     results = enterprise.users(search_for)
     assert len(results) == 1
     assert isinstance(user := results[0], UserInfo)
@@ -102,3 +147,66 @@ def test_group__no_collaboration(enterprise, enterprise_mocks):
     assert not grp.collaborations.bases
     assert not grp.collaborations.interfaces
     assert not grp.collaborations.workspaces
+
+
+@pytest.mark.parametrize(
+    "fncall,expected_size",
+    [
+        (call(), N_AUDIT_PAGES * N_AUDIT_PAGE_SIZE),
+        (call(page_limit=1), N_AUDIT_PAGE_SIZE),
+    ],
+)
+def test_audit_log(enterprise, fncall, expected_size):
+    """
+    Test that we iterate through multiple pages of the audit log. correctly
+    """
+    events = [
+        event
+        for page in enterprise.audit_log(*fncall.args, **fncall.kwargs)
+        for event in page.events
+    ]
+    assert len(events) == expected_size
+
+
+def test_audit_log__no_loop(enterprise, requests_mock):
+    """
+    Test that an empty page of events does not cause an infinite loop.
+    """
+    requests_mock.get(
+        enterprise.api.build_url(
+            f"meta/enterpriseAccounts/{enterprise.id}/auditLogEvents"
+        ),
+        json={
+            "events": [],
+            "pagination": {"previous": "dummy"},
+        },
+    )
+    events = [event for page in enterprise.audit_log() for event in page.events]
+    assert len(events) == 0
+
+
+@pytest.mark.parametrize(
+    "fncall,sortorder,offset_field",
+    [
+        (call(), "descending", "previous"),
+        (call(sort_asc=True), "ascending", "next"),
+    ],
+)
+def test_audit_log__sortorder(
+    api,
+    enterprise,
+    enterprise_mocks,
+    fncall,
+    sortorder,
+    offset_field,
+):
+    """
+    Test that we calculate sortorder and offset_field correctly
+    dpeending on whether we're ascending or descending.
+    """
+    with patch.object(api, "iterate_requests", wraps=api.iterate_requests) as m:
+        list(enterprise.audit_log(*fncall.args, **fncall.kwargs))
+
+    request = enterprise_mocks.get_audit_log.last_request
+    assert request.qs["sortorder"] == [sortorder]
+    assert m.mock_calls[-1].kwargs["offset_field"] == offset_field
diff --git a/tests/test_utils.py b/tests/test_utils.py
index 2dfc041..9a3589b 100644
--- a/tests/test_utils.py
+++ b/tests/test_utils.py
@@ -58,3 +58,32 @@ def test_attachment():
 )
 def test_id_check(func, value, expected):
     assert func(value) is expected
+
+
+@pytest.mark.parametrize(
+    "func,input,expected",
+    [
+        (utils.coerce_iso_str, None, None),
+        (utils.coerce_iso_str, "asdf", ValueError),
+        (utils.coerce_iso_str, -1, TypeError),
+        (utils.coerce_iso_str, "2023-01-01", "2023-01-01"),
+        (utils.coerce_iso_str, "2023-01-01 12:34:56", "2023-01-01 12:34:56"),
+        (utils.coerce_iso_str, date(2023, 1, 1), "2023-01-01"),
+        (
+            utils.coerce_iso_str,
+            datetime(2023, 1, 1, 12, 34, 56),
+            "2023-01-01T12:34:56",
+        ),
+        (utils.coerce_list_str, None, []),
+        (utils.coerce_list_str, "asdf", ["asdf"]),
+        (utils.coerce_list_str, ("one", "two", "three"), ["one", "two", "three"]),
+        (utils.coerce_list_str, -1, TypeError),
+    ],
+)
+def test_converter(func, input, expected):
+    if isinstance(expected, type) and issubclass(expected, Exception):
+        with pytest.raises(expected):
+            func(input)
+        return
+
+    assert func(input) == expected