diff options
| -rw-r--r-- | os_net_config/objects.py | 127 | ||||
| -rw-r--r-- | os_net_config/tests/test_objects.py | 138 |
2 files changed, 259 insertions, 6 deletions
diff --git a/os_net_config/objects.py b/os_net_config/objects.py index 45824de..10f3060 100644 --- a/os_net_config/objects.py +++ b/os_net_config/objects.py @@ -13,12 +13,34 @@ # under the License. import netaddr +from openstack.common import strutils -class NetworkObjectException(Exception): +class InvalidConfigException(ValueError): pass +def object_from_json(json): + obj_type = json.get("type") + if obj_type == "interface": + return Interface.from_json(json) + elif obj_type == "vlan": + return Vlan.from_json(json) + elif obj_type == "ovs_bridge": + return OvsBridge.from_json(json) + elif obj_type == "ovs_bond": + return OvsBond.from_json(json) + + +def _get_required_field(json, name, object_name): + field = json.get(name) + if not field: + msg = '%s JSON objects require \'%s\' to be configured.' \ + % (object_name, name) + raise InvalidConfigException(msg) + return field + + class Route(object): """Base class for network routes.""" @@ -27,17 +49,28 @@ class Route(object): self.ip_netmask = ip_netmask self.default = default + @staticmethod + def from_json(json): + next_hop = _get_required_field(json, 'next_hop', 'Route') + ip_netmask = json.get('ip_netmask', "") + default = strutils.bool_from_string(str(json.get('default', False))) + return Route(next_hop, ip_netmask, default) + class Address(object): """Base class for network addresses.""" - def __init__(self, ip_netmask, routes=[]): + def __init__(self, ip_netmask): self.ip_netmask = ip_netmask ip_nw = netaddr.IPNetwork(self.ip_netmask) self.ip = str(ip_nw.ip) self.netmask = str(ip_nw.netmask) self.version = ip_nw.version - self.routes = routes + + @staticmethod + def from_json(json): + ip_netmask = _get_required_field(json, 'ip_netmask', 'Address') + return Address(ip_netmask) class _BaseOpts(object): @@ -70,6 +103,37 @@ class _BaseOpts(object): return v6_addresses + @staticmethod + def base_opts_from_json(json): + use_dhcp = strutils.bool_from_string(str(json.get('use_dhcp', False))) + use_dhcpv6 = strutils.bool_from_string(str(json.get('use_dhcpv6', + False))) + mtu = json.get('mtu', 1500) + addresses = [] + routes = [] + + # addresses + addresses_json = json.get('addresses') + if addresses_json: + if isinstance(addresses_json, list): + for address in addresses_json: + addresses.append(Address.from_json(address)) + else: + msg = 'Addresses must be a list.' + raise InvalidConfigException(msg) + + # routes + routes_json = json.get('routes') + if routes_json: + if isinstance(routes_json, list): + for route in routes_json: + routes.append(Route.from_json(route)) + else: + msg = 'Routes must be a list.' + raise InvalidConfigException(msg) + + return (use_dhcp, use_dhcpv6, addresses, routes, mtu) + class Interface(_BaseOpts): """Base class for network interfaces.""" @@ -79,11 +143,17 @@ class Interface(_BaseOpts): super(Interface, self).__init__(name, use_dhcp, use_dhcpv6, addresses, routes, mtu) + @staticmethod + def from_json(json): + name = _get_required_field(json, 'name', 'Interface') + opts = _BaseOpts.base_opts_from_json(json) + return Interface(name, *opts) + class Vlan(_BaseOpts): """Base class for VLANs. - NOTE: the name parameter must be formated w/ vlan#### where #### + NOTE: the name parameter must be formated w/ vlan<num> where <num> matches the vlan ID being used. Example: vlan5 """ @@ -95,12 +165,19 @@ class Vlan(_BaseOpts): self.vlan_id = int(vlan_id) self.device = device + @staticmethod + def from_json(json): + device = _get_required_field(json, 'device', 'Vlan') + vlan_id = _get_required_field(json, 'vlan_id', 'Vlan') + opts = _BaseOpts.base_opts_from_json(json) + return Vlan(device, vlan_id, *opts) + class OvsBridge(_BaseOpts): """Base class for OVS bridges.""" def __init__(self, name, use_dhcp=False, use_dhcpv6=False, addresses=[], - routes=[], members=[], mtu=1500, ovs_options=None): + routes=[], mtu=1500, members=[], ovs_options=None): super(OvsBridge, self).__init__(name, use_dhcp, use_dhcpv6, addresses, routes, mtu) self.members = members @@ -109,13 +186,51 @@ class OvsBridge(_BaseOpts): member.bridge_name = name member.ovs_port = True + @staticmethod + def from_json(json): + name = _get_required_field(json, 'name', 'OvsBridge') + opts = _BaseOpts.base_opts_from_json(json) + ovs_options = json.get('ovs_options') + members = [] + + # members + members_json = json.get('members') + if members_json: + if isinstance(members_json, list): + for member in members_json: + members.append(object_from_json(member)) + else: + msg = 'Members must be a list.' + raise InvalidConfigException(msg) + + return OvsBridge(name, *opts, members=members, ovs_options=ovs_options) + class OvsBond(_BaseOpts): """Base class for OVS bonds.""" def __init__(self, name, use_dhcp=False, use_dhcpv6=False, addresses=[], - routes=[], members=[], mtu=1500, ovs_options=None): + routes=[], mtu=1500, members=[], ovs_options=None): super(OvsBond, self).__init__(name, use_dhcp, use_dhcpv6, addresses, routes, mtu) self.members = members self.ovs_options = ovs_options + + @staticmethod + def from_json(json): + name = _get_required_field(json, 'name', 'OvsBond') + opts = _BaseOpts.base_opts_from_json(json) + ovs_options = json.get('ovs_options') + members = [] + + # members + members_json = json.get('members') + if members_json: + if isinstance(members_json, list): + for member in members_json: + members.append(object_from_json(member)) + else: + msg = 'Members must be a list.' + raise InvalidConfigException(msg) + + return OvsBond(name, *opts, members=members, ovs_options=ovs_options) diff --git a/os_net_config/tests/test_objects.py b/os_net_config/tests/test_objects.py index 7ffd5b3..51a2793 100644 --- a/os_net_config/tests/test_objects.py +++ b/os_net_config/tests/test_objects.py @@ -12,10 +12,37 @@ # License for the specific language governing permissions and limitations # under the License. +import json + from os_net_config import objects from os_net_config.tests import base +class TestRoute(base.TestCase): + + def test_from_json(self): + data = '{"next_hop": "172.19.0.1", "ip_netmask": "172.19.0.0/24"}' + route = objects.Route.from_json(json.loads(data)) + self.assertEqual("172.19.0.1", route.next_hop) + self.assertEqual("172.19.0.0/24", route.ip_netmask) + self.assertEqual(False, route.default) + + def test_from_json_default_route(self): + data = '{"next_hop": "172.19.0.1", "ip_netmask": "172.19.0.0/24", ' \ + '"default": true}' + route = objects.Route.from_json(json.loads(data)) + self.assertEqual("172.19.0.1", route.next_hop) + self.assertEqual("172.19.0.0/24", route.ip_netmask) + self.assertEqual(True, route.default) + + data = '{"next_hop": "172.19.0.1", "ip_netmask": "172.19.0.0/24", ' \ + '"default": "true"}' + route = objects.Route.from_json(json.loads(data)) + self.assertEqual("172.19.0.1", route.next_hop) + self.assertEqual("172.19.0.0/24", route.ip_netmask) + self.assertEqual(True, route.default) + + class TestAddress(base.TestCase): def test_ipv4_address(self): @@ -30,6 +57,23 @@ class TestAddress(base.TestCase): self.assertEqual("ffff:ffff:ffff:ffff::", address.netmask) self.assertEqual(6, address.version) + def test_from_json(self): + data = '{"ip_netmask": "192.0.2.5/24"}' + address = objects.Address.from_json(json.loads(data)) + self.assertEqual("192.0.2.5", address.ip) + self.assertEqual("255.255.255.0", address.netmask) + self.assertEqual(4, address.version) + + def test_from_json_invalid(self): + self.assertRaises(objects.InvalidConfigException, + objects.Address.from_json, + {}) + data = '{"ip_netmask": false}' + json_data = json.loads(data) + self.assertRaises(objects.InvalidConfigException, + objects.Address.from_json, + json_data) + class TestInterface(base.TestCase): @@ -39,3 +83,97 @@ class TestInterface(base.TestCase): interface = objects.Interface('foo', addresses=[v4_addr, v6_addr]) self.assertEquals("192.168.1.1", interface.v4_addresses()[0].ip) self.assertEquals("2001:abc:a::", interface.v6_addresses()[0].ip) + + def test_from_json_dhcp(self): + data = '{"type": "interface", "name": "em1", "use_dhcp": true}' + interface = objects.object_from_json(json.loads(data)) + self.assertEqual("em1", interface.name) + self.assertEqual(True, interface.use_dhcp) + + def test_from_json_with_addresses(self): + data = """{ +"type": "interface", +"name": "em1", +"use_dhcp": false, +"mtu": 1501, +"addresses": [{ + "ip_netmask": "192.0.2.1/24" +}], +"routes": [{ + "next_hop": "192.0.2.1", + "ip_netmask": "192.0.2.1/24" +}] +} +""" + interface = objects.object_from_json(json.loads(data)) + self.assertEqual("em1", interface.name) + self.assertEqual(False, interface.use_dhcp) + self.assertEqual(False, interface.use_dhcpv6) + self.assertEqual(1501, interface.mtu) + address1 = interface.v4_addresses()[0] + self.assertEqual("192.0.2.1", address1.ip) + self.assertEqual("255.255.255.0", address1.netmask) + route1 = interface.routes[0] + self.assertEqual("192.0.2.1", route1.next_hop) + self.assertEqual("192.0.2.1/24", route1.ip_netmask) + + +class TestVlan(base.TestCase): + + def test_from_json_dhcp(self): + data = '{"type": "vlan", "device": "em1", "vlan_id": 16,' \ + '"use_dhcp": true}' + vlan = objects.object_from_json(json.loads(data)) + self.assertEqual("em1", vlan.device) + self.assertEqual(16, vlan.vlan_id) + self.assertEqual(True, vlan.use_dhcp) + + +class TestBridge(base.TestCase): + + def test_from_json_dhcp(self): + data = """{ +"type": "ovs_bridge", +"name": "br-foo", +"use_dhcp": true, +"members": [{ + "type": "interface", + "name": "em1" +}] +} +""" + bridge = objects.object_from_json(json.loads(data)) + self.assertEqual("br-foo", bridge.name) + self.assertEqual(True, bridge.use_dhcp) + interface1 = bridge.members[0] + self.assertEqual("em1", interface1.name) + self.assertEqual(True, interface1.ovs_port) + self.assertEqual("br-foo", interface1.bridge_name) + + +class TestBond(base.TestCase): + + def test_from_json_dhcp(self): + data = """{ +"type": "ovs_bond", +"name": "bond1", +"use_dhcp": true, +"members": [ + { + "type": "interface", + "name": "em1" + }, + { + "type": "interface", + "name": "em2" + } +] +} +""" + bridge = objects.object_from_json(json.loads(data)) + self.assertEqual("bond1", bridge.name) + self.assertEqual(True, bridge.use_dhcp) + interface1 = bridge.members[0] + self.assertEqual("em1", interface1.name) + interface2 = bridge.members[1] + self.assertEqual("em2", interface2.name) |