diff --git a/src/komet_node/hello.py b/src/komet_node/hello.py deleted file mode 100644 index 6cab2ee..0000000 --- a/src/komet_node/hello.py +++ /dev/null @@ -1,2 +0,0 @@ -def hello(name: str) -> str: - return f'Hello, {name}!' diff --git a/src/tests/integration/conftest.py b/src/tests/integration/conftest.py new file mode 100644 index 0000000..ab958c3 --- /dev/null +++ b/src/tests/integration/conftest.py @@ -0,0 +1,216 @@ +"""Shared fixtures and helpers for the integration tests. + +The integration tests drive a live ``StellarRpcServer`` over HTTP. This module centralises +everything they build on: the server fixture, the JSON-RPC / HTTP plumbing, wat compilation, +the JSON-shape predicates used to assert the stellar-rpc wire format, and the transaction +submit / deploy helpers. Keeping these in one place stops each feature's test file from +carrying its own drifting copy. +""" + +from __future__ import annotations + +import json +import re +import socket +import subprocess +import threading +import time +import urllib.request +from typing import TYPE_CHECKING, Any, NamedTuple + +import pytest +from stellar_sdk import Account, Keypair, Network, TransactionBuilder +from stellar_sdk.utils import sha256 + +from komet_node.server import StellarRpcServer + +if TYPE_CHECKING: + from collections.abc import Callable, Iterator + from pathlib import Path + + from stellar_sdk import xdr + +PASSPHRASE = Network.TESTNET_NETWORK_PASSPHRASE + + +# --------------------------------------------------------------------------- +# HTTP / JSON-RPC plumbing +# --------------------------------------------------------------------------- + + +def wat_to_wasm(wat_path: Path) -> bytes: + proc_res = subprocess.run(['wat2wasm', str(wat_path), '--output=/dev/stdout'], check=True, capture_output=True) + return proc_res.stdout + + +def _find_free_port() -> int: + with socket.socket() as s: + s.bind(('', 0)) + return s.getsockname()[1] + + +def _wait_for_server(host: str, port: int, timeout: float = 10.0) -> None: + deadline = time.time() + timeout + while time.time() < deadline: + try: + with socket.create_connection((host, port), timeout=0.1): + return + except OSError: + time.sleep(0.05) + raise TimeoutError(f'Server did not start on {host}:{port}') + + +def _rpc(port: int, method: str, params: dict[str, Any]) -> dict[str, Any]: + body = json.dumps({'jsonrpc': '2.0', 'id': 1, 'method': method, 'params': params}).encode() + return _post(port, body) + + +def _post(port: int, body: bytes) -> Any: + return json.loads(_post_raw(port, body)) + + +def _post_raw(port: int, body: bytes) -> bytes: + req = urllib.request.Request( + f'http://localhost:{port}', + data=body, + headers={'Content-Type': 'application/json'}, + ) + with urllib.request.urlopen(req) as resp: + return resp.read() + + +# --------------------------------------------------------------------------- +# Spec-shape predicates +# +# The official serialization rules come from the Go structs in stellar/go-stellar-sdk +# protocols/rpc (what real stellar-rpc emits): ledger sequence numbers and protocolVersion +# are JSON numbers; the close-time fields on the singular methods are int64 with Go's +# `,string` encoding, i.e. JSON strings holding a decimal integer; hashes are 64 lowercase +# hex characters. +# --------------------------------------------------------------------------- + +_HEX64_RE = re.compile(r'[0-9a-f]{64}') +_INT_STRING_RE = re.compile(r'-?[0-9]+') + + +def _is_number(value: Any) -> bool: + """True for a JSON number decoded to int (bool is a distinct JSON type).""" + return isinstance(value, int) and not isinstance(value, bool) + + +def _is_int_string(value: Any) -> bool: + """True for a JSON string holding a decimal integer (Go int64 `,string` encoding).""" + return isinstance(value, str) and _INT_STRING_RE.fullmatch(value) is not None + + +def _is_hex64(value: Any) -> bool: + return isinstance(value, str) and _HEX64_RE.fullmatch(value) is not None + + +# --------------------------------------------------------------------------- +# Server fixture +# --------------------------------------------------------------------------- + + +@pytest.fixture +def server(tmp_path: Path) -> Iterator[StellarRpcServer]: + port = _find_free_port() + srv = StellarRpcServer( + host='localhost', + port=port, + io_dir=tmp_path, + network_passphrase=PASSPHRASE, + ) + thread = threading.Thread(target=srv.serve, daemon=True) + thread.start() + _wait_for_server('localhost', port) + yield srv + srv.shutdown() + + +# --------------------------------------------------------------------------- +# Transaction submit / deploy helpers +# +# Every lifecycle test builds on the same primitives: submit a signed transaction and +# confirm it reaches SUCCESS, fund a fresh account, deploy a contract from a .wat file, and +# invoke functions on it. They compose — ``deploy_and_get_invoker`` is just the three-step +# account -> upload -> deploy setup wrapped around ``make_invoker``. +# --------------------------------------------------------------------------- + + +class Deployed(NamedTuple): + """A deployed contract instance: its C-strkey address, wasm hash, and wasm bytecode.""" + + address: str + wasm_hash: bytes + wasm_bytecode: bytes + + +def send_tx(server: StellarRpcServer, keypair: Keypair, tb: TransactionBuilder) -> tuple[str, dict[str, Any]]: + """Sign, submit, and confirm a transaction. + + Asserts the submission is accepted (PENDING) and the transaction executes to SUCCESS. + Returns ``(tx_hash, getTransaction_result)``. + """ + env = tb.set_timeout(30).build() + env.sign(keypair) + res = _rpc(server.port(), 'sendTransaction', {'transaction': env.to_xdr()}) + assert res['result']['status'] == 'PENDING' + tx_hash = res['result']['hash'] + get_res = _rpc(server.port(), 'getTransaction', {'hash': tx_hash})['result'] + assert get_res['status'] == 'SUCCESS', f'Transaction failed: {get_res}' + return tx_hash, get_res + + +def fund_account(server: StellarRpcServer) -> tuple[Keypair, Account]: + """Create a fresh, funded account; return its keypair and sequence-tracking ``Account``.""" + keypair = Keypair.random() + account = Account(keypair.public_key, sequence=0) + tb = TransactionBuilder(account, PASSPHRASE).append_create_account_op(keypair.public_key, '1000') + send_tx(server, keypair, tb) + return keypair, account + + +def deploy_contract(server: StellarRpcServer, keypair: Keypair, account: Account, wat_path: Path) -> Deployed: + """Upload ``wat_path`` and deploy an instance from it, signed by ``keypair``/``account``. + + The account must already exist. Submits two transactions (upload + create) and returns + the deployed contract's address, wasm hash, and wasm bytecode. + """ + + def builder() -> TransactionBuilder: + return TransactionBuilder(account, PASSPHRASE) + + wasm_bytecode = wat_to_wasm(wat_path) + send_tx(server, keypair, builder().append_upload_contract_wasm_op(wasm_bytecode)) + wasm_hash = sha256(wasm_bytecode) + salt = b'\x00' * 32 + send_tx(server, keypair, builder().append_create_contract_op(wasm_hash, keypair.public_key, None, salt)) + address = server.encoder.contract_address_from_deployer_address(keypair.public_key, salt) + return Deployed(address=address, wasm_hash=wasm_hash, wasm_bytecode=wasm_bytecode) + + +def make_invoker( + server: StellarRpcServer, keypair: Keypair, account: Account, contract_address: str +) -> Callable[..., str]: + """Return an ``invoke(func, args=None) -> tx_hash`` callable that asserts each call SUCCEEDs.""" + + def invoke(func: str, args: list[xdr.SCVal] | None = None) -> str: + tb = TransactionBuilder(account, PASSPHRASE).append_invoke_contract_function_op( + contract_address, func, args or [] + ) + tx_hash, _ = send_tx(server, keypair, tb) + return tx_hash + + return invoke + + +def deploy_and_get_invoker(server: StellarRpcServer, wat_path: Path) -> Callable[..., str]: + """Fund an account, upload ``wat_path``, and deploy a contract instance from it. + + Returns an ``invoke(func, args=None)`` callable that runs a contract function and returns + the executed transaction's hash, asserting the whole setup and each call reaches SUCCESS. + """ + keypair, account = fund_account(server) + deployed = deploy_contract(server, keypair, account, wat_path) + return make_invoker(server, keypair, account, deployed.address) diff --git a/src/tests/integration/test_get_events.py b/src/tests/integration/test_get_events.py index 3b562e3..8760feb 100644 --- a/src/tests/integration/test_get_events.py +++ b/src/tests/integration/test_get_events.py @@ -13,26 +13,20 @@ from __future__ import annotations -import json import re -import socket -import subprocess -import threading -import time -import urllib.request from datetime import datetime from pathlib import Path from typing import TYPE_CHECKING, Any -import pytest -from stellar_sdk import Account, Keypair, Network, StrKey, TransactionBuilder, xdr -from stellar_sdk.utils import sha256 +from stellar_sdk import StrKey, TransactionBuilder, xdr from stellar_sdk.xdr.sc_val_type import SCValType -from komet_node.server import StellarRpcServer +from .conftest import PASSPHRASE, _is_number, _rpc, deploy_contract, fund_account, send_tx if TYPE_CHECKING: - from collections.abc import Iterator + from stellar_sdk import Account, Keypair + + from komet_node.server import StellarRpcServer EVENTS_CONTRACT_WAT = (Path(__file__).parent / 'data' / 'wasm' / 'events.wat').resolve(strict=True) @@ -47,105 +41,26 @@ # --------------------------------------------------------------------------- -# Helpers (mirrored from test_server.py so this module stays self-contained) +# Helpers (the server fixture, RPC plumbing, and deploy helpers live in conftest.py) # --------------------------------------------------------------------------- -def wat_to_wasm(wat_path: Path) -> bytes: - proc_res = subprocess.run(['wat2wasm', str(wat_path), '--output=/dev/stdout'], check=True, capture_output=True) - return proc_res.stdout - - -def _find_free_port() -> int: - with socket.socket() as s: - s.bind(('', 0)) - return s.getsockname()[1] - - -def _wait_for_server(host: str, port: int, timeout: float = 10.0) -> None: - deadline = time.time() + timeout - while time.time() < deadline: - try: - with socket.create_connection((host, port), timeout=0.1): - return - except OSError: - time.sleep(0.05) - raise TimeoutError(f'Server did not start on {host}:{port}') - - -def _rpc(port: int, method: str, params: dict[str, Any]) -> dict[str, Any]: - body = json.dumps({'jsonrpc': '2.0', 'id': 1, 'method': method, 'params': params}).encode() - req = urllib.request.Request( - f'http://localhost:{port}', - data=body, - headers={'Content-Type': 'application/json'}, - ) - with urllib.request.urlopen(req) as resp: - return json.loads(resp.read()) - - -@pytest.fixture -def server(tmp_path: Path) -> Iterator[StellarRpcServer]: - port = _find_free_port() - srv = StellarRpcServer( - host='localhost', - port=port, - io_dir=tmp_path, - network_passphrase=Network.TESTNET_NETWORK_PASSPHRASE, - ) - thread = threading.Thread(target=srv.serve, daemon=True) - thread.start() - _wait_for_server('localhost', port) - yield srv - srv.shutdown() - - -def _send(server: StellarRpcServer, keypair: Keypair, tb: TransactionBuilder) -> str: - """Sign, submit, and confirm a transaction; return its hash.""" - env = tb.set_timeout(30).build() - env.sign(keypair) - res = _rpc(server.port(), 'sendTransaction', {'transaction': env.to_xdr()}) - assert res['result']['status'] == 'PENDING' - tx_hash = res['result']['hash'] - get_res = _rpc(server.port(), 'getTransaction', {'hash': tx_hash})['result'] - assert get_res['status'] == 'SUCCESS', f'Transaction failed: {get_res}' - return tx_hash - - def _deploy_events_contract(server: StellarRpcServer) -> tuple[Keypair, Account, str]: """Create an account, upload events.wat, and deploy it (ledgers 1-3). Returns the funding keypair, its (mutated) sequence-tracking account, and the C-strkey address of the deployed contract. """ - keypair = Keypair.random() - account = Account(keypair.public_key, sequence=0) - - def builder() -> TransactionBuilder: - return TransactionBuilder(account, Network.TESTNET_NETWORK_PASSPHRASE) - - _send(server, keypair, builder().append_create_account_op(keypair.public_key, '1000')) - - wasm_bytecode = wat_to_wasm(EVENTS_CONTRACT_WAT) - _send(server, keypair, builder().append_upload_contract_wasm_op(wasm_bytecode)) - - wasm_hash = sha256(wasm_bytecode) - salt = b'\x00' * 32 - _send(server, keypair, builder().append_create_contract_op(wasm_hash, keypair.public_key, None, salt)) - - contract_address = server.encoder.contract_address_from_deployer_address(keypair.public_key, salt) - return keypair, account, contract_address + keypair, account = fund_account(server) + deployed = deploy_contract(server, keypair, account, EVENTS_CONTRACT_WAT) + return keypair, account, deployed.address def _emit_event(server: StellarRpcServer, keypair: Keypair, account: Account, contract_address: str) -> str: """Invoke the contract's `emit` function; return the transaction hash.""" - tb = TransactionBuilder(account, Network.TESTNET_NETWORK_PASSPHRASE) - return _send(server, keypair, tb.append_invoke_contract_function_op(contract_address, 'emit', [])) - - -def _is_json_number(value: Any) -> bool: - """True for a JSON number decoded to int (bool is an int subclass, so exclude it).""" - return isinstance(value, int) and not isinstance(value, bool) + tb = TransactionBuilder(account, PASSPHRASE).append_invoke_contract_function_op(contract_address, 'emit', []) + tx_hash, _ = send_tx(server, keypair, tb) + return tx_hash def _assert_request_error(response: dict[str, Any]) -> None: @@ -218,14 +133,11 @@ def test_get_events_rejects_invalid_xdr_format(server: StellarRpcServer) -> None def test_get_events_no_events_returns_empty_array(server: StellarRpcServer) -> None: """A range that contains no contract events yields an empty events array, not null.""" - keypair = Keypair.random() - account = Account(keypair.public_key, sequence=0) - tb = TransactionBuilder(account, Network.TESTNET_NETWORK_PASSPHRASE) - _send(server, keypair, tb.append_create_account_op(keypair.public_key, '1000')) + fund_account(server) result = _rpc(server.port(), 'getEvents', {'startLedger': 1})['result'] assert result['events'] == [] - assert _is_json_number(result['latestLedger']) + assert _is_number(result['latestLedger']) assert result['latestLedger'] == 1 # Real stellar-rpc always populates the cursor (the position to resume scanning from). assert isinstance(result['cursor'], str) @@ -237,7 +149,7 @@ def test_get_events_returns_emitted_contract_event(server: StellarRpcServer) -> tx_hash = _emit_event(server, keypair, account, contract_address) # ledger 4 result = _rpc(server.port(), 'getEvents', {'startLedger': 1})['result'] - assert _is_json_number(result['latestLedger']) + assert _is_number(result['latestLedger']) assert result['latestLedger'] == 4 assert isinstance(result['cursor'], str) @@ -246,7 +158,7 @@ def test_get_events_returns_emitted_contract_event(server: StellarRpcServer) -> event = result['events'][0] assert event['type'] == 'contract' - assert _is_json_number(event['ledger']) + assert _is_number(event['ledger']) assert event['ledger'] == 4 # ledgerClosedAt is an ISO-8601 timestamp string. assert isinstance(event['ledgerClosedAt'], str) diff --git a/src/tests/integration/test_integration.py b/src/tests/integration/test_integration.py index 76dfa84..4de0d68 100644 --- a/src/tests/integration/test_integration.py +++ b/src/tests/integration/test_integration.py @@ -10,7 +10,6 @@ import json from typing import TYPE_CHECKING -from komet_node.hello import hello from komet_node.interpreter import NodeInterpreter if TYPE_CHECKING: @@ -19,10 +18,6 @@ import pytest -def test_hello() -> None: - assert hello('World') == 'Hello, World!' - - def test_empty_config_ignores_stray_request_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: """empty_config() must produce a clean idle state regardless of the process cwd. diff --git a/src/tests/integration/test_server.py b/src/tests/integration/test_server.py index bf90560..5bdc569 100644 --- a/src/tests/integration/test_server.py +++ b/src/tests/integration/test_server.py @@ -2,26 +2,35 @@ import importlib.metadata import json -import re import shutil -import socket -import subprocess -import threading import time -import urllib.request from pathlib import Path from typing import TYPE_CHECKING, Any -import pytest from stellar_sdk import Account, Address, Asset, Keypair, Network, StrKey, TransactionBuilder, xdr from stellar_sdk.utils import sha256 from stellar_sdk.xdr.sc_val_type import SCValType +from komet_node.scval import scval_from_json from komet_node.server import StellarRpcServer -if TYPE_CHECKING: - from collections.abc import Callable, Iterator +from .conftest import ( + PASSPHRASE, + _is_hex64, + _is_int_string, + _is_number, + _post, + _post_raw, + _rpc, + deploy_and_get_invoker, + deploy_contract, + fund_account, + make_invoker, + send_tx, + wat_to_wasm, +) +if TYPE_CHECKING: from stellar_sdk import TransactionEnvelope EMPTY_CONTRACT_WAT = (Path(__file__).parent / 'data' / 'wasm' / 'empty.wat').resolve(strict=True) @@ -31,71 +40,6 @@ STORAGE_CONTRACT_WAT = (Path(__file__).parent / 'data' / 'wasm' / 'storage.wat').resolve(strict=True) -def wat_to_wasm(wat_path: Path) -> bytes: - proc_res = subprocess.run(['wat2wasm', str(wat_path), '--output=/dev/stdout'], check=True, capture_output=True) - return proc_res.stdout - - -def _find_free_port() -> int: - with socket.socket() as s: - s.bind(('', 0)) - return s.getsockname()[1] - - -def _wait_for_server(host: str, port: int, timeout: float = 10.0) -> None: - deadline = time.time() + timeout - while time.time() < deadline: - try: - with socket.create_connection((host, port), timeout=0.1): - return - except OSError: - time.sleep(0.05) - raise TimeoutError(f'Server did not start on {host}:{port}') - - -def _rpc(port: int, method: str, params: dict[str, Any]) -> dict[str, Any]: - body = json.dumps({'jsonrpc': '2.0', 'id': 1, 'method': method, 'params': params}).encode() - return _post(port, body) - - -def _post(port: int, body: bytes) -> Any: - return json.loads(_post_raw(port, body)) - - -def _post_raw(port: int, body: bytes) -> bytes: - req = urllib.request.Request( - f'http://localhost:{port}', - data=body, - headers={'Content-Type': 'application/json'}, - ) - with urllib.request.urlopen(req) as resp: - return resp.read() - - -# Spec-shape helpers. The official serialization rules come from the Go structs in -# stellar/go-stellar-sdk protocols/rpc (what real stellar-rpc emits): ledger sequence -# numbers and protocolVersion are JSON numbers; the close-time fields on the singular -# methods are int64 with Go's `,string` encoding, i.e. JSON strings holding a decimal -# integer; hashes are 64 lowercase hex characters. - -_HEX64_RE = re.compile(r'[0-9a-f]{64}') -_INT_STRING_RE = re.compile(r'-?[0-9]+') - - -def _is_number(value: Any) -> bool: - """True for a JSON number decoded to int (bool is a distinct JSON type).""" - return isinstance(value, int) and not isinstance(value, bool) - - -def _is_int_string(value: Any) -> bool: - """True for a JSON string holding a decimal integer (Go int64 `,string` encoding).""" - return isinstance(value, str) and _INT_STRING_RE.fullmatch(value) is not None - - -def _is_hex64(value: Any) -> bool: - return isinstance(value, str) and _HEX64_RE.fullmatch(value) is not None - - def _assert_ledger_bounds(result: dict[str, Any]) -> None: """Check the latest/oldest ledger-range fields required on every getTransaction response.""" assert _is_number(result['latestLedger']) @@ -137,57 +81,6 @@ def _create_account_xdr(keypair: Keypair, account: Account) -> str: return envelope.to_xdr() -@pytest.fixture -def server(tmp_path: Path) -> Iterator[StellarRpcServer]: - port = _find_free_port() - srv = StellarRpcServer( - host='localhost', - port=port, - io_dir=tmp_path, - network_passphrase=Network.TESTNET_NETWORK_PASSPHRASE, - ) - thread = threading.Thread(target=srv.serve, daemon=True) - thread.start() - _wait_for_server('localhost', port) - yield srv - srv.shutdown() - - -def _deploy_and_get_invoker(server: StellarRpcServer, wat_path: Path) -> Callable[..., str]: - """Create an account, upload `wat_path`, and deploy a contract instance from it. - - Returns an ``invoke(func, args=None)`` callable that runs a contract function and returns - the executed transaction's hash, asserting the whole setup and each call reaches SUCCESS. - """ - keypair = Keypair.random() - account = Account(keypair.public_key, sequence=0) - - def builder() -> TransactionBuilder: - return TransactionBuilder(account, Network.TESTNET_NETWORK_PASSPHRASE) - - def send(tb: TransactionBuilder) -> str: - env = tb.set_timeout(30).build() - env.sign(keypair) - res = _rpc(server.port(), 'sendTransaction', {'transaction': env.to_xdr()}) - assert res['result']['status'] == 'PENDING' - tx_hash = res['result']['hash'] - get_res = _rpc(server.port(), 'getTransaction', {'hash': tx_hash})['result'] - assert get_res['status'] == 'SUCCESS', f'Transaction failed: {get_res}' - return tx_hash - - send(builder().append_create_account_op(keypair.public_key, '1000')) - wasm_bytecode = wat_to_wasm(wat_path) - send(builder().append_upload_contract_wasm_op(wasm_bytecode)) - salt = b'\x00' * 32 - send(builder().append_create_contract_op(sha256(wasm_bytecode), keypair.public_key, None, salt)) - contract_address = server.encoder.contract_address_from_deployer_address(keypair.public_key, salt) - - def invoke(func: str, args: list[xdr.SCVal] | None = None) -> str: - return send(builder().append_invoke_contract_function_op(contract_address, func, args or [])) - - return invoke - - def test_default_io_dir_is_a_fresh_temp_dir() -> None: """With no io_dir, the server provisions a fresh temporary directory and seeds it.""" srv = StellarRpcServer(host='localhost', port=0) @@ -520,14 +413,10 @@ def test_ledger_seq_increments(server: StellarRpcServer) -> None: account = Account(keypair.public_key, sequence=0) def send_create_account() -> None: - envelope = ( - TransactionBuilder(account, Network.TESTNET_NETWORK_PASSPHRASE) - .append_create_account_op(destination=keypair.public_key, starting_balance='1000') - .set_timeout(30) - .build() + tb = TransactionBuilder(account, PASSPHRASE).append_create_account_op( + destination=keypair.public_key, starting_balance='1000' ) - envelope.sign(keypair) - _rpc(server.port(), 'sendTransaction', {'transaction': envelope.to_xdr()}) + send_tx(server, keypair, tb) send_create_account() assert _rpc(server.port(), 'getLatestLedger', {})['result']['sequence'] == 1 @@ -537,43 +426,13 @@ def send_create_account() -> None: def test_full_lifecycle_over_http(server: StellarRpcServer) -> None: - """Full contract lifecycle through the HTTP server: account → upload → deploy → invoke.""" - keypair = Keypair.random() - account = Account(keypair.public_key, sequence=0) + """Full contract lifecycle through the HTTP server: account → upload → deploy → invoke. - def send(envelope_xdr: str) -> dict[str, Any]: - send_res = _rpc(server.port(), 'sendTransaction', {'transaction': envelope_xdr}) - assert send_res['result']['status'] == 'PENDING' - tx_hash = send_res['result']['hash'] - get_res = _rpc(server.port(), 'getTransaction', {'hash': tx_hash}) - assert get_res['result']['status'] == 'SUCCESS', f'Transaction failed: {get_res}' - return get_res['result'] - - def builder() -> TransactionBuilder: - return TransactionBuilder(account, Network.TESTNET_NETWORK_PASSPHRASE) - - def sign_and_xdr(tb: TransactionBuilder) -> str: - env = tb.set_timeout(30).build() - env.sign(keypair) - return env.to_xdr() - - # 1. Create account - send(sign_and_xdr(builder().append_create_account_op(keypair.public_key, '1000'))) - - # 2. Upload wasm - wasm_bytecode = wat_to_wasm(EMPTY_CONTRACT_WAT) - send(sign_and_xdr(builder().append_upload_contract_wasm_op(wasm_bytecode))) - - # 3. Deploy contract - from stellar_sdk.utils import sha256 - - wasm_hash = sha256(wasm_bytecode) - salt = b'\x00' * 32 - send(sign_and_xdr(builder().append_create_contract_op(wasm_hash, keypair.public_key, None, salt))) - - # 4. Invoke foo() - contract_address = server.encoder.contract_address_from_deployer_address(keypair.public_key, salt) - send(sign_and_xdr(builder().append_invoke_contract_function_op(contract_address, 'foo', []))) + Each step asserts SUCCESS inside the shared helpers; this is the end-to-end smoke test + that the whole pipeline works over HTTP, independent of any trace/return-value assertions. + """ + invoke = deploy_and_get_invoker(server, EMPTY_CONTRACT_WAT) + invoke('foo') def test_trace_transaction_retrieves_trace_by_hash(server: StellarRpcServer) -> None: @@ -624,7 +483,7 @@ def test_trace_transaction_returns_full_instruction_trace_for_foo(server: Stella result is caught. The entry/exit frames carry per-run contract and account ids, so they are checked structurally rather than by value. """ - invoke = _deploy_and_get_invoker(server, EMPTY_CONTRACT_WAT) + invoke = deploy_and_get_invoker(server, EMPTY_CONTRACT_WAT) tx_hash = invoke('foo') trace = _rpc(server.port(), 'traceTransaction', {'hash': tx_hash})['result'] @@ -662,7 +521,7 @@ def test_trace_records_have_expected_structure_and_reflect_arguments(server: Ste takes arguments the arguments are bound as locals while intermediate values build up on the stack — exercising a richer trace than the argument-less ``foo()`` case. """ - invoke = _deploy_and_get_invoker(server, ARGS_CONTRACT_WAT) + invoke = deploy_and_get_invoker(server, ARGS_CONTRACT_WAT) tx_hash = invoke( 'test_integers', [ @@ -713,15 +572,23 @@ def test_trace_records_have_expected_structure_and_reflect_arguments(server: Ste def test_call_tx_with_args(server: StellarRpcServer) -> None: - """Exercise the scval_to_json / #decodeArg pipeline for each supported SCVal type. + """The scval_to_json / #decodeArg pipeline decodes each supported SCVal arg type correctly. Uses a minimal contract (args.wat) whose functions accept various arg types and return - Void. Covers: bool, u32, i32, u64, i64, u128, i128, symbol. + Void. For each call the arguments echoed in the trace's ``callContract`` frame must + round-trip back to the exact SCVals that were sent — so a decoding bug is caught even + when the transaction still succeeds. Covers: bool, u32, i32, u64, i64, u128, i128, symbol. """ - invoke = _deploy_and_get_invoker(server, ARGS_CONTRACT_WAT) + invoke = deploy_and_get_invoker(server, ARGS_CONTRACT_WAT) - invoke('test_bool', [xdr.SCVal(type=SCValType.SCV_BOOL, b=True)]) - invoke( + def assert_args_round_trip(func: str, args: list[xdr.SCVal]) -> None: + tx_hash = invoke(func, args) + entry = _rpc(server.port(), 'traceTransaction', {'hash': tx_hash})['result'][0] + assert entry['function'] == func + assert [scval_from_json(arg) for arg in entry['args']] == args + + assert_args_round_trip('test_bool', [xdr.SCVal(type=SCValType.SCV_BOOL, b=True)]) + assert_args_round_trip( 'test_integers', [ xdr.SCVal(type=SCValType.SCV_U32, u32=xdr.Uint32(42)), @@ -730,14 +597,14 @@ def test_call_tx_with_args(server: StellarRpcServer) -> None: xdr.SCVal(type=SCValType.SCV_I64, i64=xdr.Int64(-200)), ], ) - invoke( + assert_args_round_trip( 'test_wide_integers', [ xdr.SCVal(type=SCValType.SCV_U128, u128=xdr.UInt128Parts(hi=xdr.Uint64(0), lo=xdr.Uint64(999))), xdr.SCVal(type=SCValType.SCV_I128, i128=xdr.Int128Parts(hi=xdr.Int64(0), lo=xdr.Uint64(888))), ], ) - invoke('test_symbol', [xdr.SCVal(type=SCValType.SCV_SYMBOL, sym=xdr.SCSymbol(sc_symbol=b'hello'))]) + assert_args_round_trip('test_symbol', [xdr.SCVal(type=SCValType.SCV_SYMBOL, sym=xdr.SCSymbol(sc_symbol=b'hello'))]) def test_call_tx_with_return_value(server: StellarRpcServer) -> None: @@ -748,44 +615,17 @@ def test_call_tx_with_return_value(server: StellarRpcServer) -> None: stuck in the semantics and was recorded as FAILED. ``uncheckedCallTx`` drops the return value check. """ - keypair = Keypair.random() - account = Account(keypair.public_key, sequence=0) - - def send(tb: TransactionBuilder) -> None: - env = tb.set_timeout(30).build() - env.sign(keypair) - res = _rpc(server.port(), 'sendTransaction', {'transaction': env.to_xdr()}) - assert res['result']['status'] == 'PENDING' - tx_hash = res['result']['hash'] - get_res = _rpc(server.port(), 'getTransaction', {'hash': tx_hash})['result'] - assert get_res['status'] == 'SUCCESS', f'Transaction failed: {get_res}' - - def builder() -> TransactionBuilder: - return TransactionBuilder(account, Network.TESTNET_NETWORK_PASSPHRASE) + keypair, account = fund_account(server) # ledger 1 + deployed = deploy_contract(server, keypair, account, ADDER_CONTRACT_WAT) # ledgers 2, 3 + invoke = make_invoker(server, keypair, account, deployed.address) - # Set up: create account, upload adder.wat, deploy contract - send(builder().append_create_account_op(keypair.public_key, '1000')) - - wasm_bytecode = wat_to_wasm(ADDER_CONTRACT_WAT) - send(builder().append_upload_contract_wasm_op(wasm_bytecode)) - - from stellar_sdk.utils import sha256 - - wasm_hash = sha256(wasm_bytecode) - salt = b'\x00' * 32 - send(builder().append_create_contract_op(wasm_hash, keypair.public_key, None, salt)) - - # add(2, 3) returns U32(5), not Void — the send() helper asserts SUCCESS. - contract_address = server.encoder.contract_address_from_deployer_address(keypair.public_key, salt) - send( - builder().append_invoke_contract_function_op( - contract_address, - 'add', - [ - xdr.SCVal(type=SCValType.SCV_U32, u32=xdr.Uint32(2)), - xdr.SCVal(type=SCValType.SCV_U32, u32=xdr.Uint32(3)), - ], - ) + # add(2, 3) returns U32(5), not Void — make_invoker asserts the call reaches SUCCESS. + invoke( + 'add', + [ + xdr.SCVal(type=SCValType.SCV_U32, u32=xdr.Uint32(2)), + xdr.SCVal(type=SCValType.SCV_U32, u32=xdr.Uint32(3)), + ], ) # All four transactions, including the non-Void invocation, advanced the ledger. @@ -839,20 +679,11 @@ def _assert_ledger_entry_shape(entry: dict[str, Any], expected_key: str) -> None assert type(entry['liveUntilLedgerSeq']) is int -def _send_tx(server: StellarRpcServer, keypair: Keypair, tb: TransactionBuilder) -> None: - env = tb.set_timeout(30).build() - env.sign(keypair) - res = _rpc(server.port(), 'sendTransaction', {'transaction': env.to_xdr()}) - assert res['result']['status'] == 'PENDING' - get_res = _rpc(server.port(), 'getTransaction', {'hash': res['result']['hash']})['result'] - assert get_res['status'] == 'SUCCESS', f'Transaction failed: {get_res}' - - def test_get_ledger_entries_account(server: StellarRpcServer) -> None: """An ACCOUNT ledger key resolves to an AccountEntry; unknown keys are silently dropped.""" keypair = Keypair.random() account = Account(keypair.public_key, sequence=0) - _send_tx( + send_tx( server, keypair, TransactionBuilder(account, Network.TESTNET_NETWORK_PASSPHRASE).append_create_account_op( @@ -883,24 +714,12 @@ def test_get_ledger_entries_account(server: StellarRpcServer) -> None: def test_get_ledger_entries_contract_code_and_data(server: StellarRpcServer) -> None: """CONTRACT_CODE, the CONTRACT_DATA instance entry, and persistent CONTRACT_DATA storage.""" - from stellar_sdk.utils import sha256 - - keypair = Keypair.random() - account = Account(keypair.public_key, sequence=0) - - def builder() -> TransactionBuilder: - return TransactionBuilder(account, Network.TESTNET_NETWORK_PASSPHRASE) - # Set up: create account, upload storage.wat, deploy, invoke store() which writes the # persistent storage entry U32(7) -> U32(42). - _send_tx(server, keypair, builder().append_create_account_op(keypair.public_key, '1000')) - wasm_bytecode = wat_to_wasm(STORAGE_CONTRACT_WAT) - _send_tx(server, keypair, builder().append_upload_contract_wasm_op(wasm_bytecode)) - wasm_hash = sha256(wasm_bytecode) - salt = b'\x00' * 32 - _send_tx(server, keypair, builder().append_create_contract_op(wasm_hash, keypair.public_key, None, salt)) - contract_address = server.encoder.contract_address_from_deployer_address(keypair.public_key, salt) - _send_tx(server, keypair, builder().append_invoke_contract_function_op(contract_address, 'store', [])) + keypair, account = fund_account(server) + deployed = deploy_contract(server, keypair, account, STORAGE_CONTRACT_WAT) + wasm_hash, wasm_bytecode, contract_address = deployed.wasm_hash, deployed.wasm_bytecode, deployed.address + make_invoker(server, keypair, account, contract_address)('store') code_key = _contract_code_ledger_key(wasm_hash) instance_key = _contract_data_ledger_key( @@ -1499,28 +1318,13 @@ def test_batch_of_only_notifications_returns_nothing(server: StellarRpcServer) - def _deploy_adder_contract(server: StellarRpcServer, keypair: Keypair, account: Account) -> str: """Create the account, upload adder.wat, and deploy it; return the contract address. - Submits three transactions, so the ledger sequence afterwards is 3. + Takes the caller's keypair/account (the simulate tests reuse the account's sequence to + build a follow-up unsigned invocation). Submits three transactions, so the ledger + sequence afterwards is 3. """ - from stellar_sdk.utils import sha256 - - def send(tb: TransactionBuilder) -> None: - env = tb.set_timeout(30).build() - env.sign(keypair) - res = _rpc(server.port(), 'sendTransaction', {'transaction': env.to_xdr()}) - assert res['result']['status'] == 'PENDING' - get_res = _rpc(server.port(), 'getTransaction', {'hash': res['result']['hash']})['result'] - assert get_res['status'] == 'SUCCESS', f'Transaction failed: {get_res}' - - def builder() -> TransactionBuilder: - return TransactionBuilder(account, Network.TESTNET_NETWORK_PASSPHRASE) - - send(builder().append_create_account_op(keypair.public_key, '1000')) - wasm_bytecode = wat_to_wasm(ADDER_CONTRACT_WAT) - send(builder().append_upload_contract_wasm_op(wasm_bytecode)) - wasm_hash = sha256(wasm_bytecode) - salt = b'\x00' * 32 - send(builder().append_create_contract_op(wasm_hash, keypair.public_key, None, salt)) - return server.encoder.contract_address_from_deployer_address(keypair.public_key, salt) + tb = TransactionBuilder(account, PASSPHRASE).append_create_account_op(keypair.public_key, '1000') + send_tx(server, keypair, tb) + return deploy_contract(server, keypair, account, ADDER_CONTRACT_WAT).address def _build_add_invocation(account: Account, contract_address: str) -> TransactionEnvelope: @@ -1753,15 +1557,10 @@ def test_get_transaction_invocation_reports_return_value(server: StellarRpcServe account = Account(keypair.public_key, sequence=0) def builder() -> TransactionBuilder: - return TransactionBuilder(account, Network.TESTNET_NETWORK_PASSPHRASE) + return TransactionBuilder(account, PASSPHRASE) def send(tb: TransactionBuilder) -> dict[str, Any]: - env = tb.set_timeout(30).build() - env.sign(keypair) - res = _rpc(server.port(), 'sendTransaction', {'transaction': env.to_xdr()}) - assert res['result']['status'] == 'PENDING' - get_res = _rpc(server.port(), 'getTransaction', {'hash': res['result']['hash']})['result'] - assert get_res['status'] == 'SUCCESS', f'Transaction failed: {get_res}' + _, get_res = send_tx(server, keypair, tb) # The internal K-side returnValue field must never leak into getTransaction responses. assert 'returnValue' not in get_res # Every successful receipt, whatever the operation, has a decodable txSUCCESS result. @@ -1774,8 +1573,6 @@ def send(tb: TransactionBuilder) -> dict[str, Any]: wasm_bytecode = wat_to_wasm(ADDER_CONTRACT_WAT) send(builder().append_upload_contract_wasm_op(wasm_bytecode)) - from stellar_sdk.utils import sha256 - wasm_hash = sha256(wasm_bytecode) salt = b'\x00' * 32 send(builder().append_create_contract_op(wasm_hash, keypair.public_key, None, salt)) @@ -1815,33 +1612,11 @@ def test_get_transaction_reports_empty_bytes_return_value(server: StellarRpcServ as ``"0"`` (an odd-length non-hex value that breaks the XDR rewrite and would leak the internal ``returnValue`` field to clients). """ - keypair = Keypair.random() - account = Account(keypair.public_key, sequence=0) - - def builder() -> TransactionBuilder: - return TransactionBuilder(account, Network.TESTNET_NETWORK_PASSPHRASE) - - def send(tb: TransactionBuilder) -> dict[str, Any]: - env = tb.set_timeout(30).build() - env.sign(keypair) - res = _rpc(server.port(), 'sendTransaction', {'transaction': env.to_xdr()}) - assert 'result' in res, f'sendTransaction failed: {res}' - get_res = _rpc(server.port(), 'getTransaction', {'hash': res['result']['hash']})['result'] - assert get_res['status'] == 'SUCCESS', f'Transaction failed: {get_res}' - return get_res - - send(builder().append_create_account_op(keypair.public_key, '1000')) + keypair, account = fund_account(server) + deployed = deploy_contract(server, keypair, account, BYTES_CONTRACT_WAT) - wasm_bytecode = wat_to_wasm(BYTES_CONTRACT_WAT) - send(builder().append_upload_contract_wasm_op(wasm_bytecode)) - - from stellar_sdk.utils import sha256 - - salt = b'\x00' * 32 - send(builder().append_create_contract_op(sha256(wasm_bytecode), keypair.public_key, None, salt)) - - contract_address = server.encoder.contract_address_from_deployer_address(keypair.public_key, salt) - invoke_result = send(builder().append_invoke_contract_function_op(contract_address, 'empty_bytes', [])) + invoke_tx_hash = make_invoker(server, keypair, account, deployed.address)('empty_bytes') + invoke_result = _rpc(server.port(), 'getTransaction', {'hash': invoke_tx_hash})['result'] assert 'returnValue' not in invoke_result # internal receipt field, must never be served return_value = _soroban_return_value(_tx_meta_of(invoke_result)) @@ -1860,24 +1635,19 @@ def test_get_transaction_failed_reports_error_result_xdr(server: StellarRpcServe keypair = Keypair.random() account = Account(keypair.public_key, sequence=0) - def send(tb: TransactionBuilder) -> str: - env = tb.set_timeout(30).build() - env.sign(keypair) - res = _rpc(server.port(), 'sendTransaction', {'transaction': env.to_xdr()}) - tx_hash = res['result']['hash'] - assert isinstance(tx_hash, str) - return tx_hash - def builder() -> TransactionBuilder: - return TransactionBuilder(account, Network.TESTNET_NETWORK_PASSPHRASE) + return TransactionBuilder(account, PASSPHRASE) # A successful reference transaction first, to compare receipt shapes across statuses. - ok_hash = send(builder().append_create_account_op(keypair.public_key, '1000')) - ok_result = _rpc(server.port(), 'getTransaction', {'hash': ok_hash})['result'] + _, ok_result = send_tx(server, keypair, builder().append_create_account_op(keypair.public_key, '1000')) assert ok_result['status'] == 'SUCCESS' + # A failing invocation of a never-deployed contract — submitted directly, since send_tx + # asserts SUCCESS and this transaction is expected to end up FAILED. missing_contract = StrKey.encode_contract(b'\x22' * 32) # valid C-strkey, never deployed - bad_hash = send(builder().append_invoke_contract_function_op(missing_contract, 'foo', [])) + env = builder().append_invoke_contract_function_op(missing_contract, 'foo', []).set_timeout(30).build() + env.sign(keypair) + bad_hash = _rpc(server.port(), 'sendTransaction', {'transaction': env.to_xdr()})['result']['hash'] get_result = _rpc(server.port(), 'getTransaction', {'hash': bad_hash})['result'] assert get_result['status'] == 'FAILED' diff --git a/src/tests/unit/test_unit.py b/src/tests/unit/test_unit.py index dc4c81f..9ffa57d 100644 --- a/src/tests/unit/test_unit.py +++ b/src/tests/unit/test_unit.py @@ -1,14 +1,9 @@ import pytest -from komet_node.hello import hello from komet_node.interpreter import NodeInterpreterError from komet_node.transaction import _xlm_to_stroops -def test_hello() -> None: - assert hello('World') == 'Hello, World!' - - def test_xlm_to_stroops_whole() -> None: assert _xlm_to_stroops('1000') == 10_000_000_000