class MappingRPCBlaster(object):
- putheaders = {'Content-type': 'application/json'}
- getheaders = {'Accept': 'application/json'}
+ putheaders = {"Content-type": "application/json"}
+ getheaders = {"Accept": "application/json"}
- RPC_URL_LI = 'restconf/operations/lfm-mapping-database:'
- RPC_URL_BE = 'restconf/operations/odl-mappingservice:'
+ RPC_URL_LI = "restconf/operations/lfm-mapping-database:"
+ RPC_URL_BE = "restconf/operations/odl-mappingservice:"
TIMEOUT = 10
# Template for adding mappings
add_mapping_template = {
- u'input': {
- u'mapping-record': {
- u'recordTtl': 60,
- u'action': u'NoAction',
- u'authoritative': True,
- u'eid': {
- u'address-type':
- u'ietf-lisp-address-types:ipv4-prefix-afi',
- u'ipv4-prefix': u'10.0.0.0/32'
+ u"input": {
+ u"mapping-record": {
+ u"recordTtl": 60,
+ u"action": u"NoAction",
+ u"authoritative": True,
+ u"eid": {
+ u"address-type": u"ietf-lisp-address-types:ipv4-prefix-afi",
+ u"ipv4-prefix": u"10.0.0.0/32",
},
- u'LocatorRecord': [
+ u"LocatorRecord": [
{
- u'locator-id': u'ipv4:172.16.0.0',
- u'priority': 1,
- u'weight': 1,
- u'multicastPriority': 255,
- u'multicastWeight': 0,
- u'localLocator': True,
- u'rlocProbed': False,
- u'routed': True,
- u'rloc': {
- u'address-type':
- u'ietf-lisp-address-types:ipv4-afi',
- u'ipv4': u'172.16.0.0'
- }
+ u"locator-id": u"ipv4:172.16.0.0",
+ u"priority": 1,
+ u"weight": 1,
+ u"multicastPriority": 255,
+ u"multicastWeight": 0,
+ u"localLocator": True,
+ u"rlocProbed": False,
+ u"routed": True,
+ u"rloc": {
+ u"address-type": u"ietf-lisp-address-types:ipv4-afi",
+ u"ipv4": u"172.16.0.0",
+ },
}
- ]
+ ],
}
}
}
# Template for getting mappings
- get_mapping_template = {
- u'input': {
- u'eid': {
- u'ipv4-prefix': u'10.0.0.0'
- }
- }
- }
+ get_mapping_template = {u"input": {u"eid": {u"ipv4-prefix": u"10.0.0.0"}}}
def __init__(self, host, port, start_eid, mask, start_rloc, nmappings, v):
"""
print("Using the Beryllium and later RPC URL")
rpc_url = self.RPC_URL_BE
- self.post_url_template = 'http://' + self.host + ':' \
- + self.port + '/' + rpc_url
+ self.post_url_template = "http://" + self.host + ":" + self.port + "/" + rpc_url
def mapping_from_tpl(self, eid, mask, rloc):
"""Create an add-mapping RPC input dictionary from the mapping template
Returns:
:return dict: mapping - template modified with the arguments
"""
- mapping = copy.deepcopy(
- self.add_mapping_template['input']['mapping-record'])
- mapping['eid']['ipv4-prefix'] = str(netaddr.IPAddress(eid)) \
- + '/' + mask
- mapping['LocatorRecord'][0]['locator-id'] = 'ipv4:' \
- + str(netaddr.IPAddress(rloc))
- mapping['LocatorRecord'][0]['rloc']['ipv4'] \
- = str(netaddr.IPAddress(rloc))
+ mapping = copy.deepcopy(self.add_mapping_template["input"]["mapping-record"])
+ mapping["eid"]["ipv4-prefix"] = str(netaddr.IPAddress(eid)) + "/" + mask
+ mapping["LocatorRecord"][0]["locator-id"] = "ipv4:" + str(
+ netaddr.IPAddress(rloc)
+ )
+ mapping["LocatorRecord"][0]["rloc"]["ipv4"] = str(netaddr.IPAddress(rloc))
return mapping
def send_rpc(self, session, method, body):
:return int: status_code - HTTP status code
"""
rpc_url = self.post_url_template + method
- r = session.post(rpc_url, data=body, headers=self.putheaders,
- stream=False, auth=('admin', 'admin'),
- timeout=self.TIMEOUT)
+ r = session.post(
+ rpc_url,
+ data=body,
+ headers=self.putheaders,
+ stream=False,
+ auth=("admin", "admin"),
+ timeout=self.TIMEOUT,
+ )
return r.status_code
def add_n_mappings(self):
rpc = dict(self.add_mapping_template)
increment = pow(2, 32 - int(self.mask))
for i in range(self.nmappings):
- rpc['input']['mapping-record'] = self.mapping_from_tpl(
- self.start_eid + i * increment, self.mask,
- self.start_rloc + i)
+ rpc["input"]["mapping-record"] = self.mapping_from_tpl(
+ self.start_eid + i * increment, self.mask, self.start_rloc + i
+ )
rpc_json = json.dumps(rpc)
- self.send_rpc(self.session, 'add-mapping', rpc_json)
+ self.send_rpc(self.session, "add-mapping", rpc_json)
self.session.close()
def get_n_mappings(self):
increment = pow(2, 32 - int(self.mask))
for i in range(self.nmappings):
eid = self.start_eid + i * increment
- rpc['input']['eid']['ipv4-prefix'] = str(netaddr.IPAddress(eid)) \
- + '/' + self.mask
+ rpc["input"]["eid"]["ipv4-prefix"] = (
+ str(netaddr.IPAddress(eid)) + "/" + self.mask
+ )
rpc_json = json.dumps(rpc)
- self.send_rpc(self.session, 'get-mapping', rpc_json)
+ self.send_rpc(self.session, "get-mapping", rpc_json)
self.session.close()
if __name__ == "__main__":
- parser = argparse.ArgumentParser(description='Add simple IPv4 \
- prefix-to-IPv4 locator LISP mappings to OpenDaylight')
-
- parser.add_argument('--mode', default='add',
- help='Operating mode, can be "add" or "get" \
- (default is "add")')
- parser.add_argument('--host', default='127.0.0.1',
- help='Host where ODL controller is running (default \
- is 127.0.0.1)')
- parser.add_argument('--port', default='8181',
- help='Port on which ODL\'s RESTCONF is listening \
- (default is 8181)')
- parser.add_argument('--start-eid', default='10.0.0.0',
- help='Start incrementing EID from this address \
- (default is 10.0.0.0)')
- parser.add_argument('--mask', default='32',
- help='Network mask for the IPv4 EID prefixes \
- (default is 32)')
- parser.add_argument('--start-rloc', default='172.16.0.0',
- help='Start incrementing RLOC from this address \
- (default is 172.16.0.0, ignored for "get")')
- parser.add_argument('--mappings', type=int, default=1,
- help='Number of mappings to add/get (default 1)')
- parser.add_argument('--odl-version', default='Be',
- help='OpenDaylight version, can be "Li" or "Be" \
- (default is "Be")')
+ parser = argparse.ArgumentParser(
+ description="Add simple IPv4 \
+ prefix-to-IPv4 locator LISP mappings to OpenDaylight"
+ )
+
+ parser.add_argument(
+ "--mode",
+ default="add",
+ help='Operating mode, can be "add" or "get" \
+ (default is "add")',
+ )
+ parser.add_argument(
+ "--host",
+ default="127.0.0.1",
+ help="Host where ODL controller is running (default \
+ is 127.0.0.1)",
+ )
+ parser.add_argument(
+ "--port",
+ default="8181",
+ help="Port on which ODL's RESTCONF is listening \
+ (default is 8181)",
+ )
+ parser.add_argument(
+ "--start-eid",
+ default="10.0.0.0",
+ help="Start incrementing EID from this address \
+ (default is 10.0.0.0)",
+ )
+ parser.add_argument(
+ "--mask",
+ default="32",
+ help="Network mask for the IPv4 EID prefixes \
+ (default is 32)",
+ )
+ parser.add_argument(
+ "--start-rloc",
+ default="172.16.0.0",
+ help='Start incrementing RLOC from this address \
+ (default is 172.16.0.0, ignored for "get")',
+ )
+ parser.add_argument(
+ "--mappings",
+ type=int,
+ default=1,
+ help="Number of mappings to add/get (default 1)",
+ )
+ parser.add_argument(
+ "--odl-version",
+ default="Be",
+ help='OpenDaylight version, can be "Li" or "Be" \
+ (default is "Be")',
+ )
in_args = parser.parse_args()
- mapping_rpc_blaster = MappingRPCBlaster(in_args.host, in_args.port,
- in_args.start_eid, in_args.mask,
- in_args.start_rloc,
- in_args.mappings,
- in_args.odl_version)
+ mapping_rpc_blaster = MappingRPCBlaster(
+ in_args.host,
+ in_args.port,
+ in_args.start_eid,
+ in_args.mask,
+ in_args.start_rloc,
+ in_args.mappings,
+ in_args.odl_version,
+ )
if in_args.mode == "add":
mapping_rpc_blaster.add_n_mappings()
Code Review / integration / test.git / blobdiff