test: align test suite with the JavaScript SDK baseline

Bring the Python test suite in line with the scope and strategy used by
seamapi/javascript-http: tests are organized by SDK concern, exercise the
SDK against fake-seam-connect, and assert against seeded records rather
than hand-written stubs.

Fixture:

- Start the fake directly from node_modules/.bin instead of `npm run
  start`, so teardown signals the server rather than an npm wrapper.
- Poll /health until the server is ready and fail loudly if it exits
  early, replacing the connect-retry on /_fake/default_seed that made
  startup flaky.
- Scope PORT to the subprocess instead of mutating os.environ.
- Add a recording_server fixture for the two things the fake cannot do:
  asserting what the SDK puts on the wire, and driving retry responses.

Scope:

- Drop test/workspaces, which covered generated route methods rather
  than SDK behavior.
- Move deep_attr_dict_test.py under test/ so all tests live together.

Coverage:

- Add serialization, retry, and client tests, and cover the multi
  workspace client against the fake.
- Replace the mocked niquests.Session in the headers test with
  assertions on the request the server actually received.
- Assert against seed ids instead of `len(devices) > 0`.
- Cover paginator construction, cursor validation, and last page.
- Replace the personal access token xfail with passing tests. The fake
  rejects that token on /devices/list but authorizes it on
  /devices/get, which is the route the JavaScript SDK tests use.

Two retry tests are marked xfail(strict=True): SeamHttpClient sets
self.retries after calling niquests.Session.__init__ without forwarding
it, so the mounted HTTPAdapter keeps its default and the retries option
is silently ignored.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011DzapiU8A9NMdyoTybL9xB

Author: claude

test/api_key_test.py

@@ -1,39 +1,42 @@
 import pytest
+
 from seam import Seam
 from seam.auth import SeamInvalidTokenError
 
 
-def test_seam_client_from_api_key_returns_instance_authorized_with_api_key(
-    server,
-):
+def test_seam_from_api_key_returns_instance_authorized_with_api_key(server):
     endpoint, seed = server
     seam = Seam.from_api_key(seed["seam_apikey1_token"], endpoint=endpoint)
-    devices = seam.devices.list()
 
-    assert len(devices) > 0
+    device = seam.devices.get(device_id=seed["august_device_1"])
+
+    assert device.workspace_id == seed["seed_workspace_1"]
+    assert device.device_id == seed["august_device_1"]
 
 
-def test_seam_client_constructor_returns_instance_authorized_with_api_key(
-    server,
-):
+def test_seam_constructor_returns_instance_authorized_with_api_key(server):
     endpoint, seed = server
     seam = Seam(api_key=seed["seam_apikey1_token"], endpoint=endpoint)
-    devices = seam.devices.list()
 
-    assert len(devices) > 0
+    device = seam.devices.get(device_id=seed["august_device_1"])
+
+    assert device.workspace_id == seed["seed_workspace_1"]
+    assert device.device_id == seed["august_device_1"]
 
 
-def test_seam_client_constructor_interprets_single_string_argument_as_api_key(server):
-    _, seed = server
-    seam = Seam(seed["seam_apikey1_token"])
+def test_seam_constructor_interprets_single_string_argument_as_api_key(server):
+    endpoint, seed = server
+    seam = Seam(seed["seam_apikey1_token"], endpoint=endpoint)
+
+    device = seam.devices.get(device_id=seed["august_device_1"])
 
-    assert seam is not None
+    assert device.device_id == seed["august_device_1"]
 
     with pytest.raises(SeamInvalidTokenError, match=r"api_key"):
         Seam("some-invalid-key-format")
 
 
-def test_seam_client_checks_api_key_format():
+def test_seam_checks_api_key_format():
     with pytest.raises(SeamInvalidTokenError, match=r"Unknown"):
         Seam.from_api_key("some-invalid-key-format")
 
@@ -45,3 +48,6 @@ def test_seam_client_checks_api_key_format():
 
     with pytest.raises(SeamInvalidTokenError, match=r"Access Token"):
         Seam.from_api_key("seam_at")
+
+    with pytest.raises(SeamInvalidTokenError, match=r"Publishable Key"):
+        Seam.from_api_key("seam_pk_token")

test/client_test.py

@@ -0,0 +1,43 @@
+from seam import Seam
+
+
+def test_seam_exposes_a_client_that_can_make_requests(seam: Seam, server):
+    _, seed = server
+
+    response = seam.client.post(
+        "/devices/get", json={"device_id": seed["august_device_1"]}
+    )
+
+    assert response["device"]["workspace_id"] == seed["seed_workspace_1"]
+    assert response["device"]["device_id"] == seed["august_device_1"]
+
+
+def test_seam_client_resolves_paths_against_the_endpoint(seam: Seam, server):
+    endpoint, _ = server
+
+    assert seam.client.base_url == endpoint
+
+
+def test_seam_client_sets_auth_headers(server):
+    endpoint, seed = server
+    seam = Seam.from_api_key(seed["seam_apikey1_token"], endpoint=endpoint)
+
+    assert (
+        seam.client.headers["authorization"] == f"Bearer {seed['seam_apikey1_token']}"
+    )
+
+
+def test_seam_defaults_to_waiting_for_action_attempts(server):
+    endpoint, seed = server
+    seam = Seam.from_api_key(seed["seam_apikey1_token"], endpoint=endpoint)
+
+    assert seam.defaults["wait_for_action_attempt"] is True
+
+
+def test_seam_wait_for_action_attempt_default_can_be_overridden(server):
+    endpoint, seed = server
+    seam = Seam.from_api_key(
+        seed["seam_apikey1_token"], endpoint=endpoint, wait_for_action_attempt=False
+    )
+
+    assert seam.defaults["wait_for_action_attempt"] is False

test/conftest.py

@@ -1,55 +1,183 @@
+import json
+import os
 import socket
-from urllib.parse import urljoin
-from urllib3.util import Retry
-import pytest
 import subprocess
-import os
+import threading
+import time
 from contextlib import contextmanager
-from niquests import Session
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from urllib.error import URLError
+from urllib.request import urlopen
+
+import pytest
+
 from seam import Seam
 
+SERVER_STARTUP_TIMEOUT = 30
+SERVER_SHUTDOWN_TIMEOUT = 10
+HEALTH_POLL_INTERVAL = 0.05
 
-@pytest.fixture(scope="function")
-def server():
-    port = get_port()
-    os.environ["PORT"] = str(port)
+REPO_ROOT = Path(__file__).resolve().parent.parent
+FAKE_SEAM_CONNECT_BIN = REPO_ROOT / "node_modules" / ".bin" / "fake-seam-connect"
 
-    with subprocess_popen(["npm", "run", "start"]):
-        endpoint = f"http://localhost:{port}"
-        seed = get_seed(endpoint)
-        yield endpoint, seed
 
+@pytest.fixture(name="server")
+def server_fixture():
+    """Run a fake Seam Connect server for the duration of a single test.
+
+    Yields the endpoint of the running server along with its seed, which holds
+    the ids and tokens of the seeded records.
+    """
+
+    with fake_seam_connect() as server:
+        yield server
+
+
+@pytest.fixture(name="seam")
+def seam_fixture(server):
+    """Return a Seam client authorized against a fake Seam Connect server."""
 
-@pytest.fixture(scope="function")
-def seam(server):
     endpoint, seed = server
-    seam = Seam(endpoint=endpoint, api_key=seed["seam_apikey1_token"])
 
-    yield seam
+    return Seam(api_key=seed["seam_apikey1_token"], endpoint=endpoint)
+
+
+@pytest.fixture(name="recording_server")
+def recording_server_fixture():
+    """Return a factory for a server that records requests and replays responses.
+
+    Use this only to assert on what the SDK puts on the wire, or to drive
+    responses the fake cannot produce. Prefer the fake for everything else.
+    """
+
+    return recording_server
+
+
+@contextmanager
+def recording_server(responses):
+    """Serve the given (status, body) responses, repeating the last one.
+
+    Yields the endpoint along with the list of requests received so far.
+    """
+
+    requests = []
+    remaining = list(responses)
+
+    class Handler(BaseHTTPRequestHandler):
+        protocol_version = "HTTP/1.1"
+
+        # pylint: disable-next=invalid-name
+        def do_POST(self):  # BaseHTTPRequestHandler dispatches on this name.
+            content_length = int(self.headers.get("content-length", 0))
+            raw_body = self.rfile.read(content_length)
+
+            requests.append(
+                {
+                    "path": self.path,
+                    "headers": {k.lower(): v for k, v in self.headers.items()},
+                    "body": json.loads(raw_body) if raw_body else None,
+                }
+            )
 
+            status, payload = remaining.pop(0) if len(remaining) > 1 else remaining[0]
 
-def get_port():
-    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-        s.bind(("", 0))
-        return s.getsockname()[1]
+            if isinstance(payload, str):
+                content_type = "text/plain"
+                body = payload.encode()
+            else:
+                content_type = "application/json"
+                body = json.dumps(payload).encode()
+
+            self.send_response(status)
+            self.send_header("content-type", content_type)
+            self.send_header("content-length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+        def log_message(self, *args):
+            pass
+
+    server = ThreadingHTTPServer(("localhost", 0), Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+
+    try:
+        yield f"http://localhost:{server.server_port}", requests
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5)
 
 
-# Create a custom context manager to ensure the fake server subprocess is terminated correctly
 @contextmanager
-def subprocess_popen(*args):
-    process = subprocess.Popen(*args)
+def fake_seam_connect():
+    if not FAKE_SEAM_CONNECT_BIN.exists():
+        raise RuntimeError(
+            f"Could not find {FAKE_SEAM_CONNECT_BIN}, run npm install before the tests."
+        )
+
+    port = get_unused_port()
+    endpoint = f"http://localhost:{port}"
+
+    process = subprocess.Popen(
+        [str(FAKE_SEAM_CONNECT_BIN), "--seed"],
+        cwd=REPO_ROOT,
+        env={**os.environ, "PORT": str(port)},
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+
     try:
-        yield process
+        wait_for_health(endpoint, process)
+        yield endpoint, get_seed(endpoint)
     finally:
-        process.terminate()
+        stop_process(process)
+
+
+def get_unused_port():
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.bind(("", 0))
+        return sock.getsockname()[1]
+
+
+def wait_for_health(endpoint, process):
+    deadline = time.monotonic() + SERVER_STARTUP_TIMEOUT
+
+    while time.monotonic() < deadline:
+        if process.poll() is not None:
+            raise RuntimeError(
+                f"Fake Seam Connect exited with code {process.returncode} "
+                "before becoming healthy."
+            )
+
         try:
-            process.wait(timeout=10)
-        except subprocess.TimeoutExpired:
-            process.kill()
+            with urlopen(f"{endpoint}/health") as response:
+                if response.status == 200:
+                    return
+        except (URLError, OSError):
+            pass
+
+        time.sleep(HEALTH_POLL_INTERVAL)
+
+    raise RuntimeError(
+        f"Fake Seam Connect did not become healthy within {SERVER_STARTUP_TIMEOUT}s."
+    )
 
 
 def get_seed(endpoint):
-    retries = Retry(connect=5, total=None, backoff_factor=0.1)
-    session = Session(retries=retries)
-    seed_url = urljoin(endpoint, "/_fake/default_seed")
-    return session.get(seed_url).json()
+    with urlopen(f"{endpoint}/_fake/default_seed") as response:
+        return json.load(response)
+
+
+def stop_process(process):
+    if process.poll() is not None:
+        return
+
+    process.terminate()
+
+    try:
+        process.wait(timeout=SERVER_SHUTDOWN_TIMEOUT)
+    except subprocess.TimeoutExpired:
+        process.kill()
+        process.wait()