Repository: planetlabs/planet-client-python

Issue:
implement subscriptions api & CLI create/update functions 
implement the following subscriptions api & CLI functions:
* create_subscription
* update_subscription

CLI specified in https://github.com/planetlabs/planet-client-python/blob/main/design-docs/CLI-Subscriptions.md#subscriptions-command-line-interface-specification

Discussion:

Patch:
diff --git a/planet/cli/data.py b/planet/cli/data.py
index 54057a8..a09b0b5 100644
--- a/planet/cli/data.py
+++ b/planet/cli/data.py
@@ -19,10 +19,10 @@ from contextlib import asynccontextmanager
 
 import click
 
-from planet import data_filter, exceptions, io, DataClient, Session
-
+from planet import data_filter, exceptions, io, DataClient
 from .cmds import coro, translate_exceptions
 from .io import echo_json
+from .session import CliSession
 
 pretty = click.option('--pretty', is_flag=True, help='Pretty-print output.')
 
@@ -31,7 +31,7 @@ pretty = click.option('--pretty', is_flag=True, help='Pretty-print output.')
 async def data_client(ctx):
     auth = ctx.obj['AUTH']
     base_url = ctx.obj['BASE_URL']
-    async with Session(auth=auth) as sess:
+    async with CliSession(auth=auth) as sess:
         cl = DataClient(sess, base_url=base_url)
         yield cl
 
diff --git a/planet/cli/orders.py b/planet/cli/orders.py
index 92bdc6c..f357bd6 100644
--- a/planet/cli/orders.py
+++ b/planet/cli/orders.py
@@ -20,10 +20,11 @@ from pathlib import Path
 import click
 
 import planet
-from planet import OrdersClient, Session  # allow mocking
+from planet import OrdersClient  # allow mocking
 from .cmds import coro, translate_exceptions
 from .io import echo_json
 from .options import pretty
+from .session import CliSession
 
 LOGGER = logging.getLogger(__name__)
 
@@ -32,7 +33,7 @@ LOGGER = logging.getLogger(__name__)
 async def orders_client(ctx):
     auth = ctx.obj['AUTH']
     base_url = ctx.obj['BASE_URL']
-    async with Session(auth=auth) as sess:
+    async with CliSession(auth=auth) as sess:
         cl = OrdersClient(sess, base_url=base_url)
         yield cl
 
diff --git a/planet/cli/session.py b/planet/cli/session.py
new file mode 100644
index 0000000..47baac1
--- /dev/null
+++ b/planet/cli/session.py
@@ -0,0 +1,11 @@
+"""CLI HTTP/auth sessions."""
+
+from planet.auth import AuthType
+from planet.http import Session
+
+
+class CliSession(Session):
+
+    def __init__(self, auth: AuthType = None):
+        super().__init__(auth)
+        self._client.headers.update({'X-Planet-App': 'python-cli'})
diff --git a/planet/cli/subscriptions.py b/planet/cli/subscriptions.py
index 0b9fe0b..3f465a0 100644
--- a/planet/cli/subscriptions.py
+++ b/planet/cli/subscriptions.py
@@ -4,9 +4,9 @@ import json
 
 import click
 
-from planet import Session
-from planet.cli.cmds import coro, translate_exceptions
-from planet.cli.io import echo_json
+from .cmds import coro, translate_exceptions
+from .io import echo_json
+from .session import CliSession
 from planet.clients.subscriptions import SubscriptionsClient
 
 
@@ -39,7 +39,7 @@ def subscriptions(ctx):
 @coro
 async def list_subscriptions_cmd(ctx, status, limit, pretty):
     """Prints a sequence of JSON-encoded Subscription descriptions."""
-    async with Session(auth=ctx.obj['AUTH']) as session:
+    async with CliSession(auth=ctx.obj['AUTH']) as session:
         client = SubscriptionsClient(session)
         filtered_subs = client.list_subscriptions(status=status, limit=limit)
         async for sub in filtered_subs:
@@ -75,7 +75,7 @@ def parse_request(ctx, param, value: str) -> dict:
 @coro
 async def create_subscription_cmd(ctx, request, pretty):
     """Submits a subscription request and prints the API response."""
-    async with Session(auth=ctx.obj['AUTH']) as session:
+    async with CliSession(auth=ctx.obj['AUTH']) as session:
         client = SubscriptionsClient(session)
         sub = await client.create_subscription(request)
         echo_json(sub, pretty)
@@ -89,7 +89,7 @@ async def create_subscription_cmd(ctx, request, pretty):
 @coro
 async def cancel_subscription_cmd(ctx, subscription_id, pretty):
     """Cancels a subscription and prints the API response."""
-    async with Session(auth=ctx.obj['AUTH']) as session:
+    async with CliSession(auth=ctx.obj['AUTH']) as session:
         client = SubscriptionsClient(session)
         sub = await client.cancel_subscription(subscription_id)
         echo_json(sub, pretty)
@@ -104,7 +104,7 @@ async def cancel_subscription_cmd(ctx, subscription_id, pretty):
 @coro
 async def update_subscription_cmd(ctx, subscription_id, request, pretty):
     """Cancels a subscription and prints the API response."""
-    async with Session(auth=ctx.obj['AUTH']) as session:
+    async with CliSession(auth=ctx.obj['AUTH']) as session:
         client = SubscriptionsClient(session)
         sub = await client.update_subscription(subscription_id, request)
         echo_json(sub, pretty)
@@ -125,7 +125,7 @@ async def describe_subscription_cmd(ctx, subscription_id, pretty):
     After we refactor we will change to mocking the API.
 
     """
-    async with Session(auth=ctx.obj['AUTH']) as session:
+    async with CliSession(auth=ctx.obj['AUTH']) as session:
         client = SubscriptionsClient(session)
         sub = await client.get_subscription(subscription_id)
         echo_json(sub, pretty)
@@ -161,7 +161,7 @@ async def list_subscription_results_cmd(ctx,
                                         status,
                                         limit):
     """Gets results of a subscription and prints the API response."""
-    async with Session(auth=ctx.obj['AUTH']) as session:
+    async with CliSession(auth=ctx.obj['AUTH']) as session:
         client = SubscriptionsClient(session)
         filtered_results = client.get_results(subscription_id,
                                               status=status,
diff --git a/planet/clients/subscriptions.py b/planet/clients/subscriptions.py
index e042e57..990105c 100644
--- a/planet/clients/subscriptions.py
+++ b/planet/clients/subscriptions.py
@@ -1,77 +1,79 @@
 """Planet Subscriptions API Python client."""
 
-import itertools
-from typing import AsyncIterator, Dict, Optional, Set
-import uuid
+import logging
+from typing import AsyncIterator, Optional, Set
 
-from planet.exceptions import APIError
+from httpx import URL
 
-# Collections of fake subscriptions and results for testing. Tests will
-# monkeypatch these attributes.
-_fake_subs: Dict[str, dict] = {}
-_fake_sub_results: Dict[str, list] = {}
+from planet.exceptions import APIError, ClientError, PagingError
+from planet.http import Session
+from planet.models import Paged
 
+LOGGER = logging.getLogger()
 
-async def _server_subscriptions_post(request):
-    missing_keys = {'name', 'delivery', 'source'} - request.keys()
-    if missing_keys:
-        raise RuntimeError(f"Request lacks required members: {missing_keys!r}")
 
-    id = str(uuid.uuid4())
-    _fake_subs[id] = request
-    sub = _fake_subs[id].copy()
-    sub.update(id=id)
-    return sub
+class _Pager(Paged):
+    """Navigates pages of messages about subscriptions or results."""
 
+    # This constructor overrides Paged's and takes an httpx
+    # async client as the first argument.
+    def __init__(self, client, request, limit=None):
+        self.request = request
+        self.client = client
+        self._pages = None
+        self._items = []
+        self.i = 0
+        self.limit = limit
 
-async def _server_subscriptions_get(status=None, limit=None):
-    select_subs = (dict(**sub, id=sub_id) for sub_id,
-                   sub in _fake_subs.items()
-                   if not status or sub['status'] in status)
-    filtered_subs = itertools.islice(select_subs, limit)
-    for sub in filtered_subs:
-        yield sub
+    # This method uses the instance's httpx client to build and
+    # send requests. It's less abstracted than Paged's method.
+    async def _get_pages(self):
+        LOGGER.debug('getting first page')
+        resp = await self.client.send(self.request)
+        page = resp.json()
+        yield page
 
+        next_url = self._next_link(page)
 
-async def _server_subscriptions_id_cancel_post(subscription_id):
-    sub = _fake_subs.pop(subscription_id)
-    sub.update(id=subscription_id)
-    return sub
+        while (next_url):  # pragma: no branch
+            LOGGER.debug('getting next page')
+            request = self.client.build_request('GET', next_url)
+            resp = await self.client.send(request)
+            page = resp.json()
+            yield page
 
+            # If the next URL is the same as the previous URL we will
+            # get the same response and be stuck in a page cycle. This
+            # has happened in development and could happen in the case
+            # of a bug in the production API.
+            prev_url = next_url
+            next_url = self._next_link(page)
 
-async def _server_subscriptions_id_put(subscription_id, request):
-    _fake_subs[subscription_id].update(**request)
-    sub = _fake_subs[subscription_id].copy()
-    sub.update(id=subscription_id)
-    return sub
-
-
-async def _server_subscriptions_id_get(subscription_id):
-    sub = _fake_subs[subscription_id].copy()
-    sub.update(id=subscription_id)
-    return sub
-
-
-async def _server_subscriptions_id_results_get(subscription_id,
-                                               status=None,
-                                               limit=None):
-    select_results = (result for result in _fake_sub_results[subscription_id]
-                      if not status or result['status'] in status)
-    filtered_results = itertools.islice(select_results, limit)
-    for result in filtered_results:
-        yield result
+            if next_url == prev_url:
+                raise PagingError(
+                    "Page cycle detected at {!r}".format(next_url))
 
 
 class SubscriptionsClient:
-    """The Planet Subscriptions API client.
+    """A Planet Subscriptions Service API 1.0.0 client.
 
-    TODO: make requests to the production API. Currently, the backend
-    is fake and exists at the top of this class's module.
+    The methods of this class forward request parameters to the
+    operations described in the Planet Subscriptions Service API 1.0.0
+    (https://api.planet.com/subscriptions/v1/spec) using HTTP 1.1.
 
+    The methods generally return or yield Python dicts with the same
+    structure as the JSON messages used by the service API. Many of the
+    exceptions raised by this class are categorized by HTTP client
+    (4xx) or server (5xx) errors. This client's level of abstraction is
+    low.
+
+    Attributes:
+        session (Session): an authenticated session which wraps the
+            low-level HTTP client.
     """
 
-    def __init__(self, session=None) -> None:
-        self._session = session
+    def __init__(self, session: Session) -> None:
+        self.session = session
 
     async def list_subscriptions(self,
                                  status: Optional[Set[str]] = None,
@@ -79,9 +81,9 @@ class SubscriptionsClient:
         """Get account subscriptions with optional filtering.
 
         Note:
-            The name of this method is based on the API's method name. This
-            method provides iteration over subcriptions, it does not return
-            a list.
+            The name of this method is based on the API's method name.
+            This method provides iteration over subcriptions, it does
+            not return a list.
 
         Args:
             status (Set[str]): pass subscriptions with status in this
@@ -97,18 +99,27 @@ class SubscriptionsClient:
         Raises:
             APIError: on an API server error.
             ClientError: on a client error.
-
         """
+
+        class _SubscriptionsPager(_Pager):
+            """Navigates pages of messages about subscriptions."""
+            ITEMS_KEY = 'subscriptions'
+
+        params = {'status': [val for val in status or {}]}
+        url = URL('https://api.planet.com/subscriptions/v1', params=params)
+        req = self.session._client.build_request('GET', url)
+
         try:
-            # TODO: replace with httpx request.
-            async for sub in _server_subscriptions_get(status=status,
-                                                       limit=limit):
+            async for sub in _SubscriptionsPager(self.session._client,
+                                                 req,
+                                                 limit=limit):
                 yield sub
-        except Exception as server_error:
-            # TODO: remove "from server_error" clause. It's useful
-            # during development but may invite users to depend on the
-            # value of server_error.
-            raise APIError("Subscription failure") from server_error
+        # Forward APIError. We don't strictly need this clause, but it
+        # makes our intent clear.
+        except APIError:
+            raise
+        except ClientError:  # pragma: no cover
+            raise
 
     async def create_subscription(self, request: dict) -> dict:
         """Create a Subscription.
@@ -122,43 +133,49 @@ class SubscriptionsClient:
         Raises:
             APIError: on an API server error.
             ClientError: on a client error.
-
         """
+
+        url = URL('https://api.planet.com/subscriptions/v1')
+        req = self.session._client.build_request('POST', url, json=request)
+
         try:
-            # TODO: replace with httpx request.
-            sub = await _server_subscriptions_post(request)
-        except Exception as server_error:
-            # TODO: remove "from server_error" clause. It's useful
-            # during development but may invite users to depend on the
-            # value of server_error.
-            raise APIError("Subscription failure") from server_error
+            resp = await self.session._client.send(req)
+        # Forward APIError. We don't strictly need this clause, but it
+        # makes our intent clear.
+        except APIError:
+            raise
+        except ClientError:  # pragma: no cover
+            raise
         else:
+            sub = resp.json()
             return sub
 
-    async def cancel_subscription(self, subscription_id: str) -> dict:
+    async def cancel_subscription(self, subscription_id: str) -> None:
         """Cancel a Subscription.
 
         Args:
             subscription_id (str): id of subscription to cancel.
 
         Returns:
-            dict: description of cancelled subscription.
+            None
 
         Raises:
             APIError: on an API server error.
             ClientError: on a client error.
-
         """
+        url = URL(
+            f'https://api.planet.com/subscriptions/v1/{subscription_id}/cancel'
+        )
+        req = self.session._client.build_request('POST', url)
+
         try:
-            # TODO: replace with httpx request.
-            sub = await _server_subscriptions_id_cancel_post(subscription_id)
-        except Exception as server_error:
-            # TODO: remove "from server_error" clause. It's useful
-            # during development but may invite users to depend on the
-            # value of server_error.
-            raise APIError("Subscription failure.") from server_error
-        else:
-            return sub
+            _ = await self.session._client.send(req)
+        # Forward APIError. We don't strictly need this clause, but it
+        # makes our intent clear.
+        except APIError:
+            raise
+        except ClientError:  # pragma: no cover
+            raise
 
     async def update_subscription(self, subscription_id: str,
                                   request: dict) -> dict:
@@ -174,17 +191,20 @@ class SubscriptionsClient:
         Raises:
             APIError: on an API server error.
             ClientError: on a client error.
-
         """
+        url = URL(f'https://api.planet.com/subscriptions/v1/{subscription_id}')
+        req = self.session._client.build_request('PUT', url, json=request)
+
         try:
-            # TODO: replace with httpx request.
-            sub = await _server_subscriptions_id_put(subscription_id, request)
-        except Exception as server_error:
-            # TODO: remove "from server_error" clause. It's useful
-            # during development but may invite users to depend on the
-            # value of server_error.
-            raise APIError("Subscription failure.") from server_error
+            resp = await self.session._client.send(req)
+        # Forward APIError. We don't strictly need this clause, but it
+        # makes our intent clear.
+        except APIError:
+            raise
+        except ClientError:  # pragma: no cover
+            raise
         else:
+            sub = resp.json()
             return sub
 
     async def get_subscription(self, subscription_id: str) -> dict:
@@ -199,17 +219,20 @@ class SubscriptionsClient:
         Raises:
             APIError: on an API server error.
             ClientError: on a client error.
-
         """
+        url = URL(f'https://api.planet.com/subscriptions/v1/{subscription_id}')
+        req = self.session._client.build_request('GET', url)
+
         try:
-            # TODO: replace with httpx request.
-            sub = await _server_subscriptions_id_get(subscription_id)
-        except Exception as server_error:
-            # TODO: remove "from server_error" clause. It's useful
-            # during development but may invite users to depend on the
-            # value of server_error.
-            raise APIError("Subscription failure.") from server_error
+            resp = await self.session._client.send(req)
+        # Forward APIError. We don't strictly need this clause, but it
+        # makes our intent clear.
+        except APIError:
+            raise
+        except ClientError:  # pragma: no cover
+            raise
         else:
+            sub = resp.json()
             return sub
 
     async def get_results(self,
@@ -237,15 +260,27 @@ class SubscriptionsClient:
         Raises:
             APIError: on an API server error.
             ClientError: on a client error.
-
         """
+
+        class _ResultsPager(_Pager):
+            """Navigates pages of messages about subscription results."""
+            NEXT_KEY = '_next'
+            ITEMS_KEY = 'results'
+
+        params = {'status': [val for val in status or {}]}
+        url = URL(
+            'https://api.planet.com/subscriptions/v1/{subscription_id}/results',
+            params=params)
+        req = self.session._client.build_request('GET', url)
+
         try:
-            # TODO: replace with httpx request.
-            async for result in _server_subscriptions_id_results_get(
-                    subscription_id, status=status, limit=limit):
-                yield result
-        except Exception as server_error:
-            # TODO: remove "from server_error" clause. It's useful
-            # during development but may invite users to depend on the
-            # value of server_error.
-            raise APIError("Subscription failure.") from server_error
+            async for sub in _ResultsPager(self.session._client,
+                                           req,
+                                           limit=limit):
+                yield sub
+        # Forward APIError. We don't strictly need this clause, but it
+        # makes our intent clear.
+        except APIError:
+            raise
+        except ClientError:  # pragma: no cover
+            raise
diff --git a/planet/exceptions.py b/planet/exceptions.py
index d5129a6..7fcf32f 100644
--- a/planet/exceptions.py
+++ b/planet/exceptions.py
@@ -78,7 +78,7 @@ class AuthException(ClientError):
     pass
 
 
-class PagingError(PlanetError):
+class PagingError(ClientError):
     '''For errors that occur during paging.'''
     pass
 
diff --git a/setup.cfg b/setup.cfg
index aad6c3d..14e2987 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -20,3 +20,6 @@ fail_under = 98
 [yapf]
 based_on_style = pep8
 split_all_top_level_comma_separated_values=true
+
+[flake8]
+ignore = E126,E501,W503,W504


Test patch:
diff --git a/tests/integration/test_subscriptions_api.py b/tests/integration/test_subscriptions_api.py
index 3b059a8..ffc01ee 100644
--- a/tests/integration/test_subscriptions_api.py
+++ b/tests/integration/test_subscriptions_api.py
@@ -1,36 +1,166 @@
 """Tests of the Planet Subscriptions API client."""
 
+from datetime import datetime
+from itertools import zip_longest
+import json
+
+from httpx import Response
 import pytest
+import respx
+from respx.patterns import M
 
-import planet.clients.subscriptions
 from planet.clients.subscriptions import SubscriptionsClient
-from planet.exceptions import APIError
+from planet.exceptions import APIError, PagingError, ServerError
 from planet.http import Session
 
+# Mock router representing an API that returns only a 500
+# internal server error. The router is configured outside the test
+# to help keep our test more readable.
+failing_api_mock = respx.mock()
+failing_api_mock.route(
+    M(url__startswith='https://api.planet.com/subscriptions/v1')).mock(
+        return_value=Response(500, json={
+            "code": 0, "message": "string"
+        }))
+
+
+def result_pages(status=None, size=40):
+    """Helper for creating fake subscriptions listing pages."""
+    all_subs = [{'id': str(i), 'status': 'created'} for i in range(1, 101)]
+    select_subs = (sub for sub in all_subs
+                   if not status or sub['status'] in status)
+    pages = []
+    for group in zip_longest(*[iter(select_subs)] * size):
+        subs = list(sub for sub in group if sub is not None)
+        page = {'subscriptions': subs, '_links': {}}
+        pm = datetime.now().isoformat()
+        if len(subs) == size:
+            page['_links'][
+                'next'] = f'https://api.planet.com/subscriptions/v1?pm={pm}'
+        pages.append(page)
+    return pages
+
+
+# By default, respx's mock router asserts that all configured routes
+# are followed in a test. Each of our parameterized tests follows only
+# one route, so if we want to configure routes once and reuse them, we
+# must disable the default.
+api_mock = respx.mock(assert_all_called=False)
+
+# 1. Request for status: created. Response has three pages.
+api_mock.route(M(url__startswith='https://api.planet.com/subscriptions/v1'),
+               M(params__contains={'status': 'created'})).mock(side_effect=[
+                   Response(200, json=page)
+                   for page in result_pages(status={'created'}, size=40)
+               ])
+
+# 2. Request for status: failed. Response has a single empty page.
+api_mock.route(M(url__startswith='https://api.planet.com/subscriptions/v1'),
+               M(params__contains={'status': 'failed'})).mock(
+                   side_effect=[Response(200, json={'subscriptions': []})])
+
+# 3. Request for status: queued. Response has a single empty page.
+api_mock.route(M(url__startswith='https://api.planet.com/subscriptions/v1'),
+               M(params__contains={'status': 'queued'})).mock(
+                   side_effect=[Response(200, json={'subscriptions': []})])
+
+# 4. No status requested. Response is the same as for 1.
+api_mock.route(
+    M(url__startswith='https://api.planet.com/subscriptions/v1')
+).mock(
+    side_effect=[Response(200, json=page) for page in result_pages(size=40)])
+
+
+# This mock API returns posted data to the caller. It's used to test
+# methods that rely on POST or PUT.
+def modify_response(request):
+    if request.content:
+        return Response(200, json=json.loads(request.content))
+    else:
+        return Response(200)
+
+
+modify_api_mock = respx.mock()
+modify_api_mock.route(
+    M(url__startswith='https://api.planet.com/subscriptions/v1'),
+    M(method__in=['POST', 'PUT'])).mock(side_effect=modify_response)
+
+sub_api_mock = respx.mock()
+sub_api_mock.route(
+    M(url__startswith='https://api.planet.com/subscriptions/v1'),
+    M(method__in=['GET'])).mock(
+        return_value=Response(200,
+                              json={
+                                  'id': '42',
+                                  'name': 'test',
+                                  'delivery': 'yes, please',
+                                  'source': 'test'
+                              }))
+
+
+def result_pages(status=None, size=40):
+    """Helper for creating fake result listing pages."""
+    all_results = [{'id': str(i), 'status': 'created'} for i in range(1, 101)]
+    select_results = (result for result in all_results
+                      if not status or result['status'] in status)
+    pages = []
+    for group in zip_longest(*[iter(select_results)] * size):
+        results = list(result for result in group if result is not None)
+        page = {'results': results, '_links': {}}
+        pm = datetime.now().isoformat()
+        if len(results) == size:
+            page['_links'][
+                '_next'] = f'https://api.planet.com/subscriptions/v1/42/results?pm={pm}'
+        pages.append(page)
+    return pages
+
+
+# By default, respx's mock router asserts that all configured routes
+# are followed in a test. Each of our parameterized tests follows only
+# one route, so if we want to configure routes once and reuse them, we
+# must disable the default.
+res_api_mock = respx.mock(assert_all_called=False)
+
+# 1. Request for status: created. Response has three pages.
+res_api_mock.route(
+    M(url__startswith='https://api.planet.com/subscriptions/v1'),
+    M(params__contains={'status': 'created'})).mock(side_effect=[
+        Response(200, json=page)
+        for page in result_pages(status={'created'}, size=40)
+    ])
+
+# 2. Request for status: queued. Response has a single empty page.
+res_api_mock.route(
+    M(url__startswith='https://api.planet.com/subscriptions/v1'),
+    M(params__contains={'status': 'queued'})).mock(
+        side_effect=[Response(200, json={'results': []})])
+
+# 3. No status requested. Response is the same as for 1.
+res_api_mock.route(
+    M(url__startswith='https://api.planet.com/subscriptions/v1')
+).mock(
+    side_effect=[Response(200, json=page) for page in result_pages(size=40)])
+
 
-@pytest.mark.xfail(reason="Client/server interaction not yet implemented")
 @pytest.mark.asyncio
-async def test_list_subscriptions_failure(monkeypatch):
-    """APIError is raised if there is a server error."""
-    monkeypatch.setattr(planet.clients.subscriptions, "_fake_subs", 0)
-    with pytest.raises(APIError):
+@failing_api_mock
+async def test_list_subscriptions_failure():
+    """ServerError is raised if there is an internal server error (500)."""
+    with pytest.raises(ServerError):
         async with Session() as session:
             client = SubscriptionsClient(session)
             _ = [sub async for sub in client.list_subscriptions()]
 
 
-@pytest.mark.xfail(reason="Client/server interaction not yet implemented")
 @pytest.mark.parametrize("status, count", [({"created"}, 100), ({"failed"}, 0),
                                            (None, 100)])
 @pytest.mark.asyncio
-async def test_list_subscriptions_success(status, count, monkeypatch):
+@api_mock
+async def test_list_subscriptions_success(
+    status,
+    count,
+):
     """Account subscriptions iterator yields expected descriptions."""
-    monkeypatch.setattr(planet.clients.subscriptions,
-                        '_fake_subs',
-                        {str(i): {
-                            'status': 'created'
-                        }
-                         for i in range(101)})
     async with Session() as session:
         client = SubscriptionsClient(session)
         assert len([
@@ -38,22 +168,20 @@ async def test_list_subscriptions_success(status, count, monkeypatch):
         ]) == count
 
 
-@pytest.mark.xfail(reason="Client/server interaction not yet implemented")
 @pytest.mark.asyncio
-async def test_create_subscription_failure(monkeypatch):
+@failing_api_mock
+async def test_create_subscription_failure():
     """APIError is raised if there is a server error."""
-    monkeypatch.setattr(planet.clients.subscriptions, "_fake_subs", {})
-    with pytest.raises(APIError):
+    with pytest.raises(ServerError):
         async with Session() as session:
             client = SubscriptionsClient(session)
             _ = await client.create_subscription({"lol": "wut"})
 
 
-@pytest.mark.xfail(reason="Client/server interaction not yet implemented")
 @pytest.mark.asyncio
-async def test_create_subscription_success(monkeypatch):
+@modify_api_mock
+async def test_create_subscription_success():
     """Subscription is created, description has the expected items."""
-    monkeypatch.setattr(planet.clients.subscriptions, "_fake_subs", {})
     async with Session() as session:
         client = SubscriptionsClient(session)
         sub = await client.create_subscription({
@@ -62,56 +190,39 @@ async def test_create_subscription_success(monkeypatch):
         assert sub['name'] == 'test'
 
 
-@pytest.mark.xfail(reason="Client/server interaction not yet implemented")
 @pytest.mark.asyncio
-async def test_cancel_subscription_failure(monkeypatch):
+@failing_api_mock
+async def test_cancel_subscription_failure():
     """APIError is raised if there is a server error."""
-    monkeypatch.setattr(planet.clients.subscriptions, "_fake_subs", {})
-    with pytest.raises(APIError):
+    with pytest.raises(ServerError):
         async with Session() as session:
             client = SubscriptionsClient(session)
             _ = await client.cancel_subscription("lolwut")
 
 
-@pytest.mark.xfail(reason="Client/server interaction not yet implemented")
 @pytest.mark.asyncio
-async def test_cancel_subscription_success(monkeypatch):
+@modify_api_mock
+async def test_cancel_subscription_success():
     """Subscription is canceled, description has the expected items."""
-    monkeypatch.setattr(
-        planet.clients.subscriptions,
-        "_fake_subs", {
-            "test": {
-                'name': 'test', 'delivery': 'yes, please', 'source': 'test'
-            }
-        })
     async with Session() as session:
         client = SubscriptionsClient(session)
-        sub = await client.cancel_subscription("test")
-        assert sub['name'] == 'test'
+        _ = await client.cancel_subscription("test")
 
 
-@pytest.mark.xfail(reason="Client/server interaction not yet implemented")
 @pytest.mark.asyncio
-async def test_update_subscription_failure(monkeypatch):
+@failing_api_mock
+async def test_update_subscription_failure():
     """APIError is raised if there is a server error."""
-    monkeypatch.setattr(planet.clients.subscriptions, "_fake_subs", {})
-    with pytest.raises(APIError):
+    with pytest.raises(ServerError):
         async with Session() as session:
             client = SubscriptionsClient(session)
             _ = await client.update_subscription("lolwut", {})
 
 
-@pytest.mark.xfail(reason="Client/server interaction not yet implemented")
 @pytest.mark.asyncio
-async def test_update_subscription_success(monkeypatch):
+@modify_api_mock
+async def test_update_subscription_success():
     """Subscription is created, description has the expected items."""
-    monkeypatch.setattr(
-        planet.clients.subscriptions,
-        "_fake_subs", {
-            "test": {
-                'name': 'test', 'delivery': 'yes, please', 'source': 'test'
-            }
-        })
     async with Session() as session:
         client = SubscriptionsClient(session)
         sub = await client.update_subscription(
@@ -121,56 +232,78 @@ async def test_update_subscription_success(monkeypatch):
         assert sub['delivery'] == "no, thanks"
 
 
-@pytest.mark.xfail(reason="Client/server interaction not yet implemented")
 @pytest.mark.asyncio
-async def test_get_subscription_failure(monkeypatch):
+@failing_api_mock
+async def test_get_subscription_failure():
     """APIError is raised if there is a server error."""
-    monkeypatch.setattr(planet.clients.subscriptions, "_fake_subs", {})
-    with pytest.raises(APIError):
+    with pytest.raises(ServerError):
         async with Session() as session:
             client = SubscriptionsClient(session)
             _ = await client.get_subscription("lolwut")
 
 
-@pytest.mark.xfail(reason="Client/server interaction not yet implemented")
 @pytest.mark.asyncio
+@sub_api_mock
 async def test_get_subscription_success(monkeypatch):
     """Subscription description fetched, has the expected items."""
-    monkeypatch.setattr(
-        planet.clients.subscriptions,
-        "_fake_subs", {
-            "test": {
-                'name': 'test', 'delivery': 'yes, please', 'source': 'test'
-            }
-        })
     async with Session() as session:
         client = SubscriptionsClient(session)
         sub = await client.get_subscription("test")
         assert sub['delivery'] == "yes, please"
 
 
-@pytest.mark.xfail(reason="Client/server interaction not yet implemented")
 @pytest.mark.asyncio
-async def test_get_results_failure(monkeypatch):
+@failing_api_mock
+async def test_get_results_failure():
     """APIError is raised if there is a server error."""
-    monkeypatch.setattr(planet.clients.subscriptions, "_fake_sub_results", {})
     with pytest.raises(APIError):
         async with Session() as session:
             client = SubscriptionsClient(session)
             _ = [res async for res in client.get_results("lolwut")]
 
 
-@pytest.mark.xfail(reason="Client/server interaction not yet implemented")
 @pytest.mark.asyncio
-async def test_get_results_success(monkeypatch):
+@res_api_mock
+async def test_get_results_success():
     """Subscription description fetched, has the expected items."""
-    monkeypatch.setattr(
-        planet.clients.subscriptions,
-        '_fake_sub_results',
-        {'42': [{
-            'id': f'r{i}', 'status': 'created'
-        } for i in range(101)]})
     async with Session() as session:
         client = SubscriptionsClient(session)
         results = [res async for res in client.get_results("42")]
         assert len(results) == 100
+
+
+paging_cycle_api_mock = respx.mock()
+
+# Identical next links is a hangup we want to avoid.
+paging_cycle_api_mock.route(
+    M(url__startswith='https://api.planet.com/subscriptions/v1')).mock(
+        side_effect=[
+            Response(200,
+                     json={
+                         'subscriptions': [{
+                             'id': '1'
+                         }],
+                         '_links': {
+                             "next": 'https://api.planet.com/subscriptions/v1'
+                         }
+                     }),
+            Response(200,
+                     json={
+                         'subscriptions': [{
+                             'id': '2'
+                         }],
+                         '_links': {
+                             "next": 'https://api.planet.com/subscriptions/v1'
+                         }
+                     })
+        ])
+
+
+@pytest.mark.asyncio
+@paging_cycle_api_mock
+async def test_list_subscriptions_cycle_break():
+    """PagingError is raised if there is a paging cycle."""
+    with pytest.raises(PagingError):
+        async with Session() as session:
+            client = SubscriptionsClient(session)
+            _ = [sub async for sub in client.list_subscriptions()]
diff --git a/tests/integration/test_subscriptions_cli.py b/tests/integration/test_subscriptions_cli.py
index f1eb8d9..2d7e66b 100644
--- a/tests/integration/test_subscriptions_cli.py
+++ b/tests/integration/test_subscriptions_cli.py
@@ -13,33 +13,18 @@ TODO: tests for 3 options of the planet-subscriptions-results command.
 
 """
 
-import asyncio
 import json
-import uuid
 
 from click.testing import CliRunner
 import pytest
 
 from planet.cli import cli
-import planet.cli.subscriptions
-from planet.clients.subscriptions import SubscriptionsClient
 
-
-@pytest.fixture
-def subscription_count():
-    """Return a function that counts subscriptions in the system."""
-
-    def func():
-
-        async def count_em():
-            return len([
-                sub
-                async for sub in SubscriptionsClient().list_subscriptions()
-            ])
-
-        return asyncio.run(count_em())
-
-    return func
+from test_subscriptions_api import (api_mock,
+                                    failing_api_mock,
+                                    modify_api_mock,
+                                    sub_api_mock,
+                                    res_api_mock)
 
 
 # CliRunner doesn't agree with empty options, so a list of option
@@ -54,17 +39,10 @@ def subscription_count():
                           (['--limit=1', '--status=created'], 1),
                           (['--limit=2', '--pretty', '--status=created'], 2),
                           (['--limit=1', '--status=queued'], 0)])
+@api_mock
 # Remember, parameters come before fixtures in the function definition.
-def test_subscriptions_list_options(options, expected_count, monkeypatch):
+def test_subscriptions_list_options(options, expected_count):
     """Prints the expected sequence of subscriptions."""
-
-    monkeypatch.setattr(planet.clients.subscriptions,
-                        '_fake_subs',
-                        {str(i): {
-                            'status': 'created'
-                        }
-                         for i in range(101)})
-
     # While developing it is handy to have click's command invoker
     # *not* catch exceptions, so we can use the pytest --pdb option.
     result = CliRunner().invoke(cli.main,
@@ -77,11 +55,9 @@ def test_subscriptions_list_options(options, expected_count, monkeypatch):
     assert result.output.count('"id"') == expected_count
 
 
-def test_subscriptions_create_failure(monkeypatch, subscription_count):
+@failing_api_mock
+def test_subscriptions_create_failure():
     """An invalid subscription request fails to create a new subscription."""
-
-    monkeypatch.setattr(planet.clients.subscriptions, '_fake_subs', {})
-
     # This subscription request lacks the required "delivery" and
     # "source" members.
     sub = {'name': 'lol'}
@@ -98,7 +74,7 @@ def test_subscriptions_create_failure(monkeypatch, subscription_count):
         catch_exceptions=True)
 
     assert result.exit_code == 1  # failure.
-    assert subscription_count() == 0
+    assert "Error:" in result.output
 
 
 # This subscription request has the members required by our fake API.
@@ -110,14 +86,9 @@ GOOD_SUB_REQUEST = {'name': 'lol', 'delivery': True, 'source': 'wut'}
 @pytest.mark.parametrize('cmd_arg, runner_input',
                          [('-', json.dumps(GOOD_SUB_REQUEST)),
                           (json.dumps(GOOD_SUB_REQUEST), None)])
-def test_subscriptions_create_success(cmd_arg,
-                                      runner_input,
-                                      monkeypatch,
-                                      subscription_count):
+@modify_api_mock
+def test_subscriptions_create_success(cmd_arg, runner_input):
     """Subscriptions creation succeeds with a valid subscription request."""
-
-    monkeypatch.setattr(planet.clients.subscriptions, '_fake_subs', {})
-
     # The "-" argument says "read from stdin" and the input keyword
     # argument specifies what bytes go to the runner's stdin.
     result = CliRunner().invoke(
@@ -130,8 +101,6 @@ def test_subscriptions_create_success(cmd_arg,
         catch_exceptions=True)
 
     assert result.exit_code == 0  # success.
-    assert uuid.UUID(json.loads(result.output)['id'])
-    assert subscription_count() == 1
 
 
 # Invalid JSON.
@@ -140,9 +109,8 @@ BAD_SUB_REQUEST = '{0: "lolwut"}'
 
 @pytest.mark.parametrize('cmd_arg, runner_input', [('-', BAD_SUB_REQUEST),
                                                    (BAD_SUB_REQUEST, None)])
-def test_subscriptions_bad_request(cmd_arg, runner_input, monkeypatch):
+def test_subscriptions_bad_request(cmd_arg, runner_input):
     """Short circuit and print help message if request is bad."""
-
     # The "-" argument says "read from stdin" and the input keyword
     # argument specifies what bytes go to the runner's stdin.
     result = CliRunner().invoke(
@@ -158,11 +126,9 @@ def test_subscriptions_bad_request(cmd_arg, runner_input, monkeypatch):
     assert "Request does not contain valid json" in result.output
 
 
-def test_subscriptions_cancel_failure(monkeypatch):
+@failing_api_mock
+def test_subscriptions_cancel_failure():
     """Cancel command exits gracefully from an API error."""
-
-    monkeypatch.setattr(planet.clients.subscriptions, '_fake_subs', {})
-
     result = CliRunner().invoke(
         cli.main,
         args=['subscriptions', 'cancel', '42'],
@@ -174,15 +140,9 @@ def test_subscriptions_cancel_failure(monkeypatch):
     assert result.exit_code == 1  # failure.
 
 
-def test_subscriptions_cancel_success(monkeypatch, subscription_count):
+@modify_api_mock
+def test_subscriptions_cancel_success():
     """Cancel command succeeds."""
-
-    monkeypatch.setattr(planet.clients.subscriptions,
-                        '_fake_subs', {'42': GOOD_SUB_REQUEST})
-
-    # Let's check the state of the fake API before we try to cancel.
-    assert subscription_count() == 1
-
     result = CliRunner().invoke(
         cli.main,
         args=['subscriptions', 'cancel', '42'],
@@ -192,15 +152,11 @@ def test_subscriptions_cancel_success(monkeypatch, subscription_count):
         catch_exceptions=True)
 
     assert result.exit_code == 0  # success.
-    assert json.loads(result.output)['id'] == '42'
-    assert subscription_count() == 0
 
 
-def test_subscriptions_update_failure(monkeypatch):
+@failing_api_mock
+def test_subscriptions_update_failure():
     """Update command exits gracefully from an API error."""
-
-    monkeypatch.setattr(planet.clients.subscriptions, '_fake_subs', {})
-
     result = CliRunner().invoke(
         cli.main,
         args=['subscriptions', 'update', '42', json.dumps(GOOD_SUB_REQUEST)],
@@ -212,12 +168,9 @@ def test_subscriptions_update_failure(monkeypatch):
     assert result.exit_code == 1  # failure.
 
 
-def test_subscriptions_update_success(monkeypatch):
+@modify_api_mock
+def test_subscriptions_update_success():
     """Update command succeeds."""
-
-    monkeypatch.setattr(planet.clients.subscriptions,
-                        '_fake_subs', {'42': GOOD_SUB_REQUEST})
-
     request = GOOD_SUB_REQUEST.copy()
     request['name'] = 'new_name'
 
@@ -233,11 +186,9 @@ def test_subscriptions_update_success(monkeypatch):
     assert json.loads(result.output)['name'] == 'new_name'
 
 
-def test_subscriptions_describe_failure(monkeypatch):
+@failing_api_mock
+def test_subscriptions_describe_failure():
     """Describe command exits gracefully from an API error."""
-
-    monkeypatch.setattr(planet.clients.subscriptions, '_fake_subs', {})
-
     result = CliRunner().invoke(
         cli.main,
         args=['subscriptions', 'describe', '42'],
@@ -249,12 +200,9 @@ def test_subscriptions_describe_failure(monkeypatch):
     assert result.exit_code == 1  # failure.
 
 
-def test_subscriptions_describe_success(monkeypatch):
+@sub_api_mock
+def test_subscriptions_describe_success():
     """Describe command succeeds."""
-
-    monkeypatch.setattr(planet.clients.subscriptions,
-                        '_fake_subs', {'42': GOOD_SUB_REQUEST})
-
     result = CliRunner().invoke(
         cli.main,
         args=['subscriptions', 'describe', '42'],
@@ -267,11 +215,9 @@ def test_subscriptions_describe_success(monkeypatch):
     assert json.loads(result.output)['id'] == '42'
 
 
-def test_subscriptions_results_failure(monkeypatch):
+@failing_api_mock
+def test_subscriptions_results_failure():
     """Results command exits gracefully from an API error."""
-
-    monkeypatch.setattr(planet.clients.subscriptions, '_fake_sub_results', {})
-
     result = CliRunner().invoke(
         cli.main,
         args=['subscriptions', 'results', '42'],
@@ -281,7 +227,6 @@ def test_subscriptions_results_failure(monkeypatch):
         catch_exceptions=True)
 
     assert result.exit_code == 1  # failure.
-    assert "Subscription failure." in result.output
 
 
 @pytest.mark.parametrize('options,expected_count',
@@ -289,17 +234,10 @@ def test_subscriptions_results_failure(monkeypatch):
                           (['--limit=1', '--status=created'], 1),
                           (['--limit=2', '--pretty', '--status=created'], 2),
                           (['--limit=1', '--status=queued'], 0)])
+@res_api_mock
 # Remember, parameters come before fixtures in the function definition.
-def test_subscriptions_results_success(options, expected_count, monkeypatch):
+def test_subscriptions_results_success(options, expected_count):
     """Describe command succeeds."""
-
-    monkeypatch.setattr(
-        planet.clients.subscriptions,
-        '_fake_sub_results',
-        {'42': [{
-            'id': f'r{i}', 'status': 'created'
-        } for i in range(101)]})
-
     result = CliRunner().invoke(
         cli.main,
         args=['subscriptions', 'results', '42'] + options,