Feat: AuthenticatedVoucher integrations

Author: 1yam

src/aleph/sdk/client/services/authenticated_voucher.py

@@ -0,0 +1,62 @@
+from typing import TYPE_CHECKING, Optional, overload
+
+from typing_extensions import override
+
+from aleph.sdk.types import Voucher
+
+from .voucher import Vouchers
+
+if TYPE_CHECKING:
+    from aleph.sdk.client.abstract import AuthenticatedAlephClient
+
+
+class AuthenticatedVoucher(Vouchers):
+    """
+    This service is same logic than Vouchers but allow to don't pass address
+    to use account address
+    """
+
+    def __init__(self, client: "AuthenticatedAlephClient"):
+        super().__init__(client)
+
+    @overload
+    def _resolve_address(self, address: str) -> str: ...
+
+    @overload
+    def _resolve_address(self, address: None) -> str: ...
+
+    @override
+    def _resolve_address(self, address: Optional[str] = None) -> str:
+        """
+        Resolve the address to use. Prefer the provided address, fallback to account.
+        """
+        if address:
+            return address
+        if self._client.account:
+            return self._client.account.get_address()
+
+        raise ValueError("No address provided and no account configured")
+
+    @override
+    async def get_vouchers(self, address: Optional[str] = None) -> list[Voucher]:
+        """
+        Retrieve all vouchers for the account / specific address, across EVM and Solana chains.
+        """
+        address = address or self._client.account.get_address()
+        return await super().get_vouchers(address=address)
+
+    @override
+    async def get_evm_vouchers(self, address: Optional[str] = None) -> list[Voucher]:
+        """
+        Retrieve vouchers specific to EVM chains for a specific address.
+        """
+        address = address or self._client.account.get_address()
+        return await super().get_evm_vouchers(address=address)
+
+    @override
+    async def get_solana_vouchers(self, address: Optional[str] = None) -> list[Voucher]:
+        """
+        Fetch Solana vouchers for a specific address.
+        """
+        address = address or self._client.account.get_address()
+        return await super().get_solana_vouchers(address=address)

tests/unit/services/test_authenticated_voucher.py

@@ -0,0 +1,111 @@
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from aleph.sdk.client.services.authenticated_voucher import AuthenticatedVoucher
+
+from .test_voucher import (
+    MOCK_ADDRESS,
+    MOCK_METADATA,
+    MOCK_SOLANA_ADDRESS,
+    MOCK_SOLANA_REGISTRY,
+    MOCK_VOUCHER_ID,
+)
+
+
+def test_resolve_address_with_argument():
+    client = MagicMock()
+    service = AuthenticatedVoucher(client=client)
+    assert service._resolve_address(address="custom-address") == "custom-address"
+
+
+def test_resolve_address_with_account_fallback():
+    mock_account = MagicMock()
+    mock_account.get_address.return_value = MOCK_ADDRESS
+
+    client = MagicMock()
+    client.account = mock_account
+
+    service = AuthenticatedVoucher(client=client)
+    assert service._resolve_address(address=None) == MOCK_ADDRESS
+    mock_account.get_address.assert_called_once()
+
+
+def test_resolve_address_no_address_no_account():
+    client = MagicMock()
+    client.account = None
+
+    service = AuthenticatedVoucher(client=client)
+
+    with pytest.raises(
+        ValueError, match="No address provided and no account configured"
+    ):
+        service._resolve_address(address=None)
+
+
+@pytest.mark.asyncio
+async def test_get_vouchers_fallback_to_account(
+    make_mock_aiohttp_session, mock_post_response
+):
+    mock_account = MagicMock()
+    mock_account.get_address.return_value = MOCK_ADDRESS
+
+    mock_client = MagicMock()
+    mock_client.account = mock_account
+    mock_client.get_posts = AsyncMock(return_value=mock_post_response)
+
+    service = AuthenticatedVoucher(client=mock_client)
+
+    metadata_session = make_mock_aiohttp_session(MOCK_METADATA)
+
+    with patch("aiohttp.ClientSession", return_value=metadata_session):
+        vouchers = await service.get_vouchers()
+
+    assert len(vouchers) == 1
+    assert vouchers[0].name == MOCK_METADATA["name"]
+    mock_account.get_address.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_get_evm_vouchers_fallback_to_account(
+    make_mock_aiohttp_session, mock_post_response
+):
+    mock_account = MagicMock()
+    mock_account.get_address.return_value = MOCK_ADDRESS
+
+    mock_client = MagicMock()
+    mock_client.account = mock_account
+    mock_client.get_posts = AsyncMock(return_value=mock_post_response)
+
+    service = AuthenticatedVoucher(client=mock_client)
+
+    metadata_session = make_mock_aiohttp_session(MOCK_METADATA)
+
+    with patch("aiohttp.ClientSession", return_value=metadata_session):
+        vouchers = await service.get_evm_vouchers()
+
+    assert len(vouchers) == 1
+    assert vouchers[0].id == MOCK_VOUCHER_ID
+
+
+@pytest.mark.asyncio
+async def test_get_solana_vouchers_fallback_to_account(make_mock_aiohttp_session):
+    mock_account = MagicMock()
+    mock_account.get_address.return_value = MOCK_SOLANA_ADDRESS
+
+    mock_client = MagicMock()
+    mock_client.account = mock_account
+
+    service = AuthenticatedVoucher(client=mock_client)
+
+    registry_session = make_mock_aiohttp_session(MOCK_SOLANA_REGISTRY)
+    metadata_session = make_mock_aiohttp_session(MOCK_METADATA)
+
+    with patch(
+        "aiohttp.ClientSession", side_effect=[registry_session, metadata_session]
+    ):
+        vouchers = await service.get_solana_vouchers()
+
+    assert len(vouchers) == 1
+    assert vouchers[0].id == "solticket123"
+    assert vouchers[0].name == MOCK_METADATA["name"]