[client] add JWT auth support

Author: capcom6

android_sms_gateway/client.py

@@ -15,17 +15,21 @@
 class BaseClient(abc.ABC):
     def __init__(
         self,
-        login: str,
-        password: str,
+        login: t.Optional[str],
+        password_or_token: str,
         *,
         base_url: str = DEFAULT_URL,
         encryptor: t.Optional[BaseEncryptor] = None,
     ) -> None:
-        credentials = base64.b64encode(f"{login}:{password}".encode("utf-8")).decode(
-            "utf-8"
-        )
+        if login and password_or_token:
+            auth_header = f"Basic {base64.b64encode(f'{login}:{password_or_token}'.encode()).decode()}"
+        elif password_or_token:
+            auth_header = f"Bearer {password_or_token}"
+        else:
+            raise ValueError("Either login and password or token must be provided")
+
         self.headers = {
-            "Authorization": f"Basic {credentials}",
+            "Authorization": auth_header,
             "Content-Type": "application/json",
             "User-Agent": f"android-sms-gateway/{VERSION} (client; python {sys.version_info.major}.{sys.version_info.minor})",
         }
@@ -87,14 +91,19 @@ def _decrypt(self, state: domain.MessageState) -> domain.MessageState:
 class APIClient(BaseClient):
     def __init__(
         self,
-        login: str,
-        password: str,
+        login: t.Optional[str],
+        password_or_token: str,
         *,
         base_url: str = DEFAULT_URL,
         encryptor: t.Optional[BaseEncryptor] = None,
         http: t.Optional[http.HttpClient] = None,
     ) -> None:
-        super().__init__(login, password, base_url=base_url, encryptor=encryptor)
+        super().__init__(
+            login=login,
+            password_or_token=password_or_token,
+            base_url=base_url,
+            encryptor=encryptor,
+        )
         self.http = http
         self.default_http = None
 
@@ -221,14 +230,19 @@ def health_check(self) -> dict:
 class AsyncAPIClient(BaseClient):
     def __init__(
         self,
-        login: str,
-        password: str,
+        login: t.Optional[str],
+        password_or_token: str,
         *,
         base_url: str = DEFAULT_URL,
         encryptor: t.Optional[BaseEncryptor] = None,
         http_client: t.Optional[ahttp.AsyncHttpClient] = None,
     ) -> None:
-        super().__init__(login, password, base_url=base_url, encryptor=encryptor)
+        super().__init__(
+            login=login,
+            password_or_token=password_or_token,
+            base_url=base_url,
+            encryptor=encryptor,
+        )
         self.http = http_client
         self.default_http = None
 
@@ -244,7 +258,7 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
         if self.default_http is None:
             return
 
-        await self.default_http.__aexit__(exc_type, exc_val, exc_tb)
+        self.default_http.__aexit__(exc_type, exc_val, exc_tb)
         self.http = self.default_http = None
 
     async def send(self, message: domain.Message) -> domain.MessageState:

tests/test_jwt_auth.py

@@ -0,0 +1,95 @@
+import pytest
+from android_sms_gateway.client import APIClient, AsyncAPIClient
+from android_sms_gateway.constants import DEFAULT_URL
+from android_sms_gateway.http import RequestsHttpClient
+
+
+def test_basic_auth_initialization():
+    """Test that the client can be initialized with Basic Auth (backward compatibility)."""
+    with (
+        RequestsHttpClient() as h,
+        APIClient(
+            "test_login",
+            "test_password",
+            base_url=DEFAULT_URL,
+            http=h,
+        ) as client,
+    ):
+        # Check that the Authorization header is set correctly
+        assert "Authorization" in client.headers
+        assert client.headers["Authorization"].startswith("Basic ")
+        assert client.headers["Content-Type"] == "application/json"
+        assert "User-Agent" in client.headers
+
+
+def test_jwt_auth_initialization():
+    """Test that the client can be initialized with JWT token."""
+    with (
+        RequestsHttpClient() as h,
+        APIClient(
+            login=None,
+            password_or_token="test_jwt_token",
+            base_url=DEFAULT_URL,
+            http=h,
+        ) as client,
+    ):
+        # Check that the Authorization header is set correctly
+        assert "Authorization" in client.headers
+        assert client.headers["Authorization"] == "Bearer test_jwt_token"
+        assert client.headers["Content-Type"] == "application/json"
+        assert "User-Agent" in client.headers
+
+
+def test_async_basic_auth_initialization():
+    """Test that the async client can be initialized with Basic Auth (backward compatibility)."""
+    client = AsyncAPIClient(
+        "test_login",
+        "test_password",
+        base_url=DEFAULT_URL,
+    )
+    # Check that the Authorization header is set correctly
+    assert "Authorization" in client.headers
+    assert client.headers["Authorization"].startswith("Basic ")
+    assert client.headers["Content-Type"] == "application/json"
+    assert "User-Agent" in client.headers
+
+
+def test_async_jwt_auth_initialization():
+    """Test that the async client can be initialized with JWT token."""
+    client = AsyncAPIClient(
+        login=None,
+        password_or_token="test_jwt_token",
+        base_url=DEFAULT_URL,
+    )
+    # Check that the Authorization header is set correctly
+    assert "Authorization" in client.headers
+    assert client.headers["Authorization"] == "Bearer test_jwt_token"
+    assert client.headers["Content-Type"] == "application/json"
+    assert "User-Agent" in client.headers
+
+
+def test_missing_credentials_error():
+    """Test that an error is raised when neither login/password nor jwt_token is provided."""
+    with pytest.raises(
+        ValueError,
+        match="Either login and password or token must be provided",
+    ):
+        APIClient(None, "")
+
+
+def test_missing_password_error():
+    """Test that an error is raised when login is provided but password is missing."""
+    with pytest.raises(
+        ValueError,
+        match="Either login and password or token must be provided",
+    ):
+        APIClient(login="test_login", password_or_token="")
+
+
+def test_async_missing_credentials_error():
+    """Test that an error is raised when neither login/password nor jwt_token is provided for async client."""
+    with pytest.raises(
+        ValueError,
+        match="Either login and password or token must be provided",
+    ):
+        AsyncAPIClient(None, "")