Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
result = binascii.unhexlify(ms.ToArray())
StreamManager.ReleaseStream(ms)
ms = StreamManager.GetStream(result)
reader = BinaryReader(ms)
deserialized_message = Message()
deserialized_message.Deserialize(reader)
StreamManager.ReleaseStream(ms)
dm = deserialized_message
self.assertEqual(dm.Command, 'version')
self.assertEqual(dm.Magic, settings.MAGIC)
checksum = Message.GetChecksum(dm.Payload)
self.assertEqual(checksum, dm.Checksum)
deserialized_version = IOHelper.AsSerializableWithType(dm.Payload, 'neo.Network.Payloads.VersionPayload.VersionPayload')
self.assertEqual(deserialized_version.Port, self.port)
self.assertEqual(deserialized_version.UserAgent, self.ua)
self.assertEqual(deserialized_version.Timestamp, self.payload.Timestamp)
tx = Transaction.DeserializeFrom(reader)
self.assertEqual(tx.ToArray(), self.pb_raw)
self.assertEqual(tx.Hash.ToBytes(), self.pb_hash)
json = tx.ToJson()
self.assertEqual(json['size'], 613)
self.assertEqual(json['type'], 'PublishTransaction')
contract = json['contract']
self.assertEqual(contract['author'], 'Erik Zhang')
self.assertEqual(contract['description'], 'Lock your assets until a timestamp.')
self.assertEqual(contract['code']['hash'], '0xffbd1a7ad1e2348b6b3822426f364bfb4bcce3b9')
self.assertEqual(contract['code']['returntype'], "Boolean")
self.assertEqual(contract['code']['parameters'], ['Integer', 'ByteArray', 'Signature'])
self.assertEqual(Fixed8.FromDecimal(settings.ALL_FEES['PublishTransaction']), tx.SystemFee())
parser.add_argument("-c", "--config", action="store", help="Use a specific config file")
parser.add_argument('--version', action='version',
version='neo-python v{version}'.format(version=__version__))
args = parser.parse_args()
if args.config and (args.mainnet or args.privnet):
print("Cannot use both --config and --mainnet/--privnet arguments, please use only one.")
exit(1)
if args.mainnet and args.privnet:
print("Cannot use both --mainnet and --privnet arguments")
exit(1)
# Setup depending on command line arguments. By default, the testnet settings are already loaded.
if args.config:
settings.setup(args.config)
elif args.mainnet:
settings.setup_mainnet()
elif args.privnet:
settings.setup_privnet()
# Instantiate the blockchain and subscribe to notifications
blockchain = LevelDBBlockchain(settings.chain_leveldb_path)
Blockchain.RegisterBlockchain(blockchain)
dbloop = task.LoopingCall(Blockchain.Default().PersistBlocks)
dbloop.start(.1)
ndb = NotificationDB.instance()
ndb.start()
api_server = RestApi()
version="neo-python v{version}".format(version=__version__))
args = parser.parse_args()
# Setting the datadir must come before setting the network, else the wrong path is checked at net setup.
if args.datadir:
settings.set_data_dir(args.datadir)
# Setup depending on command line arguments. By default, the testnet settings are already loaded.
if args.config:
settings.setup(args.config)
elif args.mainnet:
settings.setup_mainnet()
elif args.privnet:
try:
settings.setup_privnet(args.privnet)
except PrivnetConnectionError as e:
logger.error(str(e))
return
elif args.coznet:
settings.setup_coznet()
# Logfile settings & setup
logfile_fn = os.path.join(settings.DATA_DIR_PATH, 'prompt.log')
logfile_max_bytes = 5e7 # 50 MB
logfile_backup_count = 3 # 3 logfiles history
settings.set_logfile(logfile_fn, logfile_max_bytes, logfile_backup_count)
if args.theme:
preferences.set_theme(args.theme)
if args.verbose:
if c1 == 'on' or c1 == '1':
print("Smart contract event logging is now enabled")
settings.set_log_smart_contract_events(True)
if c1 == 'off' or c1 == '0':
print("Smart contract event logging is now disabled")
settings.set_log_smart_contract_events(False)
else:
print("Cannot configure log. Please specify on|off")
elif what == 'sc-debug-notify':
c1 = get_arg(args, 1).lower()
if c1 is not None:
if c1 == 'on' or c1 == '1':
print("Smart contract emit Notify events on execution failure is now enabled")
settings.set_emit_notify_events_on_sc_execution_error(True)
if c1 == 'off' or c1 == '0':
print("Smart contract emit Notify events on execution failure is now disabled")
settings.set_emit_notify_events_on_sc_execution_error(False)
else:
print("Cannot configure log. Please specify on|off")
elif what == 'vm-log':
c1 = get_arg(args, 1).lower()
if c1 is not None:
if c1 == 'on' or c1 == '1':
print("VM instruction execution logging is now enabled")
settings.set_log_vm_instruction(True)
if c1 == 'off' or c1 == '0':
print("VM instruction execution logging is now disabled")
settings.set_log_vm_instruction(False)
{
"type": "SmartContract.Runtime.Notify",
"contract": "f9572c5b119a6b5775a6af07f1cef5d310038f55",
"tx": "6d0f1decbf3874d08d41f2cc9e8672cd3507c962668c15793e3dd3e01fc3551c",
"block": 942369,
"addr_from": "ALULT5WpeiHnEXYFe72Yq7nRB3ZBmsBypq",
"amount": 1,
"addr_to": "APaGQT4dx4gUDApVPnbtZvChJ8UKRsZBdt",
"notify_type": "transfer"
}
],
"message": ""
}
""" % (settings.net_name, endpoints_html)
return web.Response(body=out)
# logging.basicConfig(
# level=logging.DEBUG,
# filemode='a',
# filename=logname,
# format="%(levelname)s:%(name)s:%(funcName)s:%(message)s")
from neo.Network.NodeLeader import NodeLeader
from twisted.internet import reactor, task
from neo.Core.Blockchain import Blockchain
from neo.Implementations.Blockchains.LevelDB.LevelDBBlockchain import LevelDBBlockchain
from neo.Settings import settings
blockchain = LevelDBBlockchain(settings.LEVELDB_PATH)
Blockchain.RegisterBlockchain(blockchain)
dbloop = task.LoopingCall(Blockchain.Default().PersistBlocks)
dbloop.start(.01)
NodeLeader.Instance().Start()
reactor.run()
def configure(self, args):
what = get_arg(args)
if what == 'debug':
c1 = get_arg(args, 1).lower()
if c1 is not None:
if c1 == 'on' or c1 == '1':
print("Debug logging is now enabled")
settings.set_loglevel(logging.DEBUG)
if c1 == 'off' or c1 == '0':
print("Debug logging is now disabled")
settings.set_loglevel(logging.INFO)
else:
print("Cannot configure log. Please specify on|off")
elif what == 'sc-events':
c1 = get_arg(args, 1).lower()
if c1 is not None:
if c1 == 'on' or c1 == '1':
print("Smart contract event logging is now enabled")
settings.set_log_smart_contract_events(True)
if c1 == 'off' or c1 == '0':
print("Smart contract event logging is now disabled")
settings.set_log_smart_contract_events(False)
def execute(self, arguments):
if len(arguments) != 1:
print("Please specify the required parameter")
return False
try:
flag = bool(util.strtobool(arguments[0]))
except ValueError:
print("Invalid option")
return False
settings.COMPILER_NEP_8 = flag
if flag:
print("NEP-8 compiler instruction usage is ON")
else:
print("NEP-8 compiler instruction usage is OFF")
return True
settings.set_data_dir(args.datadir)
# Network configuration depending on command line arguments. By default, the testnet settings are already loaded.
if args.config:
settings.setup(args.config)
elif args.mainnet:
settings.setup_mainnet()
elif args.testnet:
settings.setup_testnet()
elif args.privnet:
settings.setup_privnet()
elif args.coznet:
settings.setup_coznet()
if args.maxpeers:
settings.set_max_peers(args.maxpeers)
if args.syslog or args.syslog_local is not None:
# Setup the syslog facility
if args.syslog_local is not None:
print("Logging to syslog local%s facility" % args.syslog_local)
syslog_facility = SysLogHandler.LOG_LOCAL0 + args.syslog_local
else:
print("Logging to syslog user facility")
syslog_facility = SysLogHandler.LOG_USER
# Setup logzero to only use the syslog handler
logzero.syslog(facility=syslog_facility)
else:
# Setup file logging
if args.logfile:
logfile = os.path.abspath(args.logfile)