diff --git a/src/aleph/sdk/client/authenticated_http.py b/src/aleph/sdk/client/authenticated_http.py index ae4b6b04..4528a5b7 100644 --- a/src/aleph/sdk/client/authenticated_http.py +++ b/src/aleph/sdk/client/authenticated_http.py @@ -39,6 +39,7 @@ from .abstract import AuthenticatedAlephClient from .http import AlephHttpClient from .services.authenticated_port_forwarder import AuthenticatedPortForwarder +from .services.authenticated_voucher import AuthenticatedVoucher logger = logging.getLogger(__name__) @@ -86,7 +87,7 @@ async def __aenter__(self): await super().__aenter__() # Override services with authenticated versions self.port_forwarder = AuthenticatedPortForwarder(self) - + self.voucher = AuthenticatedVoucher(self) return self async def ipfs_push(self, content: Mapping) -> str: diff --git a/src/aleph/sdk/client/http.py b/src/aleph/sdk/client/http.py index a433e48d..6823178d 100644 --- a/src/aleph/sdk/client/http.py +++ b/src/aleph/sdk/client/http.py @@ -61,6 +61,7 @@ safe_getattr, ) from .abstract import AlephClient +from .services.voucher import Vouchers logger = logging.getLogger(__name__) @@ -135,6 +136,7 @@ async def __aenter__(self): self.crn = Crn(self) self.scheduler = Scheduler(self) self.instance = Instance(self) + self.voucher = Vouchers(self) return self diff --git a/src/aleph/sdk/client/services/authenticated_voucher.py b/src/aleph/sdk/client/services/authenticated_voucher.py new file mode 100644 index 00000000..48d7d73d --- /dev/null +++ b/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) diff --git a/src/aleph/sdk/client/services/voucher.py b/src/aleph/sdk/client/services/voucher.py new file mode 100644 index 00000000..30d54c3b --- /dev/null +++ b/src/aleph/sdk/client/services/voucher.py @@ -0,0 +1,164 @@ +from typing import Optional + +import aiohttp +from aiohttp import ClientResponseError +from aleph_message.models import Chain + +from aleph.sdk.conf import settings +from aleph.sdk.query.filters import PostFilter +from aleph.sdk.query.responses import Post, PostsResponse +from aleph.sdk.types import Voucher, VoucherMetadata + + +class Vouchers: + """ + This service is made to fetch voucher (SOL / EVM) + """ + + def __init__(self, client): + self._client = client + + # Utils + def _resolve_address(self, address: str) -> str: + return address # Not Authenticated client so address need to be given + + async def _fetch_voucher_update(self): + """ + Fetch the latest EVM voucher update (unfiltered). + """ + + post_filter = PostFilter( + types=["vouchers-update"], addresses=[settings.VOUCHER_SENDER] + ) + vouchers_post: PostsResponse = await self._client.get_posts( + post_filter=post_filter, page_size=1 + ) + + if not vouchers_post.posts: + return [] + + message_post: Post = vouchers_post.posts[0] + + nft_vouchers = message_post.content.get("nft_vouchers", {}) + return list(nft_vouchers.items()) # [(voucher_id, voucher_data)] + + async def _fetch_solana_voucher_list(self): + """ + Fetch full Solana voucher registry (unfiltered). + """ + try: + async with aiohttp.ClientSession() as session: + async with session.get(settings.VOUCHER_SOL_REGISTRY) as resp: + resp.raise_for_status() + return await resp.json() + except ClientResponseError: + return {} + + async def fetch_voucher_metadata( + self, metadata_id: str + ) -> Optional[VoucherMetadata]: + """ + Fetch metadata for a given voucher. + """ + url = f"https://claim.twentysix.cloud/sbt/metadata/{metadata_id}.json" + try: + async with aiohttp.ClientSession() as session: + async with session.get(url) as resp: + resp.raise_for_status() + data = await resp.json() + return VoucherMetadata.model_validate(data) + except ClientResponseError: + return None + + async def get_solana_vouchers(self, address: str) -> list[Voucher]: + """ + Fetch Solana vouchers for a specific address. + """ + resolved_address = self._resolve_address(address=address) + vouchers: list[Voucher] = [] + + registry_data = await self._fetch_solana_voucher_list() + + claimed_tickets = registry_data.get("claimed_tickets", {}) + batches = registry_data.get("batches", {}) + + for ticket_hash, ticket_data in claimed_tickets.items(): + claimer = ticket_data.get("claimer") + if claimer != resolved_address: + continue + + batch_id = ticket_data.get("batch_id") + metadata_id = None + + if str(batch_id) in batches: + metadata_id = batches[str(batch_id)].get("metadata_id") + + if metadata_id: + metadata = await self.fetch_voucher_metadata(metadata_id) + if metadata: + voucher = Voucher( + id=ticket_hash, + metadata_id=metadata_id, + name=metadata.name, + description=metadata.description, + external_url=metadata.external_url, + image=metadata.image, + icon=metadata.icon, + attributes=metadata.attributes, + ) + vouchers.append(voucher) + + return vouchers + + async def get_evm_vouchers(self, address: str) -> list[Voucher]: + """ + Retrieve vouchers specific to EVM chains for a specific address. + """ + resolved_address = self._resolve_address(address=address) + vouchers: list[Voucher] = [] + + nft_vouchers = await self._fetch_voucher_update() + for voucher_id, voucher_data in nft_vouchers: + if voucher_data.get("claimer") != resolved_address: + continue + + metadata_id = voucher_data.get("metadata_id") + metadata = await self.fetch_voucher_metadata(metadata_id) + if not metadata: + continue + + voucher = Voucher( + id=voucher_id, + metadata_id=metadata_id, + name=metadata.name, + description=metadata.description, + external_url=metadata.external_url, + image=metadata.image, + icon=metadata.icon, + attributes=metadata.attributes, + ) + vouchers.append(voucher) + return vouchers + + async def fetch_vouchers_by_chain(self, chain: Chain, address: str): + if chain == Chain.SOL: + return await self.get_solana_vouchers(address=address) + else: + return await self.get_evm_vouchers(address=address) + + async def get_vouchers(self, address: str) -> list[Voucher]: + """ + Retrieve all vouchers for the account / specific adress, across EVM and Solana chains. + """ + vouchers = [] + + # Get EVM vouchers + if address.startswith("0x") and len(address) == 42: + evm_vouchers = await self.get_evm_vouchers(address=address) + vouchers.extend(evm_vouchers) + else: + # Get Solana vouchers + solana_vouchers = await self.get_solana_vouchers(address=address) + vouchers.extend(solana_vouchers) + + return vouchers diff --git a/src/aleph/sdk/conf.py b/src/aleph/sdk/conf.py index fc852417..665edae7 100644 --- a/src/aleph/sdk/conf.py +++ b/src/aleph/sdk/conf.py @@ -92,6 +92,12 @@ class Settings(BaseSettings): ) SCHEDULER_URL: ClassVar[str] = "https://scheduler.api.aleph.cloud/" + VOUCHER_METDATA_TEMPLATE_URL: str = ( + "https://claim.twentysix.cloud/sbt/metadata/{}.json" + ) + VOUCHER_SOL_REGISTRY: str = "https://api.claim.twentysix.cloud/v1/registry/sol" + VOUCHER_SENDER: str = "0xB34f25f2c935bCA437C061547eA12851d719dEFb" + # Web3Provider settings TOKEN_DECIMALS: ClassVar[int] = 18 TX_TIMEOUT: ClassVar[int] = 60 * 3 diff --git a/src/aleph/sdk/types.py b/src/aleph/sdk/types.py index 6c1ae561..72d07e94 100644 --- a/src/aleph/sdk/types.py +++ b/src/aleph/sdk/types.py @@ -1,5 +1,6 @@ from abc import abstractmethod from datetime import datetime +from decimal import Decimal from enum import Enum from typing import Any, Dict, List, Literal, Optional, Protocol, TypeVar, Union @@ -289,3 +290,29 @@ class Ports(BaseModel): AllForwarders = RootModel[Dict[ItemHash, Ports]] + + +class VoucherAttribute(BaseModel): + value: Union[str, Decimal] + trait_type: str = Field(..., alias="trait_type") + display_type: Optional[str] = Field(None, alias="display_type") + + +class VoucherMetadata(BaseModel): + name: str + description: str + external_url: str + image: str + icon: str + attributes: list[VoucherAttribute] + + +class Voucher(BaseModel): + id: str + metadata_id: str + name: str + description: str + external_url: str + image: str + icon: str + attributes: list[VoucherAttribute] diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index 3ad0a4ad..5086703b 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -306,3 +306,65 @@ def post(self, *_args, **_kwargs): client._http_session = http_session return client + + +@pytest.fixture +def make_mock_aiohttp_session(): + def _make(mocked_json_response): + mock_response = AsyncMock() + mock_response.json.return_value = mocked_json_response + mock_response.raise_for_status.return_value = None + + session = MagicMock() + + get_cm = AsyncMock() + get_cm.__aenter__.return_value = mock_response + session.get.return_value = get_cm + + session_cm = AsyncMock() + session_cm.__aenter__.return_value = session + return session_cm + + return _make + + +# Constants needed for voucher tests +MOCK_ADDRESS = "0x1234567890123456789012345678901234567890" +MOCK_SOLANA_ADDRESS = "abcdefghijklmnopqrstuvwxyz123456789" +MOCK_METADATA_ID = "metadata123" +MOCK_VOUCHER_ID = "voucher123" +MOCK_METADATA = { + "name": "Test Voucher", + "description": "A test voucher", + "external_url": "https://example.com", + "image": "https://example.com/image.png", + "icon": "https://example.com/icon.png", + "attributes": [ + {"trait_type": "Test Trait", "value": "Test Value"}, + {"trait_type": "Numeric Trait", "value": "123", "display_type": "number"}, + ], +} + +MOCK_EVM_VOUCHER_DATA = [ + (MOCK_VOUCHER_ID, {"claimer": MOCK_ADDRESS, "metadata_id": MOCK_METADATA_ID}) +] + +MOCK_SOLANA_REGISTRY = { + "claimed_tickets": { + "solticket123": {"claimer": MOCK_SOLANA_ADDRESS, "batch_id": "batch123"} + }, + "batches": {"batch123": {"metadata_id": MOCK_METADATA_ID}}, +} + + +@pytest.fixture +def mock_post_response(): + mock_post = MagicMock() + mock_post.content = { + "nft_vouchers": { + MOCK_VOUCHER_ID: {"claimer": MOCK_ADDRESS, "metadata_id": MOCK_METADATA_ID} + } + } + posts_response = MagicMock() + posts_response.posts = [mock_post] + return posts_response diff --git a/tests/unit/services/test_authenticated_voucher.py b/tests/unit/services/test_authenticated_voucher.py new file mode 100644 index 00000000..bb83ea74 --- /dev/null +++ b/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 ..conftest 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"] diff --git a/tests/unit/services/test_voucher.py b/tests/unit/services/test_voucher.py new file mode 100644 index 00000000..4e319c25 --- /dev/null +++ b/tests/unit/services/test_voucher.py @@ -0,0 +1,112 @@ +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from aleph_message.models import Chain + +from aleph.sdk.client.services.voucher import Vouchers + +from ..conftest import ( + MOCK_ADDRESS, + MOCK_METADATA, + MOCK_SOLANA_ADDRESS, + MOCK_SOLANA_REGISTRY, + MOCK_VOUCHER_ID, +) + + +@pytest.mark.asyncio +async def test_get_evm_vouchers(mock_post_response, make_mock_aiohttp_session): + mock_client = MagicMock() + mock_client.get_posts = AsyncMock(return_value=mock_post_response) + voucher_service = Vouchers(client=mock_client) + + session = make_mock_aiohttp_session(MOCK_METADATA) + + with patch("aiohttp.ClientSession", return_value=session): + vouchers = await voucher_service.get_evm_vouchers(MOCK_ADDRESS) + + assert len(vouchers) == 1 + assert vouchers[0].id == MOCK_VOUCHER_ID + assert vouchers[0].name == MOCK_METADATA["name"] + + +@pytest.mark.asyncio +async def test_get_solana_vouchers(make_mock_aiohttp_session): + mock_client = MagicMock() + voucher_service = Vouchers(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 voucher_service.get_solana_vouchers(MOCK_SOLANA_ADDRESS) + + assert len(vouchers) == 1 + assert vouchers[0].id == "solticket123" + assert vouchers[0].name == MOCK_METADATA["name"] + + +@pytest.mark.asyncio +async def test_fetch_vouchers_by_chain_for_evm( + mock_post_response, make_mock_aiohttp_session +): + mock_client = MagicMock() + mock_client.get_posts = AsyncMock(return_value=mock_post_response) + voucher_service = Vouchers(client=mock_client) + + metadata_session = make_mock_aiohttp_session(MOCK_METADATA) + with patch("aiohttp.ClientSession", return_value=metadata_session): + vouchers = await voucher_service.fetch_vouchers_by_chain( + Chain.ETH, MOCK_ADDRESS + ) + + assert len(vouchers) == 1 + assert vouchers[0].id == "voucher123" + + +@pytest.mark.asyncio +async def test_fetch_vouchers_by_chain_for_solana(make_mock_aiohttp_session): + mock_client = MagicMock() + voucher_service = Vouchers(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 voucher_service.fetch_vouchers_by_chain( + Chain.SOL, MOCK_SOLANA_ADDRESS + ) + + assert len(vouchers) == 1 + assert vouchers[0].id == "solticket123" + + +@pytest.mark.asyncio +async def test_get_vouchers_detects_chain( + make_mock_aiohttp_session, mock_post_response +): + mock_client = MagicMock() + mock_client.get_posts = AsyncMock(return_value=mock_post_response) + voucher_service = Vouchers(client=mock_client) + + # EVM + metadata_session = make_mock_aiohttp_session(MOCK_METADATA) + with patch("aiohttp.ClientSession", return_value=metadata_session): + vouchers = await voucher_service.get_vouchers(MOCK_ADDRESS) + assert len(vouchers) == 1 + assert vouchers[0].id == "voucher123" + + # Solana + 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 voucher_service.get_vouchers(MOCK_SOLANA_ADDRESS) + assert len(vouchers) == 1 + assert vouchers[0].id == "solticket123"