Added implementations for get_token_largest_accounts() and get_token_supply()

solana/rpc/api.py

@@ -874,13 +874,19 @@ def __get_token_accounts(
         args = self._get_token_accounts_args(method, pubkey, opts, commitment)
         return self._provider.make_request(*args)
 
-    def get_token_largest_accounts(self, pubkey: Union[PublicKey, str]) -> types.RPCResponse:
-        """Returns the 20 largest accounts of a particular SPL Token type (UNSTABLE)."""
-        raise NotImplementedError("get_token_largbest_accounts not implemented")
+    def get_token_largest_accounts(
+        self, pubkey: Union[PublicKey, str], commitment: Optional[Commitment] = None
+    ) -> types.RPCResponse:
+        """Returns the 20 largest accounts of a particular SPL Token type."""
+        args = self._get_token_largest_account_args(pubkey, commitment)
+        return self._provider.make_request(*args)
 
-    def get_token_supply(self, pubkey: Union[PublicKey, str]) -> types.RPCResponse:
-        """Returns the total supply of an SPL Token type(UNSTABLE)."""
-        raise NotImplementedError("get_token_supply not implemented")
+    def get_token_supply(
+        self, pubkey: Union[PublicKey, str], commitment: Optional[Commitment] = None
+    ) -> types.RPCResponse:
+        """Returns the total supply of an SPL Token type."""
+        args = self._get_token_supply_args(pubkey, commitment)
+        return self._provider.make_request(*args)
 
     def get_transaction_count(self, commitment: Optional[Commitment] = None) -> types.RPCResponse:
         """Returns the current Transaction count from the ledger.

solana/rpc/core.py

@@ -275,6 +275,16 @@ def _get_token_accounts_args(
 
         return method, pubkey, acc_opts, rpc_opts
 
+    def _get_token_largest_account_args(
+        self, pubkey: Union[str, PublicKey], commitment: Optional[Commitment]
+    ) -> Tuple[types.RPCMethod, str, Dict[str, Commitment]]:
+        return types.RPCMethod("getTokenLargestAccounts"), str(pubkey), {self._comm_key: commitment or self._commitment}
+
+    def _get_token_supply_args(
+        self, pubkey: Union[str, PublicKey], commitment: Optional[Commitment]
+    ) -> Tuple[types.RPCMethod, str, Dict[str, Commitment]]:
+        return types.RPCMethod("getTokenSupply"), str(pubkey), {self._comm_key: commitment or self._commitment}
+
     def _get_transaction_count_args(
         self, commitment: Optional[Commitment]
     ) -> Tuple[types.RPCMethod, Dict[str, Commitment]]:

tests/conftest.py

@@ -111,6 +111,12 @@ def async_stubbed_sender_cached_blockhash() -> Account:
     return Account(bytes([3] * PublicKey.LENGTH))
 
 
+@pytest.fixture(scope="session")
+def stubbed_token_addr() -> PublicKey:
+    """An arbitrary known token address."""
+    return PublicKey("DeV3CnJsrhN5TzCZQvmuQgYn1Up8nDycqmzxZJWH9DAw")
+
+
 @pytest.fixture(scope="session")
 def freeze_authority() -> Account:
     """Arbitrary known account to be used as freeze authority."""

tests/integration/test_http_client.py

@@ -490,6 +490,20 @@ def test_get_multiple_accounts(stubbed_sender, test_http_client):
     assert_valid_response(resp)
 
 
+@pytest.mark.integration
+def test_get_token_largest_accounts(stubbed_token_addr, test_http_client):
+    """Test get token largest accounts."""
+    resp = test_http_client.get_token_largest_accounts(stubbed_token_addr.public_key())
+    assert_valid_response(resp)
+
+
+@pytest.mark.integration
+def test_get_token_supply(stubbed_token_addr, test_http_client):
+    """Test get token largest accounts."""
+    resp = test_http_client.get_token_supply(stubbed_token_addr.public_key())
+    assert_valid_response(resp)
+
+
 @pytest.mark.integration
 def test_get_vote_accounts(test_http_client):
     """Test get vote accounts."""