25632331469

Committed 10 May 2026 03:18PM UTC coverage: 82.17% (-0.02%) from 82.191%

Build # 25632331469

Build Type

Pull #365

github

Committed by

web-flow

Commit Message

Merge 2ac5e1614 into e53450fe2

Pull Request Pull Request #365: sctesting: fix unclosed file ResourceWarning's using `NeoGoNode`

Coverage Stats

3 of 4 new or added lines in 1 file covered. (75.0%)

2 existing lines in 1 file now uncovered.

10010 of 12182 relevant lines covered (82.17%)

0.82 hits per line

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

92.45

/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, ""):
                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
            self._thread.join()

        if self._process is not None and self._process.stdout is not None:
            self._process.stdout.close()
        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:
57	self.config_path = config_path	×
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✔
66	self.prog += ".exe"	×
67	self.posix = False	×
68	if not self.data_dir.joinpath(self.prog).exists():	1✔
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, ""):	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✔
UNCOV 112	break	×
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✔
UNCOV 128	self._terminate = True	×
NEW 129	self._thread.join()	×
130
131	if self._process is not None and self._process.stdout is not None:	1✔
132	self._process.stdout.close()	1✔
133	log.debug("stopped")	1✔
134
135	def reset(self):	1✔
136	# neo-go uses an in memory database so there's no need to reset anything
137	pass	1✔
138
139	def _parse_config(self):	1✔
140	with open(self.config_path) as f:	1✔
141	config: dict = list(yaml.load_all(f, yaml.FullLoader))[0]	1✔
142	data = config["ApplicationConfiguration"]	1✔
143
144	consensus_wallet_password = data["Consensus"]["UnlockWallet"]["Password"]	1✔
145	self.wallet = wallet.Wallet.from_file(	1✔
146	str(self.consensus_wallet_path.absolute()),
147	passwords=[
148	consensus_wallet_password,
149	consensus_wallet_password,
150	],
151	)
152	self.wallet.accounts[0].label = "committee-signature"	1✔
153	self.wallet.accounts[1].label = "committee"	1✔
154	self.account_committee = self.wallet.accounts[1]	1✔
155	self.account_committee = self.wallet.import_multisig_address(	1✔
156	1, [self.wallet.account_default.public_key] # type: ignore
157	)
158
159	# TODO: warn if port is :0 because then we can't tell where the RPC server is running on
160	address = data["RPC"]["Addresses"][0]	1✔
161	host = f"http://{address}"	1✔
162	self.facade = ChainFacade(rpc_host=host)	1✔
163	self.facade._emit_log_marker = True	1✔

CityOfZion / neo-mamba / 25632331469

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