From e642e9177337d57abf33faab8d470d1cc2c285d2 Mon Sep 17 00:00:00 2001 From: Cody Mitchell Date: Sun, 13 Sep 2026 23:17:47 -0500 Subject: [PATCH 1/2] Keep V3 token caches local to each client --- actiapi/v3.py | 17 ++--- tests/test_v3_authentication.py | 119 ++++++++++++++++++++++++++++++++ 2 files changed, 128 insertions(+), 8 deletions(-) create mode 100644 tests/test_v3_authentication.py diff --git a/actiapi/v3.py b/actiapi/v3.py index 3a081cc..99aab25 100644 --- a/actiapi/v3.py +++ b/actiapi/v3.py @@ -4,14 +4,12 @@ """ import logging -from collections import defaultdict from typing import Any, Dict, List, Literal, Optional, Union import requests from actiapi import ActiGraphClient -tokens: defaultdict[str, Optional[str]] = defaultdict(lambda: None) session = requests.Session() logger = logging.getLogger(__name__) @@ -23,6 +21,11 @@ class ActiGraphClientV3(ActiGraphClient): BASE_URL = "https://api.actigraphcorp.com" AUTH_API = "https://auth.actigraphcorp.com/connect/token" + def __init__(self, api_access_key: str, api_secret_key: str): + """Initialize a client with an independent token cache.""" + super().__init__(api_access_key, api_secret_key) + self._tokens: Dict[str, Optional[str]] = {} + @staticmethod def _generate_headers(token: str, raw: bool = False): headers = {} @@ -220,17 +223,15 @@ def get_sleep_summary( return results def _get_single(self, request: str, scope: str): - global tokens - if tokens[scope] is None: - tokens[scope] = self._get_access_token(scope) + if self._tokens.get(scope) is None: + self._tokens[scope] = self._get_access_token(scope) logger.info("Requesting %s", request) - headers = self._generate_headers(str(tokens[scope])) + headers = self._generate_headers(str(self._tokens[scope])) response = session.get(self.BASE_URL + request, headers=headers, stream=False) reply = validate_response(response) return reply def _get_paginated(self, request: str, scope: str): - global tokens results = [] offset = 0 limit = 100 @@ -243,7 +244,7 @@ def _get_paginated(self, request: str, scope: str): total_count = reply["totalCount"] except KeyError: - tokens[scope] = None + self._tokens.pop(scope, None) reply = self._get_single(request=paginated_request, scope=scope) if reply is None: break diff --git a/tests/test_v3_authentication.py b/tests/test_v3_authentication.py new file mode 100644 index 0000000..82342b1 --- /dev/null +++ b/tests/test_v3_authentication.py @@ -0,0 +1,119 @@ +import json +from collections import deque +from urllib.parse import parse_qs, urlsplit + +import pytest +import requests + +from actiapi.v3 import ActiGraphClientV3 + + +@pytest.fixture +def http(monkeypatch): + calls = {"auth": [], "get": [], "replies": deque()} + + def request(session, method, url, **kwargs): + response = requests.Response() + response.status_code = 200 + if method.upper() == "POST": + data = kwargs["data"] + calls["auth"].append((data["client_id"], data["scope"])) + payload = {"access_token": f"synthetic-{data['client_id']}-{data['scope']}"} + else: + calls["get"].append((url, kwargs["headers"]["Authorization"])) + if calls["replies"]: + payload = calls["replies"].popleft() + elif "offset=" in url: + payload = {"totalCount": 1, "items": [ + {"id": 1, "downloadUrl": "https://example.invalid/synthetic"} + ]} + else: + payload = {"id": 1} + response._content = json.dumps(payload).encode() + return response + + def forbidden_send(*args, **kwargs): + pytest.fail("A test attempted a real HTTP request") + + monkeypatch.setattr(requests.Session, "request", request) + monkeypatch.setattr(requests.Session, "send", forbidden_send) + return calls + + +def test_clients_use_their_own_credentials_at_the_same_scope(http): + first = ActiGraphClientV3("first", "synthetic-secret-a") + second = ActiGraphClientV3("second", "synthetic-secret-b") + for client in (first, second, first, second): + assert client.get_study_info(1) == {"id": 1} + assert [token for _, token in http["get"]] == [ + "Bearer synthetic-first-CentrePoint", + "Bearer synthetic-second-CentrePoint", + "Bearer synthetic-first-CentrePoint", + "Bearer synthetic-second-CentrePoint", + ] + assert http["auth"] == [("first", "CentrePoint"), ("second", "CentrePoint")] + + +def test_client_reuses_each_scope_independently(http): + client = ActiGraphClientV3("first", "synthetic-secret") + for _ in range(2): + for scope in ("CentrePoint", "Analytics", "DataAccess"): + client._get_single("/synthetic", scope) + assert http["auth"] == [ + ("first", "CentrePoint"), ("first", "Analytics"), ("first", "DataAccess") + ] + + +@pytest.mark.parametrize("method,args,scope", [ + ("get_study_info", (1,), "CentrePoint"), + ("get_studies", (), "CentrePoint"), + ("get_study_metadata", (1,), "CentrePoint"), + ("get_event_markers", (2, 1), "Analytics"), + ("get_minute_summary", (2, 1), "Analytics"), + ("get_daily_summary", (2, 1), "Analytics"), + ("get_sleep_summary", (2, 1), "Analytics"), + ("get_files", (2, 1), "DataAccess"), +]) +def test_public_v3_methods_keep_their_scope(http, method, args, scope): + client = ActiGraphClientV3("wrapper", "synthetic-secret") + result = getattr(client, method)(*args) + assert result + assert http["auth"] == [("wrapper", scope)] + assert http["get"][0][1] == f"Bearer synthetic-wrapper-{scope}" + + +def test_pagination_reuses_token_and_preserves_page_order(http): + http["replies"].extend([ + {"totalCount": 101, "items": [{"id": i} for i in range(100)]}, + {"totalCount": 101, "items": [{"id": 100}]}, + ]) + client = ActiGraphClientV3("pages", "synthetic-secret") + assert client.get_studies() == [{"id": i} for i in range(101)] + assert http["auth"] == [("pages", "CentrePoint")] + offsets = [parse_qs(urlsplit(url).query)["offset"] for url, _ in http["get"]] + assert offsets == [["0"], ["100"]] + + +def test_pagination_refresh_does_not_invalidate_another_client(http): + first = ActiGraphClientV3("first", "synthetic-secret-a") + second = ActiGraphClientV3("second", "synthetic-secret-b") + second.get_study_info(1) + http["replies"].extend([ + {"error": "synthetic-expired-token"}, + {"totalCount": 1, "items": [{"id": 1}]}, + ]) + assert first.get_studies() == [{"id": 1}] + second.get_study_info(1) + assert http["auth"] == [ + ("second", "CentrePoint"), ("first", "CentrePoint"), + ("first", "CentrePoint"), + ] + assert http["get"][-1][1] == "Bearer synthetic-second-CentrePoint" + + +def test_existing_malformed_page_refresh_remains_bounded(http): + http["replies"].extend([{}, {}]) + client = ActiGraphClientV3("bounded", "synthetic-secret") + with pytest.raises(KeyError, match="totalCount"): + client.get_studies() + assert len(http["get"]) == len(http["auth"]) == 2 From 088ece9b627a0e47e5e415d12b5d55b5df2e7d7b Mon Sep 17 00:00:00 2001 From: Cody Mitchell Date: Sun, 13 Sep 2026 23:43:04 -0500 Subject: [PATCH 2/2] Add bounded V3 request recovery and client sessions --- .github/workflows/tests.yml | 6 + README.rst | 52 +++++ actiapi/_transport.py | 128 +++++++++++++ actiapi/v3.py | 122 ++++++++---- pyproject.toml | 1 + tests/test_v3_authentication.py | 75 +++++--- tests/test_v3_http.py | 121 ++++++++++++ tests/test_v3_recovery.py | 324 ++++++++++++++++++++++++++++++++ 8 files changed, 764 insertions(+), 65 deletions(-) create mode 100644 actiapi/_transport.py create mode 100644 tests/test_v3_http.py create mode 100644 tests/test_v3_recovery.py diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 104cc3b..56f93f2 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -7,6 +7,8 @@ on: pull_request: types: - opened + - synchronize + - reopened branches: - 'master' @@ -32,3 +34,7 @@ jobs: run: uv sync - name: Style run: uv run make lint + - name: Offline V3 tests + run: >- + uv run pytest tests/test_v3_authentication.py tests/test_v3_recovery.py + tests/test_v3_http.py tests/test_v3.py::test_validate_empty_response diff --git a/README.rst b/README.rst index 94631db..dd1b362 100644 --- a/README.rst +++ b/README.rst @@ -31,3 +31,55 @@ Raw data user=, study_id= ) +V3 request recovery +=================== + +Each V3 client owns its tokens, cookies, and connection pool. Use a context +manager to release its HTTP connections after use: + +.. code-block:: python + + from threading import Event + from actiapi.v3 import ActiGraphClientV3 + + cancelled = Event() + with ActiGraphClientV3( + api_access_key, + api_secret_key, + timeout=(3.05, 30.0), + max_retries=2, + max_retry_delay=30.0, + cancel_event=cancelled, + ) as client: + studies = client.get_studies() + +``timeout`` specifies connection and read inactivity limits in seconds, or a +single positive value for both. GET requests can retry connection failures, +timeouts, and HTTP 429, 500, 502, 503, and 504 responses. ``max_retries`` counts +additional attempts; zero selects one attempt. Backoff starts at 0.5 seconds +and doubles within ``max_retry_delay``. + +A valid ``Retry-After`` delay is honoured within that wait limit. A longer +server delay returns the HTTP failure immediately so the caller can schedule +a later attempt. Authentication POSTs use one attempt and surface redirects +as an explicit error. A data request receiving +HTTP 401 can refresh its scope token once and try again. + +Calling ``cancelled.set()`` stops a pending retry wait and prevents subsequent +attempts. Cancellation is observed after an in-flight socket operation finishes +or times out. Connection and read inactivity limits apply to that operation. + +Successful return shapes and existing 404 handling are preserved. HTTP failures +raise ``requests.HTTPError``. Invalid JSON, missing authentication tokens, and +malformed pages raise ``actiapi.v3.InvalidResponseError``. Cancellation raises +``actiapi.v3.RequestCancelled``. Transport failures retain their Requests +exception types. Each of these errors is a ``requests.RequestException``. + +The offline tests use synthetic responses and a loopback HTTP server: + +.. code-block:: bash + + uv run pytest tests/test_v3_authentication.py tests/test_v3_recovery.py \ + tests/test_v3_http.py tests/test_v3.py::test_validate_empty_response + +The other existing tests require access to the configured live study. diff --git a/actiapi/_transport.py b/actiapi/_transport.py new file mode 100644 index 0000000..762b121 --- /dev/null +++ b/actiapi/_transport.py @@ -0,0 +1,128 @@ +"""Bounded synchronous transport for the V3 example client.""" + +import math +import time +from datetime import timezone +from email.utils import parsedate_to_datetime +from threading import Event +from typing import Optional, Tuple, Union + +import requests + + +class RequestCancelled(requests.RequestException): + """The caller cancelled a request or its retry wait.""" + + +class InvalidResponseError(requests.RequestException): + """The server response does not satisfy the expected JSON contract.""" + + +class Transport: + """Own one session with finite timeouts and bounded GET retries.""" + + RETRY_STATUSES = {429, 500, 502, 503, 504} + + def __init__( + self, + timeout: Union[float, Tuple[float, float]], + max_retries: int, + max_retry_delay: float, + cancel_event: Optional[Event], + ): + """Validate policy before creating the client session.""" + values = timeout if isinstance(timeout, tuple) else (timeout,) + if len(values) not in (1, 2) or any( + isinstance(value, bool) + or not isinstance(value, (int, float)) + or not math.isfinite(value) + or value <= 0 + for value in values + ): + raise ValueError("timeout must contain finite positive values") + if isinstance(timeout, tuple) and len(timeout) != 2: + raise ValueError("timeout tuple must contain connect and read values") + if type(max_retries) is not int or max_retries < 0: + raise ValueError("max_retries must be a nonnegative integer") + if ( + isinstance(max_retry_delay, bool) + or not isinstance(max_retry_delay, (int, float)) + or not math.isfinite(max_retry_delay) + or max_retry_delay < 0 + ): + raise ValueError("max_retry_delay must be finite and nonnegative") + self.timeout = timeout + self.max_retries = max_retries + self.max_retry_delay = max_retry_delay + self.cancel_event = cancel_event + self.session = requests.Session() + + def close(self): + """Release this client's connection pool.""" + self.session.close() + + def _check_cancelled(self): + if self.cancel_event is not None and self.cancel_event.is_set(): + raise RequestCancelled("Request cancelled") + + def _wait(self, delay): + if self.cancel_event is None: + time.sleep(delay) + elif self.cancel_event.wait(delay): + raise RequestCancelled("Request cancelled during retry wait") + + def _retry_delay(self, response, fallback): + value = response.headers.get("Retry-After") + if value is None: + return fallback + try: + delay = int(value) + if delay < 0: + return fallback + except ValueError: + try: + date = parsedate_to_datetime(value) + if date.tzinfo is None: + date = date.replace(tzinfo=timezone.utc) + delay = max(0.0, date.timestamp() - time.time()) + except (ValueError, TypeError, OverflowError): + return fallback + if delay > self.max_retry_delay: + return None + return delay + + def request(self, method, url, **kwargs): + """Retry transient GET failures within the configured attempt budget.""" + retry_limit = self.max_retries if method.upper() == "GET" else 0 + attempt = 0 + backoff = 0.5 + while True: + self._check_cancelled() + delay = min(backoff, self.max_retry_delay) + try: + response = self.session.request( + method, url, timeout=self.timeout, **kwargs + ) + except requests.exceptions.SSLError: + raise + except (requests.ConnectionError, requests.Timeout): + if attempt >= retry_limit: + raise + else: + try: + self._check_cancelled() + except RequestCancelled: + response.close() + raise + if ( + response.status_code not in self.RETRY_STATUSES + or attempt >= retry_limit + ): + return response + delay = self._retry_delay(response, delay) + if delay is None: + return response + response.close() + self._wait(delay) + attempt += 1 + backoff = min(backoff * 2, self.max_retry_delay) diff --git a/actiapi/v3.py b/actiapi/v3.py index 99aab25..cb04fc8 100644 --- a/actiapi/v3.py +++ b/actiapi/v3.py @@ -4,13 +4,18 @@ """ import logging -from typing import Any, Dict, List, Literal, Optional, Union - -import requests +from threading import Event +from typing import Any, Dict, List, Literal, Optional, Tuple, Union from actiapi import ActiGraphClient +from actiapi._transport import InvalidResponseError, RequestCancelled, Transport -session = requests.Session() +__all__ = [ + "ActiGraphClientV3", + "InvalidResponseError", + "RequestCancelled", + "validate_response", +] logger = logging.getLogger(__name__) @@ -21,10 +26,32 @@ class ActiGraphClientV3(ActiGraphClient): BASE_URL = "https://api.actigraphcorp.com" AUTH_API = "https://auth.actigraphcorp.com/connect/token" - def __init__(self, api_access_key: str, api_secret_key: str): - """Initialize a client with an independent token cache.""" + def __init__( + self, + api_access_key: str, + api_secret_key: str, + *, + timeout: Union[float, Tuple[float, float]] = (3.05, 30.0), + max_retries: int = 2, + max_retry_delay: float = 30.0, + cancel_event: Optional[Event] = None, + ): + """Initialize independent token and transport state with bounded retries.""" super().__init__(api_access_key, api_secret_key) self._tokens: Dict[str, Optional[str]] = {} + self._transport = Transport(timeout, max_retries, max_retry_delay, cancel_event) + + def close(self): + """Release this client's HTTP connections.""" + self._transport.close() + + def __enter__(self): + """Enter the client context.""" + return self + + def __exit__(self, exc_type, exc_value, traceback): + """Release connections when leaving the client context.""" + self.close() @staticmethod def _generate_headers(token: str, raw: bool = False): @@ -45,15 +72,28 @@ def _get_access_token(self, scope: str): } headers = {"Content-Type": "application/x-www-form-urlencoded"} - response = requests.post( - endpoint, data=request_body, headers=headers, verify=True - ) - try: - return response.json()["access_token"] - except KeyError: - raise RuntimeError( - "No access token! Make sure you have API_ACCESS_KEY and API_SECRET_KEY!" - ) + with self._transport.request( + "POST", + endpoint, + data=request_body, + headers=headers, + verify=True, + allow_redirects=False, + ) as response: + if 300 <= response.status_code < 400: + raise InvalidResponseError( + "Authentication endpoint returned a redirect" + ) + reply = validate_response(response) + if ( + not isinstance(reply, dict) + or not isinstance(reply.get("access_token"), str) + or not reply["access_token"] + ): + raise InvalidResponseError( + "Authentication response has no access token" + ) + return reply["access_token"] def get_files( self, @@ -223,13 +263,19 @@ def get_sleep_summary( return results def _get_single(self, request: str, scope: str): - if self._tokens.get(scope) is None: - self._tokens[scope] = self._get_access_token(scope) - logger.info("Requesting %s", request) - headers = self._generate_headers(str(self._tokens[scope])) - response = session.get(self.BASE_URL + request, headers=headers, stream=False) - reply = validate_response(response) - return reply + for refresh in range(2): + if self._tokens.get(scope) is None: + self._tokens[scope] = self._get_access_token(scope) + logger.info("Requesting %s", request) + headers = self._generate_headers(str(self._tokens[scope])) + with self._transport.request( + "GET", self.BASE_URL + request, headers=headers, stream=False + ) as response: + if response.status_code == 401: + self._tokens.pop(scope, None) + if refresh == 0: + continue + return validate_response(response) def _get_paginated(self, request: str, scope: str): results = [] @@ -237,19 +283,19 @@ def _get_paginated(self, request: str, scope: str): limit = 100 while True: paginated_request = f"{request}offset={offset}&limit={limit}" - try: - reply = self._get_single(request=paginated_request, scope=scope) - if reply is None: - break - - total_count = reply["totalCount"] - except KeyError: - self._tokens.pop(scope, None) - reply = self._get_single(request=paginated_request, scope=scope) - if reply is None: - break - - total_count = reply["totalCount"] + reply = self._get_single(request=paginated_request, scope=scope) + if reply is None: + break + if ( + not isinstance(reply, dict) + or type(reply.get("totalCount")) is not int + or reply["totalCount"] < 0 + or not isinstance(reply.get("items"), list) + ): + raise InvalidResponseError("Page requires totalCount and items") + total_count = reply["totalCount"] + if not reply["items"] and offset < total_count: + raise InvalidResponseError("Empty page before totalCount was reached") for item in reply["items"]: results.append(item) @@ -268,5 +314,9 @@ def validate_response(response): logger.error("404 Not Found!") result = None else: - result = response.json() + response.raise_for_status() + try: + result = response.json() + except ValueError as error: + raise InvalidResponseError("API response is not valid JSON") from error return result diff --git a/pyproject.toml b/pyproject.toml index 2679388..2c96993 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,6 +7,7 @@ readme = "README.rst" license = "GPL-3.0-only" maintainers = [{ name = "ActiGraph Data Science Team", email = "science@theactigraph.com" }] dynamic = ["version"] +dependencies = ["requests>=2.32.4,<3"] [project.urls] Repository = "https://github.com/actigraph/actiapi" diff --git a/tests/test_v3_authentication.py b/tests/test_v3_authentication.py index 82342b1..c05304f 100644 --- a/tests/test_v3_authentication.py +++ b/tests/test_v3_authentication.py @@ -5,12 +5,12 @@ import pytest import requests -from actiapi.v3 import ActiGraphClientV3 +from actiapi.v3 import ActiGraphClientV3, InvalidResponseError @pytest.fixture def http(monkeypatch): - calls = {"auth": [], "get": [], "replies": deque()} + calls = {"auth": [], "get": [], "replies": deque(), "statuses": deque()} def request(session, method, url, **kwargs): response = requests.Response() @@ -20,16 +20,22 @@ def request(session, method, url, **kwargs): calls["auth"].append((data["client_id"], data["scope"])) payload = {"access_token": f"synthetic-{data['client_id']}-{data['scope']}"} else: + if calls["statuses"]: + response.status_code = calls["statuses"].popleft() calls["get"].append((url, kwargs["headers"]["Authorization"])) if calls["replies"]: payload = calls["replies"].popleft() elif "offset=" in url: - payload = {"totalCount": 1, "items": [ - {"id": 1, "downloadUrl": "https://example.invalid/synthetic"} - ]} + payload = { + "totalCount": 1, + "items": [ + {"id": 1, "downloadUrl": "https://example.invalid/synthetic"} + ], + } else: payload = {"id": 1} response._content = json.dumps(payload).encode() + response._content_consumed = True return response def forbidden_send(*args, **kwargs): @@ -60,20 +66,25 @@ def test_client_reuses_each_scope_independently(http): for scope in ("CentrePoint", "Analytics", "DataAccess"): client._get_single("/synthetic", scope) assert http["auth"] == [ - ("first", "CentrePoint"), ("first", "Analytics"), ("first", "DataAccess") + ("first", "CentrePoint"), + ("first", "Analytics"), + ("first", "DataAccess"), ] -@pytest.mark.parametrize("method,args,scope", [ - ("get_study_info", (1,), "CentrePoint"), - ("get_studies", (), "CentrePoint"), - ("get_study_metadata", (1,), "CentrePoint"), - ("get_event_markers", (2, 1), "Analytics"), - ("get_minute_summary", (2, 1), "Analytics"), - ("get_daily_summary", (2, 1), "Analytics"), - ("get_sleep_summary", (2, 1), "Analytics"), - ("get_files", (2, 1), "DataAccess"), -]) +@pytest.mark.parametrize( + "method,args,scope", + [ + ("get_study_info", (1,), "CentrePoint"), + ("get_studies", (), "CentrePoint"), + ("get_study_metadata", (1,), "CentrePoint"), + ("get_event_markers", (2, 1), "Analytics"), + ("get_minute_summary", (2, 1), "Analytics"), + ("get_daily_summary", (2, 1), "Analytics"), + ("get_sleep_summary", (2, 1), "Analytics"), + ("get_files", (2, 1), "DataAccess"), + ], +) def test_public_v3_methods_keep_their_scope(http, method, args, scope): client = ActiGraphClientV3("wrapper", "synthetic-secret") result = getattr(client, method)(*args) @@ -83,10 +94,12 @@ def test_public_v3_methods_keep_their_scope(http, method, args, scope): def test_pagination_reuses_token_and_preserves_page_order(http): - http["replies"].extend([ - {"totalCount": 101, "items": [{"id": i} for i in range(100)]}, - {"totalCount": 101, "items": [{"id": 100}]}, - ]) + http["replies"].extend( + [ + {"totalCount": 101, "items": [{"id": i} for i in range(100)]}, + {"totalCount": 101, "items": [{"id": 100}]}, + ] + ) client = ActiGraphClientV3("pages", "synthetic-secret") assert client.get_studies() == [{"id": i} for i in range(101)] assert http["auth"] == [("pages", "CentrePoint")] @@ -98,22 +111,26 @@ def test_pagination_refresh_does_not_invalidate_another_client(http): first = ActiGraphClientV3("first", "synthetic-secret-a") second = ActiGraphClientV3("second", "synthetic-secret-b") second.get_study_info(1) - http["replies"].extend([ - {"error": "synthetic-expired-token"}, - {"totalCount": 1, "items": [{"id": 1}]}, - ]) + http["statuses"].extend([401, 200]) + http["replies"].extend( + [ + {"error": "synthetic-expired-token"}, + {"totalCount": 1, "items": [{"id": 1}]}, + ] + ) assert first.get_studies() == [{"id": 1}] second.get_study_info(1) assert http["auth"] == [ - ("second", "CentrePoint"), ("first", "CentrePoint"), + ("second", "CentrePoint"), + ("first", "CentrePoint"), ("first", "CentrePoint"), ] assert http["get"][-1][1] == "Bearer synthetic-second-CentrePoint" -def test_existing_malformed_page_refresh_remains_bounded(http): - http["replies"].extend([{}, {}]) +def test_malformed_page_does_not_refresh_authentication(http): + http["replies"].append({}) client = ActiGraphClientV3("bounded", "synthetic-secret") - with pytest.raises(KeyError, match="totalCount"): + with pytest.raises(InvalidResponseError, match="totalCount"): client.get_studies() - assert len(http["get"]) == len(http["auth"]) == 2 + assert len(http["get"]) == len(http["auth"]) == 1 diff --git a/tests/test_v3_http.py b/tests/test_v3_http.py new file mode 100644 index 0000000..bad2fd4 --- /dev/null +++ b/tests/test_v3_http.py @@ -0,0 +1,121 @@ +import json +import time +from collections import deque +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from threading import Thread +from urllib.parse import parse_qs, urlsplit + +import pytest +import requests + +from actiapi.v3 import ActiGraphClientV3, InvalidResponseError + + +@pytest.fixture +def server(monkeypatch): + state = {"gets": [], "auth": [], "statuses": deque()} + + class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, *args): + pass + + def reply(self, status, payload, headers=None): + body = json.dumps(payload).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + for name, value in (headers or {}).items(): + self.send_header(name, value) + self.end_headers() + try: + self.wfile.write(body) + except (BrokenPipeError, ConnectionResetError): + pass + + def do_POST(self): + assert self.path == "/token" + data = parse_qs( + self.rfile.read(int(self.headers["Content-Length"])).decode() + ) + key = data["client_id"][0] + assert key in ("first", "second") + assert data["client_secret"] == ["synthetic-secret"] + state["auth"].append(key) + if state.get("redirect_auth"): + self.reply(307, {}, {"Location": "/other"}) + return + self.reply(200, {"access_token": f"{key}:{state['auth'].count(key)}"}) + + def do_GET(self): + state["gets"].append(dict(self.headers)) + if self.path == "/slow": + time.sleep(0.15) + status = state["statuses"].popleft() if state["statuses"] else 200 + key = self.headers["Authorization"].split(" ", 1)[1].split(":", 1)[0] + headers = {"Retry-After": "0"} if status in (429, 503) else {} + if status == 200: + headers["Set-Cookie"] = f"synthetic={key}; Path=/" + self.reply(status, {"id": 1}, headers) + + httpd = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = Thread(target=httpd.serve_forever, kwargs={"poll_interval": 0.01}) + thread.start() + base = f"http://127.0.0.1:{httpd.server_port}" + monkeypatch.setattr(ActiGraphClientV3, "BASE_URL", base) + monkeypatch.setattr(ActiGraphClientV3, "AUTH_API", base + "/token") + monkeypatch.setenv("NO_PROXY", "127.0.0.1") + monkeypatch.setenv("no_proxy", "127.0.0.1") + original_send = requests.Session.send + + def loopback_only(session, request, **kwargs): + assert urlsplit(request.url).hostname == "127.0.0.1" + return original_send(session, request, **kwargs) + + monkeypatch.setattr(requests.Session, "send", loopback_only) + try: + yield state + finally: + httpd.shutdown() + httpd.server_close() + thread.join(timeout=2) + assert not thread.is_alive() + + +def test_real_http_recovery_refresh_and_client_cookie_isolation(server): + server["statuses"].extend([503, 401, 429, 200]) + with ActiGraphClientV3("first", "synthetic-secret") as first: + assert first.get_study_info(1) == {"id": 1} + with ActiGraphClientV3("second", "synthetic-secret") as second: + assert second.get_study_info(1) == {"id": 1} + assert first.get_study_info(1) == {"id": 1} + assert server["auth"] == ["first", "first", "second"] + assert len(server["gets"]) == 6 + assert [item["Authorization"] for item in server["gets"][:4]] == [ + "Bearer first:1", + "Bearer first:1", + "Bearer first:2", + "Bearer first:2", + ] + assert server["gets"][4]["Authorization"] == "Bearer second:1" + assert "Cookie" not in server["gets"][4] + assert server["gets"][5]["Cookie"] == "synthetic=first" + + +def test_real_socket_read_timeout(server): + with ActiGraphClientV3( + "first", "synthetic-secret", timeout=(1, 0.03), max_retries=0 + ) as client: + with pytest.raises(requests.ReadTimeout): + client._get_single("/slow", "CentrePoint") + assert len(server["gets"]) == 1 + + +def test_real_authentication_redirect_is_not_followed(server): + server["redirect_auth"] = True + with ActiGraphClientV3("first", "synthetic-secret") as client: + with pytest.raises(InvalidResponseError, match="redirect"): + client.get_study_info(1) + assert server["auth"] == ["first"] + assert not server["gets"] diff --git a/tests/test_v3_recovery.py b/tests/test_v3_recovery.py new file mode 100644 index 0000000..774b5a8 --- /dev/null +++ b/tests/test_v3_recovery.py @@ -0,0 +1,324 @@ +import json +from collections import deque +from email.utils import formatdate +from threading import Event, Timer +from unittest.mock import Mock + +import pytest +import requests + +from actiapi import _transport +from actiapi.v3 import ActiGraphClientV3, InvalidResponseError, RequestCancelled + + +@pytest.fixture +def http(monkeypatch): + state = { + "calls": [], + "get": deque(), + "auth": deque(), + "responses": [], + "waits": [], + "sessions": [], + } + + def request(session, method, url, **kwargs): + method = method.upper() + state["calls"].append((method, url, kwargs)) + state["sessions"].append(session) + queue = state["get"] if method == "GET" else state["auth"] + auth_count = sum(call[0] == "POST" for call in state["calls"]) + item = ( + queue.popleft() + if queue + else (200, {"access_token": f"synthetic-{auth_count}"}, {}) + ) + if isinstance(item, BaseException): + raise item + status, payload, headers = item + response = requests.Response() + response.status_code = status + response.url = url + response.headers.update(headers) + response._content = ( + payload if isinstance(payload, bytes) else json.dumps(payload).encode() + ) + response._content_consumed = True + state["responses"].append(response) + response.close = Mock(wraps=response.close) + return response + + def forbidden_send(*args, **kwargs): + pytest.fail("A test attempted a real HTTP request") + + monkeypatch.setattr(requests.Session, "request", request) + monkeypatch.setattr(requests.Session, "send", forbidden_send) + monkeypatch.setattr(_transport.time, "sleep", state["waits"].append) + return state + + +def test_default_requests_have_finite_timeouts(http): + http["get"].append((200, {"id": 1}, {})) + assert ActiGraphClientV3("synthetic", "synthetic-secret").get_study_info(1) == { + "id": 1 + } + assert all(call[2].get("timeout") == (3.05, 30.0) for call in http["calls"]) + + +def test_transient_get_recovers(http): + http["get"].extend( + [(503, b"temporarily unavailable", {"Retry-After": "0"}), (200, {"id": 1}, {})] + ) + assert ActiGraphClientV3("synthetic", "synthetic-secret").get_study_info(1) == { + "id": 1 + } + assert [call[0] for call in http["calls"]] == ["POST", "GET", "GET"] + + +def test_unauthorized_single_request_refreshes_once(http): + http["get"].extend([(401, {"error": "expired"}, {}), (200, {"id": 1}, {})]) + assert ActiGraphClientV3("synthetic", "synthetic-secret").get_study_info(1) == { + "id": 1 + } + assert [call[0] for call in http["calls"]] == ["POST", "GET", "POST", "GET"] + assert [ + call[2]["headers"]["Authorization"] + for call in http["calls"] + if call[0] == "GET" + ] == ["Bearer synthetic-1", "Bearer synthetic-2"] + + +def test_http_error_precedes_json_decoding(http): + http["get"].append((403, b"forbidden", {})) + with pytest.raises(requests.HTTPError): + ActiGraphClientV3("synthetic", "synthetic-secret").get_study_info(1) + + +@pytest.mark.parametrize("status", [429, 500, 502, 503, 504]) +def test_transient_status_retries_are_bounded(http, status): + http["get"].extend([(status, b"unavailable", {})] * 3) + with pytest.raises(requests.HTTPError): + ActiGraphClientV3("synthetic", "synthetic-secret").get_study_info(1) + assert [call[0] for call in http["calls"]] == ["POST", "GET", "GET", "GET"] + assert http["waits"] == [0.5, 1.0] + assert all(response.close.call_count == 1 for response in http["responses"]) + + +@pytest.mark.parametrize( + "error", [requests.ConnectTimeout, requests.ReadTimeout, requests.ConnectionError] +) +def test_transient_network_failure_recovers(http, error): + http["get"].extend([error("synthetic failure"), (200, {"id": 1}, {})]) + with ActiGraphClientV3("synthetic", "synthetic-secret", timeout=(1, 2)) as client: + assert client.get_study_info(1) == {"id": 1} + assert len(http["calls"]) == 3 + assert all(call[2]["timeout"] == (1, 2) for call in http["calls"]) + + +def test_network_retry_exhaustion_preserves_timeout(http): + http["get"].extend([requests.ReadTimeout("synthetic")] * 3) + with pytest.raises(requests.ReadTimeout): + ActiGraphClientV3("synthetic", "synthetic-secret").get_study_info(1) + assert len(http["calls"]) == 4 + + +def test_certificate_failure_is_not_retried(http): + http["get"].append(requests.exceptions.SSLError("synthetic")) + with pytest.raises(requests.exceptions.SSLError): + ActiGraphClientV3("synthetic", "synthetic-secret").get_study_info(1) + assert len(http["calls"]) == 2 + + +@pytest.mark.parametrize("status", [400, 403, 422]) +def test_permanent_http_failure_is_not_retried(http, status): + http["get"].append((status, b"failure", {})) + with pytest.raises(requests.HTTPError): + ActiGraphClientV3("synthetic", "synthetic-secret").get_study_info(1) + assert len(http["calls"]) == 2 and not http["waits"] + + +def test_retry_after_wait_is_honoured(http): + http["get"].extend([(429, {}, {"Retry-After": "7"}), (200, {"id": 1}, {})]) + assert ActiGraphClientV3("synthetic", "synthetic-secret").get_study_info(1) == { + "id": 1 + } + assert http["waits"] == [7] + + +def test_retry_after_http_date_is_honoured(http, monkeypatch): + monkeypatch.setattr(_transport.time, "time", lambda: 1700000000) + date = formatdate(1700000005, usegmt=True) + http["get"].extend([(503, {}, {"Retry-After": date}), (200, {"id": 1}, {})]) + assert ActiGraphClientV3("synthetic", "synthetic-secret").get_study_info(1) == { + "id": 1 + } + assert http["waits"] == [5] + + +def test_retry_after_above_budget_returns_failure_without_early_retry(http): + http["get"].append((429, {}, {"Retry-After": "31"})) + with pytest.raises(requests.HTTPError): + ActiGraphClientV3("synthetic", "synthetic-secret").get_study_info(1) + assert len(http["calls"]) == 2 and not http["waits"] + + +@pytest.mark.parametrize("value", ["bad header", "-1", "NaN"]) +def test_invalid_retry_after_uses_bounded_backoff(http, value): + http["get"].extend([(503, {}, {"Retry-After": value}), (200, {"id": 1}, {})]) + client = ActiGraphClientV3("synthetic", "synthetic-secret", max_retry_delay=0.25) + assert client.get_study_info(1) == {"id": 1} + assert http["waits"] == [0.25] + + +def test_zero_retries_disables_transient_replay(http): + http["get"].append((503, {}, {})) + with pytest.raises(requests.HTTPError): + ActiGraphClientV3( + "synthetic", "synthetic-secret", max_retries=0 + ).get_study_info(1) + assert len(http["calls"]) == 2 + + +@pytest.mark.parametrize("item", [(503, {}, {}), requests.ConnectTimeout("synthetic")]) +def test_authentication_post_is_not_replayed(http, item): + http["auth"].append(item) + with pytest.raises(requests.RequestException): + ActiGraphClientV3("synthetic", "synthetic-secret").get_study_info(1) + assert len(http["calls"]) == 1 + + +def test_repeated_401_stops_after_one_refresh(http): + http["get"].extend([(401, {}, {})] * 2) + client = ActiGraphClientV3("synthetic", "synthetic-secret") + with pytest.raises(requests.HTTPError): + client.get_study_info(1) + assert [call[0] for call in http["calls"]] == ["POST", "GET", "POST", "GET"] + assert not client._tokens + + +@pytest.mark.parametrize("status", [307, 308]) +def test_authentication_redirect_cannot_replay_credentials(http, status): + http["auth"].append((status, {}, {"Location": "https://example.invalid/other"})) + with pytest.raises(InvalidResponseError, match="redirect"): + ActiGraphClientV3("synthetic", "synthetic-secret").get_study_info(1) + assert len(http["calls"]) == 1 + assert http["calls"][0][2]["allow_redirects"] is False + + +@pytest.mark.parametrize( + "payload", + [{}, [], {"access_token": ""}, {"access_token": None}, {"access_token": 2}], +) +def test_invalid_authentication_response_is_explicit(http, payload): + http["auth"].append((200, payload, {})) + with pytest.raises(InvalidResponseError, match="access token"): + ActiGraphClientV3("synthetic", "synthetic-secret").get_study_info(1) + assert len(http["calls"]) == 1 + + +def test_successful_non_json_response_is_explicit(http): + http["get"].append((200, b"invalid JSON", {})) + with pytest.raises(InvalidResponseError, match="valid JSON"): + ActiGraphClientV3("synthetic", "synthetic-secret").get_study_info(1) + assert len(http["calls"]) == 2 + + +@pytest.mark.parametrize( + "payload", + [ + {}, + [], + {"totalCount": True, "items": []}, + {"totalCount": -1, "items": []}, + {"totalCount": 1, "items": {}}, + {"totalCount": 1, "items": []}, + ], +) +def test_malformed_pages_do_not_trigger_reauthentication(http, payload): + http["get"].append((200, payload, {})) + with pytest.raises(InvalidResponseError): + ActiGraphClientV3("synthetic", "synthetic-secret").get_studies() + assert len(http["calls"]) == 2 + + +@pytest.mark.parametrize( + "method,expected", [("get_study_info", None), ("get_studies", [])] +) +def test_404_behavior_is_preserved(http, method, expected): + http["get"].append((404, b"missing", {})) + client = ActiGraphClientV3("synthetic", "synthetic-secret") + result = ( + client.get_study_info(1) if method == "get_study_info" else client.get_studies() + ) + assert result == expected and len(http["calls"]) == 2 + + +def test_empty_page_is_preserved(http): + http["get"].append((200, {"totalCount": 0, "items": []}, {})) + assert ActiGraphClientV3("synthetic", "synthetic-secret").get_studies() == [] + + +def test_cancelled_call_sends_no_request(http): + event = Event() + event.set() + with pytest.raises(RequestCancelled): + ActiGraphClientV3( + "synthetic", "synthetic-secret", cancel_event=event + ).get_study_info(1) + assert not http["calls"] + + +def test_cancellation_interrupts_retry_wait(http): + event = Event() + http["get"].append((503, {}, {"Retry-After": "30"})) + timer = Timer(0.02, event.set) + timer.start() + try: + with pytest.raises(RequestCancelled): + ActiGraphClientV3( + "synthetic", "synthetic-secret", cancel_event=event + ).get_study_info(1) + finally: + timer.cancel() + timer.join(timeout=1) + assert len(http["calls"]) == 2 + + +def test_keyboard_interrupt_is_preserved(http): + http["get"].append(KeyboardInterrupt()) + with pytest.raises(KeyboardInterrupt): + ActiGraphClientV3("synthetic", "synthetic-secret").get_study_info(1) + assert len(http["calls"]) == 2 + + +def test_sessions_are_client_local_and_reused(http): + http["get"].extend([(200, {"id": 1}, {})] * 3) + with ActiGraphClientV3("first", "synthetic-secret") as first: + first.get_study_info(1) + first.get_study_info(1) + with ActiGraphClientV3("second", "synthetic-secret") as second: + second.get_study_info(1) + assert http["sessions"][0] is http["sessions"][1] is http["sessions"][2] + assert http["sessions"][3] is http["sessions"][4] + assert http["sessions"][0] is not http["sessions"][3] + + +@pytest.mark.parametrize( + "options", + [ + {"timeout": 0}, + {"timeout": float("inf")}, + {"timeout": (1, -1)}, + {"timeout": (1,)}, + {"timeout": None}, + {"max_retries": -1}, + {"max_retries": True}, + {"max_retries": 1.5}, + {"max_retry_delay": -1}, + {"max_retry_delay": float("nan")}, + ], +) +def test_invalid_policy_is_rejected_before_requests(http, options): + with pytest.raises(ValueError): + ActiGraphClientV3("synthetic", "synthetic-secret", **options) + assert not http["calls"]