CLI to aid integration and testing
Change-Id: If18f194e45a8fc090a6b7869bb6d81728397ec9b
diff --git a/cli/main.py b/cli/main.py
index 7b90038..e9e881a 100755
--- a/cli/main.py
+++ b/cli/main.py
@@ -14,38 +14,59 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
-import sys
-from cmd2 import Cmd, make_option, options
import readline
+import sys
+from optparse import make_option
+from time import sleep
+
import grpc
+import requests
+from cmd2 import Cmd, options
+from google.protobuf.empty_pb2 import Empty
from simplejson import dumps
+from cli.device import DeviceCli
+from cli.logical_device import LogicalDeviceCli
+from voltha.core.flow_decomposer import *
from voltha.protos import third_party
from voltha.protos import voltha_pb2
-from google.protobuf.empty_pb2 import Empty
-from google.protobuf.json_format import MessageToDict
+from voltha.protos.openflow_13_pb2 import FlowTableUpdate
+
_ = third_party
-from cli.utils import print_flows
+from cli.utils import pb2dict
-def pb2dict(pb_msg):
- d = MessageToDict(pb_msg, including_default_value_fields=1,
- preserving_proto_field_name=1)
- return d
-
+banner = """\
+ _ _ _ _ _
+ __ _____| | |_| |_ __ _ __| (_)
+ \ V / _ \ | _| ' \/ _` | / _| | |
+ \_/\___/_|\__|_||_\__,_| \__|_|_|
+(to exit type q, exit or quit or hit Ctrl-D)
+"""
class VolthaCli(Cmd):
prompt = 'voltha'
history_file_name = '.voltha_cli_history'
+
+ # Settable CLI parameters
+ voltha_grpc = 'localhost:50055'
+ voltha_sim_rest = 'localhost:18880'
max_history_lines = 500
+ default_device_id = None
+ default_logical_device_id = None
Cmd.settable.update(dict(
- voltha_grpc='Voltha GRPC endpoint in form of <host>:<port>'
+ voltha_grpc='Voltha GRPC endpoint in form of <host>:<port>',
+ voltha_sim_rest='Voltha simulation back door for testing in form '
+ 'of <host>:<port>',
+ max_history_lines='Maximum number of history lines stored across '
+ 'sessions',
+ default_device_id='Device id used when no device id is specified',
+ default_logical_device_id='Logical device id used when no device id '
+ 'is specified',
))
- voltha_grpc = 'localhost:50055'
-
def __init__(self, *args, **kw):
Cmd.__init__(self, *args, **kw)
self.prompt = '(' + self.colorize(
@@ -79,6 +100,9 @@
self.channel = grpc.insecure_channel(self.voltha_grpc)
return self.channel
+ def preloop(self):
+ self.poutput(banner)
+
def do_reset_history(self, arg):
"""Reset CLI history"""
while self.history:
@@ -111,12 +135,18 @@
def do_device(self, arg):
"""Enter device level command mode"""
- sub = DeviceCli(self.get_channel, arg)
+ device_id = arg or self.default_device_id
+ if not device_id:
+ raise Exception('<device-id> parameter needed')
+ sub = DeviceCli(self.get_channel, device_id)
sub.cmdloop()
def do_logical_device(self, arg):
"""Enter logical device level command mode"""
- sub = LogicalDeviceCli(self.get_channel, arg)
+ logical_device_id = arg or self.default_logical_device_id
+ if not logical_device_id:
+ raise Exception('<logical-device-id> parameter needed')
+ sub = LogicalDeviceCli(self.get_channel, logical_device_id)
sub.cmdloop()
def do_debug(self, arg):
@@ -130,70 +160,111 @@
res = stub.GetHealthStatus(Empty())
print dumps(pb2dict(res), indent=4)
+ def do_test(self, arg):
+ """Enter test mode, which makes a bunch on new commands available"""
+ sub = TestCli(self.history, self.get_channel)
+ sub.cmdloop()
-class DeviceCli(Cmd):
- def __init__(self, get_channel, device_id):
- Cmd.__init__(self)
+class TestCli(VolthaCli):
+
+ def __init__(self, history, get_channel):
+ VolthaCli.__init__(self)
+ self.history = history
self.get_channel = get_channel
- self.device_id = device_id
- self.prompt = '(' + self.colorize(
- self.colorize('device {}'.format(device_id), 'red'), 'bold') + ') '
-
- def get_device(self, depth=0):
- stub = voltha_pb2.VolthaLocalServiceStub(self.get_channel())
- res = stub.GetDevice(voltha_pb2.ID(id=self.device_id),
- metadata=(('get-depth', str(depth)), ))
- return res
-
- def do_show(self, arg):
- """Show detailed device information"""
- print dumps(pb2dict(self.get_device(depth=-1)),
- indent=4, sort_keys=True)
-
- def do_flows(self, arg):
- """Show flow table for device"""
- device = pb2dict(self.get_device(-1))
- print_flows(
- 'Device',
- self.device_id,
- type=device['type'],
- flows=device['flows']['items'],
- groups=device['flow_groups']['items']
- )
-
-
-class LogicalDeviceCli(Cmd):
-
- def __init__(self, get_channel, logical_device_id):
- Cmd.__init__(self)
- self.get_channel = get_channel
- self.logical_device_id = logical_device_id
- self.prompt = '(' + self.colorize(
- self.colorize('device {}'.format(logical_device_id), 'red'),
+ self.prompt = '(' + self.colorize(self.colorize('test', 'cyan'),
'bold') + ') '
- def get_logical_device(self, depth=0):
+ @options([
+ make_option('-t', '--device-type', action="store", dest='device_type',
+ help="Device type", default='simulated_olt'),
+ make_option('-m', '--mac-address', action='store', dest='mac_address',
+ default='00:0c:e2:31:40:00'),
+ make_option('-i', '--ip-address', action='store', dest='ip_address'),
+ ])
+ def do_preprovision_olt(self, arg, opts):
+ """Preprovision a new OLT with given device type"""
stub = voltha_pb2.VolthaLocalServiceStub(self.get_channel())
- res = stub.GetLogicalDevice(voltha_pb2.ID(id=self.logical_device_id),
- metadata=(('get-depth', str(depth)), ))
- return res
+ kw = dict(type=opts.device_type)
+ if opts.ip_address:
+ kw['ipv4_address'] = opts.ip_address
+ elif opts.mac_address:
+ kw['mac_address'] = opts.mac_address
+ else:
+ raise Exception('Either IP address or Mac Address is needed')
+ device = voltha_pb2.Device(**kw)
+ device = stub.CreateDevice(device)
+ print 'success (device id = {})'.format(device.id)
+ self.default_device_id = device.id
- def do_show(self, arg):
- """Show detailed logical device information"""
- print dumps(pb2dict(self.get_logical_device(depth=-1)),
- indent=4, sort_keys=True)
+ def do_activate_olt(self, arg):
+ """
+ Activate an OLT. If the <id> is not provided, it will be on the last
+ pre-provisioned OLT.
+ """
+ device_id = arg or self.default_device_id
+ print 'activating', device_id
+ stub = voltha_pb2.VolthaLocalServiceStub(self.get_channel())
+ stub.ActivateDevice(voltha_pb2.ID(id=device_id))
- def do_flows(self, arg):
- """Show flow table for logical device"""
- logical_device = pb2dict(self.get_logical_device(-1))
- print_flows(
- 'Logical Device',
- self.logical_device_id,
- type='n/a',
- flows=logical_device['flows']['items'],
- groups=logical_device['flow_groups']['items']
+ # try to acquire logical device id
+ while True:
+ device = stub.GetDevice(voltha_pb2.ID(id=device_id))
+ if device.oper_status == voltha_pb2.OperStatus.ACTIVE:
+ assert device.parent_id
+ self.default_logical_device_id = device.parent_id
+ break
+ print 'waiting for device to be activated...'
+ sleep(1)
+ print 'success (logical device id = {})'.format(
+ self.default_logical_device_id)
+
+ def do_arrive_onus(self, arg):
+ """
+ Simulate the arrival of ONUs
+ """
+ device_id = arg or self.default_device_id
+ requests.get('http://{}/devices/{}/detect_onus'.format(
+ self.voltha_sim_rest, device_id
+ ))
+
+ def do_install_eapol_flow(self, arg):
+ """
+ Install an EAPOL flow on the given logical device. If device is not
+ given, it will be applied to logical device of the last pre-provisioned
+ OLT device.
+ """
+ logical_device_id = arg or self.default_logical_device_id
+ update = FlowTableUpdate(
+ id=logical_device_id,
+ flow_mod = mk_simple_flow_mod(
+ priority=2000,
+ match_fields=[in_port(101), eth_type(0x888e)],
+ actions=[
+ push_vlan(0x8100),
+ set_field(vlan_vid(4096 + 4000)),
+ output(ofp.OFPP_CONTROLLER)
+ ]
+ )
)
+ stub = voltha_pb2.VolthaLocalServiceStub(self.get_channel())
+ res = stub.UpdateLogicalDeviceFlowTable(update)
+ print 'success', res
+
+ def do_send_simulated_upstream_eapol(self, arg):
+ """
+ Send an EAPOL upstream from a simulated OLT
+ """
+ device_id = arg or self.default_device_id
+ requests.get('http://{}/devices/{}/test_eapol_in'.format(
+ self.voltha_sim_rest, device_id
+ ))
+
+ def do_inject_eapol_start(self, arg):
+ """
+ Send out an an EAPOL start message into the given Unix interface
+ """
+ pass
if __name__ == '__main__':