hubcio commented on code in PR #4276:
URL: https://github.com/apache/iggy/pull/4276#discussion_r4237603602
##########
foreign/python/tests/test_user.py:
##########
@@ -208,6 +266,221 @@ async def test_create_user_accepts_multibyte_credentials(
await iggy_client.delete_user(created.id)
+class TestLoginIdentity:
+ """Test login returns IdentityInfo with transport-specific access_token."""
+
+ @pytest.mark.parametrize(
+ "create_client",
+ [
+ lambda:
IggyClient(f"{get_server_config()[0]}:{get_server_config()[1]}"),
+ lambda: IggyClient(
+ QuicConfig(
+ server_address=(
+
f"{get_quic_server_config()[0]}:{get_quic_server_config()[1]}"
+ ),
+ reconnection=QuicReconnectionConfig(enabled=False),
+ )
+ ),
+ lambda: IggyClient(
+ WebSocketConfig(
+ server_address=(
+ f"{get_websocket_server_config()[0]}:"
+ f"{get_websocket_server_config()[1]}"
+ ),
+ reconnection=WebSocketReconnectionConfig(enabled=False),
+ )
+ ),
+ ],
+ ids=["tcp", "quic", "websocket"],
+ )
+ @pytest.mark.asyncio
+ async def test_binary_login_returns_identity_without_access_token(
+ self, iggy_client: IggyClient, create_client: Callable[[], IggyClient]
+ ):
+ """Test binary transports return user_id and no HTTP access token."""
+ root = await iggy_client.get_user("iggy")
+ assert root is not None
+
+ client = create_client()
+ await client.connect()
+ await wait_for_ping(client)
+ identity = await client.login_user("iggy", "iggy")
+
+ assert isinstance(identity, IdentityInfo)
+ assert identity.user_id == root.id
+ assert identity.access_token is None
+
+ @pytest.mark.asyncio
+ async def test_http_login_returns_access_token(self, iggy_client:
IggyClient):
+ """Test HTTP login returns user_id and a session TokenInfo."""
+ host, port = get_http_server_config()
+ client = IggyClient(HttpConfig(api_url=f"http://{host}:{port}";))
+ await client.connect()
+ await wait_for_ping(client)
+
+ identity = await client.login_user("iggy", "iggy")
+ root = await iggy_client.get_user("iggy")
+ assert root is not None
+
+ assert isinstance(identity, IdentityInfo)
+ assert identity.user_id == root.id
+ assert identity.access_token is not None
+ assert identity.access_token.token
+ assert identity.access_token.expiry > 0
+ assert identity.access_token.token not in repr(identity)
+ assert identity.access_token.token not in repr(identity.access_token)
+
+ @pytest.mark.asyncio
+ async def test_login_with_personal_access_token(self, unique_name):
+ """Test PAT login using a token created directly through HTTP."""
+ name = unique_name(min_bytes=16, max_bytes=30)
+ http_host, http_port = get_http_server_config()
+ api = f"http://{http_host}:{http_port}";
+ login_req = urllib.request.Request( # noqa: S310
Review Comment:
simplification: this test duplicates the token setup and cleanup helpers.
reuse `_create_personal_access_token` and `_delete_personal_access_token` with
the same arguments.
##########
foreign/python/tests/test_user.py:
##########
@@ -208,6 +266,221 @@ async def test_create_user_accepts_multibyte_credentials(
await iggy_client.delete_user(created.id)
+class TestLoginIdentity:
+ """Test login returns IdentityInfo with transport-specific access_token."""
+
+ @pytest.mark.parametrize(
+ "create_client",
+ [
+ lambda:
IggyClient(f"{get_server_config()[0]}:{get_server_config()[1]}"),
+ lambda: IggyClient(
+ QuicConfig(
+ server_address=(
+
f"{get_quic_server_config()[0]}:{get_quic_server_config()[1]}"
+ ),
+ reconnection=QuicReconnectionConfig(enabled=False),
+ )
+ ),
+ lambda: IggyClient(
+ WebSocketConfig(
+ server_address=(
+ f"{get_websocket_server_config()[0]}:"
+ f"{get_websocket_server_config()[1]}"
+ ),
+ reconnection=WebSocketReconnectionConfig(enabled=False),
+ )
+ ),
+ ],
+ ids=["tcp", "quic", "websocket"],
+ )
+ @pytest.mark.asyncio
+ async def test_binary_login_returns_identity_without_access_token(
+ self, iggy_client: IggyClient, create_client: Callable[[], IggyClient]
+ ):
+ """Test binary transports return user_id and no HTTP access token."""
+ root = await iggy_client.get_user("iggy")
+ assert root is not None
+
+ client = create_client()
+ await client.connect()
+ await wait_for_ping(client)
+ identity = await client.login_user("iggy", "iggy")
+
+ assert isinstance(identity, IdentityInfo)
+ assert identity.user_id == root.id
+ assert identity.access_token is None
+
+ @pytest.mark.asyncio
+ async def test_http_login_returns_access_token(self, iggy_client:
IggyClient):
+ """Test HTTP login returns user_id and a session TokenInfo."""
+ host, port = get_http_server_config()
+ client = IggyClient(HttpConfig(api_url=f"http://{host}:{port}";))
+ await client.connect()
+ await wait_for_ping(client)
+
+ identity = await client.login_user("iggy", "iggy")
+ root = await iggy_client.get_user("iggy")
+ assert root is not None
+
+ assert isinstance(identity, IdentityInfo)
+ assert identity.user_id == root.id
+ assert identity.access_token is not None
+ assert identity.access_token.token
+ assert identity.access_token.expiry > 0
+ assert identity.access_token.token not in repr(identity)
+ assert identity.access_token.token not in repr(identity.access_token)
+
+ @pytest.mark.asyncio
+ async def test_login_with_personal_access_token(self, unique_name):
+ """Test PAT login using a token created directly through HTTP."""
+ name = unique_name(min_bytes=16, max_bytes=30)
+ http_host, http_port = get_http_server_config()
+ api = f"http://{http_host}:{http_port}";
+ login_req = urllib.request.Request( # noqa: S310
+ f"{api}/users/login",
+ data=json.dumps({"username": "iggy", "password": "iggy"}).encode(),
+ headers={"Content-Type": "application/json"},
+ method="POST",
+ )
+ with urllib.request.urlopen(login_req, timeout=10) as response: #
noqa: S310
+ jwt = json.loads(response.read())["access_token"]["token"]
+ create_req = urllib.request.Request( # noqa: S310
+ f"{api}/personal-access-tokens",
+ data=json.dumps({"name": name, "expiry": NEVER_EXPIRE}).encode(),
+ headers={
+ "Content-Type": "application/json",
+ "Authorization": f"Bearer {jwt}",
+ },
+ method="POST",
+ )
+ with urllib.request.urlopen(create_req, timeout=10) as response: #
noqa: S310
+ personal_access_token = json.loads(response.read())["token"]
+
+ try:
+ host, port = get_server_config()
+ client = IggyClient(f"{host}:{port}")
+ await client.connect()
+ identity = await client.login_with_personal_access_token(
+ personal_access_token
+ )
+
+ assert isinstance(identity, IdentityInfo)
+ assert identity.access_token is None
+ user = await client.get_user("iggy")
+ assert user is not None
+ assert identity.user_id == user.id
+ assert personal_access_token not in repr(identity)
+ finally:
+ delete_req = urllib.request.Request( # noqa: S310
+ f"{api}/personal-access-tokens/{name}",
+ headers={"Authorization": f"Bearer {jwt}"},
+ method="DELETE",
+ )
+ with urllib.request.urlopen(delete_req, timeout=10): # noqa: S310
+ pass
+
+ @pytest.mark.parametrize(
+ "token",
+ ["unknown-personal-access-token", "%%%garbled-token%%%"],
+ ids=["unknown", "garbled"],
+ )
+ @pytest.mark.asyncio
+ async def test_invalid_personal_access_token_fails(self, token):
+ """Test unknown and garbled PAT values are rejected."""
+ host, port = get_server_config()
+ client = IggyClient(f"{host}:{port}")
+ await client.connect()
+
+ with pytest.raises(RuntimeError):
+ await client.login_with_personal_access_token(token)
+
+ @pytest.mark.asyncio
+ async def test_expired_personal_access_token_fails(self, unique_name):
+ """Test an expired PAT cannot authenticate."""
+ name = unique_name(min_bytes=16, max_bytes=30)
+ http_host, http_port = get_http_server_config()
+ api = f"http://{http_host}:{http_port}";
+ jwt, personal_access_token = _create_personal_access_token(
+ api, "iggy", "iggy", name, 1
+ )
+
+ host, port = get_server_config()
+ client = IggyClient(f"{host}:{port}")
+ await client.connect()
+
+ try:
+ with pytest.raises(RuntimeError):
+ await
client.login_with_personal_access_token(personal_access_token)
+ finally:
+ _delete_personal_access_token(api, jwt, name)
Review Comment:
warning: cleanup fails with HTTP 404 if the server cleaner deletes the
expired token first. accept 404 for this cleanup and propagate other errors.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
