From f38df3257bde2c3aca9271453fa5b992cc5b882d Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 6 Aug 2026 02:47:56 +0000 Subject: [PATCH 1/2] 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 Claude-Session: https://claude.ai/code/session_011DzapiU8A9NMdyoTybL9xB --- test/api_key_test.py | 36 ++-- test/client_test.py | 43 +++++ test/conftest.py | 194 ++++++++++++++++---- {seam/utils => test}/deep_attr_dict_test.py | 0 test/env_test.py | 83 +++++---- test/headers_test.py | 60 +++--- test/paginator_test.py | 29 +++ test/personal_access_token_test.py | 87 ++++++++- test/retry_test.py | 95 ++++++++++ test/serialization_test.py | 52 ++++++ test/workspaces/workspaces_create_test.py | 17 -- test/workspaces/workspaces_test.py | 14 -- 12 files changed, 563 insertions(+), 147 deletions(-) create mode 100644 test/client_test.py rename {seam/utils => test}/deep_attr_dict_test.py (100%) create mode 100644 test/retry_test.py create mode 100644 test/serialization_test.py delete mode 100644 test/workspaces/workspaces_create_test.py delete mode 100644 test/workspaces/workspaces_test.py diff --git a/test/api_key_test.py b/test/api_key_test.py index b32004f6..5123a74a 100644 --- a/test/api_key_test.py +++ b/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") diff --git a/test/client_test.py b/test/client_test.py new file mode 100644 index 00000000..e80bd266 --- /dev/null +++ b/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 diff --git a/test/conftest.py b/test/conftest.py index 47d82c10..1c89a548 100755 --- a/test/conftest.py +++ b/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() diff --git a/seam/utils/deep_attr_dict_test.py b/test/deep_attr_dict_test.py similarity index 100% rename from seam/utils/deep_attr_dict_test.py rename to test/deep_attr_dict_test.py diff --git a/test/env_test.py b/test/env_test.py index 61d58c21..1228d40e 100644 --- a/test/env_test.py +++ b/test/env_test.py @@ -1,105 +1,116 @@ import os + import pytest + from seam import Seam from seam.options import SeamInvalidOptionsError +ENV_VARS = ("SEAM_API_KEY", "SEAM_ENDPOINT", "SEAM_API_URL") + -# Cleanup environment variables before and after each test def cleanup_env(): - os.environ.pop("SEAM_API_KEY", None) - os.environ.pop("SEAM_ENDPOINT", None) - os.environ.pop("SEAM_API_URL", None) + for name in ENV_VARS: + os.environ.pop(name, None) @pytest.fixture(autouse=True) -def run_around_tests(): +def clean_env(): + """Ensure a clean environment before and after each test in this module.""" + cleanup_env() yield cleanup_env() -def test_seam_client_constructor_uses_seam_api_key_env_variable(server): +def test_seam_constructor_uses_seam_api_key_env_variable(server): endpoint, seed = server os.environ["SEAM_API_KEY"] = seed["seam_apikey1_token"] + seam = Seam(endpoint=endpoint) + device = seam.devices.get(device_id=seed["august_device_1"]) - devices = seam.devices.list() - assert len(devices) > 0 + assert device.workspace_id == seed["seed_workspace_1"] + assert device.device_id == seed["august_device_1"] -def test_seam_client_api_key_option_overrides_env_variables(server): +def test_seam_api_key_option_overrides_env_variables(server): endpoint, seed = server os.environ["SEAM_API_KEY"] = "some-invalid-api-key-1" + seam = Seam(api_key=seed["seam_apikey1_token"], endpoint=endpoint) + device = seam.devices.get(device_id=seed["august_device_1"]) - devices = seam.devices.list() - assert len(devices) > 0 + assert device.device_id == seed["august_device_1"] -def test_seam_client_api_key_option_as_first_argument_overrides_env_variables(): +def test_seam_api_key_option_as_first_argument_overrides_env_variables(server): + endpoint, seed = server os.environ["SEAM_API_KEY"] = "some-invalid-api-key-2" - seam = Seam("seam_apikey_token") - assert seam is not None + seam = Seam(seed["seam_apikey1_token"], endpoint=endpoint) + device = seam.devices.get(device_id=seed["august_device_1"]) -def test_seam_client_constructor_requires_seam_api_key_when_passed_no_argument(): + assert device.device_id == seed["august_device_1"] + + +def test_seam_constructor_requires_seam_api_key_when_passed_no_argument(): with pytest.raises(SeamInvalidOptionsError, match=r"api_key"): Seam() -def test_seam_client_seam_endpoint_env_variable_is_used_first(server): +def test_seam_endpoint_env_variable_is_used_first(server): endpoint, seed = server os.environ["SEAM_API_URL"] = "https://example.com" os.environ["SEAM_ENDPOINT"] = endpoint + seam = Seam(api_key=seed["seam_apikey1_token"]) + device = seam.devices.get(device_id=seed["august_device_1"]) - devices = seam.devices.list() - assert len(devices) > 0 + assert device.device_id == seed["august_device_1"] -def test_seam_client_seam_api_url_env_variable_is_used_as_fallback(server): +def test_seam_api_url_env_variable_is_used_as_fallback(server): endpoint, seed = server os.environ["SEAM_API_URL"] = endpoint + seam = Seam(api_key=seed["seam_apikey1_token"]) + device = seam.devices.get(device_id=seed["august_device_1"]) - devices = seam.devices.list() - assert len(devices) > 0 + assert device.device_id == seed["august_device_1"] -def test_seam_client_endpoint_option_overrides_env_variables(server): +def test_seam_endpoint_option_overrides_env_variables(server): endpoint, seed = server os.environ["SEAM_API_URL"] = "https://example.com" os.environ["SEAM_ENDPOINT"] = "https://example.com" + seam = Seam(api_key=seed["seam_apikey1_token"], endpoint=endpoint) + device = seam.devices.get(device_id=seed["august_device_1"]) - devices = seam.devices.list() - assert len(devices) > 0 + assert device.device_id == seed["august_device_1"] -def test_seam_client_seam_endpoint_env_variable_is_used_with_from_api_key( - server, -): +def test_seam_endpoint_env_variable_is_used_with_from_api_key(server): endpoint, seed = server os.environ["SEAM_API_URL"] = "https://example.com" os.environ["SEAM_ENDPOINT"] = endpoint + seam = Seam.from_api_key(seed["seam_apikey1_token"]) + device = seam.devices.get(device_id=seed["august_device_1"]) - devices = seam.devices.list() - assert len(devices) > 0 + assert device.device_id == seed["august_device_1"] -@pytest.mark.xfail(reason="Fake does not support personal access token.") -def test_seam_client_seam_api_key_env_variable_is_ignored_with_personal_access_token( - server, -): +def test_seam_api_key_env_variable_is_ignored_with_personal_access_token(server): endpoint, seed = server - os.environ["SEAM_API_KEY"] = seed["seam_apikey1_token"] + os.environ["SEAM_API_KEY"] = "some-invalid-api-key-3" seam = Seam.from_personal_access_token( seed["seam_at1_token"], seed["seed_workspace_1"], endpoint=endpoint, ) + device = seam.devices.get(device_id=seed["august_device_1"]) - devices = seam.devices.list() - assert len(devices) > 0 + assert device.workspace_id == seed["seed_workspace_1"] + assert device.device_id == seed["august_device_1"] diff --git a/test/headers_test.py b/test/headers_test.py index e3518d98..a9f47e8e 100644 --- a/test/headers_test.py +++ b/test/headers_test.py @@ -1,41 +1,45 @@ -import niquests import uuid -from unittest.mock import patch, Mock -from seam import Seam from importlib.metadata import version + +from seam import Seam from seam.constants import LTS_VERSION -def test_seam_http_client_request(server): - endpoint, seed = server - seam = Seam.from_api_key(seed["seam_apikey1_token"], endpoint=endpoint) +def test_seam_sends_default_headers(recording_server): device_id = str(uuid.uuid4()) + responses = [(200, {"device": {"device_id": device_id}})] + + with recording_server(responses) as (endpoint, requests): + seam = Seam.from_api_key("seam_apikey_token", endpoint=endpoint) + device = seam.devices.get(device_id=device_id) - mock_response = Mock() - mock_response.status_code = 200 - mock_response.headers = {"content-type": "application/json"} - mock_response_data = {"device": {"device_id": device_id}} - mock_response.json.return_value = mock_response_data + assert device.device_id == device_id - with patch.object( - niquests.Session, "request", return_value=mock_response - ) as mock_request: - response = seam.client.post("/devices/get", json={"device_id": device_id}) + assert len(requests) == 1 + [request] = requests - mock_request.assert_called_once() - args, _ = mock_request.call_args + assert request["path"] == "/devices/get" + assert request["body"] == {"device_id": device_id} - assert args[0] == "POST" - assert args[1] == f"{endpoint}/devices/get" + assert request["headers"]["seam-sdk-name"] == "seamapi/python" + assert request["headers"]["seam-sdk-version"] == version("seam") + assert request["headers"]["seam-lts-version"] == LTS_VERSION + assert request["headers"]["authorization"] == "Bearer seam_apikey_token" + + assert Seam.lts_version == seam.lts_version + + +def test_seam_sends_workspace_header_with_personal_access_token(recording_server): + device_id = str(uuid.uuid4()) + responses = [(200, {"device": {"device_id": device_id}})] - passed_headers = mock_request.call_args.kwargs["headers"] or {} - request_headers = { - **seam.client.headers, - **passed_headers, - } + with recording_server(responses) as (endpoint, requests): + seam = Seam.from_personal_access_token( + "seam_at_token", "workspace-1", endpoint=endpoint + ) + seam.devices.get(device_id=device_id) - assert request_headers["seam-sdk-name"] == "seamapi/python" - assert request_headers["seam-sdk-version"] == version("seam") - assert request_headers["seam-lts-version"] == LTS_VERSION + [request] = requests - assert response == mock_response_data + assert request["headers"]["authorization"] == "Bearer seam_at_token" + assert request["headers"]["seam-workspace"] == "workspace-1" diff --git a/test/paginator_test.py b/test/paginator_test.py index f8ba6fa9..800cf78e 100644 --- a/test/paginator_test.py +++ b/test/paginator_test.py @@ -1,4 +1,33 @@ +import pytest + from seam import Seam +from seam.paginator import SeamPaginator + + +def test_create_paginator_returns_a_paginator(seam: Seam): + paginator = seam.create_paginator(seam.connected_accounts.list) + + assert isinstance(paginator, SeamPaginator) + + +def test_paginator_next_page_requires_a_cursor(seam: Seam): + paginator = seam.create_paginator(seam.connected_accounts.list, {"limit": 2}) + + with pytest.raises(ValueError, match=r"next_page_cursor"): + paginator.next_page(None) + + with pytest.raises(ValueError, match=r"next_page_cursor"): + paginator.next_page("") + + +def test_paginator_last_page_has_no_next_page(seam: Seam): + paginator = seam.create_paginator(seam.connected_accounts.list, {"limit": 2}) + _, first_pagination = paginator.first_page() + + _, next_pagination = paginator.next_page(first_pagination.next_page_cursor) + + assert next_pagination.has_next_page is False + assert next_pagination.next_page_cursor is None def test_paginator_first_page(seam: Seam): diff --git a/test/personal_access_token_test.py b/test/personal_access_token_test.py index 70e49fee..57ed3100 100644 --- a/test/personal_access_token_test.py +++ b/test/personal_access_token_test.py @@ -1,10 +1,44 @@ import pytest -from seam import Seam + +from seam import Seam, SeamMultiWorkspace from seam.auth import SeamInvalidTokenError -from seam.seam_multi_workspace import SeamMultiWorkspace + +# UPSTREAM: The fake rejects a personal access token on /devices/list, so these +# tests use /devices/get, which the fake does authorize. +# https://github.com/seamapi/fake-seam-connect/issues + + +def test_seam_from_personal_access_token_returns_authorized_instance(server): + endpoint, seed = server + seam = Seam.from_personal_access_token( + seed["seam_at1_token"], + seed["seed_workspace_1"], + endpoint=endpoint, + ) + + 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_checks_personal_access_token_format(): +def test_seam_constructor_returns_instance_authorized_with_personal_access_token( + server, +): + endpoint, seed = server + seam = Seam( + personal_access_token=seed["seam_at1_token"], + workspace_id=seed["seed_workspace_1"], + endpoint=endpoint, + ) + + 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_checks_personal_access_token_format(): workspace_id = "e4203e37-e569-4a5a-bfb7-e3e8de66161d" with pytest.raises(SeamInvalidTokenError, match=r"Unknown"): @@ -19,8 +53,50 @@ def test_seam_client_checks_personal_access_token_format(): with pytest.raises(SeamInvalidTokenError, match=r"JWT"): Seam.from_personal_access_token("ey", workspace_id) + with pytest.raises(SeamInvalidTokenError, match=r"Publishable Key"): + Seam.from_personal_access_token("seam_pk_token", workspace_id) + + +def test_seam_multi_workspace_from_personal_access_token_returns_authorized_instance( + server, +): + endpoint, seed = server + seam = SeamMultiWorkspace.from_personal_access_token( + seed["seam_at1_token"], endpoint=endpoint + ) -def test_seam_multi_workspace_client_checks_personal_access_token_format(): + workspaces = seam.workspaces.list() + + assert len(workspaces) > 0 + + +def test_seam_multi_workspace_constructor_returns_authorized_instance(server): + endpoint, seed = server + seam = SeamMultiWorkspace( + personal_access_token=seed["seam_at1_token"], endpoint=endpoint + ) + + workspaces = seam.workspaces.list() + + assert len(workspaces) > 0 + + +def test_seam_multi_workspace_creates_a_workspace(server): + endpoint, seed = server + seam = SeamMultiWorkspace( + personal_access_token=seed["seam_at1_token"], endpoint=endpoint + ) + + workspace = seam.workspaces.create( + name="Test Workspace", + connect_partner_name="Example Partner", + is_sandbox=True, + ) + + assert workspace.workspace_id is not None + + +def test_seam_multi_workspace_checks_personal_access_token_format(): with pytest.raises(SeamInvalidTokenError, match=r"Unknown"): SeamMultiWorkspace.from_personal_access_token("some-invalid-key-format") @@ -32,3 +108,6 @@ def test_seam_multi_workspace_client_checks_personal_access_token_format(): with pytest.raises(SeamInvalidTokenError, match=r"JWT"): SeamMultiWorkspace.from_personal_access_token("ey") + + with pytest.raises(SeamInvalidTokenError, match=r"Publishable Key"): + SeamMultiWorkspace.from_personal_access_token("seam_pk_token") diff --git a/test/retry_test.py b/test/retry_test.py new file mode 100644 index 00000000..8b77e0f5 --- /dev/null +++ b/test/retry_test.py @@ -0,0 +1,95 @@ +import niquests +import pytest +from urllib3.util import Retry + +from seam import Seam + +SERVICE_UNAVAILABLE = (503, "Service Unavailable") +DEVICES = (200, {"devices": [{"device_id": "august_device_1"}]}) + +# The retries option is currently ignored: SeamHttpClient sets self.retries after +# calling niquests.Session.__init__ without forwarding it, so the mounted +# HTTPAdapter keeps its default max_retries. Tests that depend on the option +# being honored are marked xfail until seam/client.py passes retries through. +retries_are_ignored = pytest.mark.xfail( + strict=True, + reason="SeamHttpClient does not forward retries to niquests.Session.", +) + + +@retries_are_ignored +def test_seam_retries_service_unavailable_responses(recording_server): + expected_retry_count = 2 + responses = [SERVICE_UNAVAILABLE, SERVICE_UNAVAILABLE, DEVICES] + + with recording_server(responses) as (endpoint, requests): + seam = Seam.from_api_key( + "seam_apikey_token", + endpoint=endpoint, + retries=retry_policy(total=expected_retry_count), + ) + devices = seam.devices.list() + + assert len(devices) == 1 + assert len(requests) == expected_retry_count + 1 + + +@retries_are_ignored +def test_seam_stops_retrying_once_retries_are_exhausted(recording_server): + expected_retry_count = 1 + + with recording_server([SERVICE_UNAVAILABLE]) as (endpoint, requests): + seam = Seam.from_api_key( + "seam_apikey_token", + endpoint=endpoint, + retries=retry_policy(total=expected_retry_count), + ) + + with pytest.raises(niquests.HTTPError) as exc_info: + seam.devices.list() + + assert exc_info.value.response.status_code == 503 + assert len(requests) == expected_retry_count + 1 + + +def test_seam_does_not_retry_when_retries_are_disabled(recording_server): + with recording_server([SERVICE_UNAVAILABLE]) as (endpoint, requests): + seam = Seam.from_api_key( + "seam_apikey_token", endpoint=endpoint, retries=retry_policy(total=0) + ) + + with pytest.raises(niquests.HTTPError) as exc_info: + seam.devices.list() + + assert exc_info.value.response.status_code == 503 + assert len(requests) == 1 + + +def test_seam_surfaces_service_unavailable_from_a_workspace_outage(server): + endpoint, seed = server + seam = Seam.from_api_key( + seed["seam_apikey1_token"], endpoint=endpoint, retries=retry_policy(total=1) + ) + + seam.client.post( + "/_fake/simulate_workspace_outage", + json={ + "workspace_id": seed["seed_workspace_1"], + "routes": ["/devices/list"], + }, + ) + + with pytest.raises(niquests.HTTPError) as exc_info: + seam.devices.list() + + assert exc_info.value.response.status_code == 503 + + +def retry_policy(*, total): + return Retry( + total=total, + status_forcelist=[503], + allowed_methods=["POST"], + backoff_factor=0, + raise_on_status=False, + ) diff --git a/test/serialization_test.py b/test/serialization_test.py new file mode 100644 index 00000000..40d4b0c0 --- /dev/null +++ b/test/serialization_test.py @@ -0,0 +1,52 @@ +from seam import Seam + + +def test_serializes_array_params_when_omitted(seam: Seam): + devices = seam.devices.list() + database = seam.client.get("/_fake/database") + + assert len(devices) == len(database["devices"]) + + +def test_serializes_array_params_when_none(seam: Seam): + devices = seam.devices.list(device_ids=None) + database = seam.client.get("/_fake/database") + + assert len(devices) == len(database["devices"]) + + +def test_serializes_array_params_when_empty(seam: Seam): + devices = seam.devices.list(device_ids=[]) + + assert len(devices) == 0 + + +def test_serializes_array_params_when_non_empty(server): + endpoint, seed = server + seam = Seam.from_api_key(seed["seam_apikey1_token"], endpoint=endpoint) + + devices = seam.devices.list( + device_ids=[seed["august_device_1"], seed["ecobee_device_1"]] + ) + + assert len(devices) == 2 + + device_ids = [device.device_id for device in devices] + assert seed["august_device_1"] in device_ids + assert seed["ecobee_device_1"] in device_ids + + +def test_serializes_array_params_when_explicitly_using_client(server): + endpoint, seed = server + seam = Seam.from_api_key(seed["seam_apikey1_token"], endpoint=endpoint) + + response = seam.client.post( + "/devices/list", + json={"device_ids": [seed["august_device_1"], seed["ecobee_device_1"]]}, + ) + + device_ids = [device["device_id"] for device in response["devices"]] + + assert len(device_ids) == 2 + assert seed["august_device_1"] in device_ids + assert seed["ecobee_device_1"] in device_ids diff --git a/test/workspaces/workspaces_create_test.py b/test/workspaces/workspaces_create_test.py deleted file mode 100644 index 5002129b..00000000 --- a/test/workspaces/workspaces_create_test.py +++ /dev/null @@ -1,17 +0,0 @@ -from seam import SeamMultiWorkspace - - -def test_workspaces_create(server): - endpoint, seed = server - seam = SeamMultiWorkspace( - endpoint=endpoint, - personal_access_token=seed["seam_at1_token"], - ) - - workspace = seam.workspaces.create( - name="Test Workspace", - connect_partner_name="Example Partner", - is_sandbox=True, - ) - - assert workspace.workspace_id diff --git a/test/workspaces/workspaces_test.py b/test/workspaces/workspaces_test.py deleted file mode 100644 index e392f968..00000000 --- a/test/workspaces/workspaces_test.py +++ /dev/null @@ -1,14 +0,0 @@ -from seam import Seam - - -def test_workspaces(seam: Seam): - ws = seam.workspaces.get() - assert ws.is_sandbox == True - - ws_list = seam.workspaces.list() - assert len(ws_list) > 0 - - reset_sandbox_action_attempt = seam.workspaces.reset_sandbox( - wait_for_action_attempt=False - ) - assert reset_sandbox_action_attempt.action_type == "RESET_SANDBOX_WORKSPACE" From 59a4e06030508f74d006a86cbf6fd591cac95051 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 6 Aug 2026 03:35:31 +0000 Subject: [PATCH 2/2] fix: honor the retries option SeamHttpClient assigned self.retries after calling niquests.Session.__init__, by which point the session had already mounted its adapters with the default max_retries. The retries option was therefore silently ignored: a caller passing Retry(total=5, status_forcelist=[503]) still got exactly one attempt. Pass retries through to niquests.Session so the mounted adapters are built with it. Seam and SeamMultiWorkspace default the option to None, which now falls back to DEFAULT_RETRIES rather than being dropped. Drops the xfail markers from the two retry tests that covered this. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_011DzapiU8A9NMdyoTybL9xB --- seam/client.py | 10 ++++++---- test/retry_test.py | 11 ----------- 2 files changed, 6 insertions(+), 15 deletions(-) diff --git a/seam/client.py b/seam/client.py index 2ac9f3ae..a0b4094a 100644 --- a/seam/client.py +++ b/seam/client.py @@ -47,16 +47,18 @@ def __init__( retries: Optional[Retry] = DEFAULT_RETRIES, **kwargs ): - super().__init__(**kwargs) + # niquests.Session mounts its adapters while initializing, so retries + # must be passed through here. Assigning self.retries afterwards leaves + # the mounted adapters on their default and the option has no effect. + super().__init__( + retries=DEFAULT_RETRIES if retries is None else retries, **kwargs + ) self.base_url = base_url headers = {**auth_headers, **kwargs.get("headers", {}), **SDK_HEADERS} self.headers.update(headers) - if retries: - self.retries = retries - def request(self, method, url, *args, **kwargs): url = urljoin(self.base_url, url) response = super().request(method, url, *args, **kwargs) diff --git a/test/retry_test.py b/test/retry_test.py index 8b77e0f5..d1575ac9 100644 --- a/test/retry_test.py +++ b/test/retry_test.py @@ -7,17 +7,7 @@ SERVICE_UNAVAILABLE = (503, "Service Unavailable") DEVICES = (200, {"devices": [{"device_id": "august_device_1"}]}) -# The retries option is currently ignored: SeamHttpClient sets self.retries after -# calling niquests.Session.__init__ without forwarding it, so the mounted -# HTTPAdapter keeps its default max_retries. Tests that depend on the option -# being honored are marked xfail until seam/client.py passes retries through. -retries_are_ignored = pytest.mark.xfail( - strict=True, - reason="SeamHttpClient does not forward retries to niquests.Session.", -) - -@retries_are_ignored def test_seam_retries_service_unavailable_responses(recording_server): expected_retry_count = 2 responses = [SERVICE_UNAVAILABLE, SERVICE_UNAVAILABLE, DEVICES] @@ -34,7 +24,6 @@ def test_seam_retries_service_unavailable_responses(recording_server): assert len(requests) == expected_retry_count + 1 -@retries_are_ignored def test_seam_stops_retrying_once_retries_are_exhausted(recording_server): expected_retry_count = 1