feat(wizard): add csv input flag #3

Merged
williamluke merged 1 commits from feat/csv_input into master 2022-04-26 13:07:01 +02:00
14 changed files with 431 additions and 108 deletions

View File

@ -3,10 +3,12 @@ from __future__ import annotations
# standard import # standard import
import logging import logging
from chainlib.cli.config import Config from chainlib.cli.config import Config
# local imports # local imports
from cic.contract.contract import deploy_contract, generate_contract, load_contract from cic.contract.contract import deploy_contract, generate_contract, load_contract
from cic.contract.csv import load_contract_from_csv
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
@ -20,6 +22,10 @@ def process_args(argparser):
action="store_true", action="store_true",
help="Skip Deployment", help="Skip Deployment",
) )
argparser.add_argument(
"--csv",
help="Load Voucher from CSV",
)
argparser.add_argument( argparser.add_argument(
"--target", "--target",
default="eth", default="eth",
@ -46,19 +52,24 @@ def validate_args(_args):
pass pass
ExtraArgs = {"skip_gen": str, "skip_deploy": str, "target": str, "path": str, "p": str} def execute(
config: Config,
eargs,
def execute(config: Config, eargs: ExtraArgs): ):
directory = eargs.path directory = eargs.path
target = eargs.target target = eargs.target
skip_gen = eargs.skip_gen skip_gen = eargs.skip_gen
skip_deploy = eargs.skip_deploy skip_deploy = eargs.skip_deploy
wallet_keystore = eargs.y wallet_keystore = eargs.y
csv_file = eargs.csv
if wallet_keystore: if wallet_keystore:
config.add(wallet_keystore, "WALLET_KEY_FILE", exists_ok=True) config.add(wallet_keystore, "WALLET_KEY_FILE", exists_ok=True)
if not skip_gen: if not skip_gen:
if csv_file:
contract = load_contract_from_csv(config, directory, csv_file)
else:
contract = generate_contract(directory, [target], config, interactive=True) contract = generate_contract(directory, [target], config, interactive=True)
else: else:
contract = load_contract(directory) contract = load_contract(directory)

View File

@ -5,23 +5,23 @@ proof_writer = cic.writers.KVWriter
ext_writer = cic.writers.KVWriter ext_writer = cic.writers.KVWriter
[cic] [cic]
registry_address = 0x999487d8B1EC2b2ac9F4a9D1A6D35a24D75D7EF4 registry_address = 0xe3e3431BF25b06166513019Ed7B21598D27d05dC
[meta] [meta]
url = https://meta.grassecon.org url = https://meta.sarafu.network
http_origin = http_origin =
[rpc] [rpc]
provider = https://rpc.kitabu.grassecon.org provider = http://142.93.38.53:8545
[auth] [auth]
type = gnupg type = gnupg
keyfile_path = /home/will/.config/cic/staff-client/user.asc keyfile_path =
passphrase = queenmarlena passphrase =
[wallet] [wallet]
key_file = /home/will/grassroots/cic-internal-integration/apps/contract-migration/keystore key_file =
passphrase = passphrase =
[chain] [chain]
spec = evm:kitabu:5050:sarafu spec = evm:kitabu:6060:sarafu

View File

@ -14,6 +14,7 @@ from cic.contract.base import Data, data_dir
from cic.utils import object_to_str from cic.utils import object_to_str
logg = logging.getLogger(__name__) logg = logging.getLogger(__name__)
@ -22,7 +23,9 @@ class Proof(Data):
It processes inputs from the proof.json file in the session directory. It processes inputs from the proof.json file in the session directory.
Optionally, attachment objects can be added to the proof. If added, the resulting proof digest will consists of the attachment digests added to the root digest. These are then are deterministically ordered, regardless of which order attachments were given to the constructor. Optionally, attachment objects can be added to the proof. If added, the resulting proof
digest will consists of the attachment digests added to the root digest. These are then
are deterministically ordered, regardless of which order attachments were given to the constructor.
:param path: Path to settings directory :param path: Path to settings directory
:type path: str :type path: str
@ -56,7 +59,8 @@ class Proof(Data):
if interactive: if interactive:
self.description = ( self.description = (
input(f"Enter Proof Description ({self.description}): ") or self.description input(f"Enter Proof Description ({self.description}): ")
or self.description
) )
self.namespace = ( self.namespace = (
input(f"Enter Proof Namespace ({self.namespace}): ") or self.namespace input(f"Enter Proof Namespace ({self.namespace}): ") or self.namespace
@ -67,7 +71,7 @@ class Proof(Data):
"""Load proof data from settings.""" """Load proof data from settings."""
super(Proof, self).load() super(Proof, self).load()
f = open(self.proof_path, "r") f = open(self.proof_path, "r", encoding="utf-8")
o = json.load(f) o = json.load(f)
f.close() f.close()
@ -77,7 +81,7 @@ class Proof(Data):
self.issuer = o["issuer"] self.issuer = o["issuer"]
self.proofs = o["proofs"] self.proofs = o["proofs"]
if self.extra_attachments != None: if self.extra_attachments is not None:
a = self.extra_attachments.asdict() a = self.extra_attachments.asdict()
for k in a.keys(): for k in a.keys():
self.attachments[k] = a[k] self.attachments[k] = a[k]

13
cic/contract/constants.py Normal file
View File

@ -0,0 +1,13 @@
GITABLE_CONTRACT_URL = "https://gitlab.com/cicnet/eth-erc20/-/raw/master/python/giftable_erc20_token/data/GiftableToken"
DMR_CONTRACT_URL = "https://gitlab.com/cicnet/erc20-demurrage-token/-/raw/master/python/erc20_demurrage_token/data/DemurrageTokenSingleNocap"
CONTRACT_URLS = [
{
"url": GITABLE_CONTRACT_URL,
"name": "Giftable Token",
},
{
"url": DMR_CONTRACT_URL,
"name": "Demurrage Token Single No Cap",
},
]

View File

@ -1,27 +1,27 @@
# Standard # Standard
import importlib import importlib
import json
import logging import logging
import os import os
from typing import TYPE_CHECKING, List from typing import List
import requests
# external imports
from cic_types.ext.metadata import MetadataRequestsHandler
from cic_types.ext.metadata.signer import Signer as MetadataSigner
from chainlib.chain import ChainSpec from chainlib.chain import ChainSpec
from chainlib.cli.config import Config from chainlib.cli.config import Config
# Local Modules
from cic.contract.components.attachment import Attachment from cic.contract.components.attachment import Attachment
from cic.contract.components.meta import Meta from cic.contract.components.meta import Meta
from cic.contract.components.proof import Proof from cic.contract.components.proof import Proof
from cic.contract.components.token import Token from cic.contract.components.token import Token
from cic.contract.helpers import init_writers_from_config from cic.contract.helpers import init_writers_from_config
from cic.contract.network import Network from cic.contract.network import Network
# Local Modules
from cic.contract.processor import ContractProcessor from cic.contract.processor import ContractProcessor
from cic.writers import HTTPWriter, KeyedWriterFactory, MetadataWriter, OutputWriter from cic.writers import HTTPWriter, KeyedWriterFactory, MetadataWriter
# external imports
from cic_types.ext.metadata import MetadataRequestsHandler
from cic_types.ext.metadata.signer import Signer as MetadataSigner
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
@ -85,26 +85,26 @@ def generate_contract(
print(f"Deleted {directory}") print(f"Deleted {directory}")
os.system(f"rm -rf {directory}") os.system(f"rm -rf {directory}")
os.makedirs(directory) os.makedirs(directory)
log.debug("Generating token") log.info("Generating token")
token = Token(directory, interactive=interactive) token = Token(directory, interactive=interactive)
token.start() token.start()
log.debug("Generating proof") log.info("Generating proof")
proof = Proof(directory, interactive=interactive) proof = Proof(directory, interactive=interactive)
proof.start() proof.start()
log.debug("Generating meta") log.info("Generating meta")
meta = Meta(directory, interactive=interactive) meta = Meta(directory, interactive=interactive)
meta.start() meta.start()
log.debug("Generating attachment") log.info("Generating attachment")
attachment = Attachment(directory, interactive=interactive) attachment = Attachment(directory, interactive=interactive)
log.debug("Generating network") log.info("Generating network")
network = Network(directory, targets=targets) network = Network(directory, targets=targets)
network.start() network.start()
log.debug( log.info(
f"""Populating infomation from network: f"""Populating infomation from network:
CIC_REGISTRY_ADDRESS: {config.get("CIC_REGISTRY_ADDRESS")} CIC_REGISTRY_ADDRESS: {config.get("CIC_REGISTRY_ADDRESS")}
CHAIN_SPEC: {config.get("CHAIN_SPEC")} CHAIN_SPEC: {config.get("CHAIN_SPEC")}
@ -118,7 +118,7 @@ def generate_contract(
signer_hint = config.get("WALLET_KEY_FILE") signer_hint = config.get("WALLET_KEY_FILE")
keys = cmd_mod.list_keys(config, signer_hint) keys = cmd_mod.list_keys(config, signer_hint)
if len(keys) > 1: if len(keys) > 1:
print(f"More than one key found, please select one:") print("More than one key found, please select one:")
for idx, key in enumerate(keys): for idx, key in enumerate(keys):
print(f"{idx} - {key} ") print(f"{idx} - {key} ")
selecting_key = True selecting_key = True
@ -152,7 +152,6 @@ def deploy_contract(
target: str, target: str,
contract_directory: str, contract_directory: str,
): ):
modname = f"cic.ext.{target}" modname = f"cic.ext.{target}"
cmd_mod = importlib.import_module(modname) cmd_mod = importlib.import_module(modname)

197
cic/contract/csv.py Normal file
View File

@ -0,0 +1,197 @@
import csv
import importlib
import logging
import os
from enum import IntEnum
from pathlib import Path
from typing import List
from chainlib.chain import ChainSpec
from cic.contract.components.attachment import Attachment
from cic.contract.components.meta import Meta
from cic.contract.components.proof import Proof
from cic.contract.components.token import Token
from cic.contract.constants import DMR_CONTRACT_URL
from cic.contract.contract import Contract
from cic.contract.helpers import download_file
from cic.contract.network import Network
log = logging.getLogger(__name__)
CONTRACT_CSV_HEADER = [
"issuer",
"namespace",
"voucher_name",
"symbol",
"location",
"country_code",
"supply",
"precision",
"token_type",
"demurrage",
"period_minutes",
"phone_number",
"email_address",
"sink_account",
"description",
]
class CSV_Column(IntEnum):
issuer = 0
namespace = 1
voucher_name = 2
symbol = 3
location = 4
country_code = 5
supply = 6
precision = 7
token_type = 8
demurrage = 9
period_minutes = 10
phone_number = 11
email_address = 12
sink_account = 13
description = 14
def load_contracts_from_csv(config, directory, csv_path: str) -> List[Contract]:
targets = ["eth"]
os.makedirs(directory)
contract_rows = []
bin_path = os.path.abspath(download_file(DMR_CONTRACT_URL + ".bin"))
with open(csv_path, "rt", encoding="utf-8") as file:
csvreader = csv.reader(file, delimiter=",")
for idx, row in enumerate(csvreader):
if idx == 0:
if row != CONTRACT_CSV_HEADER:
raise Exception(
f'Seems you are using the wrong csv format. Expected the header to be: \n\t {", ".join(CONTRACT_CSV_HEADER)}'
)
continue
contract_rows.append(row)
contracts = []
for idx, contract_row in enumerate(contract_rows):
issuer = contract_row[CSV_Column.issuer]
namespace = contract_row[CSV_Column.namespace]
voucher_name = contract_row[CSV_Column.voucher_name]
symbol = contract_row[CSV_Column.symbol]
location = contract_row[CSV_Column.location]
country_code = contract_row[CSV_Column.country_code]
supply = contract_row[CSV_Column.supply]
precision = contract_row[CSV_Column.precision]
token_type = contract_row[CSV_Column.token_type]
demurrage = contract_row[CSV_Column.demurrage]
period_minutes = contract_row[CSV_Column.period_minutes]
phone_number = contract_row[CSV_Column.phone_number]
email_address = contract_row[CSV_Column.email_address]
sink_account = contract_row[CSV_Column.sink_account]
description = contract_row[CSV_Column.description]
if token_type != "demurrage":
raise Exception(
f"Only demurrage tokens are supported at this time. {token_type} is not supported"
)
log.info("Generating token")
token = Token(
directory,
name=voucher_name,
symbol=symbol,
precision=precision,
supply=supply,
extra_args=[demurrage, period_minutes, sink_account],
extra_args_types=["uint256", "uint256", "address"],
code=bin_path,
)
token.start()
log.info("Generating proof")
proof = Proof(
directory,
attachments=None,
issuer=issuer,
description=description,
namespace=namespace,
)
proof.start()
log.info("Generating meta")
meta = Meta(
directory,
name=issuer,
contact={
"phone": phone_number,
"email": email_address,
},
country_code=country_code,
location=location,
)
meta.start()
log.info("Generating attachment")
attachment = Attachment(directory)
log.info("Generating network")
network = Network(directory, targets=targets)
network.start()
log.info(
f"""Populating infomation from network:
CIC_REGISTRY_ADDRESS: {config.get("CIC_REGISTRY_ADDRESS")}
CHAIN_SPEC: {config.get("CHAIN_SPEC")}
RPC_PROVIDER: {config.get("RPC_PROVIDER")}
"""
)
for target in targets:
# TODO Clean this up
modname = f"cic.ext.{target}"
cmd_mod = importlib.import_module(modname)
signer_hint = config.get("WALLET_KEY_FILE")
if signer_hint is None:
raise Exception("No Wallet Keyfile was provided")
keys = cmd_mod.list_keys(config, signer_hint)
if keys is None or len(keys) == 0:
raise Exception(f"No wallet keys found in {signer_hint}")
if len(keys) > 1:
log.warning(
f"More than one key found in the keystore. Using the first one\n - {keys[0]}"
)
key_account_address = keys[0]
m = importlib.import_module(f"cic.ext.{target}.start")
m.extension_start(
network,
registry_address=config.get("CIC_REGISTRY_ADDRESS"),
chain_spec=ChainSpec.from_chain_str(config.get("CHAIN_SPEC")),
rpc_provider=config.get("RPC_PROVIDER"),
key_account_address=key_account_address,
)
network.load()
contracts.append(
Contract(
token=token,
proof=proof,
meta=meta,
attachment=attachment,
network=network,
)
)
return contracts
def load_contract_from_csv(config, directory, csv_path: str) -> Contract:
path = Path(csv_path)
if path.is_file():
contracts = load_contracts_from_csv(config, directory, csv_path=csv_path)
if len(contracts) == 0:
raise Exception("No contracts found in CSV")
if len(contracts) > 1:
log.warning(
"Warning multiple contracts found in CSV. Only the first contract will be used"
)
else:
raise Exception("CSV file does not exist")
return contracts[0]

View File

@ -1,32 +1,25 @@
# standard imports # standard imports
import os
import logging
import sys
import json
import requests
import importlib
import tempfile
import hashlib import hashlib
import importlib
import json
import logging
import os
import sys
import tempfile
from typing import Callable, TypedDict, Union
import requests
from cic.contract.constants import CONTRACT_URLS
# local imports # local imports
from cic.writers import OutputWriter from cic.writers import WritersType
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
CONTRACTS = [
{
"url": "https://gitlab.com/cicnet/eth-erc20/-/raw/master/python/giftable_erc20_token/data/GiftableToken",
"name": "Giftable Token",
},
{
"url": "https://gitlab.com/cicnet/erc20-demurrage-token/-/raw/master/python/erc20_demurrage_token/data/DemurrageTokenSingleNocap",
"name": "Demurrage Token Single No Cap",
}
]
# Download File from Url # Download File from Url
def download_file(url: str, filename=None) -> (str, bytes): def download_file(url: str, filename=None) -> str:
directory = tempfile.gettempdir() directory = tempfile.gettempdir()
filename = filename if filename else url.split("/")[-1] filename = filename if filename else url.split("/")[-1]
log.debug(f"Downloading {filename}") log.debug(f"Downloading {filename}")
@ -38,27 +31,29 @@ def download_file(url: str, filename=None) -> (str, bytes):
log.debug(f"{filename} downloaded to {path}") log.debug(f"{filename} downloaded to {path}")
return path return path
def get_contract_args(data: list): def get_contract_args(data: list):
for item in data: for item in data:
if item["type"] == "constructor": if item["type"] == "constructor":
return item["inputs"] return item["inputs"]
raise Exception("No constructor found in contract") raise Exception("No constructor found in contract")
def select_contract(): def select_contract():
print("Contracts:") print("Contracts:")
print("\t C - Custom (path/url to contract)") print("\t C - Custom (path/url to contract)")
for idx, contract in enumerate(CONTRACTS): for idx, contract in enumerate(CONTRACT_URLS):
print(f"\t {idx} - {contract['name']}") print(f"\t {idx} - {contract['name']}")
val = input("Select contract (C,0,1..): ") val = input("Select contract (C,0,1..): ")
if val.isdigit() and int(val) < len(CONTRACTS): if val.isdigit() and int(val) < len(CONTRACT_URLS):
contract = CONTRACTS[int(val)] contract = CONTRACT_URLS[int(val)]
bin_path = os.path.abspath(download_file(contract["url"] + ".bin")) bin_path = os.path.abspath(download_file(contract["url"] + ".bin"))
json_path = download_file(contract["url"] + ".json") json_path = download_file(contract["url"] + ".json")
elif val == "C": elif val == "C":
possible_bin_location = input("Enter a path or url to a contract.bin: ") possible_bin_location = input("Enter a path or url to a contract.bin: ")
if possible_bin_location.startswith('http'): if possible_bin_location.startswith("http"):
# possible_bin_location is url # possible_bin_location is url
bin_path = download_file(possible_bin_location) bin_path = download_file(possible_bin_location)
else: else:
@ -99,25 +94,28 @@ def select_contract():
} }
Writers = { class Writers(TypedDict):
"meta": OutputWriter, meta: Union[WritersType, Callable[..., WritersType]]
"attachment": OutputWriter, attachment: Callable[..., WritersType]
"proof": OutputWriter, proof: Callable[..., WritersType]
"ext": OutputWriter, ext: Union[WritersType, Callable[..., WritersType]]
}
def init_writers_from_config(config) -> Writers: def init_writers_from_config(config) -> Writers:
writers: Writers = { writers = {}
"meta": None, writer_keys = ["meta", "attachment", "proof", "ext"]
"attachment": None, for key in writer_keys:
"proof": None,
"ext": None,
}
for key in writers:
writer_config_name = f"CIC_CORE_{key.upper()}_WRITER" writer_config_name = f"CIC_CORE_{key.upper()}_WRITER"
(module_name, attribute_name) = config.get(writer_config_name).rsplit(".", maxsplit=1) (module_name, attribute_name) = config.get(writer_config_name).rsplit(
".", maxsplit=1
)
mod = importlib.import_module(module_name) mod = importlib.import_module(module_name)
writer = getattr(mod, attribute_name) writer = getattr(mod, attribute_name)
writers[key] = writer writers[key] = writer
return writers return Writers(
meta=writers["meta"],
attachment=writers["attachment"],
proof=writers["proof"],
ext=writers["ext"],
)

View File

@ -5,10 +5,10 @@ proof_writer = cic.writers.KVWriter
ext_writer = cic.writers.KVWriter ext_writer = cic.writers.KVWriter
[cic] [cic]
registry_address = 0xcf60ebc445b636a5ab787f9e8bc465a2a3ef8299 registry_address = 0xe3e3431BF25b06166513019Ed7B21598D27d05dC
[meta] [meta]
url = https://meta.grassecon.net url = https://meta.sarafu.network
http_origin = http_origin =
[auth] [auth]

View File

@ -2,6 +2,5 @@
"name": "", "name": "",
"location": "", "location": "",
"country_code": "", "country_code": "",
"contact": { "contact": {}
}
} }

View File

@ -1,17 +1,17 @@
# standard imports # standard imports
import os
import logging
import sys
import importlib import importlib
import logging
import os
import sys
# external imports # external imports
import chainlib.cli import chainlib.cli
import cic.cmd.export as cmd_export
import cic.cmd.ext as cmd_ext
# local imports # local imports
import cic.cmd.init as cmd_init import cic.cmd.init as cmd_init
import cic.cmd.show as cmd_show import cic.cmd.show as cmd_show
import cic.cmd.ext as cmd_ext
import cic.cmd.export as cmd_export
import cic.cmd.wizard as cmd_wizard import cic.cmd.wizard as cmd_wizard
from cic.config import ensure_base_configs from cic.config import ensure_base_configs
@ -19,62 +19,80 @@ logging.basicConfig(level=logging.WARNING)
logg = logging.getLogger() logg = logging.getLogger()
script_dir = os.path.dirname(os.path.realpath(__file__)) script_dir = os.path.dirname(os.path.realpath(__file__))
data_dir = os.path.join(script_dir, '..', 'data') data_dir = os.path.join(script_dir, "..", "data")
base_config_dir = os.path.join(data_dir, 'config') base_config_dir = os.path.join(data_dir, "config")
schema_dir = os.path.join(script_dir, '..', 'schema') schema_dir = os.path.join(script_dir, "..", "schema")
user_config_dir = os.path.join(os.path.expanduser("~"), ".config", "cic", "cli", "config") user_config_dir = os.path.join(
os.path.expanduser("~"), ".config", "cic", "cli", "config"
)
arg_flags = chainlib.cli.argflag_std_read | chainlib.cli.Flag.SEQ arg_flags = chainlib.cli.argflag_std_read | chainlib.cli.Flag.SEQ
argparser = chainlib.cli.ArgumentParser( argparser = chainlib.cli.ArgumentParser(
env=os.environ, env=os.environ,
arg_flags=arg_flags, arg_flags=arg_flags,
description='CIC cli tool for generating and publishing tokens' description="CIC cli tool for generating and publishing contracts",
) )
sub = argparser.add_subparsers() sub = argparser.add_subparsers()
sub.dest = 'command' sub.dest = "command"
sub_init = sub.add_parser('init', help='initialize new cic data directory') sub_init = sub.add_parser("init", help="initialize new cic data directory")
cmd_init.process_args(sub_init) cmd_init.process_args(sub_init)
sub_show = sub.add_parser('show', help='display summary of current state of cic data directory') sub_show = sub.add_parser(
"show", help="display summary of current state of cic data directory"
)
cmd_show.process_args(sub_show) cmd_show.process_args(sub_show)
sub_export = sub.add_parser('export', help='export cic data directory state to a specified target') sub_export = sub.add_parser(
"export", help="export cic data directory state to a specified target"
)
cmd_export.process_args(sub_export) cmd_export.process_args(sub_export)
sub_ext = sub.add_parser('ext', help='extension helpers') sub_ext = sub.add_parser("ext", help="extension helpers")
cmd_ext.process_args(sub_ext) cmd_ext.process_args(sub_ext)
sub_wizard = sub.add_parser('wizard', help='An interactive wizard for creating and publishing contracts') sub_wizard = sub.add_parser(
"wizard", help="An interactive wizard for creating and publishing contracts"
)
cmd_wizard.process_args(sub_wizard) cmd_wizard.process_args(sub_wizard)
args = argparser.parse_args(sys.argv[1:]) args = argparser.parse_args(sys.argv[1:])
if args.command is None: if args.command is None:
logg.critical('Subcommand missing') logg.critical("Subcommand missing")
sys.stderr.write("\033[;91m" + 'subcommand missing' + "\033[;39m\n") sys.stderr.write("\033[;91m" + "subcommand missing" + "\033[;39m\n")
argparser.print_help(sys.stderr) argparser.print_help(sys.stderr)
sys.exit(1) sys.exit(1)
modname = f'cic.cmd.{args.command}' modname = f"cic.cmd.{args.command}"
logg.debug(f'using module {modname}') logg.debug(f"using module {modname}")
cmd_mod = importlib.import_module(modname) cmd_mod = importlib.import_module(modname)
extra_args = { extra_args = {
'p': 'RPC_PROVIDER', "p": "RPC_PROVIDER",
} }
ensure_base_configs(user_config_dir) ensure_base_configs(user_config_dir)
config = chainlib.cli.Config.from_args(args, arg_flags=arg_flags, base_config_dir=base_config_dir, extra_args=extra_args)
def main(): def main():
default_config_dir = args.config or os.path.join(user_config_dir, "mainnet")
config = chainlib.cli.Config.from_args(
args,
arg_flags=arg_flags,
base_config_dir=base_config_dir,
extra_args=extra_args,
default_config_dir=default_config_dir,
)
try: try:
cmd_mod.execute(config, args) cmd_mod.execute(config, args)
except Exception as e: except Exception as e:
logg.exception(e) logg.exception(e)
sys.stderr.write("\033[;91m" + str(e) + "\033[;39m\n") sys.stderr.write("\033[;91m" + str(e) + "\033[;39m\n")
argparser.print_help()
sys.exit(1) sys.exit(1)
if __name__ == '__main__':
if __name__ == "__main__":
main() main()

View File

@ -1,12 +1,13 @@
# standard imports # standard imports
import base64 import base64
import json
import logging import logging
import os import os
import sys import sys
import urllib.request import urllib.request
import json from typing import Type, Union
from cic_types.ext.metadata import MetadataRequestsHandler, MetadataPointer from cic_types.ext.metadata import MetadataPointer, MetadataRequestsHandler
logg = logging.getLogger(__name__) logg = logging.getLogger(__name__)
@ -25,13 +26,13 @@ class StdoutWriter(OutputWriter):
class KVWriter(OutputWriter): class KVWriter(OutputWriter):
def __init__(self, path=None, *args, **kwargs): def __init__(self, *args, path=None, **kwargs):
try: try:
os.stat(path) os.stat(path)
except FileNotFoundError: except FileNotFoundError:
os.makedirs(path) os.makedirs(path)
self.path = path self.path = path
super().__init__(*args, **kwargs)
def write(self, k, v): def write(self, k, v):
fp = os.path.join(self.path, str(k)) fp = os.path.join(self.path, str(k))
logg.debug(f"path write {fp} {str(v)}") logg.debug(f"path write {fp} {str(v)}")
@ -41,7 +42,7 @@ class KVWriter(OutputWriter):
class HTTPWriter(OutputWriter): class HTTPWriter(OutputWriter):
def __init__(self, path=None, *args, **kwargs): def __init__(self, *args, path=None, **kwargs):
super(HTTPWriter, self).__init__(*args, **kwargs) super(HTTPWriter, self).__init__(*args, **kwargs)
self.path = path self.path = path
@ -59,6 +60,7 @@ class KeyedWriter(OutputWriter):
def __init__(self, writer_keyed, writer_immutable): def __init__(self, writer_keyed, writer_immutable):
self.writer_keyed = writer_keyed self.writer_keyed = writer_keyed
self.writer_immutable = writer_immutable self.writer_immutable = writer_immutable
super().__init__()
def write(self, k, v): def write(self, k, v):
logg.debug(f"writing keywriter key: {k} value: {v}") logg.debug(f"writing keywriter key: {k} value: {v}")
@ -72,7 +74,7 @@ class KeyedWriter(OutputWriter):
class KeyedWriterFactory: class KeyedWriterFactory:
def __init__( def __init__(
self, key_writer_constructor, immutable_writer_constructor, *args, **kwargs self, key_writer_constructor, immutable_writer_constructor, *_args, **kwargs
): ):
self.key_writer_constructor = key_writer_constructor self.key_writer_constructor = key_writer_constructor
self.immutable_writer_constructor = immutable_writer_constructor self.immutable_writer_constructor = immutable_writer_constructor
@ -81,7 +83,7 @@ class KeyedWriterFactory:
logg.debug(f"adding key {k} t keyed writer factory") logg.debug(f"adding key {k} t keyed writer factory")
self.x[k] = v self.x[k] = v
def new(self, path=None, *args, **kwargs): def new(self, *_args, path=None, **_kwargs):
writer_keyed = None writer_keyed = None
writer_immutable = None writer_immutable = None
if self.key_writer_constructor is not None: if self.key_writer_constructor is not None:
@ -112,3 +114,8 @@ class MetadataWriter(OutputWriter):
r = rq.create(v) r = rq.create(v)
logg.info(f"metadata submitted at {k}") logg.info(f"metadata submitted at {k}")
return r return r
WritersType = Union[
Type[OutputWriter], Type[KeyedWriter], Type[MetadataWriter], Type[OutputWriter]
]

View File

@ -81,7 +81,7 @@ requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api" build-backend = "poetry.core.masonry.api"
[tool.pytest.ini_options] [tool.pytest.ini_options]
addopts = "--cov=cic --cov-fail-under=40 --cov-report term-missing" addopts = "--cov=cic --cov-report term-missing -v"
testpaths = ["tests"] testpaths = ["tests"]
[tool.semantic_release] [tool.semantic_release]
@ -94,3 +94,4 @@ build_command = "pip install poetry && poetry build"
hvcs = "gitea" hvcs = "gitea"
hvcs_domain = "git.grassecon.net" hvcs_domain = "git.grassecon.net"
check_build_status = false check_build_status = false

View File

@ -0,0 +1,74 @@
import os
import tempfile
from cic.contract.csv import load_contract_from_csv
from tests.base_cic import test_data_dir
def test_csv_generate():
outputs_dir = os.path.join(tempfile.mkdtemp(), "outputs")
test_csv_path = os.path.join(test_data_dir, "voucher", "bondi.csv")
contract = load_contract_from_csv(
{
"WALLET_KEY_FILE": os.path.join(test_data_dir, "keystore", "ok"),
"WALLET_PASSPHRASE": "test",
"CHAIN_SPEC": "evm:kitabu:6060:sarafu",
"RPC_PROVIDER": "http://142.93.38.53:8545",
"CIC_REGISTRY_ADDRESS": "0xe3e3431BF25b06166513019Ed7B21598D27d05dC",
},
outputs_dir,
csv_path=test_csv_path,
)
# assert len(contracts) == 1
# contract = contracts[0]
# Token
assert contract.token.name == "Bondeni"
assert contract.token.extra_args == [
"46124891913883000000000000000000",
"1440",
"0xB8830b647C01433F9492F315ddBFDc35CB3Be6A6",
]
assert contract.token.extra_args_types == ["uint256", "uint256", "address"]
# assert contract.token.code == os.path.join(test_data_dir, "contracts", "Bondi.bin")
assert contract.token.precision == '6'
assert contract.token.supply == "5025"
assert contract.token.symbol == "BONDE"
# Meta
assert contract.meta.country_code == "KE"
assert contract.meta.location == "Mutitu Kilifi"
assert contract.meta.contact == {
"email": "info@grassecon.org",
"phone": "254797782065",
}
assert contract.meta.name == "Bondeni SHG"
# Network
assert contract.network.resources["eth"]["chain_spec"] == {
"arch": "evm",
"common_name": "sarafu",
"custom": [],
"extra": {},
"fork": "kitabu",
"network_id": 6060,
}
assert contract.network.resources["eth"]["contents"] == {
"address_declarator": {
"key_account": "cc4f82f5dacde395e1e0cfc4d62827c8b8b5688c",
"reference": "f055e83f713DbFF947e923749Af9802eaffFB5f9",
},
"token": {
"key_account": "cc4f82f5dacde395e1e0cfc4d62827c8b8b5688c",
"reference": None,
},
"token_index": {
"key_account": "cc4f82f5dacde395e1e0cfc4d62827c8b8b5688c",
"reference": "5A1EB529438D8b3cA943A45a48744f4c73d1f098",
},
}
assert contract.proof.description == "1 BONDE = 1 itumbe"
assert contract.proof.namespace == "ge"
assert contract.proof.proofs == []
assert contract.proof.version() == 0

2
tests/testdata/voucher/bondi.csv vendored Normal file
View File

@ -0,0 +1,2 @@
issuer,namespace,voucher_name,symbol,location,country_code,supply,precision,token_type,demurrage,period_minutes,phone_number,email_address,sink_account,description
Bondeni SHG,ge,Bondeni,BONDE,Mutitu Kilifi,KE,5025,6,demurrage,46124891913883000000000000000000,1440,254797782065,info@grassecon.org,0xB8830b647C01433F9492F315ddBFDc35CB3Be6A6,1 BONDE = 1 itumbe
1 issuer namespace voucher_name symbol location country_code supply precision token_type demurrage period_minutes phone_number email_address sink_account description
2 Bondeni SHG ge Bondeni BONDE Mutitu Kilifi KE 5025 6 demurrage 46124891913883000000000000000000 1440 254797782065 info@grassecon.org 0xB8830b647C01433F9492F315ddBFDc35CB3Be6A6 1 BONDE = 1 itumbe