diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..c483965 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,34 @@ +name: ci + +on: + pull_request: + push: + branches: [master] + +permissions: + contents: read + +concurrency: + group: ${{ github.workflow }}-${{ github.head_ref || github.ref }} + cancel-in-progress: ${{ github.event_name == 'pull_request' }} + +jobs: + test: + runs-on: ${{ matrix.runner }} + timeout-minutes: 10 + strategy: + fail-fast: false + matrix: + python-version: ['3.8', '3.9', '3.10', '3.11', '3.12', '3.13'] + runner: [ubuntu-latest] + include: + - python-version: '3.7' + runner: ubuntu-22.04 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + cache: pip + - run: pip install -e ".[dev]" + - run: pytest diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml new file mode 100644 index 0000000..d69b2b2 --- /dev/null +++ b/.github/workflows/publish.yml @@ -0,0 +1,40 @@ +name: publish + +on: + release: + types: [published] + +permissions: + contents: read + +jobs: + build: + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: '3.12' + - run: pip install build + - run: python -m build + - uses: actions/upload-artifact@v4 + with: + name: dist + path: dist/ + + publish: + needs: build + runs-on: ubuntu-latest + environment: + name: pypi + url: https://pypi.org/p/sendbee-api + permissions: + id-token: write + attestations: write + steps: + - uses: actions/download-artifact@v4 + with: + name: dist + path: dist/ + - uses: pypa/gh-action-pypi-publish@v1.14 diff --git a/.gitignore b/.gitignore index a13c29d..dde6d73 100644 --- a/.gitignore +++ b/.gitignore @@ -132,3 +132,7 @@ dmypy.json /.DS_Store /tests/example_.py /tests/example.py + +# claude +.claude/ +CLAUDE.md diff --git a/.travis.yml b/.travis.yml deleted file mode 100644 index 2f3864c..0000000 --- a/.travis.yml +++ /dev/null @@ -1,27 +0,0 @@ -dist: xenial - -language: python - -python: - - 3.6 - - 3.7.6 - - 3.8 - - 3.9 - -install: - - pip install . - - pip install -r requirements.txt - -script: - - pytest - -deploy: - skip_cleanup: true - provider: pypi - distributions: sdist bdist_wheel - user: "sendbeedev" - password: - secure: "IfynHLv+tzb/OJlrBbKRr6tasem5Eail0CQ6w8IhUNTsLUFOeGhJrdoN+yeWzV7sBnUEdhKRGrUHVOwsxWUhanYm7X9gels473CXmos7qzwjrQyYifdJj5kNdi0A8metFdzIUbKLUOAyFgmtCOskIorMrMq+8I6T6azOOTLUyIhQZHFc4oQdevTNiOQ3hbV723mpdmlOZCdXPJA+jpzleqNvnFNOR7H1aDeI3AvHD0+yn6GY3dBRbhw7F7yt8ZIqQJ3Pt2ud818hhqqg12GSR3q4yiLY9UqAdYHPypc6ufXq2RvlBGmX4otjGm9sJGz27tpdlDh52SlzFSL1k1/EmVu1hC61CS61wCxJtQTQdDK6T9dXOuzkMR5+JVSGECzJLeKvaJDNCnVHXZ9KOKYZqne7wGJwQ5S45ireAGN1mHX4wf3nC9FP8TCtFY/gGQIaE/bzfsrRQ0ThK1xXWe8yBVZl5s/7anCg4PQXneljrGTs4286ywEmRRYacKsQ/x48KCM8HZiwCFXppg81Or8d13w0hAK038w3tVFLJTEBHN/1EToDlZ/4hEY9q5QkwpDvUEzDub05M46e0pCfp2oigrCnGdklxLzVXSaPyQzYpMIxvSGwnJPr+BiaFSD4EzkOsLE0mLsWd5lIt6hBSEEvYxqKGQKSNI+dRT8hvu5GXcA=" - skip_existing: true - on: - branch: master \ No newline at end of file diff --git a/README.md b/README.md index 64279b7..ac98bac 100644 --- a/README.md +++ b/README.md @@ -1,15 +1,6 @@ # AI Number Python API Client [![PyPI version](https://badge.fury.io/py/sendbee-api.svg)](https://badge.fury.io/py/sendbee-api) -[![Build Status](https://travis-ci.org/sendbee/sendbee-python-api-client.svg?branch=master)](https://travis-ci.org/sendbee/sendbee-python-api-client) - -![GitHub issues](https://img.shields.io/github/issues/sendbee/sendbee-python-api-client.svg) -![GitHub closed issues](https://img.shields.io/github/issues-closed/sendbee/sendbee-python-api-client.svg) -![GitHub closed pull requests](https://img.shields.io/github/issues-pr-closed/sendbee/sendbee-python-api-client.svg) - -![PyPI - Python Version](https://img.shields.io/pypi/pyversions/sendbee-api.svg) -![GitHub](https://img.shields.io/github/license/sendbee/sendbee-python-api-client.svg?color=blue) -![GitHub last commit](https://img.shields.io/github/last-commit/sendbee/sendbee-python-api-client.svg?color=blue) ## Table of contents diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..2e7783f --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,46 @@ +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "sendbee_api" +version = "1.7.2" +description = "Python client SDK for Sendbee Public API" +readme = "README.md" +requires-python = ">=3.6" +license = { text = "MIT" } +authors = [{ name = "Sendbee ltd", email = "info@sendbee.io" }] +keywords = ["sendbee", "api", "python"] +classifiers = [ + "Intended Audience :: Developers", + "Topic :: Software Development :: Build Tools", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3.6", + "Programming Language :: Python :: 3.7", + "Programming Language :: Python :: 3.8", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", +] +dependencies = [ + "click>=7.0", + "requests>=2.20.0", + "dumpit>=0.5.0", + "aenum>=2.1.2", + "ujson>=5.7.0,<6", + "cryptography>=3.2", + "curlify>=2.2.1", +] + +[project.optional-dependencies] +dev = ["pytest", "responses>=0.23"] + +[project.urls] +Homepage = "https://github.com/sendbee/sendbee-python-client" +Source = "https://github.com/sendbee/sendbee-python-client" + +[tool.setuptools.packages.find] +include = ["sendbee_api*"] +exclude = ["tests*"] diff --git a/requirements.txt b/requirements.txt index 55b033e..755876c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1 +1,4 @@ -pytest \ No newline at end of file +# Dev/test deps. Source of truth: pyproject.toml [project.optional-dependencies] dev +# Prefer `pip install -e ".[dev]"` for new setups. +pytest +responses>=0.23 diff --git a/setup.py b/setup.py deleted file mode 100644 index 71acbd9..0000000 --- a/setup.py +++ /dev/null @@ -1,47 +0,0 @@ -from setuptools import setup, find_packages - - -def readme(): - with open('README.md') as f: - return f.read() - - -setup( - name='sendbee_api', - version='1.7.1', - - description='Python client SDK for Sendbee Public API', - long_description=readme(), - long_description_content_type='text/markdown', - - url='https://github.com/sendbee/sendbee-python-client', - licence='MIT', - - author='Sendbee ltd', - author_email='info@sendbee.io', - - classifiers=[ - 'Intended Audience :: Developers', - 'Topic :: Software Development :: Build Tools', - 'License :: OSI Approved :: MIT License', - 'Programming Language :: Python :: 3.6', - 'Programming Language :: Python :: 3.7', - 'Programming Language :: Python :: 3.8', - 'Programming Language :: Python :: 3.9', - ], - keywords='sendbee api python', - - packages=find_packages(), - install_requires=[ - 'click>=7.0', - 'requests>=2.20.0', - 'dumpit>=0.5.0', - 'aenum>=2.1.2', - 'ujson==2.0.1', - 'cryptography>=3.2' - ], - - project_urls={ - 'Source': 'https://github.com/sendbee/sendbee-python-client', - }, -) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..a03fd3f --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,76 @@ +"""Shared fixtures for the sendbee-api test suite. + +Tests mock at the `requests` HTTP boundary using the `responses` library. The +SDK's documented `fake_response_path` test seam is not used because its branch +in `bind.py` returns a 2-tuple while `call()` unpacks 3 elements. +""" + +import ujson +import pytest +import responses as _responses + +from sendbee_api import SendbeeApi + + +API_KEY = "test-key" +API_SECRET = "test-secret" +BASE_URL = "https://api-v2.sendbee.io" + + +@pytest.fixture +def api_key(): + return API_KEY + + +@pytest.fixture +def api_secret(): + return API_SECRET + + +@pytest.fixture +def client(): + """Fresh SendbeeApi instance per test.""" + return SendbeeApi(API_KEY, API_SECRET) + + +@pytest.fixture +def json_body(): + """Build a Sendbee-shaped JSON envelope as a string. + + The server wraps responses as {"data": ..., "meta": ..., "warning": ...}. + The Json formatter unwraps these keys; tests pass the inner shapes here. + """ + def _build(data=None, meta=None, warning=None): + body = {} + if data is not None: + body["data"] = data + if meta is not None: + body["meta"] = meta + if warning is not None: + body["warning"] = warning + return ujson.dumps(body) + return _build + + +@pytest.fixture +def register(): + """Register a mocked HTTP response on the Sendbee base URL. + + Thin wrapper around responses.add() that prepends BASE_URL so tests stay + terse. Call as register("GET", "/contacts", body=..., status=...). + """ + def _register(method, path, body="", status=200, content_type="application/json"): + method_const = { + "GET": _responses.GET, + "POST": _responses.POST, + "PUT": _responses.PUT, + "DELETE": _responses.DELETE, + }[method.upper()] + _responses.add( + method_const, + BASE_URL + path, + body=body, + status=status, + content_type=content_type, + ) + return _register diff --git a/tests/test_api.py b/tests/test_api.py deleted file mode 100644 index 79f3119..0000000 --- a/tests/test_api.py +++ /dev/null @@ -1,7 +0,0 @@ -import unittest - - -class SendbeeApi(unittest.TestCase): - - def test_fake(self): - pass diff --git a/tests/test_auth.py b/tests/test_auth.py new file mode 100644 index 0000000..e60f94f --- /dev/null +++ b/tests/test_auth.py @@ -0,0 +1,110 @@ +"""Tests for SendbeeAuth: HMAC token generation and verification.""" + +import base64 +import hmac +import hashlib +from datetime import datetime, timezone + +import pytest + +from sendbee_api.auth import SendbeeAuth + + +def test_get_auth_token_returns_base64_encoded_timestamp_dot_hmac(): + """Token format is base64(.).""" + auth = SendbeeAuth("secret") + + token = auth.get_auth_token() + + decoded = base64.b64decode(token).decode("utf-8") + timestamp, hex_digest = decoded.split(".") + assert timestamp.isdigit() + assert len(hex_digest) == 64 + int(hex_digest, 16) + + +def test_get_auth_token_uses_hmac_sha256_of_timestamp_under_private_key(monkeypatch): + """HMAC payload is the timestamp string, key is the private key.""" + fixed_ts = 1700000000 + secret = "secret-bytes" + + class _FrozenDatetime(datetime): + @classmethod + def now(cls, tz=None): + return datetime.fromtimestamp(fixed_ts, tz=tz) + + monkeypatch.setattr("sendbee_api.auth.datetime", _FrozenDatetime) + + expected_hmac = hmac.new( + secret.encode("utf-8"), + str(fixed_ts).encode("utf-8"), + hashlib.sha256, + ).hexdigest() + + token = SendbeeAuth(secret).get_auth_token() + + decoded = base64.b64decode(token).decode("utf-8") + timestamp, hex_digest = decoded.split(".") + assert timestamp == str(fixed_ts) + assert hex_digest == expected_hmac + + +def test_get_auth_token_changes_when_timestamp_changes(monkeypatch): + """Two tokens generated at different timestamps differ.""" + secret = "secret" + timestamps = iter([1700000000, 1700000001]) + + class _CountingDatetime(datetime): + @classmethod + def now(cls, tz=None): + return datetime.fromtimestamp(next(timestamps), tz=tz) + + monkeypatch.setattr("sendbee_api.auth.datetime", _CountingDatetime) + + token_a = SendbeeAuth(secret).get_auth_token() + token_b = SendbeeAuth(secret).get_auth_token() + + assert token_a != token_b + + +def test_check_auth_token_returns_true_for_self_generated_token(): + """A token produced by get_auth_token verifies under the same secret.""" + auth = SendbeeAuth("secret") + + token = auth.get_auth_token() + + assert auth.check_auth_token(token) is True + + +def test_check_auth_token_returns_false_when_hmac_tampered(): + """A token whose HMAC has been altered fails verification.""" + auth = SendbeeAuth("secret") + token = auth.get_auth_token() + decoded = base64.b64decode(token).decode("utf-8") + timestamp, hex_digest = decoded.split(".") + tampered_hex = ("0" if hex_digest[0] != "0" else "1") + hex_digest[1:] + tampered = base64.b64encode( + f"{timestamp}.{tampered_hex}".encode("utf-8") + ).decode("utf-8") + + assert auth.check_auth_token(tampered) is False + + +def test_check_auth_token_returns_false_when_secret_differs(): + """A token signed with secret A cannot be verified with secret B.""" + token = SendbeeAuth("secret-a").get_auth_token() + other = SendbeeAuth("secret-b") + + assert other.check_auth_token(token) is False + + +def test_constructor_accepts_str_and_bytes_keys(): + """Both str and bytes private keys produce identical tokens.""" + fixed_ts = b"1700000000" + expected = hmac.new(b"secret", fixed_ts, hashlib.sha256).hexdigest() + + str_auth = SendbeeAuth("secret") + bytes_auth = SendbeeAuth(b"secret") + + assert str_auth._get_encrypted_key(fixed_ts) == expected + assert bytes_auth._get_encrypted_key(fixed_ts) == expected diff --git a/tests/test_bind_request.py b/tests/test_bind_request.py new file mode 100644 index 0000000..fbbe15a --- /dev/null +++ b/tests/test_bind_request.py @@ -0,0 +1,476 @@ +"""Tests for the bind_request factory. + +The factory in `sendbee_api/bind.py` is the core that every endpoint method +funnels through. These tests cover: request construction, query/path/body +serialization, header generation including the HMAC auth token, single vs. +multi-model response shaping, error mapping to exceptions, pagination +boundary behavior, warning surfacing, and the print_params escape hatch. + +All tests mock at the `requests` HTTP layer via the `responses` library, so +they exercise the real `_do_request` -> `_process_response` path. +""" + +import base64 +from urllib.parse import urlparse, parse_qs + +import pytest +import responses +import ujson + +from sendbee_api import constants +from sendbee_api.auth import SendbeeAuth +from sendbee_api.bind import bind_request +from sendbee_api.exceptions import ( + SendbeeRequestApiException, + PaginationException, +) +from sendbee_api.fields import TextField, NumberField, ModelField +from sendbee_api.models import Model +from sendbee_api.query_params import QueryParams + + +class _Item(Model): + """Test-only response model with two text fields.""" + _id = TextField(index="id") + _name = TextField(index="name") + + +class _DemoQueryParams(QueryParams): + """Test-only query params covering python<->wire name aliasing.""" + name = "name", "Contact name" + search_query = "search_query", "Free text filter" + tags = "tags", "Tag list" + active = "active", "Active flag" + page = "page", "Page number" + + +def _client(): + """Build a fresh client. Imported here to avoid module-load coupling.""" + from sendbee_api import SendbeeApi + return SendbeeApi("test-key", "test-secret") + + +def _make_get(api_path="/demo", **kw): + return bind_request( + api_path=api_path, + model=_Item, + query_parameters=_DemoQueryParams, + **kw, + ) + + +def _make_post(api_path="/demo", **kw): + return bind_request( + api_path=api_path, + model=_Item, + method=constants.RequestConst.POST, + query_parameters=_DemoQueryParams, + **kw, + ) + + +def _query_dict(call): + """Parse a responses.calls[i].request.url into a {param: [vals]} dict.""" + return parse_qs(urlparse(call.request.url).query) + + +# --------------------------------------------------------------------------- +# Request construction: timeout and defaults +# --------------------------------------------------------------------------- + +@responses.activate +def test_timeout_kwarg_is_consumed_and_not_sent_as_query_param(json_body, register): + """timeout=5 controls the HTTP timeout and is stripped from the query.""" + register("GET", "/demo", body=json_body(data=[], meta={"current_page": 1, "last_page": 1})) + call = _make_get() + + call(_client(), timeout=5, name="Alice") + + qs = _query_dict(responses.calls[0]) + assert "timeout" not in qs + assert qs["name"] == ["Alice"] + + +@responses.activate +def test_non_integer_timeout_falls_back_without_raising(json_body, register): + """timeout='abc' does not raise; request still succeeds.""" + register("GET", "/demo", body=json_body(data=[], meta={"current_page": 1, "last_page": 1})) + call = _make_get() + + response = call(_client(), timeout="abc", name="Alice") + + assert response.models == [] + + +@responses.activate +def test_default_parameters_are_merged_into_outgoing_query(json_body, register): + """Bind-site default_parameters appear in the request query.""" + register("GET", "/demo", body=json_body(data=[], meta={"current_page": 1, "last_page": 1})) + call = _make_get(default_parameters={"name": "default-name"}) + + call(_client()) + + qs = _query_dict(responses.calls[0]) + assert qs["name"] == ["default-name"] + + +@responses.activate +def test_call_site_kwarg_overrides_default_parameter_for_same_key(json_body, register): + """Call-site kwargs win over bind-site defaults.""" + register("GET", "/demo", body=json_body(data=[], meta={"current_page": 1, "last_page": 1})) + call = _make_get(default_parameters={"name": "default-name"}) + + call(_client(), name="Bob") + + qs = _query_dict(responses.calls[0]) + assert qs["name"] == ["Bob"] + + +# --------------------------------------------------------------------------- +# Query param name aliasing +# --------------------------------------------------------------------------- + +@responses.activate +def test_python_name_kwarg_is_translated_to_wire_name_in_url(json_body, register): + """msg_type kwarg becomes msgtype=... in the outgoing URL.""" + register("GET", "/demo", body=json_body(data=[], meta={"current_page": 1, "last_page": 1})) + call = _make_get() + + call(_client(), msg_type="simple") + + qs = _query_dict(responses.calls[0]) + assert qs["msgtype"] == ["simple"] + assert "msg_type" not in qs + + +@responses.activate +def test_wire_name_kwarg_is_preserved_as_is(json_body, register): + """Passing the wire-name directly is preserved in the URL.""" + register("GET", "/demo", body=json_body(data=[], meta={"current_page": 1, "last_page": 1})) + call = _make_get() + + call(_client(), msgtype="extended") + + qs = _query_dict(responses.calls[0]) + assert qs["msgtype"] == ["extended"] + + +@responses.activate +def test_unknown_kwarg_is_silently_dropped(json_body, register): + """A kwarg with no matching wire/python name is omitted, no error raised.""" + register("GET", "/demo", body=json_body(data=[], meta={"current_page": 1, "last_page": 1})) + call = _make_get() + + call(_client(), nonsense_param="x", name="Alice") + + qs = _query_dict(responses.calls[0]) + assert "nonsense_param" not in qs + assert qs["name"] == ["Alice"] + + +@responses.activate +def test_none_valued_kwarg_is_omitted_from_query(json_body, register): + """None values are skipped, so the param key does not appear in the URL.""" + register("GET", "/demo", body=json_body(data=[], meta={"current_page": 1, "last_page": 1})) + call = _make_get() + + call(_client(), name=None, search_query="hi") + + qs = _query_dict(responses.calls[0]) + assert "name" not in qs + assert qs["search_query"] == ["hi"] + + +# --------------------------------------------------------------------------- +# GET serialization: bool, list, path params +# --------------------------------------------------------------------------- + +@responses.activate +def test_boolean_true_is_serialized_as_one_in_query(json_body, register): + """True becomes '1' in the GET querystring.""" + register("GET", "/demo", body=json_body(data=[], meta={"current_page": 1, "last_page": 1})) + call = _make_get() + + call(_client(), active=True) + + qs = _query_dict(responses.calls[0]) + assert qs["active"] == ["1"] + + +@responses.activate +def test_boolean_false_is_serialized_as_zero_in_query(json_body, register): + """False becomes '0' in the GET querystring.""" + register("GET", "/demo", body=json_body(data=[], meta={"current_page": 1, "last_page": 1})) + call = _make_get() + + call(_client(), active=False) + + qs = _query_dict(responses.calls[0]) + assert qs["active"] == ["0"] + + +@responses.activate +def test_list_value_is_serialized_as_comma_joined_string(json_body, register): + """tags=['a','b'] is sent as tags=a,b.""" + register("GET", "/demo", body=json_body(data=[], meta={"current_page": 1, "last_page": 1})) + call = _make_get() + + call(_client(), tags=["a", "b"]) + + qs = _query_dict(responses.calls[0]) + assert qs["tags"] == ["a,b"] + + +@responses.activate +def test_positional_path_params_are_appended_to_url(json_body, register): + """Positional args build base/path/{a}/{b} URL.""" + register("GET", "/demo/abc/def", body=json_body(data=[], meta={"current_page": 1, "last_page": 1})) + call = _make_get() + + call(_client(), "abc", "def") + + actual = responses.calls[0].request.url + assert "/demo/abc/def" in actual + + +# --------------------------------------------------------------------------- +# POST/PUT/DELETE: body shape and single-model response +# --------------------------------------------------------------------------- + +@responses.activate +def test_post_sends_payload_in_json_body_not_querystring(json_body, register): + """POST puts parameters in the JSON body and leaves the URL path-only.""" + register("POST", "/demo", body=json_body(data=[{"id": "1", "name": "Alice"}], meta={"current_page": 1, "last_page": 1})) + call = _make_post() + + call(_client(), name="Alice", search_query="hi") + + request = responses.calls[0].request + assert urlparse(request.url).query == "" + body = ujson.loads(request.body) + assert body == {"name": "Alice", "search_query": "hi"} + + +@responses.activate +def test_post_returns_single_model_not_response_wrapper(json_body, register): + """POST auto-triggers single_model_response; result is a Model instance.""" + register("POST", "/demo", body=json_body(data=[{"id": "1", "name": "Alice"}], meta={"current_page": 1, "last_page": 1})) + call = _make_post() + + result = call(_client(), name="Alice") + + assert isinstance(result, _Item) + assert result.name == "Alice" + + +@responses.activate +def test_post_returns_none_when_server_returns_empty_data(json_body, register): + """Empty data list with POST yields None, not an empty Response.""" + register("POST", "/demo", body=json_body(data=[], meta={"current_page": 1, "last_page": 1})) + call = _make_post() + + result = call(_client(), name="Alice") + + assert result is None + + +@responses.activate +def test_force_single_model_response_makes_get_return_one_model(json_body, register): + """force_single_model_response=True on a GET returns one model, not a Response.""" + register("GET", "/demo", body=json_body(data=[{"id": "1", "name": "Alice"}], meta={"current_page": 1, "last_page": 1})) + call = _make_get(force_single_model_response=True) + + result = call(_client()) + + assert isinstance(result, _Item) + assert result.id == "1" + + +@responses.activate +def test_put_method_sends_request_with_json_body_and_returns_single_model(json_body, register): + """PUT path is exercised end-to-end: body in JSON, single-model response.""" + register("PUT", "/demo", body=json_body(data=[{"id": "1", "name": "x"}], meta={"current_page": 1, "last_page": 1})) + call = bind_request( + api_path="/demo", + model=_Item, + method=constants.RequestConst.PUT, + query_parameters=_DemoQueryParams, + ) + + result = call(_client(), name="x") + + assert isinstance(result, _Item) + body = ujson.loads(responses.calls[0].request.body) + assert body == {"name": "x"} + + +@responses.activate +def test_delete_method_sends_delete_request_and_returns_single_model(json_body, register): + """DELETE path returns a single model (typically ServerMessage).""" + from sendbee_api.models import ServerMessage + register("DELETE", "/demo", body=json_body(data=[{"message": "deleted"}], meta={"current_page": 1, "last_page": 1})) + call = bind_request( + api_path="/demo", + model=ServerMessage, + method=constants.RequestConst.DELETE, + query_parameters=_DemoQueryParams, + ) + + result = call(_client(), name="to-delete") + + assert isinstance(result, ServerMessage) + assert result.message == "deleted" + assert responses.calls[0].request.method == "DELETE" + + +# --------------------------------------------------------------------------- +# Headers and auth +# --------------------------------------------------------------------------- + +@responses.activate +def test_request_carries_api_key_and_auth_token_headers(json_body, register): + """X-Api-Key matches client.api_key; X-Auth-Token is a valid HMAC.""" + register("GET", "/demo", body=json_body(data=[], meta={"current_page": 1, "last_page": 1})) + call = _make_get() + client = _client() + + call(client) + + sent_headers = responses.calls[0].request.headers + assert sent_headers["X-Api-Key"] == client.api_key + token = sent_headers["X-Auth-Token"] + assert SendbeeAuth(client.api_secret).check_auth_token(token) is True + + +@responses.activate +def test_request_carries_static_accept_content_type_and_user_agent(json_body, register): + """Static SDK headers are present on every outgoing request.""" + register("GET", "/demo", body=json_body(data=[], meta={"current_page": 1, "last_page": 1})) + call = _make_get() + + call(_client()) + + headers = responses.calls[0].request.headers + assert headers["Accept"] == "application/json" + assert headers["Content-Type"] == "application/json" + assert headers["User-Agent"] == "Sendbee Python API Client" + + +# --------------------------------------------------------------------------- +# Error handling +# --------------------------------------------------------------------------- + +@responses.activate +def test_http_400_with_error_detail_raises_with_message_extracted(json_body, register): + """HTTP 400 + {error:{detail:...}} raises and surfaces the detail string.""" + register( + "GET", "/demo", + body=ujson.dumps({"error": {"detail": "Phone is invalid", "type": "validation_error"}}), + status=400, + ) + call = _make_get() + + with pytest.raises(SendbeeRequestApiException) as exc_info: + call(_client()) + + assert "Phone is invalid" in str(exc_info.value.args[0]) + assert exc_info.value.response is not None + assert exc_info.value.response.status_code == 400 + + +@responses.activate +def test_http_400_without_error_key_falls_back_to_default_error_message(json_body, register): + """A 400 body that doesn't match the {error:{detail}} shape uses default.""" + register("GET", "/demo", body=ujson.dumps({"something_else": "x"}), status=400) + call = _make_get() + + with pytest.raises(SendbeeRequestApiException) as exc_info: + call(_client()) + + assert exc_info.value.args[0] == constants.ResponseConst.DEFAULT_ERROR_MESSAGE + + +@responses.activate +def test_http_404_raises_with_not_found_message_regardless_of_body(json_body, register): + """404 uses the canonical NOT_FOUND error string.""" + register("GET", "/demo", body=ujson.dumps({"data": []}), status=404) + call = _make_get() + + with pytest.raises(SendbeeRequestApiException) as exc_info: + call(_client()) + + assert exc_info.value.args[0] == constants.ErrorConst.NOT_FOUND + + +@responses.activate +def test_ignore_error_true_swallows_4xx_and_returns_response(json_body, register): + """ignore_error=True on the bind site lets a 400 pass through.""" + register("GET", "/demo", body=json_body(data=[], meta={"current_page": 1, "last_page": 1}), status=400) + call = _make_get(ignore_error=True) + + response = call(_client()) + + assert response.status_code == 400 + + +# --------------------------------------------------------------------------- +# Pagination edge case +# --------------------------------------------------------------------------- + +@responses.activate +def test_pagination_exception_when_page_greater_than_one_returns_empty_data(json_body, register): + """current_page > 1 with empty data list raises PaginationException.""" + register("GET", "/demo", body=json_body(data=[], meta={"current_page": 2, "last_page": 1})) + call = _make_get() + + with pytest.raises(PaginationException): + call(_client(), page=2) + + +@responses.activate +def test_no_pagination_exception_on_page_one_even_when_data_empty(json_body, register): + """An empty first page is a valid 'no results', not an exception.""" + register("GET", "/demo", body=json_body(data=[], meta={"current_page": 1, "last_page": 1})) + call = _make_get() + + response = call(_client()) + + assert response.models == [] + + +# --------------------------------------------------------------------------- +# Warnings +# --------------------------------------------------------------------------- + +@responses.activate +def test_warning_in_response_is_printed_to_stdout_but_does_not_raise( + json_body, register, capsys +): + """A server warning is printed via click.secho and exposed via .warning.""" + register("GET", "/demo", body=json_body( + data=[], + meta={"current_page": 1, "last_page": 1}, + warning="deprecated-endpoint", + )) + call = _make_get() + + response = call(_client()) + + captured = capsys.readouterr() + assert "deprecated-endpoint" in captured.out + assert response.warning == "deprecated-endpoint" + + +# --------------------------------------------------------------------------- +# print_params escape hatch +# --------------------------------------------------------------------------- + +@responses.activate +def test_print_params_true_short_circuits_before_any_http_call(capsys): + """print_params=True returns without making a request.""" + call = _make_get() + + result = call(_client(), print_params=True) + + assert result is None + assert len(responses.calls) == 0 diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 0000000..d09bb4a --- /dev/null +++ b/tests/test_client.py @@ -0,0 +1,67 @@ +"""Tests for SendbeeApi construction and mixin composition.""" + +import pytest + +from sendbee_api import SendbeeApi +from sendbee_api.exceptions import SendbeeRequestApiException +from sendbee_api.contacts.client import Contacts +from sendbee_api.conversations.client import Messages +from sendbee_api.automation.client import Automation +from sendbee_api.teams.client import Teams +from sendbee_api.rate_limit.client import RateLimit + + +def test_constructor_raises_when_api_key_missing(): + """Empty api_key raises SendbeeRequestApiException.""" + with pytest.raises(SendbeeRequestApiException): + SendbeeApi(api_key="", api_secret="s") + + +def test_constructor_raises_when_api_secret_missing(): + """Empty api_secret raises SendbeeRequestApiException.""" + with pytest.raises(SendbeeRequestApiException): + SendbeeApi(api_key="k", api_secret="") + + +def test_constructor_stores_credentials_and_debug_flag(): + """Credentials and debug flag round-trip onto the instance.""" + client = SendbeeApi(api_key="k", api_secret="s", debug=True) + + assert client.api_key == "k" + assert client.api_secret == "s" + assert client.debug is True + assert client.fake_response_path is None + + +def test_client_inherits_all_five_resource_mixins(): + """Client MRO includes every resource mixin.""" + for mixin in (Contacts, Messages, Automation, Teams, RateLimit): + assert issubclass(SendbeeApi, mixin), f"{mixin.__name__} missing from MRO" + + +def test_client_exposes_endpoint_methods_from_each_mixin(): + """One representative method from each mixin is callable on the client.""" + client = SendbeeApi("k", "s") + + assert callable(client.contacts) + assert callable(client.conversations) + assert callable(client.chatbot_activity_status) + assert callable(client.teams) + assert callable(client.rate_limit_error_test) + + +def test_auth_property_returns_sendbee_auth_bound_to_api_secret(): + """client.auth is a SendbeeAuth that can verify its own tokens.""" + client = SendbeeApi("k", "secret-xyz") + + token = client.auth.get_auth_token() + + assert client.auth.check_auth_token(token) is True + + +def test_print_params_for_unknown_method_prints_red_warning(capsys): + """Calling print_params_for with a bogus name prints a warning, not raises.""" + SendbeeApi.print_params_for("not_a_real_endpoint") + + captured = capsys.readouterr() + assert "Unknown API method" in captured.out diff --git a/tests/test_endpoints_smoke.py b/tests/test_endpoints_smoke.py new file mode 100644 index 0000000..7acad8b --- /dev/null +++ b/tests/test_endpoints_smoke.py @@ -0,0 +1,177 @@ +"""End-to-end smoke tests: one happy-path per resource mixin. + +The 24 endpoint methods on `SendbeeApi` are all generated by `bind_request`, +which is covered in detail elsewhere. These tests verify the wiring per mixin +- that each `bind_request(...)` call site declares the right `api_path`, +`method`, `model`, and `QueryParams` - by exercising one representative +endpoint per mixin through a mocked HTTP response. +""" + +import responses +import ujson + +from sendbee_api.contacts.models import Contact +from sendbee_api.conversations.models import Conversation, SentMessage +from sendbee_api.automation.models import ChatbotActivityStatus +from sendbee_api.teams.models import Team +from sendbee_api.rate_limit.models import RateLimitError +from sendbee_api.models import ServerMessage + + +@responses.activate +def test_contacts_list_endpoint_returns_contact_models(client, register, json_body): + """SendbeeApi.contacts() hits /contacts and parses Contact models.""" + register("GET", "/contacts", body=json_body( + data=[{ + "id": "c1", + "name": "Alice", + "phone": "+1234", + "status": "subscribed", + "tags": [], + "contact_fields": [], + "notes": [], + }], + meta={"current_page": 1, "last_page": 1, "total": 1}, + )) + + response = client.contacts() + + assert "/contacts" in responses.calls[0].request.url + assert responses.calls[0].request.method == "GET" + assert len(response.models) == 1 + contact = response.models[0] + assert isinstance(contact, Contact) + assert contact.id == "c1" + assert contact.name == "Alice" + + +@responses.activate +def test_conversations_list_endpoint_returns_conversation_models(client, register, json_body): + """SendbeeApi.conversations() hits /conversations and parses Conversation models.""" + register("GET", "/conversations", body=json_body( + data=[{ + "id": "conv1", + "folder": "open", + "chatbot_active": True, + "platform": "whatsapp", + "contact": {"id": "c1", "name": "Alice", "phone": "+1234"}, + "last_message": { + "direction": "inbound", + "status": "delivered", + "inbound_sent_at": None, + "outbound_sent_at": None, + }, + }], + meta={"current_page": 1, "last_page": 1, "total": 1}, + )) + + response = client.conversations() + + assert "/conversations" in responses.calls[0].request.url + assert isinstance(response.models[0], Conversation) + assert response.models[0].id == "conv1" + assert response.models[0].contact.name == "Alice" + + +@responses.activate +def test_send_message_post_endpoint_returns_single_sent_message(client, register, json_body): + """send_message is a POST that returns one SentMessage, not a Response.""" + register("POST", "/conversations/messages/send", body=json_body( + data=[{ + "id": "conv1", + "message_id": "msg1", + "message_reference_id": "ref1", + "status": "queued", + }], + meta={"current_page": 1, "last_page": 1}, + )) + + result = client.send_message(phone="+1234", text="hello") + + assert isinstance(result, SentMessage) + assert result.message_id == "msg1" + assert responses.calls[0].request.method == "POST" + body = ujson.loads(responses.calls[0].request.body) + assert body == {"phone": "+1234", "text": "hello"} + + +@responses.activate +def test_chatbot_activity_status_endpoint_returns_single_status_model(client, register, json_body): + """chatbot_activity_status is force-single-model GET; returns one model.""" + register("GET", "/automation/chatbot/activity/status", body=json_body( + data=[{"conversation_id": "conv1", "chatbot_active": True}], + meta={"current_page": 1, "last_page": 1}, + )) + + result = client.chatbot_activity_status(conversation_id="conv1") + + assert isinstance(result, ChatbotActivityStatus) + assert result.conversation_id == "conv1" + assert result.chatbot_active is True + + +@responses.activate +def test_teams_list_endpoint_returns_team_models(client, register, json_body): + """SendbeeApi.teams() hits /teams/teams and parses Team models.""" + register("GET", "/teams/teams", body=json_body( + data=[{"id": "t1", "name": "Support", "members": []}], + meta={"current_page": 1, "last_page": 1, "total": 1}, + )) + + response = client.teams() + + assert "/teams/teams" in responses.calls[0].request.url + assert isinstance(response.models[0], Team) + assert response.models[0].name == "Support" + + +@responses.activate +def test_rate_limit_error_test_endpoint_returns_rate_limit_error_model(client, register, json_body): + """rate_limit_error_test hits /rate-limit/error-test on the RateLimit mixin.""" + register("GET", "/rate-limit/error-test", body=json_body( + data=[{"detail": "Too many", "error": True, "type": "rate_limit"}], + meta={"current_page": 1, "last_page": 1}, + )) + + response = client.rate_limit_error_test() + + assert "/rate-limit/error-test" in responses.calls[0].request.url + assert isinstance(response.models[0], RateLimitError) + assert response.models[0].detail == "Too many" + + +@responses.activate +def test_delete_tag_returns_server_message_ack(client, register, json_body): + """delete_tag is a DELETE; response is a ServerMessage ack.""" + register("DELETE", "/contacts/tags", body=json_body( + data=[{"message": "tag deleted"}], + meta={"current_page": 1, "last_page": 1}, + )) + + result = client.delete_tag(id="tag-uuid-1") + + assert isinstance(result, ServerMessage) + assert result.message == "tag deleted" + assert responses.calls[0].request.method == "DELETE" + + +@responses.activate +def test_subscribe_contact_default_parameters_block_automation_and_notifications( + client, register, json_body +): + """subscribe_contact's bind-site defaults set block_automation=block_notifications=True.""" + register("POST", "/contacts/subscribe", body=json_body( + data=[{ + "id": "c1", "name": "Alice", "phone": "+1234", + "status": "subscribed", "tags": [], "contact_fields": [], "notes": [], + }], + meta={"current_page": 1, "last_page": 1}, + )) + + client.subscribe_contact(phone="+1234", name="Alice") + + body = ujson.loads(responses.calls[0].request.body) + assert body["block_automation"] is True + assert body["block_notifications"] is True + assert body["phone"] == "+1234" + assert body["name"] == "Alice" diff --git a/tests/test_fields.py b/tests/test_fields.py new file mode 100644 index 0000000..2971ded --- /dev/null +++ b/tests/test_fields.py @@ -0,0 +1,105 @@ +"""Tests for Field subclasses: type coercion and fallback behavior.""" + +from datetime import datetime + +import pytest + +from sendbee_api.fields import ( + Field, + NumberField, + RealNumberField, + TextField, + BooleanField, + DatetimeField, + ListField, + ModelField, +) +from sendbee_api.models import Model + + +class _StubModel: + """Minimal stand-in that exposes the only attribute Field reads.""" + def __init__(self, item): + self.item = item + + +def _convert(field, item_dict): + field.convert_item(_StubModel(item_dict)) + return field.value + + +def test_number_field_converts_numeric_string_to_int(): + """Numeric strings parse to int.""" + assert _convert(NumberField(index="n"), {"n": "42"}) == 42 + + +def test_number_field_returns_zero_for_non_numeric_string(): + """Non-numeric input falls back to 0, not exception.""" + assert _convert(NumberField(index="n"), {"n": "abc"}) == 0 + + +def test_real_number_field_converts_decimal_string_to_float(): + """Decimal strings parse to float.""" + assert _convert(RealNumberField(index="x"), {"x": "3.14"}) == 3.14 + + +def test_real_number_field_returns_zero_point_zero_for_invalid_input(): + """Non-numeric input falls back to 0.0.""" + assert _convert(RealNumberField(index="x"), {"x": "abc"}) == 0.0 + + +def test_text_field_coerces_int_to_string(): + """Integer values are stringified.""" + assert _convert(TextField(index="t"), {"t": 7}) == "7" + + +def test_boolean_field_coerces_truthy_to_true(): + """Non-empty string is truthy.""" + assert _convert(BooleanField(index="b"), {"b": "yes"}) is True + + +def test_boolean_field_coerces_zero_to_false(): + """0 is falsy.""" + assert _convert(BooleanField(index="b"), {"b": 0}) is False + + +def test_datetime_field_parses_with_provided_format(): + """Provided strftime format parses an ISO-style timestamp.""" + field = DatetimeField(index="ts", format="%Y-%m-%d %H:%M:%S") + + value = _convert(field, {"ts": "2024-01-02 03:04:05"}) + + assert value == datetime(2024, 1, 2, 3, 4, 5) + + +def test_datetime_field_returns_raw_string_when_format_mismatches(): + """A value that doesn't match the format is returned unchanged (no raise).""" + field = DatetimeField(index="ts", format="%Y-%m-%d %H:%M:%S") + + assert _convert(field, {"ts": "not-a-date"}) == "not-a-date" + + +def test_list_field_passes_list_values_through(): + """A list input is preserved.""" + assert _convert(ListField(index="l"), {"l": ["a", "b"]}) == ["a", "b"] + + +def test_field_returns_none_when_index_missing_from_item(): + """Missing key on the item dict yields None, not KeyError.""" + assert _convert(TextField(index="missing"), {"other": "x"}) is None + + +def test_field_returns_none_when_value_is_none(): + """Explicit None passes through as None.""" + assert _convert(TextField(index="t"), {"t": None}) is None + + +def test_model_field_stores_target_model_class(): + """ModelField records the nested model class for later recursion.""" + class Inner(Model): + pass + + field = ModelField(Inner, index="nested") + + assert field.model_cls is Inner + assert field.index == "nested" diff --git a/tests/test_formatter.py b/tests/test_formatter.py new file mode 100644 index 0000000..73d26b0 --- /dev/null +++ b/tests/test_formatter.py @@ -0,0 +1,79 @@ +"""Tests for the Json formatter and FormatterFactory lookup.""" + +import ujson + +from sendbee_api.formatter import Json, FormatterFactory +from sendbee_api import constants + + +class _StubResponse: + """Minimal Response stand-in; the formatter only uses it for context.""" + + +def _formatter(): + return Json(_StubResponse()) + + +def test_format_data_unwraps_data_key_from_envelope(): + """{"data": [...]} returns [...] from format_data.""" + body = ujson.dumps({"data": [{"id": "c1"}], "meta": {}}) + + result = _formatter().format_data(body) + + assert result == [{"id": "c1"}] + + +def test_format_data_returns_dict_unchanged_when_no_data_key(): + """Envelope without `data` key passes through as the parsed dict.""" + body = ujson.dumps({"error": {"detail": "bad"}}) + + result = _formatter().format_data(body) + + assert result == {"error": {"detail": "bad"}} + + +def test_format_data_returns_default_error_when_body_is_not_valid_json(): + """Garbage input is replaced by the canonical DEFAULT_ERROR_MESSAGE.""" + result = _formatter().format_data("not-json-at-all") + + assert result == constants.ResponseConst.DEFAULT_ERROR_MESSAGE + + +def test_format_meta_returns_meta_key_when_present(): + """Envelope's meta block is returned verbatim.""" + body = ujson.dumps({ + "data": [], + "meta": {"current_page": 2, "last_page": 5, "total": 100}, + }) + + result = _formatter().format_meta(body) + + assert result == {"current_page": 2, "last_page": 5, "total": 100} + + +def test_format_meta_returns_empty_dict_when_meta_key_absent(): + """Dict body without `meta` yields {}.""" + body = ujson.dumps({"data": []}) + + assert _formatter().format_meta(body) == {} + + +def test_format_warning_returns_warning_string_when_present(): + """Warning text is returned as-is.""" + body = ujson.dumps({"data": [], "warning": "rate-limit-soon"}) + + assert _formatter().format_warning(body) == "rate-limit-soon" + + +def test_format_warning_returns_none_when_warning_absent(): + """Missing warning yields None.""" + body = ujson.dumps({"data": []}) + + assert _formatter().format_warning(body) is None + + +def test_formatter_factory_returns_json_class_for_json_key(): + """FormatterFactory exposes the Json class for the 'json' lookup.""" + formatter_cls = FormatterFactory(constants.FormatterConst.JSON).get_formatter() + + assert formatter_cls is Json diff --git a/tests/test_models.py b/tests/test_models.py new file mode 100644 index 0000000..ec53c7e --- /dev/null +++ b/tests/test_models.py @@ -0,0 +1,99 @@ +"""Tests for Model.process(): attribute translation and nested model recursion.""" + +from sendbee_api.models import Model, Meta, ServerMessage +from sendbee_api.fields import TextField, NumberField, ModelField + + +class _Tag(Model): + _name = TextField(index="name") + + +class _Contact(Model): + _id = TextField(index="id") + _name = TextField(index="name") + _tags = ModelField(_Tag, index="tags") + + +class _Wrapper(Model): + _label = TextField(index="label") + _tag = ModelField(_Tag, index="tag") + + +def test_process_strips_leading_underscore_from_class_attr_when_exposing_on_instance(): + """`_name` on the class becomes `name` on the instance after process().""" + [contact] = _Contact.process([{"id": "c1", "name": "Alice", "tags": []}]) + + assert contact.id == "c1" + assert contact.name == "Alice" + + +def test_process_returns_list_of_model_instances_one_per_input_dict(): + """process() returns one Model per input record.""" + contacts = _Contact.process([ + {"id": "c1", "name": "Alice", "tags": []}, + {"id": "c2", "name": "Bob", "tags": []}, + ]) + + assert len(contacts) == 2 + assert [c.id for c in contacts] == ["c1", "c2"] + + +def test_process_recurses_into_model_field_when_value_is_list(): + """List-valued ModelField is mapped element-wise into nested models.""" + [contact] = _Contact.process([{ + "id": "c1", + "name": "Alice", + "tags": [{"name": "vip"}, {"name": "lead"}], + }]) + + assert isinstance(contact.tags, list) + assert [t.name for t in contact.tags] == ["vip", "lead"] + + +def test_process_recurses_into_model_field_when_value_is_dict(): + """Dict-valued ModelField is mapped to a single nested model (not a list).""" + [wrapper] = _Wrapper.process([{"label": "x", "tag": {"name": "vip"}}]) + + assert not isinstance(wrapper.tag, list) + assert wrapper.tag.name == "vip" + + +def test_process_sets_attribute_to_none_when_source_value_is_none(): + """A None field value is preserved as None on the instance.""" + [contact] = _Contact.process([{"id": "c1", "name": None, "tags": []}]) + + assert contact.name is None + + +def test_getattr_raises_attribute_error_for_unknown_attribute(): + """Accessing an attribute that was never declared raises AttributeError.""" + [contact] = _Contact.process([{"id": "c1", "name": "Alice", "tags": []}]) + + try: + _ = contact.does_not_exist + except AttributeError: + return + raise AssertionError("expected AttributeError") + + +def test_meta_model_exposes_pagination_fields_as_ints(): + """Meta.process maps current/last/per/total fields to ints.""" + [meta] = Meta.process([{ + "total": "100", + "to": "20", + "from": "1", + "per_page": "20", + "last_page": "5", + "current_page": "1", + }]) + + assert meta.total == 100 + assert meta.last_page == 5 + assert meta.current_page == 1 + + +def test_server_message_exposes_message_field(): + """ServerMessage carries a single `message` text field.""" + [sm] = ServerMessage.process([{"message": "deleted"}]) + + assert sm.message == "deleted" diff --git a/tests/test_query_params.py b/tests/test_query_params.py new file mode 100644 index 0000000..9b8e4f8 --- /dev/null +++ b/tests/test_query_params.py @@ -0,0 +1,59 @@ +"""Tests for QueryParams: MultiValueEnum aliasing and default merge.""" + +from sendbee_api.query_params import QueryParams, DefaultQueryParams + + +class _DemoParams(QueryParams): + """Sample query param set for testing.""" + + name = "name", "Contact name" + search_query = "search_query", "Free text filter" + + +def test_get_params_returns_mapping_of_python_name_to_wire_name(): + """Python attribute names map to their wire-name strings.""" + params = _DemoParams.get_params() + + assert params["name"] == "name" + assert params["search_query"] == "search_query" + + +def test_get_params_includes_default_query_params_for_msgtype_and_protocol(): + """DefaultQueryParams are merged in; msgtype/protocol available everywhere.""" + params = _DemoParams.get_params() + + assert "msg_type" in params + assert params["msg_type"] == "msgtype" + assert "protocol" in params + + +def test_multi_value_enum_resolves_both_tuple_values_to_same_member(): + """Looking up by wire-name or by description string yields the same member. + + aenum.MultiValueEnum allows any element of the tuple to be a valid lookup + value, so members defined as `name = ('wire', 'desc')` can be retrieved by + either string. + """ + by_wire = DefaultQueryParams("msgtype") + by_desc = DefaultQueryParams( + "simple or extended (extended cost more credits)" + ) + + assert by_wire is by_desc + + +def test_bind_request_aliasing_uses_get_params_to_resolve_both_names(): + """bind_request's wire/python-name aliasing comes from get_params() shape. + + The dict maps python attr name (key) -> wire name (value), so it can be + used to look up the wire form from either side: kwarg matches a key + (python name) OR matches a value (wire name). + """ + params = _DemoParams.get_params() + python_names = set(params.keys()) + wire_names = set(params.values()) + + assert "search_query" in python_names + assert "search_query" in wire_names + assert "msg_type" in python_names + assert "msgtype" in wire_names diff --git a/tests/test_response.py b/tests/test_response.py new file mode 100644 index 0000000..974e149 --- /dev/null +++ b/tests/test_response.py @@ -0,0 +1,112 @@ +"""Tests for the Response wrapper: lazy unwrap and pagination helpers.""" + +import ujson + +from sendbee_api.response import Response +from sendbee_api.formatter import Json +from sendbee_api.fields import TextField +from sendbee_api.models import Model + + +class _Item(Model): + _id = TextField(index="id") + _name = TextField(index="name") + + +class _ApiRequestStub: + model = _Item + + +def _build_response(body, headers=None, status=200): + return Response(body, headers or {}, status, Json, _ApiRequestStub()) + + +def test_models_property_returns_parsed_model_instances_from_data_envelope(): + """Response.models maps each `data` entry to a model instance.""" + body = ujson.dumps({ + "data": [{"id": "1", "name": "a"}, {"id": "2", "name": "b"}], + "meta": {"current_page": 1, "last_page": 1, "total": 2}, + }) + + response = _build_response(body) + + assert [m.name for m in response.models] == ["a", "b"] + + +def test_iter_delegates_to_models(): + """`for m in response` iterates response.models.""" + body = ujson.dumps({ + "data": [{"id": "1", "name": "a"}], + "meta": {"current_page": 1, "last_page": 1, "total": 1}, + }) + + response = _build_response(body) + + assert [m.name for m in response] == ["a"] + + +def test_raw_data_returns_input_string_unchanged(): + """`raw_data` is the unparsed body.""" + body = '{"data": []}' + + response = _build_response(body) + + assert response.raw_data == body + + +def test_meta_exposes_current_and_last_page_as_int(): + """Pagination meta is parsed via Meta model into ints.""" + body = ujson.dumps({ + "data": [{"id": "1", "name": "a"}], + "meta": {"current_page": 1, "last_page": 3, "total": 25}, + }) + + response = _build_response(body) + + assert response.meta.current_page == 1 + assert response.meta.last_page == 3 + + +def test_has_next_returns_true_when_more_pages_remain(): + """has_next() is True iff current_page < last_page.""" + body = ujson.dumps({ + "data": [{"id": "1", "name": "a"}], + "meta": {"current_page": 1, "last_page": 3}, + }) + + response = _build_response(body) + + assert response.has_next() is True + + +def test_has_next_returns_false_on_final_page(): + """has_next() is False when current_page equals last_page.""" + body = ujson.dumps({ + "data": [{"id": "1", "name": "a"}], + "meta": {"current_page": 3, "last_page": 3}, + }) + + response = _build_response(body) + + assert response.has_next() is False + + +def test_next_page_returns_current_page_plus_one(): + """next_page() advances by exactly one.""" + body = ujson.dumps({ + "data": [{"id": "1", "name": "a"}], + "meta": {"current_page": 2, "last_page": 5}, + }) + + response = _build_response(body) + + assert response.next_page() == 3 + + +def test_warning_returns_server_warning_string_when_present(): + """Warning text from the envelope is exposed via .warning.""" + body = ujson.dumps({"data": [], "warning": "deprecated-endpoint"}) + + response = _build_response(body) + + assert response.warning == "deprecated-endpoint"