From 664a7bec05c0a5b57eae096124e8f63fc591ce80 Mon Sep 17 00:00:00 2001 From: Guillaume Mazoyer Date: Sat, 27 Jun 2026 12:48:42 +0200 Subject: [PATCH] Add unit test suite and enable branch coverage Cover the API client, endpoints, query layer, response/record handling, models, tokens and helpers with unit tests, and report branch coverage. - Add unit tests across api, endpoint, query, response, models, token and util - Mint auth tokens dynamically in tests instead of shipping static JWT fixtures - Fix Record list diffing so items containing the separator are not conflated - Count single-object responses as one result - Remove dead code: Request.get_openapi, Record pickle hooks, Token.__str__ - Correct the cat() docstring - Enable --cov-branch and upload coverage to Codecov with a token --- .github/workflows/tests.yml | 13 +- pyixapi/core/query.py | 21 +- pyixapi/core/response.py | 38 +- pyixapi/core/token.py | 3 - pyixapi/core/util.py | 14 +- pyproject.toml | 2 +- tests/fixtures/api/authenticate.json | 4 - .../fixtures/api/refresh_authentication.json | 4 - tests/test_api.py | 233 ++++++++-- tests/test_endpoint.py | 183 ++++++++ tests/test_models.py | 273 +++++++++++ tests/test_query.py | 433 ++++++++++++++++++ tests/test_response.py | 407 ++++++++++++++++ tests/test_token.py | 67 +++ tests/test_util.py | 37 ++ tests/util.py | 90 +++- 16 files changed, 1719 insertions(+), 103 deletions(-) delete mode 100644 tests/fixtures/api/authenticate.json delete mode 100644 tests/fixtures/api/refresh_authentication.json create mode 100644 tests/test_endpoint.py create mode 100644 tests/test_models.py create mode 100644 tests/test_query.py create mode 100644 tests/test_response.py create mode 100644 tests/test_token.py create mode 100644 tests/test_util.py diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index d6a31c3..1c038f1 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -21,6 +21,8 @@ jobs: steps: - name: Checkout uses: actions/checkout@v7 + with: + fetch-depth: 2 - name: Setup Python uses: actions/setup-python@v6 @@ -40,9 +42,18 @@ jobs: run: | uv run ruff format --check --diff . uv run ruff check --diff . - + - name: Run type checking run: uv run ty check . - name: Run Tests run: uv run pytest + + - name: Upload coverage to Codecov + if: matrix.python == '3.14' + uses: codecov/codecov-action@v5 + with: + token: ${{ secrets.CODECOV_TOKEN }} + slug: peering-manager/pyixapi + files: coverage.xml + fail_ci_if_error: false diff --git a/pyixapi/core/query.py b/pyixapi/core/query.py index ea67ef3..33ac214 100644 --- a/pyixapi/core/query.py +++ b/pyixapi/core/query.py @@ -59,10 +59,8 @@ class Request(object): Responsible for building the URL and making the HTTP(S) requests to the API. :param base: (str) Base URL passed in api() instantiation. - :param filters: (dict, optional) contains key/value pairs that correlate to the - filters a given endpoint accepts. - In (e.g. /api/v1/devices?name='test') 'name': 'test' would be in the filters - dict. + :param filters: (dict, optional) key/value pairs matching the filters an + endpoint accepts, e.g. {"name": "test"} for /devices?name=test. """ def __init__( @@ -84,17 +82,6 @@ def __init__( self.user_agent = user_agent self.proxies = proxies - def get_openapi(self) -> dict[str, Any]: - """ - Get the OpenAPI Spec. - """ - headers = {"Content-Type": "application/json;"} - req = self.http_session.get(cat(self.base, "docs/?format=openapi"), headers=headers) - if req.ok: - return req.json() - else: - raise RequestError(req) - def get_version(self) -> int: """ Get the API version of IX-API. @@ -129,7 +116,7 @@ def _make_call( add_params: dict[str, Any] | None = None, data: dict[str, Any] | None = None, ) -> Any: - if verb in ("post", "put") or verb == "delete" and data: + if verb in ("post", "put") or (verb == "delete" and data): headers: dict[str, str] = {"Content-Type": "application/json;"} else: headers = {"accept": "application/json;"} @@ -184,7 +171,7 @@ def get(self, add_params: dict[str, Any] | None = None) -> Generator[dict[str, A for i in req: yield i else: - self.count = len(req) + self.count = 1 yield req def put(self, data: dict[str, Any]) -> dict[str, Any]: diff --git a/pyixapi/core/response.py b/pyixapi/core/response.py index b3b810c..9aa1539 100644 --- a/pyixapi/core/response.py +++ b/pyixapi/core/response.py @@ -90,11 +90,11 @@ class Record(object): Create Python objects from IX-API responses. Nested dicts that represent other endpoints are also turned into - :py:class:`.Record` objects. All fields are then assigned to the object's - attributes. If a missing attribute is requested (e.g. requesting a field that's - only present on a full response on a :py:class:`.Record` made from a nested - response) then pyixapi will make a request for the full object and return the - requested value. + :py:class:`.Record` objects (when the corresponding model declares them as a + class attribute). All fields are then assigned to the object's attributes. + + Only the fields present in the response are set as attributes; accessing a + field that was not returned raises :py:exc:`AttributeError`. :examples: Default representation of the object is usually its ID and/or name: @@ -149,7 +149,6 @@ class Record(object): url: str | None = None def __init__(self, values: dict[str, Any], api: API, endpoint: Endpoint) -> None: - self._full_cache: list[Any] = [] self._init_cache: list[tuple[str, Any]] = [] self.api = api self.default_ret: type[Record] = Record @@ -179,12 +178,6 @@ def __str__(self) -> str: def __repr__(self) -> str: return str(dict(self)) - def __getstate__(self) -> dict[str, Any]: - return self.__dict__ - - def __setstate__(self, d: dict[str, Any]) -> None: - self.__dict__.update(d) - def __key__(self) -> tuple[str, ...] | tuple[str]: if hasattr(self, "id"): return (self.endpoint.name, self.id) @@ -212,8 +205,7 @@ def list_parser(key_name: str, list_item: Any) -> Any: if isinstance(list_item, dict): lookup = getattr(self.__class__, key_name, None) if not isinstance(lookup, list): - # This is *list_parser*, so if the custom model field is not - # a list (or is not defined), just return the default model + # Field not declared as a list model: use the default Record. return self.default_ret(list_item, self.api, self.endpoint) else: model = lookup[0] @@ -259,16 +251,18 @@ def serialize(self, nested: bool = False, init: bool = False) -> Any: return r def _diff(self) -> set[str]: - def fmt_dict(k: str, v: Any) -> tuple[str, Any]: + def make_hashable(v: Any) -> Any: + # Hashable, structure-preserving form for set comparison. Lists become + # tuples, not a joined string (which would collide). if isinstance(v, dict): - return k, Hashabledict(v) + return Hashabledict({k: make_hashable(val) for k, val in v.items()}) if isinstance(v, list): - return k, ",".join(map(str, v)) - return k, v + return tuple(make_hashable(i) for i in v) + return v - current = Hashabledict({fmt_dict(k, v) for k, v in self.serialize().items()}) - init = Hashabledict({fmt_dict(k, v) for k, v in self.serialize(init=True).items()}) - return set([i[0] for i in set(current.items()) ^ set(init.items())]) + current = {k: make_hashable(v) for k, v in self.serialize().items()} + init = {k: make_hashable(v) for k, v in self.serialize(init=True).items()} + return {key for key, _ in set(current.items()) ^ set(init.items())} def updates(self) -> dict[str, Any]: """ @@ -343,4 +337,4 @@ def delete(self) -> bool: user_agent=self.api.user_agent, proxies=self.api.proxies, ) - return True if r.delete() else False + return r.delete() diff --git a/pyixapi/core/token.py b/pyixapi/core/token.py index 976cf87..8529dd7 100644 --- a/pyixapi/core/token.py +++ b/pyixapi/core/token.py @@ -29,9 +29,6 @@ def __init__(self, token: str, expires_at: datetime) -> None: self.encoded: str = token # Cache signed token data self.expires_at: datetime = expires_at - def __str__(self) -> str: - return self.encoded - def __repr__(self) -> str: return f"" diff --git a/pyixapi/core/util.py b/pyixapi/core/util.py index 71065c4..4bcc6cd 100644 --- a/pyixapi/core/util.py +++ b/pyixapi/core/util.py @@ -15,14 +15,14 @@ def cat(*args: Any, separator: str = "/", trailing: str = "") -> str: If an item cannot be parsed as a string, an AttributeError will be raised. - >>> concatenate("a", "b", "c") - 'a/b/c/' - >>> concatenate("a", "b", "/c/", separator="") + >>> cat("a", "/b/", "c/") 'a/b/c' - >>> concatenate("a", "/b/", 1) - 'a/b/1/' - >>> concatenate("a", "b", "c", separator="_", trailing="_") - 'a_b_c_' + >>> cat("a", 1, "b") + 'a/1/b' + >>> cat("a", "b", "c", separator="_") + 'a_b_c' + >>> cat("a", "b", "c", trailing="/") + 'a/b/c/' """ s = separator.join([str(i).strip(separator) for i in args if str(i)]) if trailing: diff --git a/pyproject.toml b/pyproject.toml index 2b52bb9..4848500 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -59,7 +59,7 @@ invalid-return-type = "ignore" unresolved-import = "ignore" [tool.pytest.ini_options] -addopts = "--cov=pyixapi --cov-report=term-missing --cov-report=xml" +addopts = "--cov=pyixapi --cov-branch --cov-report=term-missing --cov-report=xml" [tool.coverage.run] source = ["pyixapi"] diff --git a/tests/fixtures/api/authenticate.json b/tests/fixtures/api/authenticate.json deleted file mode 100644 index 1f2d3c4..0000000 --- a/tests/fixtures/api/authenticate.json +++ /dev/null @@ -1,4 +0,0 @@ -{ - "access_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiJPbmxpbmUgSldUIEJ1aWxkZXIiLCJpYXQiOjE2NjM5NjE0MDAsImV4cCI6MTY2Mzk2MjMwMCwiYXVkIjoid3d3LmV4YW1wbGUuY29tIiwic3ViIjoidGVzdEBleGFtcGxlLmNvbSJ9.jPZ4TMh9mFGu2Ebgz-UO3bd5wrAJDyOjJaul1tq0AoI", - "refresh_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiJPbmxpbmUgSldUIEJ1aWxkZXIiLCJpYXQiOjE2NjM5NjE0MDAsImV4cCI6MTY2NDU2NjIwMCwiYXVkIjoid3d3LmV4YW1wbGUuY29tIiwic3ViIjoidGVzdEBleGFtcGxlLmNvbSJ9.NCIzCxKjceLGfHb7ozJ5KsLw0WSlWHXx3dPGx8n8P0Q" -} diff --git a/tests/fixtures/api/refresh_authentication.json b/tests/fixtures/api/refresh_authentication.json deleted file mode 100644 index 0475cca..0000000 --- a/tests/fixtures/api/refresh_authentication.json +++ /dev/null @@ -1,4 +0,0 @@ -{ - "access_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiJPbmxpbmUgSldUIEJ1aWxkZXIiLCJpYXQiOjE2NjQwNDc4MDAsImV4cCI6MTY2NDA0ODcwMCwiYXVkIjoid3d3LmV4YW1wbGUuY29tIiwic3ViIjoidGVzdEBleGFtcGxlLmNvbSJ9._PeLD8F8SlVJKZcBAQ7a5ONsFYXWl9Q_UjKJLGS-uAI", - "refresh_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiJPbmxpbmUgSldUIEJ1aWxkZXIiLCJpYXQiOjE2NjQwNDc4MDAsImV4cCI6MTY2NDY1MjYwMCwiYXVkIjoid3d3LmV4YW1wbGUuY29tIiwic3ViIjoidGVzdEBleGFtcGxlLmNvbSJ9.xHKO5CyRQbbLEAvuiuk4srb7px8G-2JScjBy-bVTqsg" -} diff --git a/tests/test_api.py b/tests/test_api.py index f4c2491..42aec87 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -3,42 +3,106 @@ from unittest.mock import patch import pyixapi +from pyixapi.core.token import Token -from .util import Response +from .util import Response, auth_response, def_args, host, make_jwt -host = "https://api.example.net/v1/" -def_args = ( - "-GnNlMD8hBuxSSUJmpbfUkss9dyOKfTV1SnZibNyyr4", - "XKq8M6NVh5lCbPJ2Ml1h7V93QNIMsGVBfM6g2nRZF-E", -) +class ApiInitTestCase(unittest.TestCase): + def test_url_trailing_slash_stripped(self) -> None: + api = pyixapi.api(host, *def_args) + self.assertFalse(api.url.endswith("/")) + + def test_init_with_pre_existing_tokens(self) -> None: + token = make_jwt(expires_in=-3600) + api = pyixapi.api(host, *def_args, access_token=token, refresh_token=token) + self.assertIsNotNone(api.access_token) + self.assertIsInstance(api.access_token, Token) + self.assertIsNotNone(api.refresh_token) + self.assertIsInstance(api.refresh_token, Token) + + def test_init_without_tokens(self) -> None: + api = pyixapi.api(host, *def_args) + self.assertIsNone(api.access_token) + self.assertIsNone(api.refresh_token) + + def test_init_with_custom_user_agent(self) -> None: + api = pyixapi.api(host, *def_args, user_agent="custom/1.0") + self.assertEqual(api.user_agent, "custom/1.0") + + def test_init_with_proxies(self) -> None: + proxies = {"http": "http://proxy:8080"} + api = pyixapi.api(host, *def_args, proxies=proxies) + self.assertEqual(api.proxies, proxies) + + def test_default_user_agent(self) -> None: + api = pyixapi.api(host, *def_args) + self.assertEqual(api.user_agent, f"pyixapi/{pyixapi.__version__}") class ApiTestCase(unittest.TestCase): - @patch( - "requests.sessions.Session.post", - return_value=Response(fixture="api/authenticate.json"), - ) - def test_authenticate(self, *_) -> None: + @patch("requests.sessions.Session.post", return_value=auth_response()) + def test_authenticate_stores_valid_tokens(self, mock_post) -> None: api = pyixapi.api(host, *def_args) r = api.authenticate() self.assertIsNotNone(r) - self.assertIsNotNone(api.access_token) - self.assertIsNotNone(api.refresh_token) + self.assertFalse(api.access_token.is_expired) + self.assertFalse(api.refresh_token.is_expired) + self.assertTrue(mock_post.call_args[0][0].endswith("/auth/token")) + self.assertEqual(mock_post.call_args[1]["json"]["api_key"], def_args[0]) + self.assertEqual(mock_post.call_args[1]["json"]["api_secret"], def_args[1]) - @patch( - "requests.sessions.Session.post", - return_value=Response(fixture="api/refresh_authentication.json"), - ) - def test_refresh_authentication(self, *_) -> None: + def test_authenticate_returns_none_when_access_token_valid(self) -> None: + api = pyixapi.api(host, *def_args) + api.access_token = Token.from_jwt(make_jwt(expires_in=3600)) + with patch("requests.sessions.Session.post") as mock_post: + result = api.authenticate() + self.assertIsNone(result) + mock_post.assert_not_called() + + @patch("requests.sessions.Session.post", return_value=auth_response()) + def test_authenticate_refreshes_when_access_expired_refresh_valid(self, mock_post) -> None: + api = pyixapi.api(host, *def_args) + api.access_token = Token.from_jwt(make_jwt(expires_in=-3600)) + api.refresh_token = Token.from_jwt(make_jwt(expires_in=3600)) + + result = api.authenticate() + + self.assertIsNotNone(result) + self.assertTrue(mock_post.call_args[0][0].endswith("/auth/refresh")) + + @patch("requests.sessions.Session.post", return_value=auth_response()) + def test_authenticate_reauthenticates_when_both_tokens_expired(self, mock_post) -> None: + api = pyixapi.api(host, *def_args) + api.access_token = Token.from_jwt(make_jwt(expires_in=-3600)) + api.refresh_token = Token.from_jwt(make_jwt(expires_in=-3600)) + + result = api.authenticate() + + self.assertIsNotNone(result) + self.assertTrue(mock_post.call_args[0][0].endswith("/auth/token")) + self.assertFalse(api.access_token.is_expired) + + @patch("requests.sessions.Session.post", return_value=auth_response()) + def test_refresh_authentication(self, mock_post) -> None: api = pyixapi.api(host, *def_args) api.authenticate() + sent_refresh_token = api.refresh_token.encoded r = api.refresh_authentication() self.assertIsNotNone(r) - self.assertIsNotNone(api.access_token) - self.assertIsNotNone(api.refresh_token) + self.assertFalse(api.access_token.is_expired) + self.assertFalse(api.refresh_token.is_expired) + self.assertTrue(mock_post.call_args[0][0].endswith("/auth/refresh")) + self.assertEqual(mock_post.call_args[1]["json"]["refresh_token"], sent_refresh_token) + + def test_refresh_authentication_raises_without_refresh_token(self) -> None: + api = pyixapi.api(host, *def_args) + api.refresh_token = None + with self.assertRaises(ValueError) as ctx: + api.refresh_authentication() + self.assertIn("No refresh token", str(ctx.exception)) class ApiVersionTestCase(unittest.TestCase): @@ -65,10 +129,7 @@ def test_api_version(self, *_) -> None: api = pyixapi.api(host, *def_args) self.assertEqual(api.version, 2) - @patch( - "requests.sessions.Session.get", - return_value=Response(content={"status": "pass", "version": 2}), - ) + @patch("requests.sessions.Session.get", return_value=Response(content={"status": "pass", "version": 2})) def test_api_version_is_cached(self, mock_get) -> None: api = pyixapi.api(host, *def_args) self.assertEqual(api.version, 2) @@ -77,14 +138,126 @@ def test_api_version_is_cached(self, mock_get) -> None: mock_get.assert_called_once() -class ApiHealthTestCase(unittest.TestCase): - class ResponseWithHealth: - ok = True +class ApiVersionDependentEndpointsTestCase(unittest.TestCase): + @patch( + "requests.sessions.Session.get", + return_value=Response(content={"status": "pass", "version": 2}), + ) + def test_accounts_v2(self, *_) -> None: + api = pyixapi.api(host, *def_args) + endpoint = api.accounts + self.assertIn("accounts", endpoint.url) + + @patch( + "requests.sessions.Session.get", + return_value=Response(status_code=404, ok=False, url=host, text="Not found"), + ) + def test_accounts_v1(self, *_) -> None: + with warnings.catch_warnings(record=True): + warnings.simplefilter("always") + api = pyixapi.api(host, *def_args) + endpoint = api.accounts + self.assertIn("customers", endpoint.url) + + @patch( + "requests.sessions.Session.get", + return_value=Response(content={"status": "pass", "version": 2}), + ) + def test_product_offerings_v2(self, *_) -> None: + api = pyixapi.api(host, *def_args) + endpoint = api.product_offerings + self.assertIn("product-offerings", endpoint.url) + + @patch( + "requests.sessions.Session.get", + return_value=Response(status_code=404, ok=False, url=host, text="Not found"), + ) + def test_product_offerings_v1(self, *_) -> None: + with warnings.catch_warnings(record=True): + warnings.simplefilter("always") + api = pyixapi.api(host, *def_args) + endpoint = api.product_offerings + self.assertIn("products", endpoint.url) + + @patch( + "requests.sessions.Session.get", + return_value=Response(content={"status": "pass", "version": 2}), + ) + def test_demarcs_raises_on_v2(self, *_) -> None: + api = pyixapi.api(host, *def_args) + with self.assertRaises(AttributeError): + api.demarcs - def json(self): - return {"status": "pass", "version": "2"} + @patch( + "requests.sessions.Session.get", + return_value=Response(status_code=404, ok=False, url=host, text="Not found"), + ) + def test_demarcs_v1(self, *_) -> None: + with warnings.catch_warnings(record=True): + warnings.simplefilter("always") + api = pyixapi.api(host, *def_args) + endpoint = api.demarcs + self.assertIn("demarcs", endpoint.url) + + +class ApiAccountTestCase(unittest.TestCase): + @patch("requests.sessions.Session.get") + def test_account(self, mock_get) -> None: + # Respond by URL so the test does not depend on call order. + def by_url(url, *args, **kwargs): + if url.endswith("/account"): + return Response(content={"id": "ACCT-001", "name": "My Account"}) + return Response(content={"status": "pass", "version": 2}) + + mock_get.side_effect = by_url + + api = pyixapi.api(host, *def_args) + account = api.account() - @patch("requests.sessions.Session.get", return_value=ResponseWithHealth()) + self.assertEqual(account.id, "ACCT-001") + self.assertTrue(any(c.args[0].endswith("/account") for c in mock_get.call_args_list)) + + +class ApiImplementationTestCase(unittest.TestCase): + @patch( + "requests.sessions.Session.get", + return_value=Response(content={"name": "Test IXP", "version": "2.7.1"}), + ) + def test_implementation(self, *_) -> None: + api = pyixapi.api(host, *def_args) + result = api.implementation() + self.assertEqual(result["name"], "Test IXP") + self.assertEqual(result["version"], "2.7.1") + + +class ApiExtensionsTestCase(unittest.TestCase): + @patch( + "requests.sessions.Session.get", + return_value=Response(content=[{"name": "ext1"}, {"name": "ext2"}]), + ) + def test_extensions(self, *_) -> None: + api = pyixapi.api(host, *def_args) + result = api.extensions() + self.assertEqual(len(result), 2) + self.assertEqual(result[0]["name"], "ext1") + + +class ApiHealthTestCase(unittest.TestCase): + @patch( + "requests.sessions.Session.get", + return_value=Response(content={"status": "pass", "version": "2"}), + ) def test_api_status(self, *_) -> None: api = pyixapi.api(host, *def_args) self.assertEqual(api.health()["status"], "pass") + + @patch( + "requests.sessions.Session.get", + return_value=Response(status_code=404, ok=False, url=host, text="Not found"), + ) + def test_health_v1_returns_empty_dict(self, *_) -> None: + with warnings.catch_warnings(record=True): + warnings.simplefilter("always") + api = pyixapi.api(host, *def_args) + result = api.health() + self.assertEqual(result, {}) diff --git a/tests/test_endpoint.py b/tests/test_endpoint.py new file mode 100644 index 0000000..02a0a5c --- /dev/null +++ b/tests/test_endpoint.py @@ -0,0 +1,183 @@ +import unittest +from unittest.mock import patch + +import pyixapi +from pyixapi.core.endpoint import Endpoint +from pyixapi.core.query import RequestError +from pyixapi.core.response import Record +from pyixapi.models import Connection + +from .util import Response, auth_response, def_args, host + + +class EndpointTestCase(unittest.TestCase): + @patch("requests.sessions.Session.post", return_value=auth_response()) + def setUp(self, *_) -> None: + self.api = pyixapi.api(host, *def_args) + self.api.authenticate() + + def test_str(self) -> None: + endpoint = Endpoint(self.api, "connections") + self.assertEqual(str(endpoint), f"{host}connections") + + @patch( + "requests.sessions.Session.get", + return_value=Response( + content=[ + {"id": "CONN-001", "name": "Connection 1"}, + {"id": "CONN-002", "name": "Connection 2"}, + ] + ), + ) + def test_all(self, *_) -> None: + result = list(self.api.connections.all()) + self.assertEqual(len(result), 2) + self.assertEqual(result[0].id, "CONN-001") + self.assertEqual(result[1].id, "CONN-002") + + @patch( + "requests.sessions.Session.get", + return_value=Response( + content=[ + {"id": "CONN-001", "name": "Connection 1"}, + ] + ), + ) + def test_filter(self, *_) -> None: + result = list(self.api.connections.filter(name="Connection 1")) + self.assertEqual(len(result), 1) + self.assertEqual(result[0].id, "CONN-001") + + @patch( + "requests.sessions.Session.get", + return_value=Response( + content=[ + {"id": "CONN-001", "name": "Connection 1"}, + {"id": "CONN-002", "name": "Connection 2"}, + ] + ), + ) + def test_all_len_then_iterate(self, *_) -> None: + # Exercises the real Request.get() -> count -> RecordSet.__len__ path. + recordset = self.api.connections.all() + self.assertEqual(len(recordset), 2) + self.assertEqual([r.id for r in recordset], ["CONN-001", "CONN-002"]) + + @patch( + "requests.sessions.Session.get", + return_value=Response(content={"id": "CONN-001", "name": "Connection 1"}), + ) + def test_get_by_id(self, *_) -> None: + result = self.api.connections.get("CONN-001") + self.assertIsNotNone(result) + self.assertEqual(result.id, "CONN-001") + + @patch( + "requests.sessions.Session.get", + return_value=Response(content=[{"id": "CONN-001", "name": "Connection 1"}]), + ) + def test_get_by_id_with_single_element_list_response(self, *_) -> None: + # A by-id lookup wrapped in a one-element list still resolves to one record. + result = self.api.connections.get("CONN-001") + self.assertIsNotNone(result) + self.assertEqual(result.id, "CONN-001") + + @patch( + "requests.sessions.Session.get", + return_value=Response(content=[{"id": "CONN-001", "name": "Connection 1"}]), + ) + def test_get_by_kwargs(self, *_) -> None: + result = self.api.connections.get(name="Connection 1") + self.assertIsNotNone(result) + self.assertEqual(result.id, "CONN-001") + + @patch( + "requests.sessions.Session.get", + return_value=Response(content=[]), + ) + def test_get_not_found(self, *_) -> None: + result = self.api.connections.get(name="Nonexistent") + self.assertIsNone(result) + + @patch( + "requests.sessions.Session.get", + return_value=Response( + content=[ + {"id": "CONN-001", "name": "Connection 1"}, + {"id": "CONN-002", "name": "Connection 2"}, + ] + ), + ) + def test_get_multiple_results_raises_error(self, *_) -> None: + with self.assertRaises(ValueError) as context: + self.api.connections.get(state="active") + self.assertIn("returned more than one result", str(context.exception)) + + @patch( + "requests.sessions.Session.get", + return_value=Response(status_code=404, ok=False), + ) + def test_get_404_returns_none(self, mock_get) -> None: + mock_response = mock_get.return_value + mock_response.url = f"{host}connections/NOTFOUND" + mock_response.text = "Not Found" + result = self.api.connections.get("NOTFOUND") + self.assertIsNone(result) + + @patch( + "requests.sessions.Session.get", + return_value=Response(status_code=500, ok=False), + ) + def test_get_500_raises_error(self, mock_get) -> None: + mock_response = mock_get.return_value + mock_response.url = f"{host}connections/ERROR" + mock_response.text = "Internal Server Error" + mock_response.reason = "Internal Server Error" + with self.assertRaises(RequestError): + self.api.connections.get("ERROR") + + @patch( + "requests.sessions.Session.post", + return_value=Response(content={"id": "CONN-003", "name": "New Connection"}), + ) + def test_create_with_dict(self, *_) -> None: + result = self.api.connections.create({"name": "New Connection"}) + self.assertEqual(result.id, "CONN-003") + self.assertEqual(result.name, "New Connection") + + @patch( + "requests.sessions.Session.post", + return_value=Response(content={"id": "CONN-004", "name": "Another Connection"}), + ) + def test_create_with_kwargs(self, *_) -> None: + result = self.api.connections.create(name="Another Connection") + self.assertEqual(result.id, "CONN-004") + self.assertEqual(result.name, "Another Connection") + + def test_return_obj_defaults_to_record(self) -> None: + endpoint = Endpoint(self.api, "test-endpoint") + self.assertEqual(endpoint.return_obj, Record) + + def test_return_obj_with_model(self) -> None: + endpoint = Endpoint(self.api, "connections", model=Connection) + self.assertEqual(endpoint.return_obj, Connection) + + @patch( + "requests.sessions.Session.get", + return_value=Response( + content=[ + {"id": "CONN-001", "name": "Connection 1"}, + ] + ), + ) + def test_filter_passes_kwargs(self, mock_get) -> None: + list(self.api.connections.filter(state="active", name="Connection 1")) + call_args = mock_get.call_args + self.assertEqual(call_args[1]["params"]["state"], "active") + self.assertEqual(call_args[1]["params"]["name"], "Connection 1") + + @patch("requests.sessions.Session.get", return_value=Response(content=[{"id": "CONN-001"}])) + def test_request_carries_authenticated_token(self, mock_get) -> None: + list(self.api.connections.all()) + headers = mock_get.call_args[1]["headers"] + self.assertEqual(headers["Authorization"], f"Bearer {self.api.access_token.encoded}") diff --git a/tests/test_models.py b/tests/test_models.py new file mode 100644 index 0000000..5f39d52 --- /dev/null +++ b/tests/test_models.py @@ -0,0 +1,273 @@ +import unittest + +from pyixapi.models import ( + IP, + MAC, + Account, + AvailabilityZone, + Connection, + Contact, + Demarc, + Device, + Facility, + MemberJoiningRule, + MetroArea, + MetroAreaNetwork, + NetworkFeature, + NetworkFeatureConfig, + NetworkService, + NetworkServiceConfig, + PoP, + Port, + PortReservation, + ProductOffering, + Role, + RoleAssignment, + RoutingFunction, +) + +from .util import Response, mock_api, mock_endpoint + +# Models whose __str__ is a simple field projection; Contact/IP/MAC have real +# logic and get dedicated tests below. +STR_CASES = [ + (Account, {"id": "ACCT-001", "name": "Test Account"}, "ACCT-001: Test Account"), + (Connection, {"id": "CONN-001", "name": "Test Connection"}, "CONN-001: Test Connection"), + (Demarc, {"id": "DEM-001"}, "DEM-001"), + (Device, {"id": "DEV-001", "name": "Router01"}, "Router01"), + (Facility, {"id": "FAC-001", "name": "Data Center 1"}, "Data Center 1"), + (NetworkFeatureConfig, {"id": "NFC-001"}, "NFC-001"), + (NetworkFeature, {"id": "NF-001"}, "NF-001"), + (NetworkServiceConfig, {"id": "NSC-001"}, "NSC-001"), + (NetworkService, {"id": "NS-001"}, "NS-001"), + (PoP, {"id": "POP-001", "name": "PoP London"}, "PoP London"), + (ProductOffering, {"id": "PROD-001", "name": "Premium Service"}, "Premium Service"), + (MemberJoiningRule, {"id": "RULE-001"}, "RULE-001"), + (MetroArea, {"id": "METRO-001", "display_name": "New York Metro"}, "New York Metro"), + (MetroAreaNetwork, {"id": "MAN-001", "name": "NYC Network"}, "NYC Network"), + (Port, {"id": "PORT-001", "name": "eth0"}, "eth0"), + (PortReservation, {"id": "RES-001"}, "RES-001"), + (Role, {"id": "ROLE-001", "name": "Administrator"}, "Administrator"), + (AvailabilityZone, {"id": "AZ-001", "name": "Zone A"}, "Zone A"), + (RoutingFunction, {"id": "RF-001"}, "RF-001"), + (RoleAssignment, {"id": "ASSIGN-001"}, "ASSIGN-001"), +] + + +class ModelStrTestCase(unittest.TestCase): + def test_str(self) -> None: + for model, values, expected in STR_CASES: + with self.subTest(model=model.__name__): + obj = model(values, mock_api(), mock_endpoint()) + self.assertEqual(str(obj), expected) + + +class ContactStrTestCase(unittest.TestCase): + """Contact.__str__ joins the populated fields and falls back to id.""" + + def test_all_fields(self) -> None: + contact = Contact( + {"id": "CONT-001", "legal_company_name": "Acme Corp", "name": "John Doe", "email": "john@example.com"}, + mock_api(), + mock_endpoint(), + ) + self.assertEqual(str(contact), "Acme Corp - John Doe - john@example.com") + + def test_partial_fields(self) -> None: + contact = Contact( + {"id": "CONT-001", "name": "John Doe", "email": "john@example.com"}, mock_api(), mock_endpoint() + ) + self.assertEqual(str(contact), "John Doe - john@example.com") + + def test_fallback_to_id(self) -> None: + contact = Contact({"id": "CONT-001"}, mock_api(), mock_endpoint()) + self.assertEqual(str(contact), "CONT-001") + + +class IPTestCase(unittest.TestCase): + """IP exposes ipaddress-derived cidr/ip/network properties.""" + + def test_ipv4(self) -> None: + ip = IP({"address": "192.0.2.1", "prefix_length": 24}, mock_api(), mock_endpoint()) + self.assertEqual(str(ip.cidr), "192.0.2.1/24") + self.assertEqual(str(ip.ip), "192.0.2.1") + self.assertEqual(str(ip.network), "192.0.2.0/24") + self.assertEqual(str(ip), "192.0.2.1/24") + + def test_ipv6(self) -> None: + ip = IP({"address": "2001:db8::1", "prefix_length": 64}, mock_api(), mock_endpoint()) + self.assertEqual(str(ip.cidr), "2001:db8::1/64") + self.assertEqual(str(ip.ip), "2001:db8::1") + self.assertEqual(str(ip.network), "2001:db8::/64") + self.assertEqual(str(ip), "2001:db8::1/64") + + +class MACTestCase(unittest.TestCase): + def test_str_is_lowercased(self) -> None: + mac = MAC({"address": "00:11:22:AA:BB:CC"}, mock_api(), mock_endpoint()) + self.assertEqual(str(mac), "00:11:22:aa:bb:cc") + + +class SubResourceTestCase(unittest.TestCase): + """ + Base class for sub-resource action tests. + + The meaningful behavior of these methods is the URL and HTTP verb they + derive from the record id and sub-path; the response body is plumbed + straight through by Request, so we assert the call, not the echo. + """ + + def setUp(self) -> None: + self.api = mock_api() + + def assert_called(self, method: str, suffix: str) -> None: + mock = getattr(self.api.http_session, method) + mock.assert_called_once() + url = mock.call_args[0][0] + self.assertTrue(url.endswith(suffix), f"{url!r} does not end with {suffix!r}") + + def params(self, method: str) -> dict: + return getattr(self.api.http_session, method).call_args[1]["params"] + + +class ConnectionSubResourceTestCase(SubResourceTestCase): + def setUp(self) -> None: + super().setUp() + self.endpoint = mock_endpoint(name="connections", url="https://api.example.net/v2/connections") + self.conn = Connection({"id": "CONN-001", "name": "C"}, self.api, self.endpoint) + + def test_cancellation_policy(self) -> None: + self.conn.cancellation_policy() + self.assert_called("get", "/connections/CONN-001/cancellation-policy") + + def test_get_loa(self) -> None: + self.conn.get_loa() + self.assert_called("get", "/connections/CONN-001/loa") + + def test_upload_loa_posts(self) -> None: + self.conn.upload_loa({"file": "base64data"}) + self.assert_called("post", "/connections/CONN-001/loa") + + def test_statistics(self) -> None: + self.conn.statistics() + self.assert_called("get", "/connections/CONN-001/statistics") + + def test_statistics_forwards_kwargs_as_params(self) -> None: + self.conn.statistics(from_date="2024-01-01", to_date="2024-12-31") + self.assertEqual(self.params("get"), {"from_date": "2024-01-01", "to_date": "2024-12-31"}) + + def test_statistics_timeseries_builds_aggregate_path(self) -> None: + self.conn.statistics_timeseries("5m", from_date="2024-01-01") + self.assert_called("get", "/connections/CONN-001/statistics/5m/timeseries") + self.assertEqual(self.params("get"), {"from_date": "2024-01-01"}) + + +class NetworkServiceSubResourceTestCase(SubResourceTestCase): + def setUp(self) -> None: + super().setUp() + self.endpoint = mock_endpoint(name="network-services", url="https://api.example.net/v2/network-services") + self.ns = NetworkService({"id": "NS-001"}, self.api, self.endpoint) + + def test_cancellation_policy(self) -> None: + self.ns.cancellation_policy() + self.assert_called("get", "/network-services/NS-001/cancellation-policy") + + def test_change_request_get(self) -> None: + self.ns.change_request() + self.assert_called("get", "/network-services/NS-001/change-request") + + def test_create_change_request_posts(self) -> None: + self.ns.create_change_request({"capacity": 1000}) + self.assert_called("post", "/network-services/NS-001/change-request") + + def test_delete_change_request_deletes(self) -> None: + self.api.http_session.delete.return_value = Response(content=None) + result = self.ns.delete_change_request() + self.assert_called("delete", "/network-services/NS-001/change-request") + self.assertTrue(result) + + def test_rtt_statistics(self) -> None: + self.ns.rtt_statistics() + self.assert_called("get", "/network-services/NS-001/rtt-statistics") + + def test_statistics(self) -> None: + self.ns.statistics() + self.assert_called("get", "/network-services/NS-001/statistics") + + def test_statistics_timeseries_builds_aggregate_path(self) -> None: + self.ns.statistics_timeseries("1h") + self.assert_called("get", "/network-services/NS-001/statistics/1h/timeseries") + + +class NetworkServiceConfigSubResourceTestCase(SubResourceTestCase): + def setUp(self) -> None: + super().setUp() + self.endpoint = mock_endpoint( + name="network-service-configs", url="https://api.example.net/v2/network-service-configs" + ) + self.nsc = NetworkServiceConfig({"id": "NSC-001"}, self.api, self.endpoint) + + def test_cancellation_policy(self) -> None: + self.nsc.cancellation_policy() + self.assert_called("get", "/network-service-configs/NSC-001/cancellation-policy") + + def test_peer_statistics(self) -> None: + self.nsc.peer_statistics() + self.assert_called("get", "/network-service-configs/NSC-001/peer-statistics") + + def test_peer_statistics_timeseries_builds_aggregate_path(self) -> None: + self.nsc.peer_statistics_timeseries("5m") + self.assert_called("get", "/network-service-configs/NSC-001/peer-statistics/5m/timeseries") + + def test_statistics(self) -> None: + self.nsc.statistics() + self.assert_called("get", "/network-service-configs/NSC-001/statistics") + + def test_statistics_timeseries_builds_aggregate_path(self) -> None: + self.nsc.statistics_timeseries("5m") + self.assert_called("get", "/network-service-configs/NSC-001/statistics/5m/timeseries") + + +class PortSubResourceTestCase(SubResourceTestCase): + def setUp(self) -> None: + super().setUp() + self.endpoint = mock_endpoint(name="ports", url="https://api.example.net/v2/ports") + self.port = Port({"id": "PORT-001", "name": "eth0"}, self.api, self.endpoint) + + def test_statistics(self) -> None: + self.port.statistics() + self.assert_called("get", "/ports/PORT-001/statistics") + + def test_statistics_timeseries_builds_aggregate_path(self) -> None: + self.port.statistics_timeseries("5m") + self.assert_called("get", "/ports/PORT-001/statistics/5m/timeseries") + + +class PortReservationSubResourceTestCase(SubResourceTestCase): + def setUp(self) -> None: + super().setUp() + self.endpoint = mock_endpoint(name="port-reservations", url="https://api.example.net/v2/port-reservations") + self.pr = PortReservation({"id": "RES-001"}, self.api, self.endpoint) + + def test_cancellation_policy(self) -> None: + self.pr.cancellation_policy() + self.assert_called("get", "/port-reservations/RES-001/cancellation-policy") + + def test_get_loa(self) -> None: + self.pr.get_loa() + self.assert_called("get", "/port-reservations/RES-001/loa") + + def test_upload_loa_posts(self) -> None: + self.pr.upload_loa({"file": "base64data"}) + self.assert_called("post", "/port-reservations/RES-001/loa") + + +class RoutingFunctionSubResourceTestCase(SubResourceTestCase): + def setUp(self) -> None: + super().setUp() + self.endpoint = mock_endpoint(name="routing-functions", url="https://api.example.net/v2/routing-functions") + self.rf = RoutingFunction({"id": "RF-001"}, self.api, self.endpoint) + + def test_cancellation_policy(self) -> None: + self.rf.cancellation_policy() + self.assert_called("get", "/routing-functions/RF-001/cancellation-policy") diff --git a/tests/test_query.py b/tests/test_query.py new file mode 100644 index 0000000..ef3671e --- /dev/null +++ b/tests/test_query.py @@ -0,0 +1,433 @@ +import json +import unittest +from unittest.mock import MagicMock + +from pyixapi.core.query import ContentError, Request, RequestError +from pyixapi.core.token import Token + +from .util import make_jwt + + +class RequestErrorTestCase(unittest.TestCase): + def test_404_error_message(self) -> None: + mock_response = MagicMock() + mock_response.status_code = 404 + mock_response.url = "https://api.example.net/v1/notfound" + mock_response.text = "Not Found" + + error = RequestError(mock_response) + self.assertEqual(error.req, mock_response) + self.assertIn("could not be found", error.message) + self.assertIn("notfound", str(error)) + + def test_401_error_message(self) -> None: + mock_response = MagicMock() + mock_response.status_code = 401 + mock_response.text = "Unauthorized" + + error = RequestError(mock_response) + self.assertIn("Authentication credentials are invalid", error.message) + + def test_500_error_with_json(self) -> None: + mock_response = MagicMock() + mock_response.status_code = 500 + mock_response.reason = "Internal Server Error" + mock_response.json.return_value = {"error": "Database connection failed"} + + error = RequestError(mock_response) + self.assertIn("500", error.message) + self.assertIn("Database connection failed", error.message) + + def test_500_error_without_json(self) -> None: + mock_response = MagicMock() + mock_response.status_code = 500 + mock_response.reason = "Internal Server Error" + mock_response.json.side_effect = ValueError("Invalid JSON") + + error = RequestError(mock_response) + self.assertIn("details were not found as JSON", error.message) + + def test_error_attribute(self) -> None: + mock_response = MagicMock() + mock_response.status_code = 404 + mock_response.url = "https://api.example.net/v1/notfound" + mock_response.text = "Not Found" + + error = RequestError(mock_response) + self.assertEqual(error.error, "Not Found") + + +class ContentErrorTestCase(unittest.TestCase): + def test_content_error(self) -> None: + mock_response = MagicMock() + mock_response.text = "Not JSON content" + + error = ContentError(mock_response) + self.assertEqual(error.req, mock_response) + self.assertIn("invalid (non-JSON) data", str(error)) + + +class RequestTestCase(unittest.TestCase): + def setUp(self) -> None: + self.http_session = MagicMock() + self.token = Token.from_jwt(make_jwt()) + + def test_url_construction_with_key(self) -> None: + request = Request( + base="https://api.example.net/v1/connections", + key="CONN-001", + http_session=self.http_session, + ) + self.assertEqual(request.url, "https://api.example.net/v1/connections/CONN-001") + + def test_url_construction_without_key(self) -> None: + request = Request( + base="https://api.example.net/v1/connections", + http_session=self.http_session, + ) + self.assertEqual(request.url, "https://api.example.net/v1/connections") + + def test_get_health(self) -> None: + mock_response = MagicMock() + mock_response.ok = True + mock_response.json.return_value = {"status": "pass", "version": "2"} + self.http_session.get.return_value = mock_response + + request = Request( + base="https://api.example.net/v1", + token=self.token, + http_session=self.http_session, + ) + result = request.get_health() + + self.assertEqual(result["status"], "pass") + self.assertEqual(result["version"], "2") + + def test_get_version_v2(self) -> None: + mock_response = MagicMock() + mock_response.ok = True + mock_response.json.return_value = {"status": "pass", "version": "2"} + self.http_session.get.return_value = mock_response + + request = Request( + base="https://api.example.net/v1", + http_session=self.http_session, + ) + version = request.get_version() + + self.assertEqual(version, 2) + + def test_get_version_v1_fallback(self) -> None: + mock_response = MagicMock() + mock_response.ok = False + self.http_session.get.return_value = mock_response + + request = Request( + base="https://api.example.net/v1", + http_session=self.http_session, + ) + version = request.get_version() + + self.assertEqual(version, 1) + + def test_get_with_list_response(self) -> None: + mock_response = MagicMock() + mock_response.ok = True + mock_response.json.return_value = [ + {"id": "1", "name": "Item 1"}, + {"id": "2", "name": "Item 2"}, + ] + self.http_session.get.return_value = mock_response + + request = Request( + base="https://api.example.net/v1/items", + http_session=self.http_session, + ) + result = list(request.get()) + + self.assertEqual(len(result), 2) + self.assertEqual(request.count, 2) + + def test_get_with_single_response(self) -> None: + mock_response = MagicMock() + mock_response.ok = True + mock_response.json.return_value = {"id": "1", "name": "Item 1"} + self.http_session.get.return_value = mock_response + + request = Request( + base="https://api.example.net/v1/items/1", + http_session=self.http_session, + ) + result = list(request.get()) + + self.assertEqual(len(result), 1) + # A single object counts as one result, not one-per-field. + self.assertEqual(request.count, 1) + + def test_get_with_filters(self) -> None: + mock_response = MagicMock() + mock_response.ok = True + mock_response.json.return_value = [] + self.http_session.get.return_value = mock_response + + request = Request( + base="https://api.example.net/v1/items", + filters={"name": "test", "state": "active"}, + http_session=self.http_session, + ) + list(request.get()) + + call_args = self.http_session.get.call_args + self.assertEqual(call_args[1]["params"]["name"], "test") + self.assertEqual(call_args[1]["params"]["state"], "active") + + def test_write_verbs_send_url_verb_and_body(self) -> None: + url = "https://api.example.net/v1/items/ITEM-001" + cases = [ + ("post", lambda r, d: r.post(d)), + ("put", lambda r, d: r.put(d)), + ("patch", lambda r, d: r.patch(d)), + ] + for verb, call in cases: + with self.subTest(verb=verb): + mock_response = MagicMock() + mock_response.ok = True + mock_response.json.return_value = {"id": "X"} + getattr(self.http_session, verb).return_value = mock_response + + request = Request(base=url, http_session=self.http_session) + data = {"name": "Item"} + result = call(request, data) + + mock = getattr(self.http_session, verb) + self.assertEqual(mock.call_args[0][0], url) + self.assertEqual(mock.call_args[1]["json"], data) + self.assertEqual(result, {"id": "X"}) + + def test_delete_success(self) -> None: + mock_response = MagicMock() + mock_response.ok = True + self.http_session.delete.return_value = mock_response + + request = Request( + base="https://api.example.net/v1/items/ITEM-001", + http_session=self.http_session, + ) + result = request.delete() + + self.assertTrue(result) + # A bodyless DELETE must not advertise a JSON request body. + headers = self.http_session.delete.call_args[1]["headers"] + self.assertNotIn("Content-Type", headers) + self.assertIn("accept", headers) + + def test_delete_with_data(self) -> None: + mock_response = MagicMock() + mock_response.ok = True + self.http_session.delete.return_value = mock_response + + request = Request( + base="https://api.example.net/v1/items", + http_session=self.http_session, + ) + result = request.delete({"ids": ["ITEM-001", "ITEM-002"]}) + + self.assertTrue(result) + # A DELETE carrying a body sets the JSON content type. + headers = self.http_session.delete.call_args[1]["headers"] + self.assertEqual(headers["Content-Type"], "application/json;") + + def test_options(self) -> None: + mock_response = MagicMock() + mock_response.ok = True + mock_response.json.return_value = { + "name": "Item List", + "actions": {"POST": {}, "GET": {}}, + } + self.http_session.options.return_value = mock_response + + request = Request( + base="https://api.example.net/v1/items", + http_session=self.http_session, + ) + result = request.options() + + self.assertIn("actions", result) + + def test_make_call_with_content_error(self) -> None: + mock_response = MagicMock() + mock_response.ok = True + mock_response.json.side_effect = json.JSONDecodeError("Error", "", 0) + self.http_session.get.return_value = mock_response + + request = Request( + base="https://api.example.net/v1/items", + http_session=self.http_session, + ) + + with self.assertRaises(ContentError): + list(request.get()) + + def test_make_call_with_token(self) -> None: + mock_response = MagicMock() + mock_response.ok = True + mock_response.json.return_value = [] + self.http_session.get.return_value = mock_response + + request = Request( + base="https://api.example.net/v1/items", + token=self.token, + http_session=self.http_session, + ) + list(request.get()) + + call_args = self.http_session.get.call_args + self.assertIn("Authorization", call_args[1]["headers"]) + self.assertTrue(call_args[1]["headers"]["Authorization"].startswith("Bearer ")) + + def test_make_call_with_user_agent(self) -> None: + mock_response = MagicMock() + mock_response.ok = True + mock_response.json.return_value = [] + self.http_session.get.return_value = mock_response + + request = Request( + base="https://api.example.net/v1/items", + http_session=self.http_session, + user_agent="TestAgent/1.0", + ) + list(request.get()) + + call_args = self.http_session.get.call_args + self.assertEqual(call_args[1]["headers"]["User-Agent"], "TestAgent/1.0") + + def test_make_call_with_proxies(self) -> None: + mock_response = MagicMock() + mock_response.ok = True + mock_response.json.return_value = [] + self.http_session.get.return_value = mock_response + + proxies = {"http": "http://proxy:8080", "https": "https://proxy:8443"} + request = Request( + base="https://api.example.net/v1/items", + http_session=self.http_session, + proxies=proxies, + ) + list(request.get()) + + call_args = self.http_session.get.call_args + self.assertEqual(call_args[1]["proxies"], proxies) + + def test_get_health_failure(self) -> None: + mock_response = MagicMock() + mock_response.ok = False + mock_response.status_code = 503 + mock_response.reason = "Service Unavailable" + self.http_session.get.return_value = mock_response + + request = Request( + base="https://api.example.net/v1", + http_session=self.http_session, + ) + + with self.assertRaises(RequestError): + request.get_health() + + def test_get_health_without_token(self) -> None: + mock_response = MagicMock() + mock_response.ok = True + mock_response.json.return_value = {"status": "pass", "version": "2"} + self.http_session.get.return_value = mock_response + + request = Request( + base="https://api.example.net/v1", + http_session=self.http_session, + ) + request.get_health() + + call_args = self.http_session.get.call_args + self.assertNotIn("Authorization", call_args[1]["headers"]) + + def test_make_call_with_url_override(self) -> None: + mock_response = MagicMock() + mock_response.ok = True + mock_response.json.return_value = {"id": "1"} + self.http_session.get.return_value = mock_response + + request = Request( + base="https://api.example.net/v1/items", + filters={"name": "test"}, + http_session=self.http_session, + ) + result = request._make_call(url_override="https://api.example.net/v1/other") + self.assertEqual(result, {"id": "1"}) + + call_args = self.http_session.get.call_args + self.assertEqual(call_args[0][0], "https://api.example.net/v1/other") + # When url_override is used, params should be empty (filters ignored) + self.assertEqual(call_args[1]["params"], {}) + + def test_get_with_add_params(self) -> None: + mock_response = MagicMock() + mock_response.ok = True + mock_response.json.return_value = [{"id": "1"}] + self.http_session.get.return_value = mock_response + + request = Request( + base="https://api.example.net/v1/items", + http_session=self.http_session, + ) + list(request.get(add_params={"from": "2024-01-01"})) + + call_args = self.http_session.get.call_args + self.assertEqual(call_args[1]["params"]["from"], "2024-01-01") + + def test_verb_failures_raise_request_error(self) -> None: + cases = [ + ("post", lambda r: r.post({"name": "Invalid"})), + ("put", lambda r: r.put({"name": "Invalid"})), + ("patch", lambda r: r.patch({"name": "Invalid"})), + ("delete", lambda r: r.delete()), + ("options", lambda r: r.options()), + ] + for verb, call in cases: + with self.subTest(verb=verb): + mock_response = MagicMock() + mock_response.ok = False + mock_response.status_code = 400 + mock_response.reason = "Bad Request" + getattr(self.http_session, verb).return_value = mock_response + + request = Request(base="https://api.example.net/v1/items", http_session=self.http_session) + with self.assertRaises(RequestError): + call(request) + + def test_post_sets_content_type_header(self) -> None: + mock_response = MagicMock() + mock_response.ok = True + mock_response.json.return_value = {"id": "NEW-001"} + self.http_session.post.return_value = mock_response + + request = Request( + base="https://api.example.net/v1/items", + http_session=self.http_session, + ) + request.post({"name": "Item"}) + + call_args = self.http_session.post.call_args + self.assertEqual(call_args[1]["headers"]["Content-Type"], "application/json;") + + def test_get_sets_accept_header(self) -> None: + mock_response = MagicMock() + mock_response.ok = True + mock_response.json.return_value = [] + self.http_session.get.return_value = mock_response + + request = Request( + base="https://api.example.net/v1/items", + http_session=self.http_session, + ) + list(request.get()) + + call_args = self.http_session.get.call_args + self.assertEqual(call_args[1]["headers"]["accept"], "application/json;") diff --git a/tests/test_response.py b/tests/test_response.py new file mode 100644 index 0000000..c5150d5 --- /dev/null +++ b/tests/test_response.py @@ -0,0 +1,407 @@ +import unittest +from typing import TYPE_CHECKING, cast +from unittest.mock import MagicMock + +from pyixapi.core.response import Record, RecordSet, get_return + +from .util import Response, mock_api, mock_endpoint + +if TYPE_CHECKING: + from pyixapi.core.query import Request + + +class GetReturnTestCase(unittest.TestCase): + def test_get_return_with_record(self) -> None: + record = Record({"id": "TEST-001"}, mock_api(), mock_endpoint()) + result = get_return(record) + self.assertEqual(result, "TEST-001") + + def test_get_return_with_non_record(self) -> None: + result = get_return("simple_string") + self.assertEqual(result, "simple_string") + + result = get_return(42) + self.assertEqual(result, 42) + + +class RecordSetTestCase(unittest.TestCase): + def setUp(self) -> None: + self.api = mock_api() + self.endpoint = mock_endpoint() + self.endpoint.api = self.api + self.endpoint.return_obj = Record + + def test_iteration(self) -> None: + request = MagicMock() + request.get.return_value = iter( + [ + {"id": "ITEM-001", "name": "Item 1"}, + {"id": "ITEM-002", "name": "Item 2"}, + ] + ) + recordset = RecordSet(self.endpoint, request) + + items = list(recordset) + self.assertEqual(len(items), 2) + self.assertEqual(items[0].id, "ITEM-001") + self.assertEqual(items[1].id, "ITEM-002") + + def test_len_with_count(self) -> None: + request = MagicMock() + request.get.return_value = iter([]) + request.count = 5 + + recordset = RecordSet(self.endpoint, request) + self.assertEqual(len(recordset), 5) + + def test_len_with_deferred_count(self) -> None: + class MockRequest: + def get(self): + self.count = 1 + yield {"id": "ITEM-001"} + + request = MockRequest() + recordset = RecordSet(self.endpoint, cast("Request", request)) + + length = len(recordset) + self.assertEqual(length, 1) + + items = list(recordset) + self.assertEqual(len(items), 1) + + def test_len_empty(self) -> None: + request = MagicMock() + request.get.return_value = iter([]) + del request.count + + recordset = RecordSet(self.endpoint, request) + self.assertEqual(len(recordset), 0) + + def test_len_then_iterate_uses_cache(self) -> None: + """When len() is called first, cached items should still be iterable.""" + + class MockRequest: + def get(self): + self.count = 2 + yield {"id": "ITEM-001"} + yield {"id": "ITEM-002"} + + request = MockRequest() + recordset = RecordSet(self.endpoint, cast("Request", request)) + + length = len(recordset) + self.assertEqual(length, 2) + + items = list(recordset) + self.assertEqual(len(items), 2) + ids = {item.id for item in items} + self.assertEqual(ids, {"ITEM-001", "ITEM-002"}) + + +class RecordTestCase(unittest.TestCase): + def setUp(self) -> None: + self.api = mock_api() + self.endpoint = mock_endpoint(name="connections", url="https://api.example.net/v1/connections") + + def test_str_with_id(self) -> None: + record = Record({"id": "CONN-001"}, self.api, self.endpoint) + self.assertEqual(str(record), "CONN-001") + + def test_str_without_id(self) -> None: + record = Record({}, self.api, self.endpoint) + self.assertEqual(str(record), str(self.endpoint)) + + def test_iteration(self) -> None: + record = Record( + {"id": "CONN-001", "name": "Connection 1", "state": "active"}, + self.api, + self.endpoint, + ) + items = dict(record) + self.assertEqual(items["id"], "CONN-001") + self.assertEqual(items["name"], "Connection 1") + self.assertEqual(items["state"], "active") + + def test_getitem(self) -> None: + record = Record({"id": "CONN-001", "name": "Connection 1"}, self.api, self.endpoint) + self.assertEqual(record["id"], "CONN-001") + self.assertEqual(record["name"], "Connection 1") + + def test_missing_attribute_raises(self) -> None: + # Absent fields are not lazy-fetched. + record = Record({"id": "CONN-001"}, self.api, self.endpoint) + with self.assertRaises(AttributeError): + record.not_in_response + + def test_equality(self) -> None: + record1 = Record({"id": "CONN-001"}, self.api, self.endpoint) + record2 = Record({"id": "CONN-001"}, self.api, self.endpoint) + record3 = Record({"id": "CONN-002"}, self.api, self.endpoint) + + self.assertEqual(record1, record2) + self.assertNotEqual(record1, record3) + self.assertNotEqual(record1, "CONN-001") + + def test_hash(self) -> None: + record1 = Record({"id": "CONN-001"}, self.api, self.endpoint) + record2 = Record({"id": "CONN-001"}, self.api, self.endpoint) + + self.assertEqual(hash(record1), hash(record2)) + + records = {record1, record2} + self.assertEqual(len(records), 1) + + def test_serialize_simple(self) -> None: + record = Record( + {"id": "CONN-001", "name": "Connection 1", "capacity": 1000}, + self.api, + self.endpoint, + ) + serialized = record.serialize() + self.assertEqual(serialized["id"], "CONN-001") + self.assertEqual(serialized["name"], "Connection 1") + self.assertEqual(serialized["capacity"], 1000) + + def test_serialize_nested(self) -> None: + nested_record = Record({"id": "NESTED-001"}, self.api, self.endpoint) + record = Record({"id": "CONN-001", "nested": nested_record}, self.api, self.endpoint) + + serialized = record.serialize() + self.assertEqual(serialized["nested"], "NESTED-001") + + def test_serialize_with_list(self) -> None: + nested1 = Record({"id": "ITEM-001"}, self.api, self.endpoint) + nested2 = Record({"id": "ITEM-002"}, self.api, self.endpoint) + record = Record({"id": "CONN-001", "items": [nested1, nested2]}, self.api, self.endpoint) + + serialized = record.serialize() + self.assertEqual(serialized["items"], ["ITEM-001", "ITEM-002"]) + + def test_serialize_nested_flag(self) -> None: + record = Record({"id": "CONN-001"}, self.api, self.endpoint) + serialized = record.serialize(nested=True) + self.assertEqual(serialized, "CONN-001") + + def test_updates_no_changes(self) -> None: + record = Record({"id": "CONN-001", "name": "Connection 1"}, self.api, self.endpoint) + updates = record.updates() + self.assertEqual(updates, {}) + + def test_updates_with_changes(self) -> None: + record = Record({"id": "CONN-001", "name": "Connection 1"}, self.api, self.endpoint) + record.name = "Connection 1 Updated" + updates = record.updates() + self.assertIn("name", updates) + self.assertEqual(updates["name"], "Connection 1 Updated") + + def test_save_patches_only_changed_fields(self) -> None: + record = Record({"id": "CONN-001", "name": "Connection 1"}, self.api, self.endpoint) + record.name = "Connection 1 Updated" + + self.api.http_session.patch.return_value = Response(content={"id": "CONN-001", "name": "Updated"}) + result = record.save() + + self.assertTrue(result) + self.api.http_session.patch.assert_called_once() + self.assertEqual(self.api.http_session.patch.call_args[1]["json"], {"name": "Connection 1 Updated"}) + + def test_save_refreshes_from_response(self) -> None: + record = Record({"id": "CONN-001", "name": "Connection 1"}, self.api, self.endpoint) + record.name = "Connection 1 Updated" + + self.api.http_session.patch.return_value = Response( + content={"id": "CONN-001", "name": "Normalised", "state": "production"} + ) + result = record.save() + self.assertTrue(result) + self.assertEqual(record.name, "Normalised") + self.assertEqual(record.state, "production") + self.assertEqual(record.updates(), {}) + + def test_save_no_changes(self) -> None: + record = Record({"id": "CONN-001", "name": "Connection 1"}, self.api, self.endpoint) + result = record.save() + self.assertFalse(result) + + def test_save_with_non_dict_response_skips_refresh(self) -> None: + record = Record({"id": "CONN-001", "name": "Original"}, self.api, self.endpoint) + record.name = "Modified" + + self.api.http_session.patch.return_value = Response(content=["unexpected"]) + result = record.save() + + self.assertTrue(result) + self.assertEqual(record.name, "Modified") + + def test_update(self) -> None: + record = Record({"id": "CONN-001", "name": "Connection 1"}, self.api, self.endpoint) + + self.api.http_session.patch.return_value = Response(content={"id": "CONN-001", "name": "New Name"}) + result = record.update({"name": "New Name"}) + + self.assertTrue(result) + self.api.http_session.patch.assert_called_once() + self.assertEqual(self.api.http_session.patch.call_args[1]["json"], {"name": "New Name"}) + + def test_delete(self) -> None: + record = Record({"id": "CONN-001"}, self.api, self.endpoint) + + self.api.http_session.delete.return_value = Response(content=None) + result = record.delete() + + self.assertTrue(result) + self.assertTrue(self.api.http_session.delete.call_args[0][0].endswith("/connections/CONN-001")) + + def test_repr(self) -> None: + record = Record({"id": "CONN-001", "name": "Connection 1"}, self.api, self.endpoint) + repr_str = repr(record) + self.assertIn("CONN-001", repr_str) + self.assertIn("Connection 1", repr_str) + + def test_iter_serializes_nested_record(self) -> None: + class WithNested(Record): + nested = Record + + record = WithNested( + {"id": "CONN-001", "nested": {"id": "NESTED-001", "name": "Nested"}}, self.api, self.endpoint + ) + items = dict(record) + self.assertIsInstance(items["nested"], dict) + self.assertEqual(items["nested"]["id"], "NESTED-001") + + def test_iter_serializes_list_of_records(self) -> None: + record = Record({"id": "CONN-001", "items": [{"id": "ITEM-001"}, {"id": "ITEM-002"}]}, self.api, self.endpoint) + items = dict(record) + self.assertIsInstance(items["items"], list) + self.assertEqual([i["id"] for i in items["items"]], ["ITEM-001", "ITEM-002"]) + + def test_key_without_id(self) -> None: + record = Record({}, self.api, self.endpoint) + key = record.__key__() + self.assertEqual(key, ("connections",)) + + def test_key_with_id(self) -> None: + record = Record({"id": "CONN-001"}, self.api, self.endpoint) + key = record.__key__() + self.assertEqual(key, ("connections", "CONN-001")) + + def test_parse_values_dict_with_class_attribute_model(self) -> None: + class CustomRecord(Record): + nested = Record + + record = CustomRecord( + {"id": "CONN-001", "nested": {"id": "N-001", "name": "Nested"}}, + self.api, + self.endpoint, + ) + self.assertIsInstance(record.nested, Record) + self.assertEqual(record.nested.id, "N-001") + + def test_parse_values_list_with_dicts_default_model(self) -> None: + record = Record( + {"id": "CONN-001", "children": [{"id": "C-001"}, {"id": "C-002"}]}, + self.api, + self.endpoint, + ) + self.assertEqual(len(record.children), 2) + self.assertIsInstance(record.children[0], Record) + self.assertEqual(record.children[0].id, "C-001") + + def test_parse_values_list_with_class_attribute_model(self) -> None: + class CustomRecord(Record): + children = [Record] + + record = CustomRecord( + {"id": "CONN-001", "children": [{"id": "C-001"}]}, + self.api, + self.endpoint, + ) + self.assertIsInstance(record.children[0], Record) + self.assertEqual(record.children[0].id, "C-001") + + def test_serialize_init(self) -> None: + record = Record({"id": "CONN-001", "name": "Original"}, self.api, self.endpoint) + record.name = "Modified" + + serialized = record.serialize(init=True) + self.assertEqual(serialized["name"], "Original") + + serialized = record.serialize() + self.assertEqual(serialized["name"], "Modified") + + def test_serialize_init_with_nested_record(self) -> None: + class WithNested(Record): + nested = Record + + record = WithNested({"id": "CONN-001", "nested": {"id": "N-001"}}, self.api, self.endpoint) + # setattr to satisfy the type checker (nested is a class attr). + setattr(record, "nested", Record({"id": "N-002"}, self.api, self.endpoint)) + + self.assertEqual(record.serialize(init=True)["nested"], "N-001") + self.assertEqual(record.serialize()["nested"], "N-002") + + def test_diff_with_dict_values(self) -> None: + record = Record( + {"id": "CONN-001", "meta": {"key": "value"}}, + self.api, + self.endpoint, + ) + record.meta = {"key": "changed"} + + diff = record._diff() + self.assertIn("meta", diff) + + def test_updates_detects_nested_dict_value_change(self) -> None: + # Value-only change in a nested dict (Hashabledict hashes keys only). + record = Record({"id": "CONN-001", "meta": {"vlan": 10, "label": "primary"}}, self.api, self.endpoint) + record.meta = {"vlan": 20, "label": "primary"} + + updates = record.updates() + self.assertIn("meta", updates) + self.assertEqual(updates["meta"], {"vlan": 20, "label": "primary"}) + + def test_diff_distinguishes_list_with_comma_element(self) -> None: + # Lists compare structurally, so ["a,b"] differs from ["a", "b"]. + record = Record({"id": "CONN-001", "tags": ["a,b"]}, self.api, self.endpoint) + record.tags = ["a", "b"] + self.assertIn("tags", record._diff()) + + def test_diff_with_list_values(self) -> None: + record = Record( + {"id": "CONN-001", "tags": ["a", "b"]}, + self.api, + self.endpoint, + ) + record.tags = ["a", "c"] + + diff = record._diff() + self.assertIn("tags", diff) + + def test_updates_with_falsy_id(self) -> None: + record = Record({"id": "", "name": "Test"}, self.api, self.endpoint) + record.name = "Modified" + updates = record.updates() + self.assertEqual(updates, {}) + + def test_save_patch_returns_falsy(self) -> None: + record = Record({"id": "CONN-001", "name": "Original"}, self.api, self.endpoint) + record.name = "Modified" + + self.api.http_session.patch.return_value = Response(content={}) + result = record.save() + self.assertFalse(result) + + def test_update_no_changes(self) -> None: + record = Record({"id": "CONN-001", "name": "Connection 1"}, self.api, self.endpoint) + result = record.update({"name": "Connection 1"}) + self.assertFalse(result) + + def test_make_request_builds_correct_url(self) -> None: + self.api.user_agent = "test/1.0" + + record = Record({"id": "CONN-001"}, self.api, self.endpoint) + request = record._make_request("statistics") + + self.assertIn("CONN-001", request.url) + self.assertIn("statistics", request.url) + self.assertIn("connections", request.url) diff --git a/tests/test_token.py b/tests/test_token.py new file mode 100644 index 0000000..ba22ca3 --- /dev/null +++ b/tests/test_token.py @@ -0,0 +1,67 @@ +import unittest +import warnings +from datetime import datetime, timedelta, timezone + +from pyixapi.core.token import InvalidTokenException, Token, TokenException + +from .util import sample_jwt, sample_jwt_exp + + +class TokenTestCase(unittest.TestCase): + def test_from_jwt_valid(self) -> None: + token = Token.from_jwt(sample_jwt) + + self.assertEqual(token.encoded, sample_jwt) + self.assertEqual(token.expires_at, datetime.fromtimestamp(sample_jwt_exp, tz=timezone.utc)) + + def test_from_jwt_invalid_format(self) -> None: + with self.assertRaises(TokenException): + Token.from_jwt("not-a-valid-jwt") + + def test_from_jwt_missing_exp(self) -> None: + token_str = ( + "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9." + "eyJpc3MiOiJPbmxpbmUgSldUIEJ1aWxkZXIifQ.m5rN8mF8H3fC9CYWJLQEWh0m5J0wQ4H1qv-LrK9Cq6M" + ) + with self.assertRaises(InvalidTokenException): + Token.from_jwt(token_str) + + def test_repr(self) -> None: + token = Token.from_jwt(sample_jwt) + repr_str = repr(token) + self.assertIn("Token", repr_str) + self.assertIn("ttl=", repr_str) + + def test_ttl_future(self) -> None: + future_time = datetime.now(timezone.utc) + timedelta(hours=1) + token = Token("test_token", future_time) + + self.assertGreater(token.ttl, 3500) + self.assertLess(token.ttl, 3700) + + def test_ttl_past(self) -> None: + past_time = datetime.now(timezone.utc) - timedelta(hours=1) + token = Token("test_token", past_time) + + self.assertEqual(token.ttl, 0) + + def test_is_expired_false(self) -> None: + future_time = datetime.now(timezone.utc) + timedelta(hours=1) + token = Token("test_token", future_time) + self.assertFalse(token.is_expired) + + def test_is_expired_true(self) -> None: + past_time = datetime.now(timezone.utc) - timedelta(hours=1) + token = Token("test_token", past_time) + self.assertTrue(token.is_expired) + + def test_issued_at_deprecated(self) -> None: + token = Token.from_jwt(sample_jwt) + + with warnings.catch_warnings(record=True) as w: + warnings.simplefilter("always") + issued = token.issued_at + self.assertIsInstance(issued, datetime) + self.assertTrue(len(w) > 0) + self.assertTrue(issubclass(w[0].category, DeprecationWarning)) + self.assertIn("deprecated", str(w[0].message)) diff --git a/tests/test_util.py b/tests/test_util.py new file mode 100644 index 0000000..38350a6 --- /dev/null +++ b/tests/test_util.py @@ -0,0 +1,37 @@ +import unittest + +from pyixapi.core.util import Hashabledict, cat + + +class HashabledictTestCase(unittest.TestCase): + def test_equal_dicts_share_hash(self) -> None: + self.assertEqual(hash(Hashabledict({"k": "v"})), hash(Hashabledict({"k": "v"}))) + + def test_hash_is_keys_only_but_equality_is_by_value(self) -> None: + # Hash depends on keys only; value differences are caught by __eq__, so + # same-keys/different-values dicts remain distinct set members. + a = Hashabledict({"k": "v1"}) + b = Hashabledict({"k": "v2"}) + self.assertEqual(hash(a), hash(b)) + self.assertNotEqual(a, b) + self.assertEqual(len({a, b}), 2) + + +class CatTestCase(unittest.TestCase): + def test_basic_concatenation(self) -> None: + self.assertEqual(cat("a", "b", "c"), "a/b/c") + + def test_coerces_non_strings(self) -> None: + self.assertEqual(cat("a", 1, "b", 2), "a/1/b/2") + + def test_edge_separators_stripped(self) -> None: + self.assertEqual(cat("a", "/b/", "c/"), "a/b/c") + + def test_empty_parts_skipped(self) -> None: + self.assertEqual(cat("a", "", "b"), "a/b") + + def test_custom_separator(self) -> None: + self.assertEqual(cat("a", "b", "c", separator="_"), "a_b_c") + + def test_custom_trailing(self) -> None: + self.assertEqual(cat("a", "b", "c", trailing="/"), "a/b/c/") diff --git a/tests/util.py b/tests/util.py index 506fd4b..7fa9860 100644 --- a/tests/util.py +++ b/tests/util.py @@ -1,30 +1,92 @@ import json -from pathlib import Path +from datetime import datetime, timedelta, timezone from typing import Any +from unittest.mock import MagicMock + +import jwt + +host = "https://api.example.net/v1/" +def_args = ( + "-GnNlMD8hBuxSSUJmpbfUkss9dyOKfTV1SnZibNyyr4", + "XKq8M6NVh5lCbPJ2Ml1h7V93QNIMsGVBfM6g2nRZF-E", +) + +# JWT with a fixed exp claim, for asserting the decoded expiry. +sample_jwt = "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiJPbmxpbmUgSldUIEJ1aWxkZXIiLCJpYXQiOjE2NjM5NjE0MDAsImV4cCI6MTY2Mzk2MjMwMCwiYXVkIjoid3d3LmV4YW1wbGUuY29tIiwic3ViIjoidGVzdEBleGFtcGxlLmNvbSJ9.jPZ4TMh9mFGu2Ebgz-UO3bd5wrAJDyOjJaul1tq0AoI" # noqa: E501 +sample_jwt_exp = 1663962300 class Response(object): def __init__( self, - fixture: str | None, status_code: int = 200, ok: bool = True, content: Any = None, + url: str = "", + text: str = "", + reason: str = "", ) -> None: self.status_code = status_code - - if content: - self.content = json.dumps(content) - elif fixture: - self.content = self.load_fixture(fixture) - else: - self.content = "" - + self.url = url + self.text = text + self.reason = reason + self.content = json.dumps(content) if content is not None else "" self.ok = ok - def load_fixture(self, path: str) -> str: - f = Path("tests/fixtures") / path - return f.read_text() - def json(self) -> Any: return json.loads(self.content) + + +def make_jwt(expires_in: int = 3600, issued_ago: int = 0) -> str: + """ + Build a signed JWT whose ``exp`` claim is ``expires_in`` seconds from now. + + Use a positive ``expires_in`` for a valid token and a negative one for an + already-expired token. This keeps auth tests deterministic instead of + relying on hardcoded tokens whose expiry drifts into the past. + """ + now = datetime.now(timezone.utc) + payload = { + "iat": int((now - timedelta(seconds=issued_ago)).timestamp()), + "exp": int((now + timedelta(seconds=expires_in)).timestamp()), + "sub": "test@example.com", + } + # 32+ byte key so PyJWT does not emit an InsecureKeyLengthWarning (tests only). + return jwt.encode(payload, "pyixapi-test-signing-key-not-secret", algorithm="HS256") + + +def auth_response(access_expires_in: int = 3600, refresh_expires_in: int = 86400) -> Response: + """Build a :class:`Response` mimicking the ``/auth/token`` payload.""" + return Response( + content={ + "access_token": make_jwt(access_expires_in), + "refresh_token": make_jwt(refresh_expires_in), + } + ) + + +def mock_api() -> MagicMock: + """ + Create a MagicMock API with the standard attributes pre-configured. + + The returned mock has a mock http_session suitable for verifying HTTP calls + made by Request._make_call() without patching the Request class itself. + """ + api = MagicMock() + api.http_session = MagicMock() + api.access_token = MagicMock() + api.access_token.encoded = "fake-token" + api.user_agent = "pyixapi/test" + api.proxies = None + return api + + +def mock_endpoint( + name: str = "test", + url: str = "https://api.example.net/v2/test", +) -> MagicMock: + """Create a MagicMock Endpoint with name and url attributes.""" + endpoint = MagicMock() + endpoint.name = name + endpoint.url = url + return endpoint