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/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..d1575ac9 --- /dev/null +++ b/test/retry_test.py @@ -0,0 +1,84 @@ +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"}]}) + + +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 + + +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"