commit 07a6530764da77cd088457c8b8c6f6be31136c19 Author: Daniel Van Der Maden Date: Wed Dec 25 18:01:46 2019 -0800 init diff --git a/README.md b/README.md new file mode 100644 index 0000000..56bb046 --- /dev/null +++ b/README.md @@ -0,0 +1 @@ +PyHmy is a wrapper tool to use Harmony's CLI in python. \ No newline at end of file diff --git a/pyhmy/__init__.py b/pyhmy/__init__.py new file mode 100644 index 0000000..0354413 --- /dev/null +++ b/pyhmy/__init__.py @@ -0,0 +1,14 @@ +# +# This package relies on the fact that there is a CLI binary from some path +# starting at the root of the file script importing this package. +# This package requires the pexpect package: https://pypi.org/project/pexpect/ . +# + +from pyhmy.cli import HmyCLI, get_environment +import os + +# Find the CLI binary for HmyCLI. +for root, dirs, files in os.walk(os.path.curdir): + if "hmy" in files: + HmyCLI.hmy_binary_path = os.path.join(root, "hmy") + break diff --git a/pyhmy/cli.py b/pyhmy/cli.py new file mode 100644 index 0000000..1240565 --- /dev/null +++ b/pyhmy/cli.py @@ -0,0 +1,164 @@ +import subprocess +import pexpect +import json +import os +import shutil +import re + + +def get_environment() -> dict: + """ + Fetches the environment variables from the 'setup_bls_build_flags.sh' script + in the harmony main repo. Also fetches the 'HOME' environment variable for HmyCLI. + """ + go_path = subprocess.check_output(["go", "env", "GOPATH"]).decode().strip() + setup_script_path = f"{go_path}/src/github.com/harmony-one/harmony/scripts/setup_bls_build_flags.sh" + response = subprocess.check_output(["bash", setup_script_path, "-v"], timeout=5) + environment = json.loads(response) + environment["HOME"] = os.environ.get("HOME") + return environment + + +class HmyCLI: + hmy_binary_path = None # This class attr should be set by the __init__.py of this module. + + def __init__(self, environment, hmy_binary_path=None): + """ + :param environment: Dictionary of environment variables to be used in the CLI + :param hmy_binary_path: An optional path to the harmony binary; defaults to + class attribute. + """ + if hmy_binary_path: + assert os.path.isfile(hmy_binary_path) + self.hmy_binary_path = hmy_binary_path.replace("./", "") + if not self.hmy_binary_path: + raise FileNotFoundError("Path to harmony CLI binary is not found") + self.environment = environment + self.version = "" + self.keystore_path = "" + self._addresses = {} + self._set_version() + self._set_keystore_path() + self._sync_addresses() + + def __repr__(self) -> str: + return f"<{self.version} @ {self.hmy_binary_path}>" + + def _set_version(self) -> None: + """ + Internal method to set this instance's version according to the binary's version. + """ + proc = subprocess.Popen([self.hmy_binary_path, "version"], env=self.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() + + def _set_keystore_path(self) -> None: + """ + Internal method to set this instance's keystore path with the binary's keystore path. + """ + response = self.single_call("hmy keys location").strip() + if not os.path.exists(response): + os.mkdir(response) + self.keystore_path = response + + def _sync_addresses(self) -> None: + """ + Internal method to sync this instance's address with the binary's keystore 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(f"Name or Address not found on first line of key list") + for line in lines[1:]: + if line: + columns = line.split("\t") + if len(columns) != 2: + raise ValueError("Unexpected format of keys list") + name, address = columns + self._addresses[name.strip()] = address + + def check_address(self, address) -> bool: + """ + :param address: A 'one1...' address. + :return: T/F If the address is in the CLI's keystore. + """ + self._sync_addresses() + return address in self._addresses.values() + + def get_address(self, name) -> str: + """ + :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] + else: + self._sync_addresses() + return self._addresses.get(name, None) + + def get_accounts(self, address) -> list: + """ + :param address: The 'one1...' address + :return: A list of account names associated with + """ + self._sync_addresses() + return [acc for acc, addr in self._addresses.items() if address == addr] + + def remove_account(self, name) -> None: + """ + :param name: The alias of a key used in the CLI's keystore. + """ + if not self.get_address(name): + return + try: + shutil.rmtree(f"{self.keystore_path}/{name}") + except (shutil.Error, FileNotFoundError) as err: + raise RuntimeError(f"Failed to delete dir: {self.keystore_path}/{name}\n" + f"\tException: {err}") from err + del self._addresses[name] + + def remove_address(self, address) -> None: + """ + :param address: The 'one1...' address to be removed. + """ + for name in self.get_names(address): + self.remove_account(name) + + def single_call(self, command, timeout=30) -> str: + """ + :param command: String fo 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 + """ + command_toks = command.split(" ") + if re.match(".*hmy", command_toks[0]): + command_toks = command_toks[1:] + command_toks = [self.hmy_binary_path] + command_toks + try: + response = subprocess.check_output(command_toks, env=self.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=30) -> pexpect.pty_spawn.spawn: + """ + :param command: String fo command to execute on CLI + :param timeout: Optional timeout in seconds + :return: A pexpect child program + :raises: RuntimeError if bad command + """ + command_toks = command.split(" ") + 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) + except (pexpect.ExceptionPexpect, pexpect.TIMEOUT) as err: + raise RuntimeError(f"Bad arguments for CLI.\n " + f"\tException: {err}") from err + return proc diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..7299034 --- /dev/null +++ b/setup.py @@ -0,0 +1,17 @@ +import pathlib +from setuptools import setup + +HERE = pathlib.Path(__file__).parent +README = (HERE / "README.md").read_text() + +setup( + name='pyhmy', + version='0.11', + long_description=README, + long_description_content_type="text/markdown", + author='Daniel Van Der Maden', + author_email='daniel@harmony.one', + url="http://harmony.one/", + packages=['pyhmy'], + install_requires=['pexpect'], +) \ No newline at end of file