218 lines
8.3 KiB
Python
218 lines
8.3 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import requests
|
|
|
|
from common.client import IklimClient
|
|
from common.config import Settings
|
|
from common.security import create_signed_headers
|
|
|
|
|
|
def jwt(expiration: int) -> str:
|
|
def encode(value: dict[str, object]) -> str:
|
|
return base64.urlsafe_b64encode(
|
|
json.dumps(value, separators=(",", ":")).encode()
|
|
).decode().rstrip("=")
|
|
|
|
return f"{encode({'alg': 'none'})}.{encode({'exp': expiration})}.signature"
|
|
|
|
|
|
def response(status: int, body: object, headers: dict[str, str] | None = None):
|
|
result = requests.Response()
|
|
result.status_code = status
|
|
result._content = json.dumps(body).encode()
|
|
result.headers.update(headers or {})
|
|
result.reason = "test response"
|
|
return result
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self, responses: list[requests.Response]) -> None:
|
|
self.responses = iter(responses)
|
|
self.calls: list[dict[str, object]] = []
|
|
|
|
def request(self, method, url, **kwargs):
|
|
self.calls.append({"method": method, "url": url, **kwargs})
|
|
return next(self.responses)
|
|
|
|
|
|
class ClientTest(unittest.TestCase):
|
|
@staticmethod
|
|
def settings(env_file: Path, *, max_retries: int = 0) -> Settings:
|
|
return Settings(
|
|
env_file=env_file,
|
|
base_url="https://api-test.iklim.co",
|
|
hmac_secret="test-secret",
|
|
accept_language="tr",
|
|
timeout_seconds=30,
|
|
token_expiry_skew_seconds=60,
|
|
max_retries=max_retries,
|
|
retry_base_seconds=2,
|
|
)
|
|
|
|
def test_retry_reuses_idempotency_and_regenerates_nonce(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
env_file = Path(directory) / ".env"
|
|
env_file.write_text("", encoding="utf-8")
|
|
settings = self.settings(env_file, max_retries=3)
|
|
session = FakeSession(
|
|
[response(500, {"message": "temporary"}), response(200, {"ok": True})]
|
|
)
|
|
waits: list[float] = []
|
|
with patch.dict(
|
|
os.environ,
|
|
{"IKLIM_ACCESS_TOKEN": jwt(int(time.time()) + 3600)},
|
|
clear=True,
|
|
):
|
|
client = IklimClient(settings, session=session, sleep=waits.append)
|
|
result = client.request_json("POST", "/v1/test", json_body={"x": 1})
|
|
|
|
self.assertEqual(result.data, {"ok": True})
|
|
self.assertEqual(waits, [2])
|
|
first_headers = session.calls[0]["headers"]
|
|
second_headers = session.calls[1]["headers"]
|
|
self.assertEqual(
|
|
first_headers["X-Idempotency-Key"],
|
|
second_headers["X-Idempotency-Key"],
|
|
)
|
|
self.assertNotEqual(first_headers["X-Nonce"], second_headers["X-Nonce"])
|
|
self.assertTrue(first_headers["Authorization"].startswith("Bearer "))
|
|
|
|
def test_hmac_matches_postman_formula(self) -> None:
|
|
body = '{"username":"name@domain.com","password":"password"}'
|
|
timestamp = "1752751106704"
|
|
with patch("common.security.time.time", return_value=1752751106.704), patch(
|
|
"common.security.uuid.uuid4",
|
|
return_value="684a0dca-bd6a-4056-a449-2567f9847f9c",
|
|
):
|
|
headers = create_signed_headers(
|
|
method="POST",
|
|
path_with_query="/v1/auth/login",
|
|
body=body,
|
|
secret="test-secret",
|
|
accept_language="tr",
|
|
idempotency_key="777edc03-ad49-4c17-be6b-9baf05a1b9e0",
|
|
)
|
|
expected = hmac.new(
|
|
b"test-secret",
|
|
f"POST|/v1/auth/login|{timestamp}|{body}".encode(),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
self.assertEqual(headers["X-Signature"], expected)
|
|
self.assertEqual(headers["X-Timestamp"], timestamp)
|
|
|
|
def test_unauthorized_refreshes_token_and_retries_request(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
env_file = Path(directory) / ".env"
|
|
env_file.write_text("", encoding="utf-8")
|
|
settings = self.settings(env_file)
|
|
now = int(time.time())
|
|
old_access = jwt(now + 3600)
|
|
old_refresh = jwt(now + 7200)
|
|
new_access = jwt(now + 10800)
|
|
new_refresh = jwt(now + 14400)
|
|
session = FakeSession(
|
|
[
|
|
response(401, {"message": "expired on server"}),
|
|
response(
|
|
200,
|
|
{"accessToken": new_access, "refreshToken": new_refresh},
|
|
),
|
|
response(200, {"ok": True}),
|
|
]
|
|
)
|
|
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"IKLIM_ACCESS_TOKEN": old_access,
|
|
"IKLIM_REFRESH_TOKEN": old_refresh,
|
|
},
|
|
clear=True,
|
|
):
|
|
client = IklimClient(settings, session=session, sleep=lambda _: None)
|
|
result = client.request_json("GET", "/v1/protected")
|
|
self.assertEqual(os.environ["IKLIM_ACCESS_TOKEN"], new_access)
|
|
|
|
self.assertEqual(result.data, {"ok": True})
|
|
self.assertTrue(session.calls[0]["headers"]["Authorization"].endswith(old_access))
|
|
self.assertNotIn("Authorization", session.calls[1]["headers"])
|
|
self.assertTrue(session.calls[2]["headers"]["Authorization"].endswith(new_access))
|
|
self.assertTrue(session.calls[1]["url"].endswith("/v1/auth/refresh"))
|
|
|
|
def test_expired_access_token_is_refreshed_before_request(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
env_file = Path(directory) / ".env"
|
|
env_file.write_text("", encoding="utf-8")
|
|
now = int(time.time())
|
|
new_access = jwt(now + 3600)
|
|
new_refresh = jwt(now + 7200)
|
|
session = FakeSession(
|
|
[
|
|
response(
|
|
200,
|
|
{"accessToken": new_access, "refreshToken": new_refresh},
|
|
),
|
|
response(200, {"ok": True}),
|
|
]
|
|
)
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
"IKLIM_ACCESS_TOKEN": jwt(now - 1),
|
|
"IKLIM_REFRESH_TOKEN": jwt(now + 3600),
|
|
},
|
|
clear=True,
|
|
):
|
|
client = IklimClient(
|
|
self.settings(env_file), session=session, sleep=lambda _: None
|
|
)
|
|
result = client.request_json("GET", "/v1/protected")
|
|
|
|
self.assertEqual(result.data, {"ok": True})
|
|
self.assertTrue(session.calls[0]["url"].endswith("/v1/auth/refresh"))
|
|
self.assertNotIn("Authorization", session.calls[0]["headers"])
|
|
self.assertTrue(session.calls[1]["headers"]["Authorization"].endswith(new_access))
|
|
|
|
def test_missing_tokens_trigger_automatic_login(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
env_file = Path(directory) / ".env"
|
|
env_file.write_text("", encoding="utf-8")
|
|
now = int(time.time())
|
|
access = jwt(now + 3600)
|
|
refresh = jwt(now + 7200)
|
|
session = FakeSession(
|
|
[
|
|
response(200, {"accessToken": access, "refreshToken": refresh}),
|
|
response(200, {"ok": True}),
|
|
]
|
|
)
|
|
with patch.dict(
|
|
os.environ,
|
|
{"IKLIM_USERNAME": "user@example.com", "IKLIM_PASSWORD": "secret"},
|
|
clear=True,
|
|
):
|
|
client = IklimClient(
|
|
self.settings(env_file), session=session, sleep=lambda _: None
|
|
)
|
|
result = client.request_json("GET", "/v1/protected")
|
|
|
|
self.assertEqual(result.data, {"ok": True})
|
|
self.assertTrue(session.calls[0]["url"].endswith("/v1/auth/login"))
|
|
self.assertNotIn("Authorization", session.calls[0]["headers"])
|
|
self.assertTrue(session.calls[1]["headers"]["Authorization"].endswith(access))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|