starred/chia-blockchain
1
0
Fork 0
mirror of https://github.com/Chia-Network/chia-blockchain.git synced 2026-09-05 02:24:21 -05:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #11690 from AmineKhaldi/merge_nft1_soon_tm_into_release

Browse Source 
Merge nft1_soon_tm into release/1.4.0
This commit is contained in:
William Allen
2022-05-31 14:45:44 -05:00
committed by GitHub
parent 84738ab89e 8e709ae0d3
commit 7b104f3212
21 changed files with 703 additions and 221 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/test-install-scripts.yml
-10
View File
@@ -197,16 +197,6 @@ jobs:
apt-get --yes update
apt-get install --yes git lsb-release sudo
# @TODO this step can be removed once Python 3.10 is supported
# Python 3.10 is now the default in bookworm, so install 3.9 specifically so install does not fail
- name: Prepare debian:bookworm
if: ${{ matrix.distribution.name == 'debian:bookworm' }}
env:
DEBIAN_FRONTEND: noninteractive
run: |
apt-get update -y
apt-get install -y python3.9-venv
- name: Prepare Fedora
if: ${{ matrix.distribution.type == 'fedora' }}
run: |
chia/cmds/wallet.py
+3 -3
View File
@@ -573,7 +573,7 @@ def nft_add_uri_cmd(
@click.option("-f", "--fingerprint", help="Set the fingerprint to specify which wallet to use", type=int)
@click.option("-i", "--id", help="Id of the NFT wallet to use", type=int, required=True)
@click.option("-ni", "--nft-coin-id", help="Id of the NFT coin to transfer", type=str, required=True)
@click.option("-aa", "--artist-address", help="Target artist's wallet address", type=str, required=True)
@click.option("-ta", "--target-address", help="Target recipient wallet address", type=str, required=True)
@click.option(
"-m",
"--fee",
@@ -588,7 +588,7 @@ def nft_transfer_cmd(
fingerprint: int,
id: int,
nft_coin_id: str,
artist_address: str,
target_address: str,
fee: str,
) -> None:
import asyncio
@@ -597,7 +597,7 @@ def nft_transfer_cmd(
extra_params = {
"wallet_id": id,
"nft_coin_id": nft_coin_id,
"artist_address": artist_address,
"target_address": target_address,
"fee": fee,
}
asyncio.run(execute_with_wallet(wallet_rpc_port, fingerprint, extra_params, transfer_nft))
chia/cmds/wallet_funcs.py
+26 -5
View File
@@ -681,11 +681,31 @@ async def create_nft_wallet(args: Dict, wallet_client: WalletRpcClient, fingerpr
async def mint_nft(args: Dict, wallet_client: WalletRpcClient, fingerprint: int) -> None:
try:
wallet_id = args["wallet_id"]
artist_address = args["artist_address"]
royalty_address = args.get("royalty_address", None)
target_address = args.get("target_address", None)
hash = args["hash"]
uris = args["uris"]
meta_hash = args.get("meta_hash", None)
meta_uris = args.get("meta_uris", None)
license_hash = args.get("license_hash", None)
license_uris = args.get("license_uris", None)
series_total = args.get("series_total", None)
series_number = args.get("series_number", None)
fee = args["fee"]
response = await wallet_client.mint_nft(wallet_id, artist_address, hash, uris, fee)
response = await wallet_client.mint_nft(
wallet_id,
royalty_address,
target_address,
hash,
uris,
meta_hash,
meta_uris,
license_hash,
license_uris,
series_total,
series_number,
fee,
)
spend_bundle = response["spend_bundle"]
print(f"NFT minted Successfully with spend bundle: {spend_bundle}")
except Exception as e:
@@ -698,7 +718,8 @@ async def add_uri_to_nft(args: Dict, wallet_client: WalletRpcClient, fingerprint
nft_coin_id = args["nft_coin_id"]
uri = args["uri"]
fee = args["fee"]
response = await wallet_client.add_uri_to_nft(wallet_id, nft_coin_id, uri, fee)
key = args.get("meta_uri", "u")
response = await wallet_client.add_uri_to_nft(wallet_id, nft_coin_id, key, uri, fee)
spend_bundle = response["spend_bundle"]
print(f"URI added successfully with spend bundle: {spend_bundle}")
except Exception as e:
@@ -709,9 +730,9 @@ async def transfer_nft(args: Dict, wallet_client: WalletRpcClient, fingerprint:
try:
wallet_id = args["wallet_id"]
nft_coin_id = args["nft_coin_id"]
artist_address = args["artist_address"]
target_address = args["target_address"]
fee = args["fee"]
response = await wallet_client.transfer_nft(wallet_id, nft_coin_id, artist_address, fee)
response = await wallet_client.transfer_nft(wallet_id, nft_coin_id, target_address, fee)
spend_bundle = response["spend_bundle"]
print(f"NFT transferred successfully with spend bundle: {spend_bundle}")
except Exception as e:
chia/rpc/wallet_rpc_api.py
+115 -13
View File
@@ -11,12 +11,14 @@ from chia.consensus.block_rewards import calculate_base_farmer_reward
from chia.pools.pool_wallet import PoolWallet
from chia.pools.pool_wallet_info import FARMING_TO_POOL, PoolState, PoolWalletInfo, create_pool_state
from chia.protocols.protocol_message_types import ProtocolMessageTypes
from chia.protocols.wallet_protocol import CoinState
from chia.server.outbound_message import NodeType, make_msg
from chia.simulator.simulator_protocol import FarmNewBlockProtocol
from chia.types.announcement import Announcement
from chia.types.blockchain_format.coin import Coin, coin_as_list
from chia.types.blockchain_format.program import Program
from chia.types.blockchain_format.sized_bytes import bytes32
from chia.types.coin_spend import CoinSpend
from chia.types.spend_bundle import SpendBundle
from chia.util.bech32m import decode_puzzle_hash, encode_puzzle_hash
from chia.util.byte_types import hexstr_to_bytes
@@ -35,8 +37,10 @@ from chia.wallet.derive_keys import (
match_address_to_sk,
)
from chia.wallet.did_wallet.did_wallet import DIDWallet
from chia.wallet.nft_wallet.nft_puzzles import get_nft_info_from_puzzle
from chia.wallet.nft_wallet.nft_wallet import NFTWallet
from chia.wallet.nft_wallet import nft_puzzles
from chia.wallet.nft_wallet.nft_info import NFTInfo
from chia.wallet.nft_wallet.nft_wallet import NFTWallet, NFTCoinInfo
from chia.wallet.nft_wallet.uncurry_nft import UncurriedNFT
from chia.wallet.outer_puzzles import AssetType
from chia.wallet.puzzle_drivers import PuzzleInfo
from chia.wallet.rl_wallet.rl_wallet import RLWallet
@@ -131,6 +135,7 @@ class WalletRpcApi:
# NFT Wallet
"/nft_mint_nft": self.nft_mint_nft,
"/nft_get_nfts": self.nft_get_nfts,
"/nft_get_info": self.nft_get_info,
"/nft_transfer_nft": self.nft_transfer_nft,
"/nft_add_uri": self.nft_add_uri,
# RL wallet
@@ -1322,22 +1327,43 @@ class WalletRpcApi:
wallet_id = uint32(request["wallet_id"])
assert self.service.wallet_state_manager
nft_wallet: NFTWallet = self.service.wallet_state_manager.wallets[wallet_id]
assert nft_wallet.type() == WalletType.NFT.value, nft_wallet.type()
address = request.get("artist_address")
if isinstance(address, str):
puzzle_hash = decode_puzzle_hash(address)
elif address is None:
puzzle_hash = await nft_wallet.standard_wallet.get_new_puzzlehash()
assert nft_wallet.type() == WalletType.NFT.value
royalty_address = request.get("royalty_address")
if isinstance(royalty_address, str):
royalty_puzhash = decode_puzzle_hash(royalty_address)
elif royalty_address is None:
royalty_puzhash = await nft_wallet.standard_wallet.get_new_puzzlehash()
else:
puzzle_hash = address
royalty_puzhash = royalty_address
target_address = request.get("target_address")
if isinstance(target_address, str):
target_puzhash = decode_puzzle_hash(target_address)
elif target_address is None:
target_puzhash = await nft_wallet.standard_wallet.get_new_puzzlehash()
else:
target_puzhash = target_address
if "uris" not in request:
return {"success": False, "error": "Data URIs is required"}
if not isinstance(request["uris"], list):
return {"success": False, "error": "Data URIs must be a list"}
if not isinstance(request.get("meta_uris", []), list):
return {"success": False, "error": "Metadata URIs must be a list"}
if not isinstance(request.get("license_uris", []), list):
return {"success": False, "error": "License URIs must be a list"}
metadata = Program.to(
[
("u", request["uris"]),
("h", hexstr_to_bytes(request["hash"])),
("mu", request.get("meta_uris", [])),
("mh", hexstr_to_bytes(request.get("meta_hash", "00"))),
("lu", request.get("license_uris", [])),
("lh", hexstr_to_bytes(request.get("license_hash", "00"))),
("sn", uint64(request.get("series_number", 1))),
("st", uint64(request.get("series_total", 1))),
]
)
fee = uint64(request.get("fee", 0))
spend_bundle = await nft_wallet.generate_new_nft(metadata, puzzle_hash, fee=fee)
spend_bundle = await nft_wallet.generate_new_nft(metadata, royalty_puzhash, target_puzhash, fee=fee)
return {"wallet_id": wallet_id, "success": True, "spend_bundle": spend_bundle}
async def nft_get_nfts(self, request) -> Dict:
@@ -1347,7 +1373,7 @@ class WalletRpcApi:
nfts = nft_wallet.get_current_nfts()
nft_info_list = []
for nft in nfts:
nft_info_list.append(get_nft_info_from_puzzle(nft.full_puzzle, nft.coin))
nft_info_list.append(nft_puzzles.get_nft_info_from_puzzle(nft))
return {"wallet_id": wallet_id, "success": True, "nft_list": nft_info_list}
async def nft_transfer_nft(self, request):
@@ -1368,15 +1394,91 @@ class WalletRpcApi:
log.exception(f"Failed to transfer NFT: {e}")
return {"success": False, "error": str(e)}
async def nft_get_info(self, request: Dict) -> Optional[Dict]:
assert self.service.wallet_state_manager is not None
if "coin_id" not in request:
return {"success": False, "error": "Coin ID is required."}
coin_id = bytes32.from_hexstr(request["coin_id"])
peer = self.service.wallet_state_manager.wallet_node.get_full_node_peer()
if peer is None:
return {"success": False, "error": "Cannot find a full node peer."}
# Get coin state
coin_state_list: List[CoinState] = await self.service.wallet_state_manager.wallet_node.get_coin_state(
[coin_id], peer=peer
)
if coin_state_list is None or len(coin_state_list) < 1:
return {"success": False, "error": f"Coin record 0x{coin_id.hex()} not found"}
coin_state: CoinState = coin_state_list[0]
if request.get("latest", True):
# Find the unspent coin
while coin_state.spent_height is not None:
coin_state_list = await self.service.wallet_state_manager.wallet_node.fetch_children(
peer, coin_state.coin.name()
)
odd_coin = 0
for coin in coin_state_list:
if coin.coin.amount % 2 == 1:
odd_coin += 1
if odd_coin > 1:
return {"success": False, "error": "This is not a singleton, multiple children coins found."}
coin_state = coin_state_list[0]
# Get parent coin
parent_coin_state_list: List[CoinState] = await self.service.wallet_state_manager.wallet_node.get_coin_state(
[coin_state.coin.parent_coin_info], peer=peer
)
if parent_coin_state_list is None or len(parent_coin_state_list) < 1:
return {
"success": False,
"error": f"Parent coin record 0x{coin_state.coin.parent_coin_info.hex()} not found",
}
parent_coin_state: CoinState = parent_coin_state_list[0]
coin_spend: CoinSpend = await self.service.wallet_state_manager.wallet_node.fetch_puzzle_solution(
peer, parent_coin_state.spent_height, parent_coin_state.coin
)
# convert to NFTInfo
try:
# Check if the metadata is updated
inner_solution: Program = Program.from_bytes(bytes(coin_spend.solution)).rest().rest().first().first()
full_puzzle: Program = Program.from_bytes(bytes(coin_spend.puzzle_reveal))
update_condition = None
try:
for condition in inner_solution.rest().first().rest().as_iter():
if condition.first().as_int() == -24:
update_condition = condition
break
except Exception:
log.info(f"Inner solution is not a metadata updater solution: {inner_solution}")
if update_condition is not None:
uncurried_nft: UncurriedNFT = UncurriedNFT.uncurry(full_puzzle)
metadata: Program = uncurried_nft.metadata
metadata = nft_puzzles.update_metadata(metadata, update_condition)
# Note: This is not the actual unspent NFT full puzzle.
# There is no way to rebuild the full puzzle in a different wallet.
# But it shouldn't have impact on generating the NFTInfo, since inner_puzzle is not used there.
full_puzzle = nft_puzzles.create_full_puzzle(
uncurried_nft.singleton_launcher_id,
metadata,
uncurried_nft.metadata_updater_hash,
uncurried_nft.inner_puzzle,
)
nft_info: NFTInfo = nft_puzzles.get_nft_info_from_puzzle(NFTCoinInfo(coin_state.coin, None, full_puzzle))
except Exception as e:
return {"success": False, "error": f"The coin is not a NFT. {e}"}
else:
return {"success": True, "nft_info": nft_info}
async def nft_add_uri(self, request) -> Dict:
assert self.service.wallet_state_manager is not None
wallet_id = uint32(request["wallet_id"])
uri = request["uri"]
# Note metadata updater can only add one uri for one field per spend.
# If you want to add multiple uris for one field, you need to spend multiple times.
nft_wallet: NFTWallet = self.service.wallet_state_manager.wallets[wallet_id]
try:
uri = request["uri"]
key = request["key"]
nft_coin_info = nft_wallet.get_nft_coin_by_id(bytes32.from_hexstr(request["nft_coin_id"]))
fee = uint64(request.get("fee", 0))
spend_bundle = await nft_wallet.update_metadata(nft_coin_info, uri, fee=fee)
spend_bundle = await nft_wallet.update_metadata(nft_coin_info, key, uri, fee=fee)
return {"wallet_id": wallet_id, "success": True, "spend_bundle": spend_bundle}
except Exception as e:
log.exception(f"Failed to update NFT metadata: {e}")
chia/rpc/wallet_rpc_client.py
+41 -9
View File
@@ -583,27 +583,59 @@ class WalletRpcClient(RpcClient):
response = await self.fetch("create_new_wallet", request)
return response
async def mint_nft(self, wallet_id, artist_address, hash, uris, fee):
async def mint_nft(
self,
wallet_id,
royalty_address,
target_address,
hash,
uris,
meta_hash="00",
meta_uris=[],
license_hash="00",
license_uris=[],
series_total=1,
series_number=1,
fee=0,
):
request: Dict[str, Any] = {
"wallet_id": wallet_id,
"artist_address": artist_address,
"royalty_address": royalty_address,
"target_address": target_address,
"hash": hash,
"uris": uris,
"meta_hash": meta_hash,
"meta_uris": meta_uris,
"license_hash": license_hash,
"license_uris": license_uris,
"series_number": series_number,
"series_total": series_total,
"fee": fee,
}
response = await self.fetch("nft_mint_nft", request)
return response
async def add_uri_to_nft(self, wallet_id, nft_coin_id, uri, fee):
request: Dict[str, Any] = {"wallet_id": wallet_id, "nft_coin_id": nft_coin_id, "uri": uri, "fee": fee}
response = await self.fetch("nft_add_uri", request)
return response
async def transfer_nft(self, wallet_id, nft_coin_id, artist_address, fee):
async def add_uri_to_nft(self, wallet_id, nft_coin_id, key, uri, fee):
request: Dict[str, Any] = {
"wallet_id": wallet_id,
"nft_coin_id": nft_coin_id,
"target_address": artist_address,
"uri": uri,
"key": key,
"fee": fee,
}
response = await self.fetch("nft_add_uri", request)
return response
async def get_nft_info(self, coin_id: bytes32, latest: bool = True):
request: Dict[str, Any] = {"coin_id": coin_id.hex(), "latest": latest}
response = await self.fetch("nft_get_info", request)
return response
async def transfer_nft(self, wallet_id, nft_coin_id, target_address, fee):
request: Dict[str, Any] = {
"wallet_id": wallet_id,
"nft_coin_id": nft_coin_id,
"target_address": target_address,
"fee": fee,
}
response = await self.fetch("nft_transfer_nft", request)
chia/wallet/did_wallet/did_wallet.py
+17 -10
View File
@@ -260,9 +260,8 @@ class DIDWallet:
self.wallet_info = await wallet_state_manager.user_store.create_wallet(
name, WalletType.DISTRIBUTED_ID.value, info_as_string, in_transaction=True
)
await self.wallet_state_manager.add_new_wallet(self, self.wallet_info.id)
await self.wallet_state_manager.update_wallet_puzzle_hashes(self.wallet_info.id)
await self.wallet_state_manager.add_new_wallet(self, self.wallet_info.id, in_transaction=True)
await self.wallet_state_manager.update_wallet_puzzle_hashes(self.wallet_info.id, in_transaction=True)
await self.load_parent(self.did_info)
self.log.info(f"New DID wallet created {info_as_string}.")
if self.wallet_info is None:
@@ -461,8 +460,13 @@ class DIDWallet:
"""
# full_puz = did_wallet_puzzles.create_fullpuz(innerpuz, origin.name())
# All additions in this block here:
new_puzhash = await self.get_new_did_inner_hash()
new_pubkey = bytes((await self.wallet_state_manager.get_unused_derivation_record(self.wallet_info.id)).pubkey)
new_pubkey = bytes(
(
await self.wallet_state_manager.get_unused_derivation_record(self.wallet_info.id, in_transaction=True)
).pubkey
)
new_puzhash = puzzle_for_pk(new_pubkey).get_tree_hash()
parent_info = None
assert did_info.origin_coin is not None
assert did_info.current_inner is not None
@@ -481,8 +485,7 @@ class DIDWallet:
did_info.current_inner.get_tree_hash(),
coin.amount,
)
await self.add_parent(coin.name(), future_parent, False)
await self.add_parent(coin.name(), future_parent, True)
if children_state.spent_height != children_state.created_height:
did_info = DIDInfo(
did_info.origin_coin,
@@ -496,7 +499,8 @@ class DIDWallet:
False,
did_info.metadata,
)
await self.save_info(did_info, False)
await self.save_info(did_info, True)
assert children_state.created_height
puzzle_solution_request = wallet_protocol.RequestPuzzleSolution(
coin.parent_coin_info, children_state.created_height
@@ -514,7 +518,7 @@ class DIDWallet:
parent_innerpuz.get_tree_hash(),
parent_state.coin.amount,
)
await self.add_parent(coin.parent_coin_info, parent_info, False)
await self.add_parent(coin.parent_coin_info, parent_info, True)
assert parent_info is not None
def puzzle_for_pk(self, pubkey: G1Element) -> Program:
@@ -1179,13 +1183,16 @@ class DIDWallet:
async def generate_eve_spend(self, coin: Coin, full_puzzle: Program, innerpuz: Program):
assert self.did_info.origin_coin is not None
uncurried = did_wallet_puzzles.uncurry_innerpuz(innerpuz)
assert uncurried is not None
p2_puzzle = uncurried[0]
# innerpuz solution is (mode p2_solution)
p2_solution = self.standard_wallet.make_solution(
primaries=[
{
"puzzlehash": innerpuz.get_tree_hash(),
"amount": uint64(coin.amount),
"memos": [innerpuz.get_tree_hash()],
"memos": [p2_puzzle.get_tree_hash()],
}
]
)
chia/wallet/did_wallet/did_wallet_puzzles.py
+1 -2
View File
@@ -1,4 +1,3 @@
from clvm_tools.binutils import assemble
from chia.types.blockchain_format.sized_bytes import bytes32
from chia.types.blockchain_format.program import Program
from typing import List, Optional, Tuple, Iterator, Dict
@@ -205,7 +204,7 @@ def metadata_to_program(metadata: Dict) -> Program:
"""
kv_list = []
for key, value in metadata.items():
kv_list.append((assemble(key), assemble(value)))
kv_list.append((key, value))
return Program.to(kv_list)
chia/wallet/nft_wallet/nft_info.py
+44 -12
View File
@@ -1,8 +1,15 @@
from dataclasses import dataclass
from typing import List
from typing import List, Optional
from chia.util.ints import uint64
from chia.types.blockchain_format.coin import Coin
from chia.types.blockchain_format.program import Program
from chia.types.blockchain_format.sized_bytes import bytes32
from chia.util.ints import uint32, uint64
from chia.util.streamable import Streamable, streamable
from chia.wallet.lineage_proof import LineageProof
from chia.wallet.puzzles.load_clvm import load_clvm
LAUNCHER_PUZZLE = load_clvm("singleton_launcher.clvm")
@streamable
@@ -10,10 +17,10 @@ from chia.util.streamable import Streamable, streamable
class NFTInfo(Streamable):
"""NFT Info for displaying NFT on the UI"""
launcher_id: str
launcher_id: bytes32
"""Launcher coin ID"""
nft_coin_id: str
nft_coin_id: bytes32
"""Current NFT coin ID"""
did_owner: str
@@ -25,26 +32,51 @@ class NFTInfo(Streamable):
data_uris: List[str]
""" A list of content URIs"""
data_hash: str
data_hash: bytes
"""Hash of the content"""
metadata_uris: List[str]
"""A list of metadata URIs"""
metadata_hash: str
metadata_hash: bytes
"""Hash of the metadata"""
license_uris: List[str]
"""A list of license URIs"""
license_hash: str
license_hash: bytes
"""Hash of the license"""
version: str
"""Current NFT version"""
edition_count: uint64
series_total: uint64
"""How many NFTs in the current series"""
edition_number: uint64
series_number: uint64
"""Number of the current NFT in the series"""
updater_puzhash: bytes32
"""Puzzle hash of the metadata updater in hex"""
chain_info: str
"""Information saved on the chain in hex"""
pending_transaction: bool = False
"""Indicate if the NFT is pending for a transaction"""
launcher_puzhash: bytes32 = LAUNCHER_PUZZLE.get_tree_hash()
"""Puzzle hash of the singleton launcher in hex"""
@streamable
@dataclass(frozen=True)
class NFTCoinInfo(Streamable):
coin: Coin
lineage_proof: Optional[LineageProof]
full_puzzle: Program
pending_transaction: bool = False
@streamable
@dataclass(frozen=True)
class NFTWalletInfo(Streamable):
my_nft_coins: List[NFTCoinInfo]
did_wallet_id: Optional[uint32] = None
chia/wallet/nft_wallet/nft_puzzles.py
+82 -19
View File
@@ -1,10 +1,12 @@
import logging
from typing import Any, Dict, List
from clvm_tools.binutils import disassemble
from chia.types.blockchain_format.coin import Coin
from chia.types.blockchain_format.program import Program
from chia.types.blockchain_format.sized_bytes import bytes32
from chia.util.ints import uint64
from chia.wallet.nft_wallet.nft_info import NFTInfo
from chia.wallet.nft_wallet.nft_info import NFTCoinInfo, NFTInfo
from chia.wallet.nft_wallet.uncurry_nft import UncurriedNFT
from chia.wallet.puzzles.load_clvm import load_clvm
@@ -70,32 +72,93 @@ def create_full_puzzle(
return full_puzzle
def get_nft_info_from_puzzle(puzzle: Program, nft_coin: Coin) -> NFTInfo:
def get_nft_info_from_puzzle(nft_coin_info: NFTCoinInfo) -> NFTInfo:
"""
Extract NFT info from a full puzzle
:param puzzle: NFT full puzzle
:param nft_coin: NFT coin
:param nft_coin_info NFTCoinInfo in local database
:return: NFTInfo
"""
# TODO Update this method after the NFT code finalized
uncurried_nft: UncurriedNFT = UncurriedNFT.uncurry(puzzle)
data_uris = []
uncurried_nft: UncurriedNFT = UncurriedNFT.uncurry(nft_coin_info.full_puzzle)
data_uris: List[str] = []
for uri in uncurried_nft.data_uris.as_python():
data_uris.append(str(uri, "utf-8"))
meta_uris: List[str] = []
for uri in uncurried_nft.meta_uris.as_python():
meta_uris.append(str(uri, "utf-8"))
license_uris: List[str] = []
for uri in uncurried_nft.license_uris.as_python():
license_uris.append(str(uri, "utf-8"))
nft_info = NFTInfo(
uncurried_nft.singleton_launcher_id.as_python().hex().upper(),
nft_coin.name().hex().upper(),
uncurried_nft.owner_did.as_python().hex().upper(),
uncurried_nft.singleton_launcher_id.as_python(),
nft_coin_info.coin.name(),
uncurried_nft.owner_did.as_python(),
uint64(uncurried_nft.trade_price_percentage.as_int()),
data_uris,
uncurried_nft.data_hash.as_python().hex().upper(),
[],
"",
[],
"",
"NFT0",
uint64(1),
uint64(1),
uncurried_nft.data_hash.as_python(),
meta_uris,
uncurried_nft.meta_hash.as_python(),
license_uris,
uncurried_nft.license_hash.as_python(),
uint64(uncurried_nft.series_total.as_int()),
uint64(uncurried_nft.series_total.as_int()),
uncurried_nft.metadata_updater_hash.as_python(),
disassemble(uncurried_nft.metadata),
nft_coin_info.pending_transaction,
)
return nft_info
def metadata_to_program(metadata: Dict[bytes, Any]) -> Program:
"""
Convert the metadata dict to a Chialisp program
:param metadata: User defined metadata
:return: Chialisp program
"""
kv_list = []
for key, value in metadata.items():
kv_list.append((key, value))
program: Program = Program.to(kv_list)
return program
def program_to_metadata(program: Program) -> Dict[bytes, Any]:
"""
Convert a program to a metadata dict
:param program: Chialisp program contains the metadata
:return: Metadata dict
"""
metadata = {}
for kv_pair in program.as_iter():
metadata[kv_pair.first().as_atom()] = kv_pair.rest().as_python()
return metadata
def prepend_value(key: bytes, value: Program, metadata: Dict[bytes, Any]) -> None:
"""
Prepend a value to a list in the metadata
:param key: Key of the field
:param value: Value want to add
:param metadata: Metadata
:return:
"""
if value != Program.to(0):
if metadata[key] == b"":
metadata[key] = [value.as_python()]
else:
metadata[key].insert(0, value.as_python())
def update_metadata(metadata: Program, update_condition: Program) -> Program:
"""
Apply conditions of metadata updater to the previous metadata
:param metadata: Previous metadata
:param update_condition: Update metadata conditions
:return: Updated metadata
"""
new_metadata: Dict[bytes, Any] = program_to_metadata(metadata)
uri: Program = update_condition.rest().rest().first()
prepend_value(uri.first().as_python(), uri.rest(), new_metadata)
return metadata_to_program(new_metadata)
chia/wallet/nft_wallet/nft_wallet.py
+44 -43
View File
@@ -1,14 +1,12 @@
import json
import logging
import time
from dataclasses import dataclass
from secrets import token_bytes
from typing import Any, Dict, List, Optional, Set, Type, TypeVar
from blspy import AugSchemeMPL, G1Element, G2Element
from clvm.casts import int_from_bytes, int_to_bytes
from chia.clvm.singleton import SINGLETON_TOP_LAYER_MOD
from chia.protocols.wallet_protocol import CoinState
from chia.server.outbound_message import NodeType
from chia.server.ws_connection import WSChiaConnection
@@ -20,11 +18,11 @@ from chia.types.coin_spend import CoinSpend
from chia.types.spend_bundle import SpendBundle
from chia.util.condition_tools import conditions_dict_for_solution, pkm_pairs_for_conditions_dict
from chia.util.ints import uint8, uint32, uint64, uint128
from chia.util.streamable import Streamable, streamable
from chia.wallet.derivation_record import DerivationRecord
from chia.wallet.lineage_proof import LineageProof
from chia.wallet.nft_wallet import nft_puzzles
from chia.wallet.nft_wallet.nft_puzzles import LAUNCHER_PUZZLE, NFT_METADATA_UPDATER, NFT_STATE_LAYER_MOD_HASH
from chia.wallet.nft_wallet.nft_info import NFTCoinInfo, NFTWalletInfo
from chia.wallet.nft_wallet.nft_puzzles import NFT_METADATA_UPDATER, NFT_STATE_LAYER_MOD_HASH
from chia.wallet.nft_wallet.uncurry_nft import UncurriedNFT
from chia.wallet.puzzles.load_clvm import load_clvm
from chia.wallet.puzzles.p2_delegated_puzzle_or_hidden_puzzle import (
@@ -48,28 +46,6 @@ _T_NFTWallet = TypeVar("_T_NFTWallet", bound="NFTWallet")
OFFER_MOD = load_clvm("settlement_payments.clvm")
@streamable
@dataclass(frozen=True)
class NFTCoinInfo(Streamable):
coin: Coin
lineage_proof: LineageProof
full_puzzle: Program
@streamable
@dataclass(frozen=True)
class NFTWalletInfo(Streamable):
my_nft_coins: List[NFTCoinInfo]
did_wallet_id: Optional[uint32] = None
def create_fullpuz(innerpuz: Program, genesis_id: bytes32) -> Program:
mod_hash = SINGLETON_TOP_LAYER_MOD.get_tree_hash()
# singleton_struct = (MOD_HASH . (LAUNCHER_ID . LAUNCHER_PUZZLE_HASH))
singleton_struct = Program.to((mod_hash, (genesis_id, LAUNCHER_PUZZLE.get_tree_hash())))
return SINGLETON_TOP_LAYER_MOD.curry(singleton_struct, innerpuz)
class NFTWallet:
wallet_state_manager: Any
log: logging.Logger
@@ -217,6 +193,7 @@ class NFTWallet:
singleton_id = bytes32(uncurried_nft.singleton_launcher_id.atom)
metadata = uncurried_nft.metadata
new_inner_puzzle = None
update_condition = None
parent_inner_puzhash = uncurried_nft.nft_state_layer.get_tree_hash()
self.log.debug("Before spend metadata: %s %s \n%s", metadata, singleton_id, disassemble(solution))
for condition in solution.rest().first().rest().as_iter():
@@ -228,16 +205,7 @@ class NFTWallet:
self.log.debug("Checking condition code: %r", condition_code)
if condition_code == -24:
# metadata update
# (-24 (meta updater puzzle) url)
metadata_list = list(metadata.as_python())
new_metadata = []
for metadata_entry in metadata_list:
key = metadata_entry[0]
if key == b"u":
new_metadata.append((b"u", [condition.rest().rest().first().atom] + list(metadata_entry[1:])))
else:
new_metadata.append((b"h", metadata_entry[1]))
metadata = Program.to(new_metadata)
update_condition = condition
elif condition_code == 51 and int_from_bytes(condition.rest().rest().first().atom) == 1:
puzhash = bytes32(condition.rest().first().atom)
self.log.debug("Got back puzhash from solution: %s", puzhash)
@@ -254,6 +222,8 @@ class NFTWallet:
raise ValueError("Invalid condition")
if new_inner_puzzle is None:
raise ValueError("Invalid puzzle")
if update_condition is not None:
metadata = nft_puzzles.update_metadata(metadata, update_condition)
parent_coin = None
coin_record = await self.wallet_state_manager.coin_store.get_coin_record(coin_name)
if coin_record is None:
@@ -270,7 +240,7 @@ class NFTWallet:
child_puzzle: Program = nft_puzzles.create_full_puzzle(
singleton_id,
metadata,
bytes32(uncurried_nft.metdata_updater_hash.atom),
bytes32(uncurried_nft.metadata_updater_hash.atom),
new_inner_puzzle,
)
self.log.debug(
@@ -339,8 +309,13 @@ class NFTWallet:
return provenance_puzzle
async def generate_new_nft(
self, metadata: Program, target_puzzle_hash: bytes32 = None, fee: uint64 = uint64(0)
self,
metadata: Program,
royalty_puzzle_hash: bytes32 = None,
target_puzzle_hash: bytes32 = None,
fee: uint64 = uint64(0),
) -> Optional[SpendBundle]:
# TODO Set royalty address after NFT1 chialisp finished
"""
This must be called under the wallet state manager lock
"""
@@ -471,11 +446,13 @@ class NFTWallet:
async def _make_nft_transaction(
self, nft_coin_info: NFTCoinInfo, inner_solution: Program, fee: uint64 = uint64(0)
) -> TransactionRecord:
# Update NFT status
await self.update_coin_status(nft_coin_info.coin.name(), True)
coin = nft_coin_info.coin
amount = coin.amount
full_puzzle = nft_coin_info.full_puzzle
lineage_proof = nft_coin_info.lineage_proof
assert lineage_proof is not None
self.log.debug("Inner solution: %r", disassemble(inner_solution))
full_solution = Program.to(
[
@@ -516,21 +493,23 @@ class NFTWallet:
return nft_record
async def update_metadata(
self, nft_coin_info: NFTCoinInfo, uri: str, fee: uint64 = uint64(0)
self, nft_coin_info: NFTCoinInfo, key: str, uri: str, fee: uint64 = uint64(0)
) -> Optional[SpendBundle]:
coin = nft_coin_info.coin
# we're not changing it
uncurried_nft = UncurriedNFT.uncurry(nft_coin_info.full_puzzle)
puzzle_hash = uncurried_nft.inner_puzzle.get_tree_hash()
condition_list = [make_create_coin_condition(puzzle_hash, coin.amount, [puzzle_hash])]
condition_list.append([int_to_bytes(-24), NFT_METADATA_UPDATER, uri.encode("utf-8")])
condition_list.append([int_to_bytes(-24), NFT_METADATA_UPDATER, (key, uri)])
self.log.info("Attempting to add a url to NFT coin %s in the metadata: %s", nft_coin_info, uri)
self.log.info(
"Attempting to add urls to NFT coin %s in the metadata: %s", nft_coin_info, uncurried_nft.metadata
)
inner_solution = solution_for_conditions(condition_list)
nft_tx_record = await self._make_nft_transaction(nft_coin_info, inner_solution, fee)
await self.standard_wallet.push_transaction(nft_tx_record)
self.wallet_state_manager.state_changed("nft_coin_updated", self.wallet_info.id)
return nft_tx_record.spend_bundle
async def transfer_nft(
@@ -550,11 +529,33 @@ class NFTWallet:
inner_solution = solution_for_conditions(condition_list)
nft_tx_record = await self._make_nft_transaction(nft_coin_info, inner_solution, fee)
await self.standard_wallet.push_transaction(nft_tx_record)
self.wallet_state_manager.state_changed("nft_coin_transferred", self.wallet_info.id)
return nft_tx_record.spend_bundle
def get_current_nfts(self) -> List[NFTCoinInfo]:
return self.nft_wallet_info.my_nft_coins
async def update_coin_status(
self, coin_id: bytes32, pending_transaction: bool, in_transaction: bool = False
) -> None:
my_nft_coins = self.nft_wallet_info.my_nft_coins
target_nft: Optional[NFTCoinInfo] = None
for coin_info in my_nft_coins:
if coin_info.coin.name() == coin_id:
target_nft = coin_info
my_nft_coins.remove(coin_info)
if target_nft is None:
raise ValueError(f"NFT coin {coin_id} doesn't exist.")
my_nft_coins.append(
NFTCoinInfo(target_nft.coin, target_nft.lineage_proof, target_nft.full_puzzle, pending_transaction)
)
new_nft_wallet_info = NFTWalletInfo(
my_nft_coins,
self.nft_wallet_info.did_wallet_id,
)
await self.save_info(new_nft_wallet_info, in_transaction=in_transaction)
async def save_info(self, nft_info: NFTWalletInfo, in_transaction: bool) -> None:
self.nft_wallet_info = nft_info
current_info = self.wallet_info
chia/wallet/nft_wallet/uncurry_nft.py
+35 -4
View File
@@ -38,7 +38,7 @@ class UncurriedNFT:
owner_did: Program
"""Owner's DID"""
metdata_updater_hash: Program
metadata_updater_hash: Program
"""Metadata updater puzzle hash"""
transfer_program_hash: Program
@@ -61,6 +61,12 @@ class UncurriedNFT:
"""
data_uris: Program
data_hash: Program
meta_uris: Program
meta_hash: Program
license_uris: Program
license_hash: Program
series_number: Program
series_total: Program
inner_puzzle: Program
"""NFT state layer inner puzzle"""
@@ -89,14 +95,33 @@ class UncurriedNFT:
raise ValueError(f"Cannot uncurry NFT puzzle, failed on NFT state layer: Mod {mod}")
try:
# Set nft parameters
(nft_mod_hash, metadata, metdata_updater_hash, inner_puzzle) = curried_args.as_iter()
(nft_mod_hash, metadata, metadata_updater_hash, inner_puzzle) = curried_args.as_iter()
data_uris = Program.to([])
data_hash = Program.to(0)
meta_uris = Program.to([])
meta_hash = Program.to(0)
license_uris = Program.to([])
license_hash = Program.to(0)
series_number = Program.to(1)
series_total = Program.to(1)
# Set metadata
for kv_pair in metadata.as_iter():
if kv_pair.first().as_atom() == b"u":
data_uris = kv_pair.rest()
if kv_pair.first().as_atom() == b"h":
data_hash = kv_pair.rest()
if kv_pair.first().as_atom() == b"mu":
meta_uris = kv_pair.rest()
if kv_pair.first().as_atom() == b"mh":
meta_hash = kv_pair.rest()
if kv_pair.first().as_atom() == b"lu":
license_uris = kv_pair.rest()
if kv_pair.first().as_atom() == b"lh":
license_hash = kv_pair.rest()
if kv_pair.first().as_atom() == b"sn":
series_number = kv_pair.rest()
if kv_pair.first().as_atom() == b"st":
series_total = kv_pair.rest()
except Exception as e:
raise ValueError(f"Cannot uncurry NFT state layer: Args {curried_args}") from e
return cls(
@@ -109,7 +134,13 @@ class UncurriedNFT:
metadata=metadata,
data_uris=data_uris,
data_hash=data_hash,
metdata_updater_hash=metdata_updater_hash,
meta_uris=meta_uris,
meta_hash=meta_hash,
license_uris=license_uris,
license_hash=license_hash,
series_number=series_number,
series_total=series_total,
metadata_updater_hash=metadata_updater_hash,
inner_puzzle=inner_puzzle,
# TODO Set/Remove following fields after NFT1 implemented
owner_did=Program.to([]),
chia/wallet/puzzles/nft_metadata_updater_default.clvm
+18 -9
View File
@@ -1,21 +1,30 @@
(mod (CURRENT_METADATA METADATA_UPDATER_PUZZLE_HASH solution)
(mod (CURRENT_METADATA METADATA_UPDATER_PUZZLE_HASH (key . new_url))
; METADATA and METADATA_UPDATER_PUZZLE_HASH are passed in as truths from the layer above
; This program returns ((new_metadata new_metadata_updater_puzhash) conditions)
; once we find 'u' we don't need to continue looping
(defun add_url (METADATA new_url)
; Add uri to a field
(defun add_url (METADATA key new_url)
(if METADATA
(if (= (f (f METADATA)) 'u')
(c (c 'u' (c new_url (r (f METADATA)))) (r METADATA))
(c (f METADATA) (add_url (r METADATA) new_url))
(if (= (f (f METADATA)) key)
(c (c key (c new_url (r (f METADATA)))) (r METADATA))
(c (f METADATA) (add_url (r METADATA) key new_url))
)
()
)
)
; main
; returns ((new_metadata new_metadata_updater_puzhash) conditions)
(list (list (if solution (add_url CURRENT_METADATA solution) CURRENT_METADATA) METADATA_UPDATER_PUZZLE_HASH) 0)
(list
(list
(if (all key new_url)
(if (any (= key "mu") (= key "lu") (= key "u"))
(add_url CURRENT_METADATA key new_url)
CURRENT_METADATA
)
CURRENT_METADATA
)
METADATA_UPDATER_PUZZLE_HASH)
0
)
)
chia/wallet/puzzles/nft_metadata_updater_default.clvm.hex
+1 -1
View File
@@ -1 +1 @@
ff02ffff01ff04ffff04ffff02ffff03ff17ffff01ff02ff02ffff04ff02ffff04ff05ffff04ff17ff8080808080ffff010580ff0180ffff04ff0bff808080ffff01ff808080ffff04ffff01ff02ffff03ff05ffff01ff02ffff03ffff09ff11ffff017580ffff01ff04ffff04ffff0175ffff04ff0bff198080ff0d80ffff01ff04ff09ffff02ff02ffff04ff02ffff04ff0dffff04ff0bff80808080808080ff0180ff8080ff0180ff018080
ff02ffff01ff04ffff04ffff02ffff03ffff22ff27ff3780ffff01ff02ffff03ffff21ffff09ff27ffff01826d7580ffff09ff27ffff01826c7580ffff09ff27ffff01758080ffff01ff02ff02ffff04ff02ffff04ff05ffff04ff27ffff04ff37ff808080808080ffff010580ff0180ffff010580ff0180ffff04ff0bff808080ffff01ff808080ffff04ffff01ff02ffff03ff05ffff01ff02ffff03ffff09ff11ff0b80ffff01ff04ffff04ff0bffff04ff17ff198080ff0d80ffff01ff04ff09ffff02ff02ffff04ff02ffff04ff0dffff04ff0bffff04ff17ff8080808080808080ff0180ff8080ff0180ff018080
chia/wallet/puzzles/nft_metadata_updater_default.clvm.hex.sha256tree
+1 -1
View File
@@ -1 +1 @@
81970d352e6a39a241eaf8ca510a0e669e40d778ba612621c60a50ef6cf29c7b
fe8a4b4e27a2e29a4d3fc7ce9d527adbcaccbab6ada3903ccf3ba9a769d2d78b
chia/wallet/wallet.py
+2 -2
View File
@@ -181,8 +181,8 @@ class Wallet:
public_key = await self.hack_populate_secret_key_for_puzzle_hash(puzzle_hash)
return puzzle_for_pk(bytes(public_key))
async def get_new_puzzle(self) -> Program:
dr = await self.wallet_state_manager.get_unused_derivation_record(self.id())
async def get_new_puzzle(self, in_transaction: bool = False) -> Program:
dr = await self.wallet_state_manager.get_unused_derivation_record(self.id(), in_transaction=in_transaction)
return puzzle_for_pk(bytes(dr.pubkey))
async def get_puzzle_hash(self, new: bool) -> bytes32:
chia/wallet/wallet_state_manager.py
+2 -2
View File
@@ -326,7 +326,7 @@ class WalletStateManager:
if unused > 0:
await self.puzzle_store.set_used_up_to(uint32(unused - 1), in_transaction)
async def update_wallet_puzzle_hashes(self, wallet_id):
async def update_wallet_puzzle_hashes(self, wallet_id, in_transaction=False):
derivation_paths: List[DerivationRecord] = []
target_wallet = self.wallets[wallet_id]
last: Optional[uint32] = await self.puzzle_store.get_last_derivation_path_for_wallet(wallet_id)
@@ -353,7 +353,7 @@ class WalletStateManager:
False,
)
)
await self.puzzle_store.add_derivation_paths(derivation_paths)
await self.puzzle_store.add_derivation_paths(derivation_paths, in_transaction=in_transaction)
async def get_unused_derivation_record(
self, wallet_id: uint32, in_transaction=False, hardened=False
mypy.ini
+1 -1
View File
File diff suppressed because one or more lines are too long
tests/wallet/did_wallet/test_did.py
+109 -33
View File
@@ -23,8 +23,12 @@ async def get_wallet_num(wallet_manager):
class TestDIDWallet:
@pytest.mark.parametrize(
"trusted",
[True, False],
)
@pytest.mark.asyncio
async def test_creation_from_backup_file(self, self_hostname, three_wallet_nodes):
async def test_creation_from_backup_file(self, self_hostname, three_wallet_nodes, trusted):
num_blocks = 5
full_nodes, wallets = three_wallet_nodes
full_node_api = full_nodes[0]
@@ -39,7 +43,20 @@ class TestDIDWallet:
ph = await wallet_0.get_new_puzzlehash()
ph1 = await wallet_1.get_new_puzzlehash()
ph2 = await wallet_2.get_new_puzzlehash()
if trusted:
wallet_node_0.config["trusted_peers"] = {
full_node_api.full_node.server.node_id.hex(): full_node_api.full_node.server.node_id.hex()
}
wallet_node_1.config["trusted_peers"] = {
full_node_api.full_node.server.node_id.hex(): full_node_api.full_node.server.node_id.hex()
}
wallet_node_2.config["trusted_peers"] = {
full_node_api.full_node.server.node_id.hex(): full_node_api.full_node.server.node_id.hex()
}
else:
wallet_node_0.config["trusted_peers"] = {}
wallet_node_1.config["trusted_peers"] = {}
wallet_node_2.config["trusted_peers"] = {}
await server_0.start_client(PeerInfo(self_hostname, uint16(full_node_server._port)), None)
await server_1.start_client(PeerInfo(self_hostname, uint16(full_node_server._port)), None)
await server_2.start_client(PeerInfo(self_hostname, uint16(full_node_server._port)), None)
@@ -172,8 +189,12 @@ class TestDIDWallet:
await time_out_assert(45, did_wallet_2.get_confirmed_balance, 0)
await time_out_assert(45, did_wallet_2.get_unconfirmed_balance, 0)
@pytest.mark.parametrize(
"trusted",
[True, False],
)
@pytest.mark.asyncio
async def test_did_recovery_with_multiple_backup_dids(self, self_hostname, two_wallet_nodes):
async def test_did_recovery_with_multiple_backup_dids(self, self_hostname, two_wallet_nodes, trusted):
num_blocks = 5
full_nodes, wallets = two_wallet_nodes
full_node_api = full_nodes[0]
@@ -184,7 +205,16 @@ class TestDIDWallet:
wallet2 = wallet_node_2.wallet_state_manager.main_wallet
ph = await wallet.get_new_puzzlehash()
if trusted:
wallet_node.config["trusted_peers"] = {
full_node_api.full_node.server.node_id.hex(): full_node_api.full_node.server.node_id.hex()
}
wallet_node_2.config["trusted_peers"] = {
full_node_api.full_node.server.node_id.hex(): full_node_api.full_node.server.node_id.hex()
}
else:
wallet_node.config["trusted_peers"] = {}
wallet_node_2.config["trusted_peers"] = {}
await server_2.start_client(PeerInfo(self_hostname, uint16(server_1._port)), None)
await server_3.start_client(PeerInfo(self_hostname, uint16(server_1._port)), None)
@@ -311,8 +341,12 @@ class TestDIDWallet:
await time_out_assert(15, did_wallet_3.get_confirmed_balance, 0)
await time_out_assert(15, did_wallet_3.get_unconfirmed_balance, 0)
@pytest.mark.parametrize(
"trusted",
[True, False],
)
@pytest.mark.asyncio
async def test_did_recovery_with_empty_set(self, self_hostname, two_wallet_nodes):
async def test_did_recovery_with_empty_set(self, self_hostname, two_wallet_nodes, trusted):
num_blocks = 5
full_nodes, wallets = two_wallet_nodes
full_node_api = full_nodes[0]
@@ -322,7 +356,16 @@ class TestDIDWallet:
wallet = wallet_node.wallet_state_manager.main_wallet
ph = await wallet.get_new_puzzlehash()
if trusted:
wallet_node.config["trusted_peers"] = {
full_node_api.full_node.server.node_id.hex(): full_node_api.full_node.server.node_id.hex()
}
wallet_node_2.config["trusted_peers"] = {
full_node_api.full_node.server.node_id.hex(): full_node_api.full_node.server.node_id.hex()
}
else:
wallet_node.config["trusted_peers"] = {}
wallet_node_2.config["trusted_peers"] = {}
await server_2.start_client(PeerInfo(self_hostname, uint16(server_1._port)), None)
await server_3.start_client(PeerInfo(self_hostname, uint16(server_1._port)), None)
@@ -368,8 +411,12 @@ class TestDIDWallet:
else:
assert False
@pytest.mark.parametrize(
"trusted",
[True, False],
)
@pytest.mark.asyncio
async def test_did_attest_after_recovery(self, self_hostname, two_wallet_nodes):
async def test_did_attest_after_recovery(self, self_hostname, two_wallet_nodes, trusted):
num_blocks = 5
full_nodes, wallets = two_wallet_nodes
full_node_api = full_nodes[0]
@@ -379,7 +426,16 @@ class TestDIDWallet:
wallet = wallet_node.wallet_state_manager.main_wallet
wallet2 = wallet_node_2.wallet_state_manager.main_wallet
ph = await wallet.get_new_puzzlehash()
if trusted:
wallet_node.config["trusted_peers"] = {
full_node_api.full_node.server.node_id.hex(): full_node_api.full_node.server.node_id.hex()
}
wallet_node_2.config["trusted_peers"] = {
full_node_api.full_node.server.node_id.hex(): full_node_api.full_node.server.node_id.hex()
}
else:
wallet_node.config["trusted_peers"] = {}
wallet_node_2.config["trusted_peers"] = {}
await server_2.start_client(PeerInfo(self_hostname, uint16(server_1._port)), None)
await server_3.start_client(PeerInfo(self_hostname, uint16(server_1._port)), None)
for i in range(1, num_blocks):
@@ -533,8 +589,12 @@ class TestDIDWallet:
"with_recovery",
[True, False],
)
@pytest.mark.parametrize(
"trusted",
[True, False],
)
@pytest.mark.asyncio
async def test_did_transfer(self, two_wallet_nodes, with_recovery):
async def test_did_transfer(self, two_wallet_nodes, with_recovery, trusted):
num_blocks = 5
full_nodes, wallets = two_wallet_nodes
full_node_api = full_nodes[0]
@@ -545,13 +605,16 @@ class TestDIDWallet:
wallet2 = wallet_node_2.wallet_state_manager.main_wallet
ph = await wallet.get_new_puzzlehash()
wallet_node.config["trusted_peers"] = {
full_node_api.full_node.server.node_id.hex(): full_node_api.full_node.server.node_id.hex()
}
wallet_node_2.config["trusted_peers"] = {
full_node_api.full_node.server.node_id.hex(): full_node_api.full_node.server.node_id.hex()
}
if trusted:
wallet_node.config["trusted_peers"] = {
full_node_api.full_node.server.node_id.hex(): full_node_api.full_node.server.node_id.hex()
}
wallet_node_2.config["trusted_peers"] = {
full_node_api.full_node.server.node_id.hex(): full_node_api.full_node.server.node_id.hex()
}
else:
wallet_node.config["trusted_peers"] = {}
wallet_node_2.config["trusted_peers"] = {}
await server_2.start_client(PeerInfo("localhost", uint16(server_1._port)), None)
await server_3.start_client(PeerInfo("localhost", uint16(server_1._port)), None)
@@ -615,8 +678,12 @@ class TestDIDWallet:
assert metadata["Twitter"] == "Test"
assert metadata["GitHub"] == "测试"
@pytest.mark.parametrize(
"trusted",
[True, False],
)
@pytest.mark.asyncio
async def test_update_recovery_list(self, two_wallet_nodes):
async def test_update_recovery_list(self, two_wallet_nodes, trusted):
num_blocks = 5
full_nodes, wallets = two_wallet_nodes
full_node_api = full_nodes[0]
@@ -626,13 +693,16 @@ class TestDIDWallet:
wallet = wallet_node.wallet_state_manager.main_wallet
ph = await wallet.get_new_puzzlehash()
wallet_node.config["trusted_peers"] = {
full_node_api.full_node.server.node_id.hex(): full_node_api.full_node.server.node_id.hex()
}
wallet_node_2.config["trusted_peers"] = {
full_node_api.full_node.server.node_id.hex(): full_node_api.full_node.server.node_id.hex()
}
if trusted:
wallet_node.config["trusted_peers"] = {
full_node_api.full_node.server.node_id.hex(): full_node_api.full_node.server.node_id.hex()
}
wallet_node_2.config["trusted_peers"] = {
full_node_api.full_node.server.node_id.hex(): full_node_api.full_node.server.node_id.hex()
}
else:
wallet_node.config["trusted_peers"] = {}
wallet_node_2.config["trusted_peers"] = {}
await server_2.start_client(PeerInfo("localhost", uint16(server_1._port)), None)
await server_3.start_client(PeerInfo("localhost", uint16(server_1._port)), None)
@@ -670,8 +740,12 @@ class TestDIDWallet:
assert did_wallet_1.did_info.backup_ids[0] == bytes(ph)
assert did_wallet_1.did_info.num_of_backup_ids_needed == 1
@pytest.mark.parametrize(
"trusted",
[True, False],
)
@pytest.mark.asyncio
async def test_update_metadata(self, two_wallet_nodes):
async def test_update_metadata(self, two_wallet_nodes, trusted):
num_blocks = 5
full_nodes, wallets = two_wallet_nodes
full_node_api = full_nodes[0]
@@ -680,14 +754,16 @@ class TestDIDWallet:
wallet_node_2, server_3 = wallets[1]
wallet = wallet_node.wallet_state_manager.main_wallet
ph = await wallet.get_new_puzzlehash()
wallet_node.config["trusted_peers"] = {
full_node_api.full_node.server.node_id.hex(): full_node_api.full_node.server.node_id.hex()
}
wallet_node_2.config["trusted_peers"] = {
full_node_api.full_node.server.node_id.hex(): full_node_api.full_node.server.node_id.hex()
}
if trusted:
wallet_node.config["trusted_peers"] = {
full_node_api.full_node.server.node_id.hex(): full_node_api.full_node.server.node_id.hex()
}
wallet_node_2.config["trusted_peers"] = {
full_node_api.full_node.server.node_id.hex(): full_node_api.full_node.server.node_id.hex()
}
else:
wallet_node.config["trusted_peers"] = {}
wallet_node_2.config["trusted_peers"] = {}
await server_2.start_client(PeerInfo("localhost", uint16(server_1._port)), None)
await server_3.start_client(PeerInfo("localhost", uint16(server_1._port)), None)
tests/wallet/nft_wallet/test_nft_clvm.py
+14 -4
View File
@@ -1,4 +1,5 @@
from chia.types.blockchain_format.program import INFINITE_COST, Program
from chia.types.blockchain_format.sized_bytes import bytes32
from chia.wallet.puzzles.load_clvm import load_clvm
from chia.wallet.puzzles.p2_delegated_puzzle_or_hidden_puzzle import puzzle_for_pk, solution_for_conditions
from chia.wallet.puzzles.puzzle_utils import make_create_coin_condition
@@ -58,11 +59,17 @@ def test_update_metadata() -> None:
my_amount = 1
destination: Program = puzzle_for_pk(int_to_public_key(2))
condition_list = [make_create_coin_condition(destination.get_tree_hash(), my_amount, [])]
condition_list.append([-24, NFT_METADATA_UPDATER, "https://www.chia.net/img/branding/chia-logo-2.svg"])
condition_list.append([-24, NFT_METADATA_UPDATER, ("mu", "https://url2")])
metadata = [
("u", ["https://www.chia.net/img/branding/chia-logo.svg"]),
("h", 0xD4584AD463139FA8C0D9F68F4B59F185),
("mu", []),
("mh", 0xD4584AD463139FA8C0D9F68F4B59F185),
("lu", ["https://www.chia.net/img/branding/chia-logo.svg"]),
("lh", 0xD4584AD463139FA8C0D9F68F4B59F185),
]
solution = Program.to(
[
NFT_STATE_LAYER_MOD_HASH,
@@ -77,16 +84,19 @@ def test_update_metadata() -> None:
)
metadata = [
("u", ["https://www.chia.net/img/branding/chia-logo-2.svg", "https://www.chia.net/img/branding/chia-logo.svg"]),
("u", ["https://www.chia.net/img/branding/chia-logo.svg"]),
("h", 0xD4584AD463139FA8C0D9F68F4B59F185),
("mu", ["https://url2"]),
("mh", 0xD4584AD463139FA8C0D9F68F4B59F185),
("lu", ["https://www.chia.net/img/branding/chia-logo.svg"]),
("lh", 0xD4584AD463139FA8C0D9F68F4B59F185),
]
cost, res = NFT_STATE_LAYER_MOD.run_with_cost(INFINITE_COST, solution)
assert res.first().first().as_int() == 73
assert res.first().rest().first().as_int() == 1
assert res.rest().rest().first().first().as_int() == 51
assert (
res.rest().rest().first().rest().first().as_atom()
bytes32(res.rest().rest().first().rest().first().as_atom())
== NFT_STATE_LAYER_MOD.curry(
NFT_STATE_LAYER_MOD_HASH, metadata, NFT_METADATA_UPDATER.get_tree_hash(), destination
).get_tree_hash()
tests/wallet/nft_wallet/test_nft_wallet.py
+91 -38
View File
@@ -4,6 +4,7 @@ import asyncio
from typing import Any
import pytest
from clvm_tools.binutils import disassemble
from chia.consensus.block_rewards import calculate_base_farmer_reward, calculate_pool_reward
from chia.full_node.mempool_manager import MempoolManager
@@ -13,7 +14,8 @@ from chia.simulator.simulator_protocol import FarmNewBlockProtocol
from chia.types.blockchain_format.program import Program
from chia.types.blockchain_format.sized_bytes import bytes32
from chia.types.peer_info import PeerInfo
from chia.util.ints import uint16, uint32
from chia.util.byte_types import hexstr_to_bytes
from chia.util.ints import uint16, uint32, uint64
from chia.wallet.nft_wallet.nft_wallet import NFTWallet
from chia.wallet.util.compute_memos import compute_memos
from chia.wallet.util.wallet_types import WalletType
@@ -29,7 +31,7 @@ async def tx_in_pool(mempool: MempoolManager, tx_id: bytes32) -> bool:
@pytest.mark.parametrize(
"trusted",
[True],
[True, False],
)
@pytest.mark.asyncio
async def test_nft_wallet_creation_automatically(two_wallet_nodes: Any, trusted: Any) -> None:
@@ -92,21 +94,21 @@ async def test_nft_wallet_creation_automatically(two_wallet_nodes: Any, trusted:
for i in range(1, num_blocks):
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(ph1))
await time_out_assert(5, len, 1, nft_wallet_0.nft_wallet_info.my_nft_coins)
await time_out_assert(15, len, 1, nft_wallet_0.nft_wallet_info.my_nft_coins)
coins = nft_wallet_0.nft_wallet_info.my_nft_coins
assert len(coins) == 1, "nft not generated"
sb = await nft_wallet_0.transfer_nft(coins[0], ph1)
assert sb is not None
await time_out_assert_not_none(5, full_node_api.full_node.mempool_manager.get_spendbundle, sb.name())
await time_out_assert_not_none(15, full_node_api.full_node.mempool_manager.get_spendbundle, sb.name())
for i in range(1, num_blocks):
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(ph1))
assert len(wallet_node_1.wallet_state_manager.wallets) == 2
await time_out_assert(15, len, 2, wallet_node_1.wallet_state_manager.wallets)
# Get the new NFT wallet
nft_wallets = await wallet_node_1.wallet_state_manager.get_all_wallet_info_entries(WalletType.NFT)
assert len(nft_wallets) == 1
nft_wallet_1: NFTWallet = wallet_node_1.wallet_state_manager.wallets[nft_wallets[0].id]
await time_out_assert(5, len, 1, nft_wallet_1.nft_wallet_info.my_nft_coins)
await time_out_assert(15, len, 1, nft_wallet_1.nft_wallet_info.my_nft_coins)
coins = nft_wallet_0.nft_wallet_info.my_nft_coins
assert len(coins) == 0
coins = nft_wallet_1.nft_wallet_info.my_nft_coins
@@ -115,11 +117,11 @@ async def test_nft_wallet_creation_automatically(two_wallet_nodes: Any, trusted:
@pytest.mark.parametrize(
"trusted",
[True],
[True, False],
)
@pytest.mark.asyncio
async def test_nft_wallet_creation_and_transfer(two_wallet_nodes: Any, trusted: Any) -> None:
num_blocks = 5
num_blocks = 2
full_nodes, wallets = two_wallet_nodes
full_node_api: FullNodeSimulator = full_nodes[0]
full_node_server = full_node_api.server
@@ -171,6 +173,8 @@ async def test_nft_wallet_creation_and_transfer(two_wallet_nodes: Any, trusted:
]
)
await time_out_assert(10, wallet_0.get_unconfirmed_balance, 2000000000000)
await time_out_assert(10, wallet_0.get_confirmed_balance, 2000000000000)
sb = await nft_wallet_0.generate_new_nft(metadata)
assert sb
# ensure hints are generated
@@ -178,11 +182,9 @@ async def test_nft_wallet_creation_and_transfer(two_wallet_nodes: Any, trusted:
await time_out_assert_not_none(15, full_node_api.full_node.mempool_manager.get_spendbundle, sb.name())
for i in range(1, num_blocks):
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(ph1))
coins = nft_wallet_0.nft_wallet_info.my_nft_coins
assert len(coins) == 1, "nft not generated"
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(ph))
await time_out_assert(10, len, 1, nft_wallet_0.nft_wallet_info.my_nft_coins)
metadata = Program.to(
[
("u", ["https://www.test.net/logo.svg"]),
@@ -190,18 +192,22 @@ async def test_nft_wallet_creation_and_transfer(two_wallet_nodes: Any, trusted:
]
)
await time_out_assert(10, wallet_0.get_unconfirmed_balance, 4000000000000 - 1)
await time_out_assert(10, wallet_0.get_confirmed_balance, 4000000000000 - 1)
sb = await nft_wallet_0.generate_new_nft(metadata)
assert sb
# ensure hints are generated
assert compute_memos(sb)
await time_out_assert_not_none(15, full_node_api.full_node.mempool_manager.get_spendbundle, sb.name())
await time_out_assert_not_none(10, full_node_api.full_node.mempool_manager.get_spendbundle, sb.name())
await time_out_assert(30, wallet_node_0.wallet_state_manager.lock.locked, False)
for i in range(1, num_blocks):
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(ph1))
await time_out_assert(15, len, 2, nft_wallet_0.nft_wallet_info.my_nft_coins)
coins = nft_wallet_0.nft_wallet_info.my_nft_coins
assert len(coins) == 2, "nft not generated"
await time_out_assert(15, wallet_0.get_pending_change_balance, 0)
nft_wallet_1 = await NFTWallet.create_new_nft_wallet(
wallet_node_1.wallet_state_manager, wallet_1, name="NFT WALLET 2"
)
@@ -215,12 +221,12 @@ async def test_nft_wallet_creation_and_transfer(two_wallet_nodes: Any, trusted:
for i in range(1, num_blocks):
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(ph1))
coins = nft_wallet_0.nft_wallet_info.my_nft_coins
assert len(coins) == 1
await time_out_assert(15, len, 1, nft_wallet_0.nft_wallet_info.my_nft_coins)
await time_out_assert(15, len, 1, nft_wallet_1.nft_wallet_info.my_nft_coins)
coins = nft_wallet_1.nft_wallet_info.my_nft_coins
assert len(coins) == 1
await time_out_assert(15, wallet_1.get_pending_change_balance, 0)
# Send it back to original owner
nsb = await nft_wallet_1.transfer_nft(coins[0], ph)
assert nsb is not None
@@ -231,13 +237,17 @@ async def test_nft_wallet_creation_and_transfer(two_wallet_nodes: Any, trusted:
for i in range(1, num_blocks):
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(ph1))
await time_out_assert(5, len, 2, nft_wallet_0.nft_wallet_info.my_nft_coins)
await time_out_assert(5, len, 0, nft_wallet_1.nft_wallet_info.my_nft_coins)
for i in range(1, num_blocks):
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(ph))
await time_out_assert(30, wallet_node_0.wallet_state_manager.lock.locked, False)
await time_out_assert(15, len, 2, nft_wallet_0.nft_wallet_info.my_nft_coins)
await time_out_assert(15, len, 0, nft_wallet_1.nft_wallet_info.my_nft_coins)
@pytest.mark.parametrize(
"trusted",
[True],
[True, False],
)
@pytest.mark.asyncio
async def test_nft_wallet_rpc_creation_and_list(two_wallet_nodes: Any, trusted: Any) -> None:
@@ -269,7 +279,12 @@ async def test_nft_wallet_rpc_creation_and_list(two_wallet_nodes: Any, trusted:
for i in range(1, num_blocks):
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(ph))
funds = sum(
[calculate_pool_reward(uint32(i)) + calculate_base_farmer_reward(uint32(i)) for i in range(1, num_blocks - 1)]
)
await time_out_assert(10, wallet_0.get_unconfirmed_balance, funds)
await time_out_assert(10, wallet_0.get_confirmed_balance, funds)
api_0 = WalletRpcApi(wallet_node_0)
nft_wallet_0 = await api_0.create_new_wallet(dict(wallet_type="nft_wallet", name="NFT WALLET 1"))
assert isinstance(nft_wallet_0, dict)
@@ -289,9 +304,20 @@ async def test_nft_wallet_rpc_creation_and_list(two_wallet_nodes: Any, trusted:
assert tr1.get("success")
sb = tr1["spend_bundle"]
await time_out_assert_not_none(5, full_node_api.full_node.mempool_manager.get_spendbundle, sb.name())
await time_out_assert_not_none(15, full_node_api.full_node.mempool_manager.get_spendbundle, sb.name())
for i in range(1, num_blocks):
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(ph))
time_left = 5.0
while time_left > 0:
coins_response = await api_0.nft_get_nfts(dict(wallet_id=nft_wallet_0_id))
if len(coins_response.get("nft_list", [])) > 0:
break
await asyncio.sleep(0.5)
time_left -= 0.5
else:
raise AssertionError("NFT not minted")
await time_out_assert(15, wallet_1.get_pending_change_balance, 0)
tr2 = await api_0.nft_mint_nft(
{
"wallet_id": nft_wallet_0_id,
@@ -303,7 +329,7 @@ async def test_nft_wallet_rpc_creation_and_list(two_wallet_nodes: Any, trusted:
assert isinstance(tr2, dict)
assert tr2.get("success")
sb = tr2["spend_bundle"]
await time_out_assert_not_none(5, full_node_api.full_node.mempool_manager.get_spendbundle, sb.name())
await time_out_assert_not_none(15, full_node_api.full_node.mempool_manager.get_spendbundle, sb.name())
for i in range(1, num_blocks):
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(ph))
time_left = 5.0
@@ -319,7 +345,7 @@ async def test_nft_wallet_rpc_creation_and_list(two_wallet_nodes: Any, trusted:
uris.append(coin.to_json_dict()["data_uris"][0])
assert len(uris) == 2
assert "https://chialisp.com/img/logo.svg" in uris
assert bytes32.fromhex(coins[1].to_json_dict()["nft_coin_id"]) in [x.name() for x in sb.additions()]
assert bytes32.fromhex(coins[1].to_json_dict()["nft_coin_id"][2:]) in [x.name() for x in sb.additions()]
except AssertionError:
if time_left < 0:
raise
@@ -329,7 +355,7 @@ async def test_nft_wallet_rpc_creation_and_list(two_wallet_nodes: Any, trusted:
@pytest.mark.parametrize(
"trusted",
[True],
[True, False],
)
@pytest.mark.asyncio
async def test_nft_wallet_rpc_update_metadata(two_wallet_nodes: Any, trusted: Any) -> None:
@@ -361,6 +387,14 @@ async def test_nft_wallet_rpc_update_metadata(two_wallet_nodes: Any, trusted: An
await server_0.start_client(PeerInfo("localhost", uint16(full_node_server._port)), None)
await server_1.start_client(PeerInfo("localhost", uint16(full_node_server._port)), None)
funds = sum(
[calculate_pool_reward(uint32(i)) + calculate_base_farmer_reward(uint32(i)) for i in range(1, num_blocks - 1)]
)
await time_out_assert(10, wallet_0.get_unconfirmed_balance, funds)
await time_out_assert(10, wallet_0.get_confirmed_balance, funds)
api_0 = WalletRpcApi(wallet_node_0)
await time_out_assert(10, wallet_node_0.wallet_state_manager.synced, True)
await time_out_assert(10, wallet_node_1.wallet_state_manager.synced, True)
@@ -382,7 +416,7 @@ async def test_nft_wallet_rpc_update_metadata(two_wallet_nodes: Any, trusted: An
assert resp.get("success")
sb = resp["spend_bundle"]
await time_out_assert_not_none(5, full_node_api.full_node.mempool_manager.get_spendbundle, sb.name())
await time_out_assert_not_none(15, full_node_api.full_node.mempool_manager.get_spendbundle, sb.name())
for i in range(1, num_blocks):
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(ph))
time_left = 5.0
@@ -397,21 +431,34 @@ async def test_nft_wallet_rpc_update_metadata(two_wallet_nodes: Any, trusted: An
assert coins_response.get("success")
coins = coins_response["nft_list"]
coin = coins[0].to_json_dict()
assert coin["data_hash"] == "0xd4584ad463139fa8c0d9f68f4b59f185"
assert coin["chain_info"] == disassemble(
Program.to(
[
("u", ["https://www.chia.net/img/branding/chia-logo.svg"]),
("h", hexstr_to_bytes("0xD4584AD463139FA8C0D9F68F4B59F185")),
("mu", []),
("mh", hexstr_to_bytes("00")),
("lu", []),
("lh", hexstr_to_bytes("00")),
("sn", uint64(1)),
("st", uint64(1)),
]
)
)
nft_coin_id = coin["nft_coin_id"]
await time_out_assert(15, wallet_0.get_pending_change_balance, 0)
# add another URI
tr1 = await api_0.nft_add_uri(
{
"wallet_id": nft_wallet_0_id,
"nft_coin_id": nft_coin_id,
"hash": "0xD4584AD463139FA8C0D9F68F4B59F185",
"uri": "https://www.chia.net/img/branding/chia-logo-white.svg",
}
{"wallet_id": nft_wallet_0_id, "nft_coin_id": nft_coin_id, "uri": "http://metadata", "key": "mu"}
)
assert isinstance(tr1, dict)
assert tr1.get("success")
coins_response = await api_0.nft_get_nfts(dict(wallet_id=nft_wallet_0_id))
assert coins_response["nft_list"][0].pending_transaction
sb = tr1["spend_bundle"]
await time_out_assert_not_none(5, full_node_api.full_node.mempool_manager.get_spendbundle, sb.name())
await time_out_assert_not_none(15, full_node_api.full_node.mempool_manager.get_spendbundle, sb.name())
for i in range(1, num_blocks):
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(ph))
# check that new URI was added
@@ -425,28 +472,33 @@ async def test_nft_wallet_rpc_update_metadata(two_wallet_nodes: Any, trusted: An
assert len(coins) == 1
coin = coins[0].to_json_dict()
uris = coin["data_uris"]
assert len(uris) == 2
assert "https://www.chia.net/img/branding/chia-logo-white.svg" in uris
assert len(uris) == 1
assert "https://www.chia.net/img/branding/chia-logo.svg" in uris
assert len(coin["metadata_uris"]) == 1
assert "http://metadata" == coin["metadata_uris"][0]
assert len(coin["license_uris"]) == 0
except AssertionError:
if time_left < 0:
raise
await asyncio.sleep(0.5)
time_left -= 0.5
# add yet another URI
await time_out_assert(15, wallet_0.get_pending_change_balance, 0)
nft_coin_id = coin["nft_coin_id"]
tr1 = await api_0.nft_add_uri(
{
"wallet_id": nft_wallet_0_id,
"nft_coin_id": nft_coin_id,
"hash": "0xD4584AD463139FA8C0D9F68F4B59F185",
"uri": "https://www.chia.net/img/branding/chia-logo-more-white.svg",
"uri": "http://data",
"key": "u",
}
)
assert isinstance(tr1, dict)
assert tr1.get("success")
sb = tr1["spend_bundle"]
await time_out_assert_not_none(5, full_node_api.full_node.mempool_manager.get_spendbundle, sb.name())
await time_out_assert_not_none(15, full_node_api.full_node.mempool_manager.get_spendbundle, sb.name())
for i in range(1, num_blocks):
await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(ph))
time_left = 5.0
@@ -459,8 +511,9 @@ async def test_nft_wallet_rpc_update_metadata(two_wallet_nodes: Any, trusted: An
assert len(coins) == 1
coin = coins[0].to_json_dict()
uris = coin["data_uris"]
assert len(uris) == 3
assert "https://www.chia.net/img/branding/chia-logo-more-white.svg" in uris
assert len(uris) == 2
assert len(coin["metadata_uris"]) == 1
assert "http://data" == coin["data_uris"][0]
except AssertionError:
if time_left < 0:
raise
tests/wallet/rpc/test_wallet_rpc.py
+56
View File
@@ -33,6 +33,7 @@ from chia.wallet.cat_wallet.cat_constants import DEFAULT_CATS
from chia.wallet.cat_wallet.cat_wallet import CATWallet
from chia.wallet.derive_keys import master_sk_to_wallet_sk, master_sk_to_wallet_sk_unhardened
from chia.wallet.did_wallet.did_wallet import DIDWallet
from chia.wallet.nft_wallet.nft_wallet import NFTWallet
from chia.wallet.trading.trade_status import TradeStatus
from chia.wallet.transaction_record import TransactionRecord
from chia.wallet.transaction_sorting import SortKey
@@ -822,6 +823,61 @@ async def test_did_endpoints(wallet_rpc_environment: WalletRpcTestEnvironment):
assert metadata["Twitter"] == "Https://test"
@pytest.mark.asyncio
async def test_nft_endpoints(wallet_rpc_environment: WalletRpcTestEnvironment):
env: WalletRpcTestEnvironment = wallet_rpc_environment
wallet_1_node: WalletNode = env.wallet_1.node
wallet_1_rpc: WalletRpcClient = env.wallet_1.rpc_client
wallet_2: Wallet = env.wallet_2.wallet
wallet_2_node: WalletNode = env.wallet_2.node
wallet_2_rpc: WalletRpcClient = env.wallet_2.rpc_client
full_node_api: FullNodeSimulator = env.full_node.api
await generate_funds(env.full_node.api, env.wallet_1, 5)
res = await wallet_1_rpc.create_new_nft_wallet(None)
nft_wallet_id = res["wallet_id"]
res = await wallet_1_rpc.mint_nft(
nft_wallet_id,
None,
None,
"0xD4584AD463139FA8C0D9F68F4B59F185",
["https://www.chia.net/img/branding/chia-logo.svg"],
)
assert res["success"]
for _ in range(3):
await farm_transaction_block(full_node_api, wallet_1_node)
assert wallet_1_node.wallet_state_manager is not None
nft_wallet: NFTWallet = wallet_1_node.wallet_state_manager.wallets[nft_wallet_id]
nft_id = nft_wallet.get_current_nfts()[0].coin.name()
nft_info = (await wallet_1_rpc.get_nft_info(nft_id))["nft_info"]
assert nft_info["nft_coin_id"][2:] == nft_wallet.get_current_nfts()[0].coin.name().hex()
addr = encode_puzzle_hash(await wallet_2.get_new_puzzlehash(), "txch")
res = await wallet_1_rpc.transfer_nft(nft_wallet_id, nft_id.hex(), addr, 0)
assert res["success"]
for _ in range(3):
await farm_transaction_block(full_node_api, wallet_1_node)
assert wallet_2_node.wallet_state_manager is not None
nft_wallet_id_1 = (
await wallet_2_node.wallet_state_manager.get_all_wallet_info_entries(wallet_type=WalletType.NFT)
)[0].id
nft_wallet_1: NFTWallet = wallet_2_node.wallet_state_manager.wallets[nft_wallet_id_1]
nft_info_1 = (await wallet_1_rpc.get_nft_info(nft_id, False))["nft_info"]
assert nft_info_1 == nft_info
nft_info_1 = (await wallet_1_rpc.get_nft_info(nft_id))["nft_info"]
assert nft_info_1["nft_coin_id"][2:] == nft_wallet_1.get_current_nfts()[0].coin.name().hex()
# Cross-check NFT
nft_info_2 = (await wallet_2_rpc.list_nfts(nft_wallet_id_1))["nft_list"][0]
assert nft_info_1 == nft_info_2
@pytest.mark.asyncio
async def test_key_and_address_endpoints(wallet_rpc_environment: WalletRpcTestEnvironment):
env: WalletRpcTestEnvironment = wallet_rpc_environment