PoS_EVM
/
pywiki
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[cli] BREAKING CHANGE - Refactor to static module

Browse Source
pull/2/head
Daniel Van Der Maden 5 years ago
parent 162abebc77
commit 08991c1a48
1 changed files with 176 additions and 184 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 180
      pyhmy/cli.py

180
pyhmy/cli.py
Unescape View File

@ -4,133 +4,115 @@ import os
import shutil
import re
from .util import get_bls_build_variables
from .util import get_bls_build_variables, get_gopath
DEFAULT_BIN_FILENAME = 'hmy'
_addresses = {} # Internal address keystore, not guaranteed to be up-to-date unless needed.
_binary_path = "hmy" # Internal binary path.
_environment = os.environ.copy() # Internal environment dict for Subprocess & Pexpect.
def get_environment():
def _get_default_hmy_binary_path(file_name="hmy"):
"""
:returns All the environment variables needed to run the CLI with dynamic linking.
Internal function to get the binary path by looking for the first file with
the same name as the param in the current working directory.
Note that this assumes that the BLS & MCL repo are in the appropriate directory
as stated here: https://github.com/harmony-one/harmony/blob/master/README.md
:param file_name: The file name to look for.
"""
environment = {"HOME": os.environ.get("HOME")}
environment.update(get_bls_build_variables())
return environment
assert '/' not in file_name, "file name must not be a path."
for root, dirs, files in os.walk(os.getcwd()):
if file_name in files:
return os.path.join(root, file_name)
return ""
class HmyCLI:
def __init__(self, environment, hmy_binary_path=None):
def _sync_addresses():
"""
:param environment: Dictionary of environment variables to be used when calling the CLI.
:param hmy_binary_path: The optional path to the CLI binary.
Internal function to sync address with the binary's keystore addresses.
"""
self.environment = environment
self.hmy_binary_path = ""
self.version = ""
self.keystore_path = ""
self._addresses = {}
if hmy_binary_path:
assert os.path.isfile(hmy_binary_path), f"{hmy_binary_path} is not a file"
self.hmy_binary_path = hmy_binary_path
else:
self._set_default_hmy_binary_path()
self._set_version()
self._set_keystore_path()
self._sync_addresses()
global _addresses
curr_addresses = {}
response = single_call("hmy keys list")
lines = response.split("\n")
if "NAME" not in lines[0] or "ADDRESS" not in lines[0]:
raise ValueError("Name or Address not found on first line of key list")
if lines[1] != "":
raise ValueError("Unknown format: No blank line between label and data")
for line in lines[2:]:
columns = line.split("\t")
if len(columns) != 2:
break # Done iterating through all of the addresses.
name, address = columns
curr_addresses[name.strip()] = address
_addresses = curr_addresses
def __repr__(self):
return f"<{self.version} @ {self.hmy_binary_path}>"
def _set_default_hmy_binary_path(self, file_name=DEFAULT_BIN_FILENAME):
def set_binary_path(path):
"""
Internal method to set the binary path by looking for the first file with
the same name as the param in the current working directory.
:param file_name: The file name to look for.
:param path: The path of the CLI binary to use.
"""
assert '/' not in file_name, "file name must not be a path."
for root, dirs, files in os.walk(os.getcwd()):
if file_name in files:
self.hmy_binary_path = os.path.join(root, file_name)
break
assert self.hmy_binary_path, f"CLI binary `{file_name}` not found in current working directory."
global _binary_path
assert os.path.isfile(path), f"`{path}` is not a file"
_binary_path = path
_sync_addresses()
def _set_version(self):
def get_version():
"""
Internal method to set this instance's version according to the binary's version.
:return: The version string of the CLI binary.
"""
proc = subprocess.Popen([self.hmy_binary_path, "version"], env=self.environment,
proc = subprocess.Popen([_binary_path, "version"], env=_environment,
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
out, err = proc.communicate()
if not err:
raise RuntimeError(f"Could not get version.\n"
f"\tGot exit code {proc.returncode}. Expected non-empty error message.")
self.version = err.decode().strip()
return err.decode().strip()
def _set_keystore_path(self):
def get_account_keystore_path():
"""
Internal method to set this instance's keystore path with the binary's keystore path.
:return: The account keystore path of the CLI binary.
"""
response = self.single_call("hmy keys location").strip()
response = single_call("hmy keys location").strip()
if not os.path.exists(response):
os.mkdir(response)
self.keystore_path = response
return response
def _sync_addresses(self):
"""
Internal method to sync this instance's address with the binary's keystore addresses.
"""
addresses = {}
response = self.single_call("hmy keys list")
lines = response.split("\n")
if "NAME" not in lines[0] or "ADDRESS" not in lines[0]:
raise ValueError("Name or Address not found on first line of key list")
if lines[1] != "":
raise ValueError("Unknown format: No blank line between label and data")
for line in lines[2:]:
columns = line.split("\t")
if len(columns) != 2:
break # Done iterating through all of the addresses.
name, address = columns
addresses[name.strip()] = address
self._addresses = addresses
def check_address(self, address):
def check_address(address):
"""
:param address: A 'one1...' address.
:return: Boolean of if the address is in the CLI's keystore.
"""
if address in self._addresses.values():
if address in _addresses.values():
return True
else:
self._sync_addresses()
return address in self._addresses.values()
_sync_addresses()
return address in _addresses.values()
def get_address(self, name):
def get_address(name):
"""
:param name: The alias of a key used in the CLI's keystore.
:return: The associated 'one1...' address.
"""
if name in self._addresses:
return self._addresses[name]
if name in _addresses:
return _addresses[name]
else:
self._sync_addresses()
return self._addresses.get(name, None)
_sync_addresses()
return _addresses.get(name, None)
def get_accounts(self, address):
def get_accounts(address):
"""
:param address: The 'one1...' address
:return: A list of account names associated with
:return: A list of account names associated with the param
"""
self._sync_addresses()
return [acc for acc, addr in self._addresses.items() if address == addr]
_sync_addresses()
return [acc for acc, addr in _addresses.items() if address == addr]
def remove_account(self, name):
def remove_account(name):
"""
Note that this edits the keystore directly since there is currently no
way to remove an address using the CLI.
@ -138,25 +120,28 @@ class HmyCLI:
:param name: The alias of a key used in the CLI's keystore.
:raises RuntimeError: If it failed to remove an account.
"""
if not self.get_address(name):
if not get_address(name):
return
keystore_path = f"{get_account_keystore_path()}/{name}"
try:
shutil.rmtree(f"{self.keystore_path}/{name}")
shutil.rmtree(keystore_path)
except (shutil.Error, FileNotFoundError) as err:
raise RuntimeError(f"Failed to delete dir: {self.keystore_path}/{name}\n"
raise RuntimeError(f"Failed to delete dir: {keystore_path}\n"
f"\tException: {err}") from err
del self._addresses[name]
del _addresses[name]
def remove_address(self, address):
def remove_address(address):
"""
:param address: The 'one1...' address to be removed.
"""
for name in self.get_accounts(address):
self.remove_account(name)
for name in get_accounts(address):
remove_account(name)
def single_call(self, command, timeout=60):
def single_call(command, timeout=60):
"""
:param command: String fo command to execute on CLI
:param command: String of command to execute on CLI
:param timeout: Optional timeout in seconds
:returns: Decoded string of response from hmy CLI call
:raises: RuntimeError if bad command
@ -164,15 +149,16 @@ class HmyCLI:
command_toks = command.split(" ")
if re.match(".*hmy", command_toks[0]):
command_toks = command_toks[1:]
command_toks = [self.hmy_binary_path] + command_toks
command_toks = [_binary_path] + command_toks
try:
response = subprocess.check_output(command_toks, env=self.environment, timeout=timeout).decode()
response = subprocess.check_output(command_toks, env=_environment, timeout=timeout).decode()
except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as err:
raise RuntimeError(f"Bad arguments for CLI.\n "
f"\tException: {err}") from err
return response
def expect_call(self, command, timeout=60):
def expect_call(command, timeout=60):
"""
:param command: String fo command to execute on CLI
:param timeout: Optional timeout in seconds
@ -183,8 +169,14 @@ class HmyCLI:
if re.match(".*hmy", command_toks[0]):
command_toks = command_toks[1:]
try:
proc = pexpect.spawn(f"{self.hmy_binary_path}", command_toks, env=self.environment, timeout=timeout)
proc = pexpect.spawn(f"{_binary_path}", command_toks, env=_environment, timeout=timeout)
except (pexpect.ExceptionPexpect, pexpect.TIMEOUT) as err:
raise RuntimeError(f"Bad arguments for CLI.\n "
f"\tException: {err}") from err
return proc
_binary_path = _get_default_hmy_binary_path()
if os.path.exists(f"{get_gopath()}/src/github.com/harmony-one/bls") \
and os.path.exists(f"{get_gopath()}/src/github.com/harmony-one/mcl"): # Check prevents needless import fail.
_environment.update(get_bls_build_variables()) # Needed if using dynamically linked CLI binary.

Write Preview
Loading…
Cancel
Save