Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@cmd2.options([], arg_desc="")
def do_disconnect(self, line, opts):
if self.client is None:
logger.error("PBAPClient is not even connected.. Connect and then try disconnect")
sys.exit(2)
logger.debug("Disconnecting pbap client with pbap server")
self.client.disconnect()
self.client = None
self.prompt = self.colorize("pbap> ", "yellow")
@options([
make_option('-d', '--device-id', action="store", dest='device_id', type='string',
help='ONU Device ID', default=None),
])
def do_show_alarms(self, _line, opts):
""" Show contents of the alarm table"""
device_id = opts.device_id or self.device_id
try:
alarm_db = self.get_alarm_table(device_id, depth=-1)
if alarm_db is None:
self.poutput(self.colorize('Alarm Table for ONU {} is not currently available'
.format(device_id), 'red'))
return
except Exception: # UnboundLocalError if Device ID not found in DB
self.poutput(self.colorize('Failed to get Alarm Table for ONU {}'
@options([make_option('-b', '--backend',type="string", help="Connect to backend, values: r for router, s for smppcm and r,s for both"),
make_option('--router_host', type="string", help="Router host, default 127.0.0.1"),
make_option('--router_port', type="int", help="Router port, default 8988"),
make_option('--smppcm_host', type="string", help="SMPPCM host, default 127.0.0.1"),
make_option('--smppcm_port', type="int", help="SMPPCM host, default 8989"),])
def do_connect(self, arg, opts):
'Start a connection to router (router) or smpp client manager (smppcm)'
if opts.backend is None:
self.print_stdout('Argument -b required.')
return
if opts.backend not in ['r', 's', 'r,s']:
self.print_stdout('Invalid argument (%s) for command connect, must be "r", "s" or "r,s".' % opts.backend)
return
# @todo: host and port must be configurable
router_host = '127.0.0.1'
@options([
make_option('-o', '--out', action="store_true", default=True, help="Create out-going link from current node"),
make_option('-i', '--in', action="store_true", help="Create in-coming link to current node"),
make_option('-b', '--bi', action="store_true", help="Create bi-directional link to/from current node"),
make_option('-c', '--current', type="string", help="Specific a different node id to use than the current/working node"),
make_option('-p', '--prompt', action="store_true", help="Prompt for link values")
])
def do_poortego_ln(self, arg, opts):
"""Command to add link"""
# TODO - also account for link properties
poortego_ln(self.my_interface, arg, opts)
@options([
make_option('-c', '--csv', action="store_true", help="Import data from CSV file"),
make_option('-m', '--maltego', action="store_true", help="Import data from Maltego file"),
make_option('-s', '--stix', action="store_true", help="Import data from STIX file"),
])
def do_poortego_import(self, arg, opts=None):
"""Command to import data (STIX, CSV, IOC, etc.)"""
# TODO
poortego_import(self, arg, opts)
@options([
make_option('-d', '--device-id', action="store", dest='device_id', type='string',
help='ONU Device ID', default=None),
])
def do_show_msg_types(self, _line, opts):
""" Show supported OMCI Message Types"""
device_id = opts.device_id or self.device_id
try:
mib_db = self.get_device_mib(device_id, depth=1)
if mib_db is None:
self.poutput(self.colorize('Message Types for ONU {} are not currently available'
.format(device_id), 'red'))
return
mib = self._device_to_dict(mib_db)
@options([
make_option("-n", "--name", action = "store", type = "string", help = "the label used by node")
])
def do_delete_node(self, arg, opts):
"""Remove a node of topology"""
exist = self.dataplane.exist_node(opts.name)
if exist:
node = self.dataplane.get_node(opts.name)
node.delete()
self.dataplane.del_node(name = opts.name)
print("the node {name} has removed".format(name = opts.name))
else:
print("the node was not found")
@options([make_option('-c', '--channel', action="store",
help="Channel on which to grant"),
make_option('-a', '--auth-key', dest="auth_key",
action="store",
help="Auth Key"),
make_option('-r', '--read-enabled', dest='read',
action='store_true',
default=False, help="Read ?"),
make_option('-w', '--write-enabled', dest='write',
action='store_true',
default=False, help="Write ?"),
make_option('-t', '--ttl', action="store",
default=5, help="TTL"),
make_option('-p', '--presence', action="store_true",
dest="presence",
default=False, help="Grant on presence channel ?")
])
@options([],arg_desc="[debug|info|critical]")
def do_debug(self, level, opts=None):
print level
@options([
make_option('-i', '--filter-id', action="store", dest='filter_id')
])
def do_delete(self, line, opts):
if not opts.filter_id:
self.poutput(self.colorize('Error: ', 'red') + 'Specify ' + \
self.colorize(self.colorize('"filter id"', 'blue'),
'bold') + ' to update')
return
stub = self.get_stub()
stub.DeleteAlarmFilter(voltha_pb2.ID(id=opts.filter_id))