ci: test on free-threaded Python and stress a shared RESTClient

.github/workflows/test.yml

@@ -14,7 +14,7 @@ jobs:
       fail-fast: false
       matrix:
         os: [ubuntu-latest, windows-latest]
-        python-version: ['3.10', '3.11']
+        python-version: ['3.10', '3.11', '3.12', '3.13', '3.14']
     runs-on: ${{ matrix.os }}
     name: ${{ matrix.os }} Unit test ${{ matrix.python-version }}
     steps:
@@ -32,3 +32,41 @@ jobs:
         run: poetry install
       - name: Unit tests
         run: make test
+  test-free-threaded:
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ['3.13t', '3.14t']
+    runs-on: ubuntu-latest
+    name: ubuntu-latest Unit test ${{ matrix.python-version }} (PYTHON_GIL=0)
+    steps:
+      - uses: actions/checkout@v3
+      - name: Setup Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Setup Poetry
+        # pipx is Poetry's recommended installation method, and it keeps Poetry
+        # on the runner's regular Python, independent of the interpreter under test
+        run: pipx install poetry
+      - name: Install pypi deps
+        # the dev group (linters, docs, orjson for the examples) is not needed
+        # to run the tests, and orjson does not support free-threaded Python
+        run: |
+          poetry env use python${{ matrix.python-version }}
+          poetry install --only main,test
+      - name: Unit tests
+        env:
+          PYTHON_GIL: '0'
+        run: |
+          poetry run python - <<'EOF'
+          import sys, sysconfig
+          state = (
+              f"version={sys.version!r} "
+              f"Py_GIL_DISABLED={sysconfig.get_config_var('Py_GIL_DISABLED')!r} "
+              f"_is_gil_enabled={sys._is_gil_enabled()!r}"
+          )
+          print(state)
+          assert not sys._is_gil_enabled(), f"the GIL is enabled: {state}"
+          EOF
+          make test

pyproject.toml

@@ -29,7 +29,10 @@ urllib3 = ">=1.26.9"
 websockets = ">=14.0"
 certifi = ">=2022.5.18,<2027.0.0"
 
-[tool.poetry.dev-dependencies]
+[tool.poetry.group.test.dependencies]
+pook = "^2.1.6"
+
+[tool.poetry.group.dev.dependencies]
 black = "^26.3.1"
 mypy = "^2.4"
 types-urllib3 = "^1.26.25"
@@ -39,7 +42,6 @@ sphinx-rtd-theme = "^3.1.0"
 sphinx-autodoc-typehints = "^2.3.0"
 types-certifi = "^2021.10.8"
 types-setuptools = "^84.0.0"
-pook = "^2.1.6"
 orjson = "^3.11.6"
 
 [build-system]

test_rest/base.py

@@ -3,6 +3,7 @@
 import os
 import re
 import unittest
+from typing import Set
 import pook  # type: ignore
 
 # mocks are stored in file tree
@@ -32,6 +33,12 @@
 
 
 class BaseTest(unittest.TestCase):
+    # Paths (as listed in `mocks`) whose mock answers any number of requests,
+    # for tests that repeat a request. Every other mock answers once: the
+    # pagination tests rely on that, because a first-page mock also matches
+    # the request for the next page.
+    persistent_mocks: Set[str] = set()
+
     @classmethod
     def setUpClass(cls):
         cls.maxDiff = None
@@ -40,4 +47,15 @@ def setUpClass(cls):
         for m in mocks:
             url = cls.c.BASE + m[0]
             # print("register", url)
-            pook.get(url, reply=200, response_body=m[1])
+            pook.get(
+                url,
+                reply=200,
+                response_body=m[1],
+                persist=m[0] in cls.persistent_mocks,
+            )
+
+    @classmethod
+    def tearDownClass(cls):
+        # stop intercepting urllib3 and drop the mocks, so they don't leak into
+        # later test classes
+        pook.off()

test_rest/test_concurrency.py

@@ -0,0 +1,145 @@
+import json
+import threading
+import unittest
+from concurrent.futures import ThreadPoolExecutor
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from typing import Any, Dict, List, Optional, Tuple
+from urllib.parse import parse_qsl, urlparse
+
+from massive import RESTClient
+from massive.rest.models.request import (
+    RequestOptionBuilder,
+    X_MASSIVE_EDGE_ID,
+    X_MASSIVE_EDGE_IP_ADDRESS,
+)
+
+
+class EchoServer(ThreadingHTTPServer):
+    """Answers every GET with what it received.
+
+    It counts the requests and records the most requests it had in flight at
+    once, which shows whether the client's threads really overlapped.
+    """
+
+    daemon_threads = True
+
+    def __init__(self):
+        super().__init__(("127.0.0.1", 0), EchoHandler)
+        self.lock = threading.Lock()
+        self.count = 0
+        self.in_flight = 0
+        self.peak_in_flight = 0
+
+
+class EchoHandler(BaseHTTPRequestHandler):
+    # keep-alive, so the client reuses pooled connections across threads
+    protocol_version = "HTTP/1.1"
+    server: EchoServer
+
+    def do_GET(self):
+        with self.server.lock:
+            self.server.count += 1
+            self.server.in_flight += 1
+            self.server.peak_in_flight = max(
+                self.server.peak_in_flight, self.server.in_flight
+            )
+        try:
+            url = urlparse(self.path)
+            body = json.dumps(
+                {
+                    "path": url.path,
+                    "query": dict(parse_qsl(url.query)),
+                    "edge_id": self.headers.get(X_MASSIVE_EDGE_ID),
+                    "edge_ip": self.headers.get(X_MASSIVE_EDGE_IP_ADDRESS),
+                    "authorization": self.headers.get("Authorization"),
+                }
+            ).encode()
+            self.send_response(200)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+        finally:
+            with self.server.lock:
+                self.server.in_flight -= 1
+
+    def log_message(self, format, *args):
+        pass
+
+
+class ConcurrencyTest(unittest.TestCase):
+    """One RESTClient shared by many threads against a local HTTP server."""
+
+    threads = 16
+    calls = 200
+
+    def setUp(self):
+        self.httpd = EchoServer()
+        thread = threading.Thread(target=self.httpd.serve_forever, daemon=True)
+        thread.start()
+        self.addCleanup(thread.join)
+        self.addCleanup(self.httpd.server_close)
+        self.addCleanup(self.httpd.shutdown)
+        host, port = self.httpd.server_address[:2]
+        self.base = f"http://{host!s}:{port}"
+
+    def test_shared_client(self):
+        c = RESTClient("key", base=self.base, retries=0, maxsize=self.threads)
+        headers_before = list(c.headers.items())
+        barrier = threading.Barrier(self.threads, timeout=10)
+
+        def run(t: int) -> List[Tuple[Dict[str, Any], Dict[str, Any]]]:
+            # start every thread together, so their requests overlap
+            barrier.wait()
+            results = []
+            for i in range(self.calls):
+                ticker = f"T{t}N{i}"
+                date = f"2024-01-{t + 1:02}"
+                edge_id: Optional[str] = None
+                edge_ip: Optional[str] = None
+                if i % 2:
+                    # Odd calls send per-request headers. They go through _get,
+                    # the method every endpoint calls, because the endpoint
+                    # methods also copy `options` into the query string.
+                    edge_id, edge_ip = f"edge-{t}-{i}", f"10.0.{t}.{i % 256}"
+                    resp = c._get(
+                        path=f"/v3/reference/tickers/{ticker}",
+                        params={"ticker": ticker, "date": date},
+                        raw=True,
+                        options=RequestOptionBuilder(
+                            edge_id=edge_id, edge_ip_address=edge_ip
+                        ),
+                    )
+                else:
+                    # raw=True returns the response, so the echoed request can
+                    # be checked instead of a deserialized TickerDetails
+                    resp = c.get_ticker_details(ticker, date=date, raw=True)
+                expected = {
+                    "path": f"/v3/reference/tickers/{ticker}",
+                    "query": {"ticker": ticker, "date": date},
+                    "edge_id": edge_id,
+                    "edge_ip": edge_ip,
+                    "authorization": "Bearer key",
+                }
+                results.append((expected, json.loads(resp.data)))
+            return results
+
+        with ThreadPoolExecutor(self.threads) as pool:
+            futures = [pool.submit(run, t) for t in range(self.threads)]
+            results = [r for f in futures for r in f.result()]
+
+        # every response answers its own request: the path and query name the
+        # thread and call, and the edge headers are the ones that call sent
+        self.assertEqual(len(results), self.threads * self.calls)
+        for expected, got in results:
+            self.assertEqual(got, expected)
+        self.assertEqual(self.httpd.count, self.threads * self.calls)
+        # the threads really shared the client at the same time
+        self.assertGreater(self.httpd.peak_in_flight, 1)
+        # per-request headers must not leak into the client's default headers:
+        # same keys, same values, same order
+        self.assertEqual(list(c.headers.items()), headers_before)
+
+
+if __name__ == "__main__":
+    unittest.main()

test_rest/test_dividends.py

@@ -3,6 +3,9 @@
 
 
 class DividendsTest(BaseTest):
+    # test_list_stocks_dividends_ticker_any_of requests this twice
+    persistent_mocks = {"/stocks/v1/dividends?ticker.any_of=AAPL%2CVZ"}
+
     def test_list_dividends(self):
         dividends = [d for d in self.c.list_dividends()]
         expected = [

test_rest/test_snapshots.py

@@ -26,6 +26,9 @@
 
 
 class SnapshotsTest(BaseTest):
+    # test_get_snapshot_indices requests this twice
+    persistent_mocks = {"/v3/snapshot/indices?ticker.any_of=SPX%2CAPx%2CAPy"}
+
     def test_list_universal_snapshots(self):
         expected = [
             UniversalSnapshot(