-
Notifications
You must be signed in to change notification settings - Fork 47
/
Copy pathtest_utils.py
109 lines (96 loc) · 4.05 KB
/
test_utils.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
# Copyright Frank V. Castellucci
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# http://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# -*- coding: utf-8 -*-
"""Simplify routines for test usage."""
from pysui.abstracts.client_keypair import KeyPair, SignatureScheme
from pysui.sui.sui_config import SuiConfig
from pysui.sui.sui_crypto import MultiSig
from pysui.sui.sui_txn import sync_transaction
from pysui.sui.sui_txresults.complex_tx import TxResponse
from pysui.sui.sui_txresults.single_tx import SuiCoinObject
from pysui.sui.sui_types.address import SuiAddress
from pysui.sui.sui_clients.sync_client import SuiClient
STANDARD_BUDGET: str = "5500000"
def first_addy_keypair_for(
*,
cfg: SuiConfig,
sigtype: SignatureScheme = SignatureScheme.ED25519,
not_in: list[str] = None,
) -> tuple[str, KeyPair]:
"""Get first address and keypair that matches keypair scheme."""
not_in = not_in if not_in else []
filtered = [
(k, v)
for (k, v) in cfg.addresses_and_keys.items()
if v.scheme == sigtype
]
if filtered:
for candidate in filtered:
if candidate[0] not in not_in:
return candidate
raise ValueError(f"No keypair type of {sigtype.as_str()}")
def gas_not_in(
client: SuiClient, for_addy: SuiAddress = None, not_in: list[str] = None
) -> SuiCoinObject:
"""Get gas object that is not in collection."""
for_addy = for_addy if for_addy else client.config.active_address
result = client.get_gas(for_addy)
not_in = not_in if not_in else []
if result.is_ok():
for agas in result.result_data.data:
if agas.coin_object_id not in not_in:
return agas
else:
print(result.result_string)
raise ValueError(result.result_string)
def gen_ms(config: SuiConfig) -> MultiSig:
"""."""
_, ed_key = first_addy_keypair_for(cfg=config)
_, k1_key = first_addy_keypair_for(
cfg=config, sigtype=SignatureScheme.SECP256K1
)
_, r1_key = first_addy_keypair_for(
cfg=config, sigtype=SignatureScheme.SECP256R1
)
multi_sig = MultiSig([ed_key, k1_key, r1_key], [1, 2, 3], 3)
return multi_sig
def publish_and_result(txb: sync_transaction) -> tuple[str, str]:
"""Utility."""
# Execute the transaction
tx_result = txb.execute()
package_id: str = None
upgrade_cap_id: str = None
if tx_result.is_ok():
if hasattr(tx_result.result_data, "to_json"):
# Get the result data and iterate through object changes
tx_response: TxResponse = tx_result.result_data
for object_change in tx_response.object_changes:
match object_change["type"]:
# Found our newly published package_id
case "published":
package_id = object_change["packageId"]
case "created":
# Found our newly created UpgradeCap
if object_change["objectType"].endswith("UpgradeCap"):
upgrade_cap_id = object_change["objectId"]
case "mutated":
# On upgrades, UpgradeCap is mutated
if object_change["objectType"].endswith("UpgradeCap"):
upgrade_cap_id = object_change["objectId"]
case _:
pass
else:
raise ValueError(
f"Non-standard result found {tx_result.result_data}"
)
else:
raise ValueError(f"Error encoundered {tx_result.result_string}")
return (package_id, upgrade_cap_id)