25177770475

Committed 30 Apr 2026 04:44PM UTC coverage: 83.331% (+2.6%) from 80.771%

Build # 25177770475

Build Type

push

github

Committed by

web-flow

Commit Message

sctesting: migrate boa-test-constructor (#353)

Coverage Stats

217 of 296 new or added lines in 3 files covered. (73.31%)

5174 of 6209 relevant lines covered (83.33%)

0.83 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.15

/neo3/sctesting/node.py

import json
import pathlib
import threading
import subprocess
import shlex
import logging
import sys
import time
import yaml
import platform
import re
from neo3.core import types
from neo3.wallet import wallet, account
from neo3.api.wrappers import ChainFacade
from neo3.api.helpers.txbuilder import START_IGNORE_RUNTIMELOG, STOP_IGNORE_RUNTIMELOG
from typing import Optional
from dataclasses import dataclass


log = logging.getLogger("neogo")
log.addHandler(logging.StreamHandler(sys.stdout))

RE_RUNTIME_LOG = re.compile(r"INFO\truntime log\t(\{.*})")

RE_CAPTURE_START_IGNORE_MARKER = re.compile(
    r"INFO\truntime log\t{\"tx\": \"(.*?)\", \"script\": \".*?\", \"msg\": \""
    + START_IGNORE_RUNTIMELOG
    + '"'
)

RE_CAPTURE_STOP_IGNORE_MARKER = re.compile(
    r"INFO\truntime log\t{\"tx\": \"(.*?)\", \"script\": \".*?\", \"msg\": \""
    + STOP_IGNORE_RUNTIMELOG
    + '"'
)


@dataclass
class RuntimeLog:
    txid: types.UInt256
    contract: types.UInt160
    msg: str


class NeoGoNode:
    wallet: wallet.Wallet
    account_committee: account.Account
    facade: ChainFacade
    runtime_logs: list[RuntimeLog]

    def __init__(self, config_path: Optional[str] = None):
        self.data_dir = pathlib.Path(__file__).parent.joinpath("data")
        if config_path is None:
            self.config_path = str(self.data_dir.joinpath("protocol.unittest.yml"))
            self.consensus_wallet_path = self.data_dir.joinpath("wallet1_solo.json")
        else:
            self.config_path = config_path
            self.consensus_wallet_path = pathlib.Path(config_path).parent.joinpath(
                "wallet1_solo.json"
            )

        self.system = platform.system().lower()
        self.prog = "neogo"
        self.posix = True
        if self.system == "windows":
            self.prog += ".exe"
            self.posix = False
        if not self.data_dir.joinpath(self.prog).exists():
            raise FileNotFoundError(
                f"Internal required file '{self.prog}' not found. If you installed from source run this command once `python scripts/download-node.py`"
            )

        self._thread: Optional[threading.Thread] = None
        self._process: Optional[subprocess.Popen[str]] = None
        self._ready = False
        self._terminate = False
        self._parse_config()
        self.runtime_logs = []

    def start(self):
        log.debug("starting")

        cmd = f"{self.data_dir}/{self.prog} node --config-file {self.config_path} --relative-path {self.data_dir}"

        self._process = subprocess.Popen(
            shlex.split(cmd, posix=self.posix),
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            bufsize=1,
            text=True,
            shell=False,
        )

        def process_stdout(process):
            capture = True
            for output in iter(process.stdout.readline, b""):
                if "RPC server already started" in output:
                    self._ready = True
                    # WARNING: do not terminate this loop. stdout must be read as long as the process lives otherwise
                    # we'll eventually hit the PIPE buffer limit and hang the child process.
                if RE_CAPTURE_START_IGNORE_MARKER.match(output) is not None:
                    capture = False
                elif RE_CAPTURE_STOP_IGNORE_MARKER.match(output) is not None:
                    capture = True
                elif (match := RE_RUNTIME_LOG.match(output)) is not None and capture:
                    logline = json.loads(match.group(1))
                    txid = types.UInt256.from_string(logline["tx"])
                    contract = types.UInt160.from_string(logline["script"])
                    msg = logline["msg"]
                    self.runtime_logs.append(RuntimeLog(txid, contract, msg))
                if self._terminate:
                    break

        self._thread = threading.Thread(target=process_stdout, args=(self._process,))
        self._thread.start()

        while not self._ready:
            time.sleep(0.0001)
        log.debug("running")

    def stop(self):
        log.debug("stopping")
        if self._process is not None:
            self._process.kill()
            self._process.wait()

        if self._thread is not None and self._thread.is_alive():
            self._terminate = True
        log.debug("stopped")

    def reset(self):
        # neo-go uses an in memory database so there's no need to reset anything
        pass

    def _parse_config(self):
        with open(self.config_path) as f:
            config: dict = list(yaml.load_all(f, yaml.FullLoader))[0]
            data = config["ApplicationConfiguration"]

            consensus_wallet_password = data["Consensus"]["UnlockWallet"]["Password"]
            self.wallet = wallet.Wallet.from_file(
                str(self.consensus_wallet_path.absolute()),
                passwords=[
                    consensus_wallet_password,
                    consensus_wallet_password,
                ],
            )
            self.wallet.accounts[0].label = "committee-signature"
            self.wallet.accounts[1].label = "committee"
            self.account_committee = self.wallet.accounts[1]
            self.account_committee = self.wallet.import_multisig_address(
                1, [self.wallet.account_default.public_key]  # type: ignore
            )

            # TODO: warn if port is :0 because then we can't tell where the RPC server is running on
            address = data["RPC"]["Addresses"][0]
            host = f"http://{address}"
            self.facade = ChainFacade(rpc_host=host)
            self.facade._emit_log_marker = True

1	import json	1✔
2	import pathlib	1✔
3	import threading	1✔
4	import subprocess	1✔
5	import shlex	1✔
6	import logging	1✔
7	import sys	1✔
8	import time	1✔
9	import yaml	1✔
10	import platform	1✔
11	import re	1✔
12	from neo3.core import types	1✔
13	from neo3.wallet import wallet, account	1✔
14	from neo3.api.wrappers import ChainFacade	1✔
15	from neo3.api.helpers.txbuilder import START_IGNORE_RUNTIMELOG, STOP_IGNORE_RUNTIMELOG	1✔
16	from typing import Optional	1✔
17	from dataclasses import dataclass	1✔
18
19
20	log = logging.getLogger("neogo")	1✔
21	log.addHandler(logging.StreamHandler(sys.stdout))	1✔
22
23	RE_RUNTIME_LOG = re.compile(r"INFO\truntime log\t(\{.*})")	1✔
24
25	RE_CAPTURE_START_IGNORE_MARKER = re.compile(	1✔
26	r"INFO\truntime log\t{\"tx\": \"(.?)\", \"script\": \".?\", \"msg\": \""
27	+ START_IGNORE_RUNTIMELOG
28	+ '"'
29	)
30
31	RE_CAPTURE_STOP_IGNORE_MARKER = re.compile(	1✔
32	r"INFO\truntime log\t{\"tx\": \"(.?)\", \"script\": \".?\", \"msg\": \""
33	+ STOP_IGNORE_RUNTIMELOG
34	+ '"'
35	)
36
37
38	@dataclass	1✔
39	class RuntimeLog:	1✔
40	txid: types.UInt256	1✔
41	contract: types.UInt160	1✔
42	msg: str	1✔
43
44
45	class NeoGoNode:	1✔
46	wallet: wallet.Wallet	1✔
47	account_committee: account.Account	1✔
48	facade: ChainFacade	1✔
49	runtime_logs: list[RuntimeLog]	1✔
50
51	def __init__(self, config_path: Optional[str] = None):	1✔
52	self.data_dir = pathlib.Path(__file__).parent.joinpath("data")	1✔
53	if config_path is None:	1✔
54	self.config_path = str(self.data_dir.joinpath("protocol.unittest.yml"))	1✔
55	self.consensus_wallet_path = self.data_dir.joinpath("wallet1_solo.json")	1✔
56	else:
NEW 57	self.config_path = config_path	×
NEW 58	self.consensus_wallet_path = pathlib.Path(config_path).parent.joinpath(	×
59	"wallet1_solo.json"
60	)
61
62	self.system = platform.system().lower()	1✔
63	self.prog = "neogo"	1✔
64	self.posix = True	1✔
65	if self.system == "windows":	1✔
NEW 66	self.prog += ".exe"	×
NEW 67	self.posix = False	×
68	if not self.data_dir.joinpath(self.prog).exists():	1✔
NEW 69	raise FileNotFoundError(	×
70	f"Internal required file '{self.prog}' not found. If you installed from source run this command once `python scripts/download-node.py`"
71	)
72
73	self._thread: Optional[threading.Thread] = None	1✔
74	self._process: Optional[subprocess.Popen[str]] = None	1✔
75	self._ready = False	1✔
76	self._terminate = False	1✔
77	self._parse_config()	1✔
78	self.runtime_logs = []	1✔
79
80	def start(self):	1✔
81	log.debug("starting")	1✔
82
83	cmd = f"{self.data_dir}/{self.prog} node --config-file {self.config_path} --relative-path {self.data_dir}"	1✔
84
85	self._process = subprocess.Popen(	1✔
86	shlex.split(cmd, posix=self.posix),
87	stdout=subprocess.PIPE,
88	stderr=subprocess.STDOUT,
89	bufsize=1,
90	text=True,
91	shell=False,
92	)
93
94	def process_stdout(process):	1✔
95	capture = True	1✔
96	for output in iter(process.stdout.readline, b""):	1✔
97	if "RPC server already started" in output:	1✔
98	self._ready = True	1✔
99	# WARNING: do not terminate this loop. stdout must be read as long as the process lives otherwise
100	# we'll eventually hit the PIPE buffer limit and hang the child process.
101	if RE_CAPTURE_START_IGNORE_MARKER.match(output) is not None:	1✔
102	capture = False	1✔
103	elif RE_CAPTURE_STOP_IGNORE_MARKER.match(output) is not None:	1✔
104	capture = True	1✔
105	elif (match := RE_RUNTIME_LOG.match(output)) is not None and capture:	1✔
106	logline = json.loads(match.group(1))	1✔
107	txid = types.UInt256.from_string(logline["tx"])	1✔
108	contract = types.UInt160.from_string(logline["script"])	1✔
109	msg = logline["msg"]	1✔
110	self.runtime_logs.append(RuntimeLog(txid, contract, msg))	1✔
111	if self._terminate:	1✔
112	break	1✔
113
114	self._thread = threading.Thread(target=process_stdout, args=(self._process,))	1✔
115	self._thread.start()	1✔
116
117	while not self._ready:	1✔
118	time.sleep(0.0001)	1✔
119	log.debug("running")	1✔
120
121	def stop(self):	1✔
122	log.debug("stopping")	1✔
123	if self._process is not None:	1✔
124	self._process.kill()	1✔
125	self._process.wait()	1✔
126
127	if self._thread is not None and self._thread.is_alive():	1✔
128	self._terminate = True	1✔
129	log.debug("stopped")	1✔
130
131	def reset(self):	1✔
132	# neo-go uses an in memory database so there's no need to reset anything
133	pass	1✔
134
135	def _parse_config(self):	1✔
136	with open(self.config_path) as f:	1✔
137	config: dict = list(yaml.load_all(f, yaml.FullLoader))[0]	1✔
138	data = config["ApplicationConfiguration"]	1✔
139
140	consensus_wallet_password = data["Consensus"]["UnlockWallet"]["Password"]	1✔
141	self.wallet = wallet.Wallet.from_file(	1✔
142	str(self.consensus_wallet_path.absolute()),
143	passwords=[
144	consensus_wallet_password,
145	consensus_wallet_password,
146	],
147	)
148	self.wallet.accounts[0].label = "committee-signature"	1✔
149	self.wallet.accounts[1].label = "committee"	1✔
150	self.account_committee = self.wallet.accounts[1]	1✔
151	self.account_committee = self.wallet.import_multisig_address(	1✔
152	1, [self.wallet.account_default.public_key] # type: ignore
153	)
154
155	# TODO: warn if port is :0 because then we can't tell where the RPC server is running on
156	address = data["RPC"]["Addresses"][0]	1✔
157	host = f"http://{address}"	1✔
158	self.facade = ChainFacade(rpc_host=host)	1✔
159	self.facade._emit_log_marker = True	1✔

CityOfZion / neo-mamba / 25177770475

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous