tsarevvv/nmstate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

black: Cover tests/lib/nm

Browse Source 
Signed-off-by: Edward Haas <edwardh@redhat.com>
This commit is contained in:
Edward Haas 2019-06-16 09:52:38 +03:00
parent 9c3648d932
commit dd3ae4227b
18 changed files with 1353 additions and 1402 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

975
tests/lib/metadata_test.py
View File

File diff suppressed because it is too large Load Diff

62
tests/lib/netapplier_test.py
View File

@ -55,27 +55,23 @@ def test_iface_admin_state_change(netinfo_nm_mock, netapplier_nm_mock):
'name': 'foo',
'type': 'unknown',
'state': 'up',
'ipv4': {
'enabled': False,
},
'ipv6': {
'enabled': False,
},
'ipv4': {'enabled': False},
'ipv6': {'enabled': False},
}
]
}
desired_config = copy.deepcopy(current_config)
current_iface0 = current_config[INTERFACES][0]
netinfo_nm_mock.device.list_devices.return_value = ['one-item']
netinfo_nm_mock.translator.Nm2Api.get_common_device_info.return_value = (
current_config[INTERFACES][0])
current_iface0
)
netinfo_nm_mock.bond.is_bond_type_id.return_value = False
netinfo_nm_mock.ovs.is_ovs_bridge_type_id.return_value = False
netinfo_nm_mock.ovs.is_ovs_port_type_id.return_value = False
netinfo_nm_mock.ipv4.get_info.return_value = (
current_config[INTERFACES][0]['ipv4'])
netinfo_nm_mock.ipv6.get_info.return_value = (
current_config[INTERFACES][0]['ipv6'])
netinfo_nm_mock.ipv4.get_info.return_value = current_iface0['ipv4']
netinfo_nm_mock.ipv6.get_info.return_value = current_iface0['ipv6']
netinfo_nm_mock.ipv4.get_route_running.return_value = []
netinfo_nm_mock.ipv4.get_route_config.return_value = []
netinfo_nm_mock.ipv6.get_route_running.return_value = []
@ -86,14 +82,17 @@ def test_iface_admin_state_change(netinfo_nm_mock, netapplier_nm_mock):
applier_mock = netapplier_nm_mock.applier
ifaces_conf_new = (
applier_mock.prepare_new_ifaces_configuration.return_value)
applier_mock.prepare_new_ifaces_configuration.return_value
)
ifaces_conf_edit = (
applier_mock.prepare_edited_ifaces_configuration.return_value)
applier_mock.prepare_edited_ifaces_configuration.return_value
)
applier_mock.set_ifaces_admin_state.assert_has_calls(
[
mock.call([], con_profiles=ifaces_conf_new),
mock.call(desired_config[INTERFACES],
con_profiles=ifaces_conf_edit)
mock.call(
desired_config[INTERFACES], con_profiles=ifaces_conf_edit
),
]
)
@ -114,12 +113,10 @@ def test_add_new_bond(netinfo_nm_mock, netapplier_nm_mock):
'link-aggregation': {
'mode': 'balance-rr',
'slaves': [],
'options': {
'miimon': 200,
}
'options': {'miimon': 200},
},
'ipv4': {},
'ipv6': {}
'ipv6': {},
}
]
}
@ -143,34 +140,27 @@ def test_edit_existing_bond(netinfo_nm_mock, netapplier_nm_mock):
'link-aggregation': {
'mode': 'balance-rr',
'slaves': [],
'options': {
'miimon': '100',
}
},
'ipv4': {
'enabled': False,
},
'ipv6': {
'enabled': False,
'options': {'miimon': '100'},
},
'ipv4': {'enabled': False},
'ipv6': {'enabled': False},
}
]
}
current_iface0 = current_config[INTERFACES][0]
netinfo_nm_mock.device.list_devices.return_value = ['one-item']
netinfo_nm_mock.translator.Nm2Api.get_common_device_info.return_value = {
'name': current_config[INTERFACES][0]['name'],
'type': current_config[INTERFACES][0]['type'],
'state': current_config[INTERFACES][0]['state'],
'name': current_iface0['name'],
'type': current_iface0['type'],
'state': current_iface0['state'],
}
netinfo_nm_mock.bond.is_bond_type_id.return_value = True
netinfo_nm_mock.translator.Nm2Api.get_bond_info.return_value = {
'link-aggregation': current_config[INTERFACES][0]['link-aggregation']
'link-aggregation': current_iface0['link-aggregation']
}
netinfo_nm_mock.ipv4.get_info.return_value = (
current_config[INTERFACES][0]['ipv4'])
netinfo_nm_mock.ipv6.get_info.return_value = (
current_config[INTERFACES][0]['ipv6'])
netinfo_nm_mock.ipv4.get_info.return_value = current_iface0['ipv4']
netinfo_nm_mock.ipv6.get_info.return_value = current_iface0['ipv6']
netinfo_nm_mock.ipv4.get_route_running.return_value = []
netinfo_nm_mock.ipv4.get_route_config.return_value = []
netinfo_nm_mock.ipv6.get_route_running.return_value = []

60
tests/lib/netinfo_test.py
View File

@ -41,37 +41,27 @@ def nm_dns_mock():
def test_netinfo_show_generic_iface(nm_mock, nm_dns_mock):
current_config = {
DNS.KEY: {
DNS.RUNNING: {},
DNS.CONFIG: {}
},
ROUTES: {
'config': [],
'running': []
},
DNS.KEY: {DNS.RUNNING: {}, DNS.CONFIG: {}},
ROUTES: {'config': [], 'running': []},
INTERFACES: [
{
'name': 'foo',
'type': 'unknown',
'state': 'up',
'ipv4': {
'enabled': False,
},
'ipv6': {
'enabled': False,
},
'ipv4': {'enabled': False},
'ipv6': {'enabled': False},
}
]
],
}
current_iface0 = current_config[INTERFACES][0]
nm_mock.device.list_devices.return_value = ['one-item']
nm_mock.translator.Nm2Api.get_common_device_info.return_value = (
current_config[INTERFACES][0])
current_iface0
)
nm_mock.bond.is_bond_type_id.return_value = False
nm_mock.ipv4.get_info.return_value = (
current_config[INTERFACES][0]['ipv4'])
nm_mock.ipv6.get_info.return_value = (
current_config[INTERFACES][0]['ipv6'])
nm_mock.ipv4.get_info.return_value = current_iface0['ipv4']
nm_mock.ipv6.get_info.return_value = current_iface0['ipv6']
nm_mock.ipv4.get_route_running.return_value = []
nm_mock.ipv4.get_route_config.return_value = []
nm_mock.ipv6.get_route_running.return_value = []
@ -86,14 +76,8 @@ def test_netinfo_show_generic_iface(nm_mock, nm_dns_mock):
def test_netinfo_show_bond_iface(nm_mock, nm_dns_mock):
current_config = {
DNS.KEY: {
DNS.RUNNING: {},
DNS.CONFIG: {}
},
ROUTES: {
'config': [],
'running': []
},
DNS.KEY: {DNS.RUNNING: {}, DNS.CONFIG: {}},
ROUTES: {'config': [], 'running': []},
INTERFACES: [
{
'name': 'bond99',
@ -102,18 +86,12 @@ def test_netinfo_show_bond_iface(nm_mock, nm_dns_mock):
'link-aggregation': {
'mode': 'balance-rr',
'slaves': [],
'options': {
'miimon': '100',
}
},
'ipv4': {
'enabled': False,
},
'ipv6': {
'enabled': False,
'options': {'miimon': '100'},
},
'ipv4': {'enabled': False},
'ipv6': {'enabled': False},
}
]
],
}
nm_mock.device.list_devices.return_value = ['one-item']
@ -126,10 +104,8 @@ def test_netinfo_show_bond_iface(nm_mock, nm_dns_mock):
nm_mock.translator.Nm2Api.get_bond_info.return_value = {
'link-aggregation': current_config[INTERFACES][0]['link-aggregation']
}
nm_mock.ipv4.get_info.return_value = (
current_config[INTERFACES][0]['ipv4'])
nm_mock.ipv6.get_info.return_value = (
current_config[INTERFACES][0]['ipv6'])
nm_mock.ipv4.get_info.return_value = current_config[INTERFACES][0]['ipv4']
nm_mock.ipv6.get_info.return_value = current_config[INTERFACES][0]['ipv6']
nm_mock.ipv4.get_route_running.return_value = []
nm_mock.ipv4.get_route_config.return_value = []
nm_mock.ipv6.get_route_running.return_value = []

150
tests/lib/nm/applier_test.py
View File

@ -66,17 +66,15 @@ def test_create_new_ifaces(con_profile_mock):
nm.applier.create_new_ifaces(con_profiles)
for con_profile in con_profiles:
con_profile.add.assert_has_calls(
[mock.call(save_to_disk=True)])
con_profile.add.assert_has_calls([mock.call(save_to_disk=True)])
@mock.patch.object(nm.translator.Api2Nm, 'get_iface_type',
staticmethod(lambda t: t))
def test_prepare_new_ifaces_configuration(nm_bond_mock,
nm_connection_mock,
nm_ipv4_mock,
nm_ipv6_mock,
nm_ovs_mock):
@mock.patch.object(
nm.translator.Api2Nm, 'get_iface_type', staticmethod(lambda t: t)
)
def test_prepare_new_ifaces_configuration(
nm_bond_mock, nm_connection_mock, nm_ipv4_mock, nm_ipv6_mock, nm_ovs_mock
):
nm_ovs_mock.translate_bridge_options.return_value = {}
nm_ovs_mock.translate_port_options.return_value = {}
@ -86,7 +84,7 @@ def test_prepare_new_ifaces_configuration(nm_bond_mock,
'type': 'ethernet',
'state': 'up',
metadata.MASTER: 'bond99',
metadata.MASTER_TYPE: 'bond'
metadata.MASTER_TYPE: 'bond',
},
{
'name': 'bond99',
@ -95,37 +93,35 @@ def test_prepare_new_ifaces_configuration(nm_bond_mock,
'link-aggregation': {
'mode': 'balance-rr',
'slaves': ['eth0'],
'options': {
'miimon': 120
}
}
}
'options': {'miimon': 120},
},
},
]
nm.applier.prepare_new_ifaces_configuration(ifaces_desired_state)
con_setting = nm_connection_mock.ConnectionSetting.return_value
con_setting.set_master.assert_has_calls(
[
mock.call('bond99', 'bond'),
mock.call(None, None)
],
any_order=True
[mock.call('bond99', 'bond'), mock.call(None, None)], any_order=True
)
con_profile = nm_connection_mock.ConnectionProfile.return_value
con_profile.create.assert_has_calls(
[
mock.call([
nm_ipv4_mock.create_setting.return_value,
nm_ipv6_mock.create_setting.return_value,
con_setting.setting,
]),
mock.call([
nm_ipv4_mock.create_setting.return_value,
nm_ipv6_mock.create_setting.return_value,
con_setting.setting,
nm_bond_mock.create_setting.return_value,
])
mock.call(
[
nm_ipv4_mock.create_setting.return_value,
nm_ipv6_mock.create_setting.return_value,
con_setting.setting,
]
),
mock.call(
[
nm_ipv4_mock.create_setting.return_value,
nm_ipv6_mock.create_setting.return_value,
con_setting.setting,
nm_bond_mock.create_setting.return_value,
]
),
]
)
@ -138,116 +134,93 @@ def test_edit_existing_ifaces_with_profile(con_profile_mock, nm_device_mock):
for con_profile in con_profiles:
con_profile.commit.assert_has_calls(
[mock.call(nmdev=nm_device_mock.get_device_by_name.return_value)])
[mock.call(nmdev=nm_device_mock.get_device_by_name.return_value)]
)
@mock.patch.object(nm.connection, 'ConnectionProfile')
def test_edit_existing_ifaces_without_profile(con_profile_mock,
nm_device_mock):
def test_edit_existing_ifaces_without_profile(
con_profile_mock, nm_device_mock
):
con_profiles = [mock.MagicMock(), mock.MagicMock()]
con_profile_mock.return_value.profile = None
nm.applier.edit_existing_ifaces(con_profiles)
for con_profile in con_profiles:
con_profile.add.assert_has_calls(
[mock.call(save_to_disk=True)])
con_profile.add.assert_has_calls([mock.call(save_to_disk=True)])
@mock.patch.object(nm.translator.Api2Nm, 'get_iface_type',
staticmethod(lambda t: t))
def test_prepare_edited_ifaces_configuration(nm_device_mock,
nm_connection_mock,
nm_ipv4_mock,
nm_ipv6_mock,
nm_ovs_mock):
@mock.patch.object(
nm.translator.Api2Nm, 'get_iface_type', staticmethod(lambda t: t)
)
def test_prepare_edited_ifaces_configuration(
nm_device_mock, nm_connection_mock, nm_ipv4_mock, nm_ipv6_mock, nm_ovs_mock
):
nm_ovs_mock.translate_bridge_options.return_value = {}
nm_ovs_mock.translate_port_options.return_value = {}
ifaces_desired_state = [
{
'name': 'eth0',
'type': 'ethernet',
'state': 'up',
}
{'name': 'eth0', 'type': 'ethernet', 'state': 'up'}
]
cons = nm.applier.prepare_edited_ifaces_configuration(ifaces_desired_state)
assert len(cons) == 1
con_profile = nm_connection_mock.ConnectionProfile.return_value
con_profile.update.assert_has_calls(
[
mock.call(con_profile),
]
)
con_profile.update.assert_has_calls([mock.call(con_profile)])
class TestIfaceAdminStateControl(object):
def test_set_ifaces_admin_state_up(self, nm_device_mock):
ifaces_desired_state = [
{
'name': 'eth0',
'type': 'ethernet',
'state': 'up',
}
{'name': 'eth0', 'type': 'ethernet', 'state': 'up'}
]
con_profile = mock.MagicMock()
con_profile.devname = ifaces_desired_state[0]['name']
nm.applier.set_ifaces_admin_state(ifaces_desired_state, [con_profile])
nm_device_mock.modify.assert_called_once_with(
nm_device_mock.get_device_by_name.return_value,
con_profile.profile)
nm_device_mock.get_device_by_name.return_value, con_profile.profile
)
def test_set_ifaces_admin_state_down(self, nm_device_mock):
ifaces_desired_state = [
{
'name': 'eth0',
'type': 'ethernet',
'state': 'down',
}
{'name': 'eth0', 'type': 'ethernet', 'state': 'down'}
]
nm.applier.set_ifaces_admin_state(ifaces_desired_state)
nm_device_mock.deactivate.assert_called_once_with(
nm_device_mock.get_device_by_name.return_value)
nm_device_mock.get_device_by_name.return_value
)
nm_device_mock.delete.assert_called_once_with(
nm_device_mock.get_device_by_name.return_value)
nm_device_mock.get_device_by_name.return_value
)
def test_set_ifaces_admin_state_absent(self, nm_device_mock):
ifaces_desired_state = [
{
'name': 'eth0',
'type': 'ethernet',
'state': 'absent',
}
{'name': 'eth0', 'type': 'ethernet', 'state': 'absent'}
]
nm.applier.set_ifaces_admin_state(ifaces_desired_state)
nm_device_mock.deactivate.assert_called_once_with(
nm_device_mock.get_device_by_name.return_value)
nm_device_mock.get_device_by_name.return_value
)
nm_device_mock.delete.assert_called_once_with(
nm_device_mock.get_device_by_name.return_value)
nm_device_mock.get_device_by_name.return_value
)
def test_set_bond_and_its_slaves_admin_state_up(self,
nm_device_mock,
nm_bond_mock):
def test_set_bond_and_its_slaves_admin_state_up(
self, nm_device_mock, nm_bond_mock
):
ifaces_desired_state = [
{
'name': 'bond0',
'type': 'bond',
'state': 'up',
'link-aggregation': {
'mode': '802.3ad',
'slaves': ['eth0']
},
'link-aggregation': {'mode': '802.3ad', 'slaves': ['eth0']},
},
{
'name': 'eth0',
'type': 'ethernet',
'state': 'up',
}
{'name': 'eth0', 'type': 'ethernet', 'state': 'up'},
]
nm_device_mock.get_device_by_name = lambda devname: devname
@ -262,11 +235,12 @@ class TestIfaceAdminStateControl(object):
slave_con_profile.devname = ifaces_desired_state[1]['name']
nm.applier.set_ifaces_admin_state(
ifaces_desired_state, [bond_con_profile, slave_con_profile])
ifaces_desired_state, [bond_con_profile, slave_con_profile]
)
expected_calls = [
mock.call(bond, bond_con_profile.profile),
mock.call(slaves[0], slave_con_profile.profile)
mock.call(slaves[0], slave_con_profile.profile),
]
actual_calls = nm_device_mock.modify.mock_calls
assert sorted(expected_calls) == sorted(actual_calls)

17
tests/lib/nm/bond_test.py
View File

@ -32,15 +32,12 @@ def test_create_setting(NM_mock):
bond_setting_mock = NM_mock.SettingBond.new.return_value
bond_setting_mock.add_option.return_value = True
options = {
'mode': 'balance-rr',
'miimon': '100',
}
options = {'mode': 'balance-rr', 'miimon': '100'}
nm.bond.create_setting(options)
bond_setting_mock.add_option.assert_has_calls(
[mock.call('mode', 'balance-rr'), mock.call('miimon', '100')],
any_order=True
any_order=True,
)
@ -49,10 +46,7 @@ def test_create_setting_with_invalid_bond_option(NM_mock):
bond_setting_mock = NM_mock.SettingBond.new.return_value
bond_setting_mock.add_option.return_value = False
options = {
'mode': 'balance-rr',
'foo': '100',
}
options = {'mode': 'balance-rr', 'foo': '100'}
with pytest.raises(NmstateValueError):
nm.bond.create_setting(options)
@ -70,13 +64,14 @@ def test_get_bond_info(con_profile_mock, dev_mock):
info = nm.bond.get_bond_info(dev_mock)
con_profile_mock.return_value.import_by_device.assert_called_once_with(
dev_mock)
dev_mock
)
connection_mock = con_profile_mock.return_value.profile
opts_mock = connection_mock.get_setting_bond.return_value.props.options
expected_info = {
'slaves': dev_mock.get_slaves.return_value,
'options': opts_mock
'options': opts_mock,
}
assert expected_info == info

12
tests/lib/nm/connection_test.py
View File

@ -48,7 +48,8 @@ def test_create_profile(NM_mock):
con_profile_mock = NM_mock.SimpleConnection.new.return_value
con_profile_mock.add_setting.assert_has_calls(
[mock.call(settings[0]), mock.call(settings[1])])
[mock.call(settings[0]), mock.call(settings[1])]
)
assert con_profile_mock == con_profile.profile
@ -102,7 +103,8 @@ def test_create_setting(NM_mock):
assert con_setting.setting.props.type == 'iface-type'
assert con_setting.setting.props.autoconnect is True
assert con_setting.setting.props.autoconnect_slaves == (
NM_mock.SettingConnectionAutoconnectSlaves.YES)
NM_mock.SettingConnectionAutoconnectSlaves.YES
)
def test_duplicate_settings(NM_mock):
@ -135,5 +137,7 @@ def test_get_device_connection():
con = nm.connection.ConnectionProfile()
con.import_by_device(dev_mock)
assert (dev_mock.get_active_connection.return_value.props.connection ==
con.profile)
assert (
dev_mock.get_active_connection.return_value.props.connection
== con.profile
)

137
tests/lib/nm/dns_test.py
View File

@ -34,7 +34,7 @@ def _get_test_dns_v4():
return {
nm_dns.DNS_METADATA_PRIORITY: 40,
DNS.SERVER: ['8.8.8.8', '1.1.1.1'],
DNS.SEARCH: ['example.org', 'example.com']
DNS.SEARCH: ['example.org', 'example.com'],
}
@ -42,30 +42,28 @@ def _get_test_dns_v6():
return {
nm_dns.DNS_METADATA_PRIORITY: 40,
DNS.SERVER: ['2001:4860:4860::8888', '2606:4700:4700::1111'],
DNS.SEARCH: ['example.net', 'example.edu']
DNS.SEARCH: ['example.net', 'example.edu'],
}
parametrize_ip_ver = pytest.mark.parametrize(
'nm_ip',
[(nm_ipv4), (nm_ipv6)],
ids=['ipv4', 'ipv6'])
'nm_ip', [(nm_ipv4), (nm_ipv6)], ids=['ipv4', 'ipv6']
)
parametrize_ip_ver_dns = pytest.mark.parametrize(
'nm_ip, get_test_dns_func',
[(nm_ipv4, _get_test_dns_v4),
(nm_ipv6, _get_test_dns_v6)],
ids=['ipv4', 'ipv6'])
[(nm_ipv4, _get_test_dns_v4), (nm_ipv6, _get_test_dns_v6)],
ids=['ipv4', 'ipv6'],
)
@parametrize_ip_ver
def test_add_dns_empty(nm_ip):
dns_conf = {}
setting_ip = nm_ip.create_setting({
'enabled': True,
nm_dns.DNS_METADATA: dns_conf
}, base_con_profile=None)
setting_ip = nm_ip.create_setting(
{'enabled': True, nm_dns.DNS_METADATA: dns_conf}, base_con_profile=None
)
_assert_dns(setting_ip, dns_conf)
@ -73,10 +71,9 @@ def test_add_dns_empty(nm_ip):
@parametrize_ip_ver_dns
def test_add_dns(nm_ip, get_test_dns_func):
dns_conf = get_test_dns_func()
setting_ip = nm_ip.create_setting({
'enabled': True,
nm_dns.DNS_METADATA: dns_conf
}, base_con_profile=None)
setting_ip = nm_ip.create_setting(
{'enabled': True, nm_dns.DNS_METADATA: dns_conf}, base_con_profile=None
)
_assert_dns(setting_ip, dns_conf)
@ -85,10 +82,9 @@ def test_add_dns(nm_ip, get_test_dns_func):
def test_add_dns_duplicate_server(nm_ip, get_test_dns_func):
dns_conf = get_test_dns_func()
dns_conf[DNS.SERVER] = [dns_conf[DNS.SERVER][0], dns_conf[DNS.SERVER][0]]
setting_ip = nm_ip.create_setting({
'enabled': True,
nm_dns.DNS_METADATA: dns_conf
}, base_con_profile=None)
setting_ip = nm_ip.create_setting(
{'enabled': True, nm_dns.DNS_METADATA: dns_conf}, base_con_profile=None
)
dns_conf[DNS.SERVER] = [dns_conf[DNS.SERVER][0]]
_assert_dns(setting_ip, dns_conf)
@ -98,10 +94,9 @@ def test_add_dns_duplicate_server(nm_ip, get_test_dns_func):
def test_add_dns_duplicate_search(nm_ip, get_test_dns_func):
dns_conf = get_test_dns_func()
dns_conf[DNS.SEARCH] = [dns_conf[DNS.SEARCH][0], dns_conf[DNS.SEARCH][0]]
setting_ip = nm_ip.create_setting({
'enabled': True,
nm_dns.DNS_METADATA: dns_conf
}, base_con_profile=None)
setting_ip = nm_ip.create_setting(
{'enabled': True, nm_dns.DNS_METADATA: dns_conf}, base_con_profile=None
)
dns_conf[DNS.SEARCH] = [dns_conf[DNS.SEARCH][0]]
_assert_dns(setting_ip, dns_conf)
@ -110,16 +105,15 @@ def test_add_dns_duplicate_search(nm_ip, get_test_dns_func):
@parametrize_ip_ver_dns
def test_clear_dns(nm_ip, get_test_dns_func):
dns_conf = get_test_dns_func()
setting_ip = nm_ip.create_setting({
'enabled': True,
nm_dns.DNS_METADATA: dns_conf
}, base_con_profile=None)
setting_ip = nm_ip.create_setting(
{'enabled': True, nm_dns.DNS_METADATA: dns_conf}, base_con_profile=None
)
con_profile = nm_connection.ConnectionProfile()
con_profile.create([setting_ip])
new_setting_ip = nm_ip.create_setting({
'enabled': True,
nm_dns.DNS_METADATA: {}
}, base_con_profile=con_profile.profile)
new_setting_ip = nm_ip.create_setting(
{'enabled': True, nm_dns.DNS_METADATA: {}},
base_con_profile=con_profile.profile,
)
_assert_dns(new_setting_ip, {})
@ -127,67 +121,78 @@ def test_clear_dns(nm_ip, get_test_dns_func):
def _assert_dns(setting_ip, dns_conf):
assert setting_ip.props.dns == dns_conf.get(DNS.SERVER, [])
assert setting_ip.props.dns_search == dns_conf.get(DNS.SEARCH, [])
priority = dns_conf.get(nm_dns.DNS_METADATA_PRIORITY,
nm_dns.DEFAULT_DNS_PRIORITY)
priority = dns_conf.get(
nm_dns.DNS_METADATA_PRIORITY, nm_dns.DEFAULT_DNS_PRIORITY
)
assert setting_ip.props.dns_priority == priority
def test_find_interfaces_for_dns_with_ipv6_and_ipv4_static_gateways():
iface_routes = _get_test_iface_routes()
assert ((TEST_IPV4_GATEWAY_IFACE, TEST_IPV6_GATEWAY_IFACE) ==
nm_dns.find_interfaces_for_name_servers(iface_routes))
assert (
TEST_IPV4_GATEWAY_IFACE,
TEST_IPV6_GATEWAY_IFACE,
) == nm_dns.find_interfaces_for_name_servers(iface_routes)
def test_find_interfaces_for_dns_with_ipv6_static_gateway_only():
iface_routes = {
TEST_IPV6_GATEWAY_IFACE: _get_test_ipv6_gateway(),
TEST_STATIC_ROUTE_IFACE: _get_test_static_routes()
TEST_STATIC_ROUTE_IFACE: _get_test_static_routes(),
}
assert ((None, TEST_IPV6_GATEWAY_IFACE) ==
nm_dns.find_interfaces_for_name_servers(iface_routes))
assert (
None,
TEST_IPV6_GATEWAY_IFACE,
) == nm_dns.find_interfaces_for_name_servers(iface_routes)
def test_find_interfaces_for_dns_with_ipv4_static_gateway_only():
iface_routes = {
TEST_IPV4_GATEWAY_IFACE: _get_test_ipv4_gateway(),
TEST_STATIC_ROUTE_IFACE: _get_test_static_routes()
TEST_STATIC_ROUTE_IFACE: _get_test_static_routes(),
}
assert ((TEST_IPV4_GATEWAY_IFACE, None) ==
nm_dns.find_interfaces_for_name_servers(iface_routes))
assert (
TEST_IPV4_GATEWAY_IFACE,
None,
) == nm_dns.find_interfaces_for_name_servers(iface_routes)
def test_find_interfaces_for_dns_with_no_routes():
iface_routes = {}
assert ((None, None) ==
nm_dns.find_interfaces_for_name_servers(iface_routes))
assert (None, None) == nm_dns.find_interfaces_for_name_servers(
iface_routes
)
def test_find_interfaces_for_dns_with_no_gateway():
iface_routes = {
TEST_STATIC_ROUTE_IFACE: _get_test_static_routes()
}
assert ((None, None) ==
nm_dns.find_interfaces_for_name_servers(iface_routes))
iface_routes = {TEST_STATIC_ROUTE_IFACE: _get_test_static_routes()}
assert (None, None) == nm_dns.find_interfaces_for_name_servers(
iface_routes
)
def _get_test_ipv4_gateway():
return [{
Route.DESTINATION: '0.0.0.0/0',
Route.METRIC: 200,
Route.NEXT_HOP_ADDRESS: '192.0.2.1',
Route.NEXT_HOP_INTERFACE: TEST_IPV4_GATEWAY_IFACE,
Route.TABLE_ID: 54
}]
return [
{
Route.DESTINATION: '0.0.0.0/0',
Route.METRIC: 200,
Route.NEXT_HOP_ADDRESS: '192.0.2.1',
Route.NEXT_HOP_INTERFACE: TEST_IPV4_GATEWAY_IFACE,
Route.TABLE_ID: 54,
}
]
def _get_test_ipv6_gateway():
return [{
Route.DESTINATION: '::/0',
Route.METRIC: 201,
Route.NEXT_HOP_ADDRESS: '2001:db8:2::f',
Route.NEXT_HOP_INTERFACE: TEST_IPV6_GATEWAY_IFACE,
Route.TABLE_ID: 54
}]
return [
{
Route.DESTINATION: '::/0',
Route.METRIC: 201,
Route.NEXT_HOP_ADDRESS: '2001:db8:2::f',
Route.NEXT_HOP_INTERFACE: TEST_IPV6_GATEWAY_IFACE,
Route.TABLE_ID: 54,
}
]
def _get_test_static_routes():
@ -197,15 +202,15 @@ def _get_test_static_routes():
Route.METRIC: 201,
Route.NEXT_HOP_ADDRESS: '2001:db8:2::f',
Route.NEXT_HOP_INTERFACE: TEST_STATIC_ROUTE_IFACE,
Route.TABLE_ID: 54
Route.TABLE_ID: 54,
},
{
Route.DESTINATION: '198.51.100.0/24',
Route.METRIC: 201,
Route.NEXT_HOP_ADDRESS: '192.0.2.1',
Route.NEXT_HOP_INTERFACE: TEST_STATIC_ROUTE_IFACE,
Route.TABLE_ID: 54
}
Route.TABLE_ID: 54,
},
]

67
tests/lib/nm/ipv4_test.py
View File

@ -34,29 +34,29 @@ def test_create_setting_without_config(NM_mock):
ipv4_setting = nm.ipv4.create_setting(config=None, base_con_profile=None)
assert ipv4_setting == NM_mock.SettingIP4Config.new.return_value
assert (ipv4_setting.props.method ==
NM_mock.SETTING_IP4_CONFIG_METHOD_DISABLED)
assert (
ipv4_setting.props.method == NM_mock.SETTING_IP4_CONFIG_METHOD_DISABLED
)
def test_create_setting_with_ipv4_disabled(NM_mock):
ipv4_setting = nm.ipv4.create_setting(config={'enabled': False},
base_con_profile=None)
ipv4_setting = nm.ipv4.create_setting(
config={'enabled': False}, base_con_profile=None
)
assert (ipv4_setting.props.method ==
NM_mock.SETTING_IP4_CONFIG_METHOD_DISABLED)
assert (
ipv4_setting.props.method == NM_mock.SETTING_IP4_CONFIG_METHOD_DISABLED
)
def test_create_setting_without_addresses(NM_mock):
ipv4_setting = nm.ipv4.create_setting(
config={
'enabled': True,
'address': [],
},
base_con_profile=None
config={'enabled': True, 'address': []}, base_con_profile=None
)
assert (ipv4_setting.props.method ==
NM_mock.SETTING_IP4_CONFIG_METHOD_DISABLED)
assert (
ipv4_setting.props.method == NM_mock.SETTING_IP4_CONFIG_METHOD_DISABLED
)
def test_create_setting_with_static_addresses(NM_mock):
@ -69,21 +69,28 @@ def test_create_setting_with_static_addresses(NM_mock):
}
ipv4_setting = nm.ipv4.create_setting(config=config, base_con_profile=None)
assert (ipv4_setting.props.method ==
NM_mock.SETTING_IP4_CONFIG_METHOD_MANUAL)
assert (
ipv4_setting.props.method == NM_mock.SETTING_IP4_CONFIG_METHOD_MANUAL
)
NM_mock.IPAddress.new.assert_has_calls(
[
mock.call(nm.ipv4.socket.AF_INET,
config['address'][0]['ip'],
config['address'][0]['prefix-length']),
mock.call(nm.ipv4.socket.AF_INET,
config['address'][1]['ip'],
config['address'][1]['prefix-length'])
mock.call(
nm.ipv4.socket.AF_INET,
config['address'][0]['ip'],
config['address'][0]['prefix-length'],
),
mock.call(
nm.ipv4.socket.AF_INET,
config['address'][1]['ip'],
config['address'][1]['prefix-length'],
),
]
)
NM_mock.SettingIP4Config.new.return_value.add_address.assert_has_calls(
[mock.call(NM_mock.IPAddress.new.return_value),
mock.call(NM_mock.IPAddress.new.return_value)]
[
mock.call(NM_mock.IPAddress.new.return_value),
mock.call(NM_mock.IPAddress.new.return_value),
]
)
@ -113,7 +120,8 @@ def test_get_info_with_ipv4_config(NM_mock):
set_ip_conf = mock.MagicMock()
remote_conn_mock.get_setting_ip4_config.return_value = set_ip_conf
set_ip_conf.get_method.return_value = (
NM_mock.SETTING_IP4_CONFIG_METHOD_MANUAL)
NM_mock.SETTING_IP4_CONFIG_METHOD_MANUAL
)
set_ip_conf.props.never_default = False
set_ip_conf.props.ignore_auto_dns = False
set_ip_conf.props.ignore_auto_routes = False
@ -128,23 +136,22 @@ def test_get_info_with_ipv4_config(NM_mock):
'ip': address_mock.get_address.return_value,
'prefix-length': int(address_mock.get_prefix.return_value),
}
]
],
}
def test_create_setting_with_base_con_profile(NM_mock):
config = {
'enabled': True,
'address': [
{'ip': IPV4_ADDRESS1, 'prefix-length': 24},
],
'address': [{'ip': IPV4_ADDRESS1, 'prefix-length': 24}],
}
base_con_profile_mock = mock.MagicMock()
config_mock = base_con_profile_mock.get_setting_ip4_config.return_value
config_dup_mock = config_mock.duplicate.return_value
nm.ipv4.create_setting(config=config,
base_con_profile=base_con_profile_mock)
nm.ipv4.create_setting(
config=config, base_con_profile=base_con_profile_mock
)
base_con_profile_mock.get_setting_ip4_config.assert_called_once_with()
config_mock.duplicate.assert_called_once_with()

81
tests/lib/nm/ipv6_test.py
View File

@ -39,33 +39,34 @@ def test_create_setting_without_config(NM_mock):
ipv6_setting = nm.ipv6.create_setting(config=None, base_con_profile=None)
assert ipv6_setting == NM_mock.SettingIP6Config.new.return_value
assert (ipv6_setting.props.method ==
NM_mock.SETTING_IP6_CONFIG_METHOD_IGNORE)
assert (
ipv6_setting.props.method == NM_mock.SETTING_IP6_CONFIG_METHOD_IGNORE
)
def test_create_setting_with_ipv6_disabled(NM_mock):
NM_mock.SettingIP6Config.new().props.addresses = []
ipv6_setting = nm.ipv6.create_setting(config={'enabled': False},
base_con_profile=None)
ipv6_setting = nm.ipv6.create_setting(
config={'enabled': False}, base_con_profile=None
)
assert (ipv6_setting.props.method ==
NM_mock.SETTING_IP6_CONFIG_METHOD_IGNORE)
assert (
ipv6_setting.props.method == NM_mock.SETTING_IP6_CONFIG_METHOD_IGNORE
)
def test_create_setting_without_addresses(NM_mock):
NM_mock.SettingIP6Config.new().props.addresses = []
ipv6_setting = nm.ipv6.create_setting(
config={
'enabled': True,
'address': [],
},
base_con_profile=None
config={'enabled': True, 'address': []}, base_con_profile=None
)
assert (ipv6_setting.props.method ==
NM_mock.SETTING_IP6_CONFIG_METHOD_LINK_LOCAL)
assert (
ipv6_setting.props.method
== NM_mock.SETTING_IP6_CONFIG_METHOD_LINK_LOCAL
)
def test_create_setting_with_static_addresses(NM_mock):
@ -78,21 +79,28 @@ def test_create_setting_with_static_addresses(NM_mock):
}
ipv6_setting = nm.ipv6.create_setting(config=config, base_con_profile=None)
assert (ipv6_setting.props.method ==
NM_mock.SETTING_IP6_CONFIG_METHOD_MANUAL)
assert (
ipv6_setting.props.method == NM_mock.SETTING_IP6_CONFIG_METHOD_MANUAL
)
NM_mock.IPAddress.new.assert_has_calls(
[
mock.call(nm.ipv6.socket.AF_INET6,
config['address'][0]['ip'],
config['address'][0]['prefix-length']),
mock.call(nm.ipv6.socket.AF_INET6,
config['address'][1]['ip'],
config['address'][1]['prefix-length'])
mock.call(
nm.ipv6.socket.AF_INET6,
config['address'][0]['ip'],
config['address'][0]['prefix-length'],
),
mock.call(
nm.ipv6.socket.AF_INET6,
config['address'][1]['ip'],
config['address'][1]['prefix-length'],
),
]
)
NM_mock.SettingIP6Config.new.return_value.add_address.assert_has_calls(
[mock.call(NM_mock.IPAddress.new.return_value),
mock.call(NM_mock.IPAddress.new.return_value)]
[
mock.call(NM_mock.IPAddress.new.return_value),
mock.call(NM_mock.IPAddress.new.return_value),
]
)
@ -122,7 +130,8 @@ def test_get_info_with_ipv6_config(NM_mock):
set_ip_conf = mock.MagicMock()
remote_conn_mock.get_setting_ip6_config.return_value = set_ip_conf
set_ip_conf.get_method.return_value = (
NM_mock.SETTING_IP6_CONFIG_METHOD_MANUAL)
NM_mock.SETTING_IP6_CONFIG_METHOD_MANUAL
)
set_ip_conf.props.never_default = False
set_ip_conf.props.ignore_auto_dns = False
set_ip_conf.props.ignore_auto_routes = False
@ -138,7 +147,7 @@ def test_get_info_with_ipv6_config(NM_mock):
'ip': address_mock.get_address.return_value,
'prefix-length': int(address_mock.get_prefix.return_value),
}
]
],
}
@ -152,13 +161,16 @@ def test_create_setting_with_link_local_addresses(NM_mock):
}
ipv6_setting = nm.ipv6.create_setting(config=config, base_con_profile=None)
assert (ipv6_setting.props.method ==
NM_mock.SETTING_IP6_CONFIG_METHOD_MANUAL)
assert (
ipv6_setting.props.method == NM_mock.SETTING_IP6_CONFIG_METHOD_MANUAL
)
NM_mock.IPAddress.new.assert_has_calls(
[
mock.call(nm.ipv6.socket.AF_INET6,
config['address'][1]['ip'],
config['address'][1]['prefix-length'])
mock.call(
nm.ipv6.socket.AF_INET6,
config['address'][1]['ip'],
config['address'][1]['prefix-length'],
)
]
)
NM_mock.SettingIP6Config.new.return_value.add_address.assert_has_calls(
@ -169,16 +181,15 @@ def test_create_setting_with_link_local_addresses(NM_mock):
def test_create_setting_with_base_con_profile(NM_mock):
config = {
'enabled': True,
'address': [
{'ip': IPV6_ADDRESS1, 'prefix-length': 24},
],
'address': [{'ip': IPV6_ADDRESS1, 'prefix-length': 24}],
}
base_con_profile_mock = mock.MagicMock()
config_mock = base_con_profile_mock.get_setting_ip6_config.return_value
config_dup_mock = config_mock.duplicate.return_value
nm.ipv6.create_setting(config=config,
base_con_profile=base_con_profile_mock)
nm.ipv6.create_setting(
config=config, base_con_profile=base_con_profile_mock
)
base_con_profile_mock.get_setting_ip6_config.assert_called_once_with()
config_mock.duplicate.assert_called_once_with()

24
tests/lib/nm/ovs_test.py
View File

@ -68,15 +68,15 @@ def test_get_ovs_info_without_ports(nm_connection_mock, NM_mock):
'fail-mode': '',
'mcast-snooping-enable': False,
'rstp': False,
'stp': False
'stp': False,
},
}
assert expected_info == info
def test_get_ovs_info_with_ports_without_interfaces(nm_connection_mock,
nm_device_mock,
NM_mock):
def test_get_ovs_info_with_ports_without_interfaces(
nm_connection_mock, nm_device_mock, NM_mock
):
bridge_device = mock.MagicMock()
port_device = mock.MagicMock()
_mock_port_profile(nm_connection_mock)
@ -92,15 +92,15 @@ def test_get_ovs_info_with_ports_without_interfaces(nm_connection_mock,
'fail-mode': '',
'mcast-snooping-enable': False,
'rstp': False,
'stp': False
'stp': False,
},
}
assert expected_info == info
def test_get_ovs_info_with_ports_with_interfaces(nm_connection_mock,
nm_device_mock,
NM_mock):
def test_get_ovs_info_with_ports_with_interfaces(
nm_connection_mock, nm_device_mock, NM_mock
):
bridge_device = mock.MagicMock()
port_device = mock.MagicMock()
bridge_active_con = mock.MagicMock()
@ -108,8 +108,9 @@ def test_get_ovs_info_with_ports_with_interfaces(nm_connection_mock,
nm_device_mock.get_device_by_name.return_value = port_device
_mock_port_profile(nm_connection_mock)
nm_connection_mock.get_device_active_connection = (
lambda dev:
bridge_active_con if dev == bridge_device else port_active_con
lambda dev: bridge_active_con
if dev == bridge_device
else port_active_con
)
bridge_active_con.props.master = bridge_device
port_active_con.props.master = port_device
@ -135,7 +136,8 @@ def test_create_bridge_setting(NM_mock):
assert bridge_setting.props.fail_mode == options['fail-mode']
assert bridge_setting.props.mcast_snooping_enable == (
options['mcast-snooping-enable'])
options['mcast-snooping-enable']
)
assert bridge_setting.props.rstp_enable == options['rstp']
assert bridge_setting.props.stp_enable == options['stp']

179
tests/lib/nm/route_test.py
View File

@ -33,35 +33,38 @@ IPV4_ROUTE1 = {
Route.DESTINATION: '198.51.100.0/24',
Route.METRIC: 103,
Route.NEXT_HOP_ADDRESS: '192.0.2.1',
Route.TABLE_ID: 50
Route.TABLE_ID: 50,
}
IPV4_ROUTE2 = {
Route.DESTINATION: '203.0.113.0/24',
Route.METRIC: 103,
Route.NEXT_HOP_ADDRESS: '192.0.2.2',
Route.TABLE_ID: 51
Route.TABLE_ID: 51,
}
IPV6_ROUTE1 = {
Route.DESTINATION: '2001:db8:a::/64',
Route.METRIC: 103,
Route.NEXT_HOP_ADDRESS: '2001:db8:1::a',
Route.TABLE_ID: 50
Route.TABLE_ID: 50,
}
IPV6_ROUTE2 = {
Route.DESTINATION: '2001:db8:b::/64',
Route.METRIC: 103,
Route.NEXT_HOP_ADDRESS: '2001:db8:1::b',
Route.TABLE_ID: 51
Route.TABLE_ID: 51,
}
parametrize_ip_ver_routes = pytest.mark.parametrize(
'nm_ip, routes',
[(nm_ipv4, [IPV4_ROUTE1, IPV4_ROUTE2]),
(nm_ipv6, [IPV6_ROUTE1, IPV6_ROUTE2])],
ids=['ipv4', 'ipv6'])
[
(nm_ipv4, [IPV4_ROUTE1, IPV4_ROUTE2]),
(nm_ipv6, [IPV6_ROUTE1, IPV6_ROUTE2]),
],
ids=['ipv4', 'ipv6'],
)
def _get_test_ipv4_gateways():
@ -70,14 +73,14 @@ def _get_test_ipv4_gateways():
Route.DESTINATION: '0.0.0.0/0',
Route.METRIC: 103,
Route.NEXT_HOP_ADDRESS: '192.0.2.1',
Route.TABLE_ID: 52
Route.TABLE_ID: 52,
},
{
Route.DESTINATION: '0.0.0.0/0',
Route.METRIC: 101,
Route.NEXT_HOP_ADDRESS: '192.0.2.2',
Route.TABLE_ID: 53
}
Route.TABLE_ID: 53,
},
]
@ -87,55 +90,57 @@ def _get_test_ipv6_gateways():
Route.DESTINATION: '::/0',
Route.METRIC: 103,
Route.NEXT_HOP_ADDRESS: '2001:db8:1::f',
Route.TABLE_ID: 52
Route.TABLE_ID: 52,
},
{
Route.DESTINATION: '::/0',
Route.METRIC: 101,
Route.NEXT_HOP_ADDRESS: '2001:db8:1::e',
Route.TABLE_ID: 53
}
Route.TABLE_ID: 53,
},
]
parametrize_ip_ver_routes_gw = pytest.mark.parametrize(
'nm_ip, routes, gateways',
[(nm_ipv4, [IPV4_ROUTE1, IPV4_ROUTE2], _get_test_ipv4_gateways()),
(nm_ipv6, [IPV6_ROUTE1, IPV6_ROUTE2], _get_test_ipv6_gateways())],
ids=['ipv4', 'ipv6'])
[
(nm_ipv4, [IPV4_ROUTE1, IPV4_ROUTE2], _get_test_ipv4_gateways()),
(nm_ipv6, [IPV6_ROUTE1, IPV6_ROUTE2], _get_test_ipv6_gateways()),
],
ids=['ipv4', 'ipv6'],
)
@parametrize_ip_ver_routes
def test_add_multiple_route(nm_ip, routes):
setting_ip = nm_ip.create_setting({
'enabled': True,
metadata.ROUTES: routes
}, base_con_profile=None)
setting_ip = nm_ip.create_setting(
{'enabled': True, metadata.ROUTES: routes}, base_con_profile=None
)
assert [_nm_route_to_dict(r) for r in setting_ip.props.routes] == routes
@parametrize_ip_ver_routes
def test_add_duplicate_routes(nm_ip, routes):
setting_ip = nm_ip.create_setting({
'enabled': True,
metadata.ROUTES: [routes[0], routes[0]]
}, base_con_profile=None)
assert ([_nm_route_to_dict(r) for r in setting_ip.props.routes] ==
[routes[0]])
setting_ip = nm_ip.create_setting(
{'enabled': True, metadata.ROUTES: [routes[0], routes[0]]},
base_con_profile=None,
)
assert [_nm_route_to_dict(r) for r in setting_ip.props.routes] == [
routes[0]
]
@parametrize_ip_ver_routes
def test_clear_route(nm_ip, routes):
setting_ip = nm_ip.create_setting({
'enabled': True,
metadata.ROUTES: routes
}, base_con_profile=None)
setting_ip = nm_ip.create_setting(
{'enabled': True, metadata.ROUTES: routes}, base_con_profile=None
)
con_profile = nm_connection.ConnectionProfile()
con_profile.create([setting_ip])
new_setting_ip = nm_ip.create_setting({
'enabled': True,
metadata.ROUTES: []
}, base_con_profile=con_profile.profile)
new_setting_ip = nm_ip.create_setting(
{'enabled': True, metadata.ROUTES: []},
base_con_profile=con_profile.profile,
)
assert not [_nm_route_to_dict(r) for r in new_setting_ip.props.routes]
@ -145,12 +150,13 @@ def test_add_route_without_metric(nm_ip, routes):
route_with_default_metric[Route.METRIC] = Route.USE_DEFAULT_METRIC
route_without_metric = copy.deepcopy(routes[0])
del route_without_metric[Route.METRIC]
setting_ip = nm_ip.create_setting({
'enabled': True,
metadata.ROUTES: [route_without_metric]
}, base_con_profile=None)
assert ([_nm_route_to_dict(r) for r in setting_ip.props.routes] ==
[route_with_default_metric])
setting_ip = nm_ip.create_setting(
{'enabled': True, metadata.ROUTES: [route_without_metric]},
base_con_profile=None,
)
assert [_nm_route_to_dict(r) for r in setting_ip.props.routes] == [
route_with_default_metric
]
@parametrize_ip_ver_routes
@ -159,56 +165,62 @@ def test_add_route_without_table_id(nm_ip, routes):
route_with_default_table_id[Route.TABLE_ID] = Route.USE_DEFAULT_ROUTE_TABLE
route_without_table_id = copy.deepcopy(routes[0])
del route_without_table_id[Route.TABLE_ID]
setting_ip = nm_ip.create_setting({
'enabled': True,
metadata.ROUTES: [route_without_table_id]
}, base_con_profile=None)
assert ([_nm_route_to_dict(r) for r in setting_ip.props.routes] ==
[route_with_default_table_id])
setting_ip = nm_ip.create_setting(
{'enabled': True, metadata.ROUTES: [route_without_table_id]},
base_con_profile=None,
)
assert [_nm_route_to_dict(r) for r in setting_ip.props.routes] == [
route_with_default_table_id
]
@parametrize_ip_ver_routes_gw
def test_change_gateway(nm_ip, routes, gateways):
desired_routes = routes + gateways[:1]
setting_ip = nm_ip.create_setting({
'enabled': True,
metadata.ROUTES: desired_routes
}, base_con_profile=None)
setting_ip = nm_ip.create_setting(
{'enabled': True, metadata.ROUTES: desired_routes},
base_con_profile=None,
)
assert [_nm_route_to_dict(r) for r in setting_ip.props.routes] == routes
assert _get_gateway(setting_ip) == gateways[0]
@pytest.mark.xfail(raises=NmstateNotImplementedError,
strict=True,
reason='Network Manager Bug: '
'https://bugzilla.redhat.com/1707396')
@pytest.mark.xfail(
raises=NmstateNotImplementedError,
strict=True,
reason='Network Manager Bug: ' 'https://bugzilla.redhat.com/1707396',
)
@parametrize_ip_ver_routes_gw
def test_add_two_gateway(nm_ip, routes, gateways):
nm_ip.create_setting({
'enabled': True,
metadata.ROUTES: routes + gateways
}, base_con_profile=None)
nm_ip.create_setting(
{'enabled': True, metadata.ROUTES: routes + gateways},
base_con_profile=None,
)
@pytest.mark.xfail(raises=NmstateNotImplementedError,
strict=True,
reason='Network Manager Bug: '
'https://bugzilla.redhat.com/1707396')
@pytest.mark.xfail(
raises=NmstateNotImplementedError,
strict=True,
reason='Network Manager Bug: ' 'https://bugzilla.redhat.com/1707396',
)
@parametrize_ip_ver_routes_gw
def test_add_duplicate_gateways(nm_ip, routes, gateways):
nm_ip.create_setting({
'enabled': True,
metadata.ROUTES: routes + [gateways[0], gateways[0]]
}, base_con_profile=None)
nm_ip.create_setting(
{
'enabled': True,
metadata.ROUTES: routes + [gateways[0], gateways[0]],
},
base_con_profile=None,
)
@parametrize_ip_ver_routes_gw
def test_change_gateway_without_metric(nm_ip, routes, gateways):
del gateways[0][Route.METRIC]
setting_ip = nm_ip.create_setting({
'enabled': True,
metadata.ROUTES: routes + [gateways[0]]
}, base_con_profile=None)
setting_ip = nm_ip.create_setting(
{'enabled': True, metadata.ROUTES: routes + [gateways[0]]},
base_con_profile=None,
)
gateways[0][Route.METRIC] = Route.USE_DEFAULT_METRIC
assert [_nm_route_to_dict(r) for r in setting_ip.props.routes] == routes
assert _get_gateway(setting_ip) == gateways[0]
@ -217,10 +229,10 @@ def test_change_gateway_without_metric(nm_ip, routes, gateways):
@parametrize_ip_ver_routes_gw
def test_change_gateway_without_table_id(nm_ip, routes, gateways):
del gateways[0][Route.TABLE_ID]
setting_ip = nm_ip.create_setting({
'enabled': True,
metadata.ROUTES: routes + [gateways[0]]
}, base_con_profile=None)
setting_ip = nm_ip.create_setting(
{'enabled': True, metadata.ROUTES: routes + [gateways[0]]},
base_con_profile=None,
)
gateways[0][Route.TABLE_ID] = Route.USE_DEFAULT_ROUTE_TABLE
assert [_nm_route_to_dict(r) for r in setting_ip.props.routes] == routes
@ -229,16 +241,16 @@ def test_change_gateway_without_table_id(nm_ip, routes, gateways):
@parametrize_ip_ver_routes_gw
def test_clear_gateway(nm_ip, routes, gateways):
setting_ip = nm_ip.create_setting({
'enabled': True,
metadata.ROUTES: routes + gateways[:1]
}, base_con_profile=None)
setting_ip = nm_ip.create_setting(
{'enabled': True, metadata.ROUTES: routes + gateways[:1]},
base_con_profile=None,
)
con_profile = nm_connection.ConnectionProfile()
con_profile.create([setting_ip])
setting_ip = nm_ip.create_setting({
'enabled': True,
metadata.ROUTES: routes
}, base_con_profile=con_profile.profile)
setting_ip = nm_ip.create_setting(
{'enabled': True, metadata.ROUTES: routes},
base_con_profile=con_profile.profile,
)
assert [_nm_route_to_dict(r) for r in setting_ip.props.routes] == routes
assert not setting_ip.get_gateway()
assert setting_ip.get_route_table() == Route.USE_DEFAULT_ROUTE_TABLE
@ -247,7 +259,8 @@ def test_clear_gateway(nm_ip, routes, gateways):
def _nm_route_to_dict(nm_route):
dst = '{ip}/{prefix}'.format(
ip=nm_route.get_dest(), prefix=nm_route.get_prefix())
ip=nm_route.get_dest(), prefix=nm_route.get_prefix()
)
next_hop = nm_route.get_next_hop() or ''
metric = int(nm_route.get_metric())
table_id_variant = nm_route.get_attribute('table')

40
tests/lib/nm/translator_test.py
View File

@ -58,19 +58,15 @@ def test_api2nm_get_iface_type(NM_mock):
assert NM_mock.SETTING_WIRED_SETTING_NAME == nm_type
@mock.patch.object(nm.translator.Api2Nm, 'get_iface_type',
staticmethod(lambda t: t))
@mock.patch.object(
nm.translator.Api2Nm, 'get_iface_type', staticmethod(lambda t: t)
)
def test_api2nm_bond_options():
bond_options = {
'name': 'bond99',
'type': 'bond',
'state': 'up',
'link-aggregation': {
'mode': 'balance-rr',
'options': {
'miimon': 120
}
}
'link-aggregation': {'mode': 'balance-rr', 'options': {'miimon': 120}},
}
nm_bond_options = nm.translator.Api2Nm.get_bond_options(bond_options)
@ -86,11 +82,7 @@ def test_nm2api_common_device_info():
}
info = nm.translator.Nm2Api.get_common_device_info(devinfo)
expected_info = {
'name': 'devname',
'state': 'down',
'type': 'unknown',
}
expected_info = {'name': 'devname', 'state': 'down', 'type': 'unknown'}
assert expected_info == info
@ -98,23 +90,19 @@ def test_nm2api_bond_info():
slaves_mock = [mock.MagicMock(), mock.MagicMock()]
bondinfo = {
'slaves': slaves_mock,
'options': {
'mode': 'balance-rr',
'miimon': 120,
}
'options': {'mode': 'balance-rr', 'miimon': 120},
}
info = nm.translator.Nm2Api.get_bond_info(bondinfo)
expected_info = {
'link-aggregation':
{
'mode': 'balance-rr',
'slaves': [slaves_mock[0].props.interface,
slaves_mock[1].props.interface],
'options': {
'miimon': 120
}
}
'link-aggregation': {
'mode': 'balance-rr',
'slaves': [
slaves_mock[0].props.interface,
slaves_mock[1].props.interface,
],
'options': {'miimon': 120},
}
}
assert expected_info == info

19
tests/lib/nm/user_test.py
View File

@ -36,16 +36,21 @@ def test_create_no_setting(NM_mock):
def test_create_setting_duplicate(NM_mock):
base_profile = mock.MagicMock()
setting = nm.user.create_setting({'description': 'test_interface'},
base_profile)
setting = nm.user.create_setting(
{'description': 'test_interface'}, base_profile
)
base_profile.get_setting_by_name.assert_called_with(
NM_mock.SETTING_USER_SETTING_NAME)
assert setting == \
base_profile.get_setting_by_name.return_value.duplicate.return_value
NM_mock.SETTING_USER_SETTING_NAME
)
assert (
setting
== base_profile.get_setting_by_name.return_value.duplicate.return_value
)
def test_create_setting_description(NM_mock):
setting = nm.user.create_setting({'description': 'test_interface'}, None)
assert setting == NM_mock.SettingUser.new.return_value
setting.set_data.assert_called_with('nmstate.interface.description',
'test_interface')
setting.set_data.assert_called_with(
'nmstate.interface.description', 'test_interface'
)

105
tests/lib/nm/wired_test.py
View File

@ -39,15 +39,18 @@ def test_create_setting_duplicate(NM_mock):
setting = nm.wired.create_setting(
{schema.Ethernet.CONFIG_SUBTREE: {schema.Ethernet.SPEED: 1000}},
base_profile
base_profile,
)
assert (
setting
== base_profile.get_setting_wired.return_value.duplicate.return_value
)
assert setting == \
base_profile.get_setting_wired.return_value.duplicate.return_value
def test_create_setting_mac(NM_mock):
setting = nm.wired.create_setting(
{schema.Interface.MAC: '01:23:45:67:89:ab'}, None)
{schema.Interface.MAC: '01:23:45:67:89:ab'}, None
)
assert setting == NM_mock.SettingWired.new.return_value
assert setting.props.cloned_mac_address == '01:23:45:67:89:ab'
@ -58,17 +61,24 @@ def test_create_setting_mtu(NM_mock):
assert setting.props.mtu == 1500
@mock.patch.object(nm.wired, 'minimal_ethtool',
return_value={'speed': 1337, 'duplex': 'full',
'auto-negotiation': 'mocked'})
@mock.patch.object(
nm.wired,
'minimal_ethtool',
return_value={
'speed': 1337,
'duplex': 'full',
'auto-negotiation': 'mocked',
},
)
def test_create_setting_auto_negotiation_False(ethtool_mock, NM_mock):
setting = nm.wired.create_setting(
{
schema.Interface.NAME: 'nmstate_test',
schema.Ethernet.CONFIG_SUBTREE:
{schema.Ethernet.AUTO_NEGOTIATION: False}
schema.Ethernet.CONFIG_SUBTREE: {
schema.Ethernet.AUTO_NEGOTIATION: False
},
},
None
None,
)
assert setting == NM_mock.SettingWired.new.return_value
assert setting.props.auto_negotiate is False
@ -80,10 +90,11 @@ def test_create_setting_auto_negotiation_False(ethtool_mock, NM_mock):
def test_create_setting_only_auto_negotiation_True(NM_mock):
setting = nm.wired.create_setting(
{
schema.Ethernet.CONFIG_SUBTREE:
{schema.Ethernet.AUTO_NEGOTIATION: True}
schema.Ethernet.CONFIG_SUBTREE: {
schema.Ethernet.AUTO_NEGOTIATION: True
}
},
None
None,
)
assert setting == NM_mock.SettingWired.new.return_value
assert setting.props.auto_negotiate is True
@ -97,10 +108,10 @@ def test_create_setting_auto_negotiation_speed_duplex(NM_mock):
schema.Ethernet.CONFIG_SUBTREE: {
schema.Ethernet.AUTO_NEGOTIATION: True,
schema.Ethernet.SPEED: 1000,
schema.Ethernet.DUPLEX: schema.Ethernet.FULL_DUPLEX
schema.Ethernet.DUPLEX: schema.Ethernet.FULL_DUPLEX,
}
},
None
None,
)
assert setting == NM_mock.SettingWired.new.return_value
assert setting.props.auto_negotiate is True
@ -113,19 +124,25 @@ def test_create_setting_speed_duplex(NM_mock):
{
schema.Ethernet.CONFIG_SUBTREE: {
schema.Ethernet.SPEED: 1000,
schema.Ethernet.DUPLEX: schema.Ethernet.FULL_DUPLEX
schema.Ethernet.DUPLEX: schema.Ethernet.FULL_DUPLEX,
}
},
None
None,
)
assert setting == NM_mock.SettingWired.new.return_value
assert setting.props.speed == 1000
assert setting.props.duplex == schema.Ethernet.FULL_DUPLEX
@mock.patch.object(nm.wired, 'minimal_ethtool',
return_value={'speed': 1500, 'duplex': 'unknown',
'auto-negotiation': True})
@mock.patch.object(
nm.wired,
'minimal_ethtool',
return_value={
'speed': 1500,
'duplex': 'unknown',
'auto-negotiation': True,
},
)
def test_get_info_with_invalid_duplex(ethtool_mock, NM_mock):
dev_mock = mock.MagicMock()
dev_mock.get_iface.return_value = 'nmstate_test'
@ -137,12 +154,11 @@ def test_get_info_with_invalid_duplex(ethtool_mock, NM_mock):
assert info == {
schema.Interface.MAC: dev_mock.get_hw_address.return_value,
schema.Interface.MTU: dev_mock.get_mtu.return_value
schema.Interface.MTU: dev_mock.get_mtu.return_value,
}
class TestWiredSetting(object):
def test_identity(self):
state = {}
obj1 = obj2 = nm.wired.WiredSetting(state)
@ -163,19 +179,13 @@ class TestWiredSetting(object):
assert not obj
def test_relevant_keys_with_false_values_is_false(self):
state = {
schema.Interface.MTU: 0,
schema.Interface.MAC: '',
}
state = {schema.Interface.MTU: 0, schema.Interface.MAC: ''}
obj = nm.wired.WiredSetting(state)
assert not obj
def test_partial_relevant_keys_is_true(self):
state = {
schema.Interface.MTU: 1500,
schema.Interface.MAC: 'abc',
}
state = {schema.Interface.MTU: 1500, schema.Interface.MAC: 'abc'}
obj = nm.wired.WiredSetting(state)
assert obj
@ -189,10 +199,7 @@ class TestWiredSetting(object):
assert not (obj1 != obj2)
def test_equality_for_partial_states(self):
state = {
schema.Interface.MTU: 1500,
schema.Interface.MAC: 'abc',
}
state = {schema.Interface.MTU: 1500, schema.Interface.MAC: 'abc'}
obj1 = nm.wired.WiredSetting(state)
obj2 = nm.wired.WiredSetting(state)
@ -200,14 +207,8 @@ class TestWiredSetting(object):
assert not (obj1 != obj2)
def test_inequality_for_partial_states(self):
state1 = {
schema.Interface.MTU: 1500,
schema.Interface.MAC: 'abc',
}
state2 = {
schema.Interface.MTU: 1000,
schema.Interface.MAC: 'abc',
}
state1 = {schema.Interface.MTU: 1500, schema.Interface.MAC: 'abc'}
state2 = {schema.Interface.MTU: 1000, schema.Interface.MAC: 'abc'}
obj1 = nm.wired.WiredSetting(state1)
obj2 = nm.wired.WiredSetting(state2)
@ -215,10 +216,7 @@ class TestWiredSetting(object):
assert not (obj1 == obj2)
def test_inequality_for_partial_states_with_missing_properties(self):
state1 = {
schema.Interface.MTU: 1500,
schema.Interface.MAC: 'abc',
}
state1 = {schema.Interface.MTU: 1500, schema.Interface.MAC: 'abc'}
state2 = {schema.Interface.MAC: 'abc'}
obj1 = nm.wired.WiredSetting(state1)
@ -228,24 +226,15 @@ class TestWiredSetting(object):
assert not (obj1 == obj2)
def test_hash_unique(self):
state = {
schema.Interface.MTU: 1500,
schema.Interface.MAC: 'abc',
}
state = {schema.Interface.MTU: 1500, schema.Interface.MAC: 'abc'}
obj1 = nm.wired.WiredSetting(state)
obj2 = nm.wired.WiredSetting(state)
assert hash(obj1) == hash(obj2)
def test_behaviour_with_set(self):
state1 = {
schema.Interface.MTU: 1500,
schema.Interface.MAC: 'abc',
}
state2 = {
schema.Interface.MTU: 1500,
schema.Interface.MAC: 'abc',
}
state1 = {schema.Interface.MTU: 1500, schema.Interface.MAC: 'abc'}
state2 = {schema.Interface.MTU: 1500, schema.Interface.MAC: 'abc'}
state3 = {schema.Interface.MAC: 'abc'}
obj1 = nm.wired.WiredSetting(state1)

56
tests/lib/schema_validation_test.py
View File

@ -62,7 +62,7 @@ COMMON_DATA = {
'out-multicast-pkts': 0,
'out-discards': 0,
'out-errors': 0,
}
},
}
],
ROUTES: {
@ -72,7 +72,7 @@ COMMON_DATA = {
'metric': 100,
'destination': '0.0.0.0/0',
'next-hop-interface': 'eth0',
'next-hop-address': '192.0.2.1'
'next-hop-address': '192.0.2.1',
}
],
'running': [
@ -81,32 +81,20 @@ COMMON_DATA = {
'metric': 100,
'destination': '::/0',
'next-hop-interface': 'eth0',
'next-hop-address': 'fe80::1'
'next-hop-address': 'fe80::1',
}
]
],
},
DNS.KEY: {
DNS.RUNNING: {
DNS.SERVER: [
"2001:db8::1",
"192.0.2.1"
],
DNS.SEARCH: [
"example.com",
"example.org"
]
DNS.SERVER: ["2001:db8::1", "192.0.2.1"],
DNS.SEARCH: ["example.com", "example.org"],
},
DNS.CONFIG: {
DNS.SERVER: [
"2001:db8::1",
"192.0.2.1"
],
DNS.SEARCH: [
"example.com",
"example.org"
]
}
}
DNS.SERVER: ["2001:db8::1", "192.0.2.1"],
DNS.SEARCH: ["example.com", "example.org"],
},
},
}
@ -116,7 +104,6 @@ def default_data():
class TestIfaceCommon(object):
def test_valid_instance(self, default_data):
libnmstate.validator.validate(default_data)
@ -136,20 +123,16 @@ class TestIfaceCommon(object):
class TestIfaceTypeEthernet(object):
def test_valid_ethernet_with_auto_neg(self, default_data):
default_data[INTERFACES][0].update({
'type': 'ethernet',
'auto-negotiation': True,
})
default_data[INTERFACES][0].update(
{'type': 'ethernet', 'auto-negotiation': True}
)
libnmstate.validator.validate(default_data)
def test_valid_ethernet_without_auto_neg(self, default_data):
default_data[INTERFACES][0].update({
'auto-negotiation': False,
'link-speed': 1000,
'duplex': 'full',
})
default_data[INTERFACES][0].update(
{'auto-negotiation': False, 'link-speed': 1000, 'duplex': 'full'}
)
libnmstate.validator.validate(default_data)
def test_valid_without_auto_neg_and_missing_speed(self, default_data):
@ -158,10 +141,9 @@ class TestIfaceTypeEthernet(object):
not a valid configuration, however, this is not handled by the schema
at the moment, deferring the handling to the application code.
"""
default_data[INTERFACES][0].update({
'type': 'ethernet',
'auto-negotiation': False,
})
default_data[INTERFACES][0].update(
{'type': 'ethernet', 'auto-negotiation': False}
)
del default_data[INTERFACES][0]['link-speed']
libnmstate.validator.validate(default_data)

439
tests/lib/state_test.py
View File

@ -27,18 +27,18 @@ from libnmstate.schema import Route
parametrize_route_property = pytest.mark.parametrize(
'route_property', [
'route_property',
[
Route.TABLE_ID,
Route.DESTINATION,
Route.NEXT_HOP_INTERFACE,
Route.NEXT_HOP_ADDRESS,
Route.METRIC
]
Route.METRIC,
],
)
class TestAssertIfaceState(object):
def test_desired_is_identical_to_current(self):
desired_state = self._base_state
current_state = self._base_state
@ -75,84 +75,62 @@ class TestAssertIfaceState(object):
current_state = self._base_state
desired_state.interfaces['foo-name']['ipv4'] = {
'address': [
{
'ip': '192.168.122.10',
'prefix-length': 24
},
{
'ip': '192.168.121.10',
'prefix-length': 24
},
{'ip': '192.168.122.10', 'prefix-length': 24},
{'ip': '192.168.121.10', 'prefix-length': 24},
],
'enabled': True
'enabled': True,
}
current_state.interfaces['foo-name']['ipv4'] = {
'address': [
{
'ip': '192.168.121.10',
'prefix-length': 24
},
{
'ip': '192.168.122.10',
'prefix-length': 24
},
{'ip': '192.168.121.10', 'prefix-length': 24},
{'ip': '192.168.122.10', 'prefix-length': 24},
],
'enabled': True
'enabled': True,
}
desired_state.interfaces['foo-name']['ipv6'] = {
'address': [
{
'ip': '2001::2',
'prefix-length': 64
},
{
'ip': '2001::1',
'prefix-length': 64
}
{'ip': '2001::2', 'prefix-length': 64},
{'ip': '2001::1', 'prefix-length': 64},
],
'enabled': True
'enabled': True,
}
current_state.interfaces['foo-name']['ipv6'] = {
'address': [
{
'ip': '2001::1',
'prefix-length': 64
},
{
'ip': '2001::2',
'prefix-length': 64
}
{'ip': '2001::1', 'prefix-length': 64},
{'ip': '2001::2', 'prefix-length': 64},
],
'enabled': True
'enabled': True,
}
desired_state.verify_interfaces(current_state)
@property
def _base_state(self):
return state.State({
Interface.KEY: [
{
'name': 'foo-name',
'type': 'foo-type',
'state': 'up',
'bridge': {
'port': [
{'name': 'eth0', 'type': 'system'}
]
return state.State(
{
Interface.KEY: [
{
'name': 'foo-name',
'type': 'foo-type',
'state': 'up',
'bridge': {
'port': [{'name': 'eth0', 'type': 'system'}]
},
}
}
]
})
]
}
)
@property
def _extra_state(self):
return state.State({
Interface.KEY: [
{'name': 'eth0', 'state': 'up', 'type': 'unknown'},
{'name': 'eth1', 'state': 'up', 'type': 'unknown'}
]
})
return state.State(
{
Interface.KEY: [
{'name': 'eth0', 'state': 'up', 'type': 'unknown'},
{'name': 'eth1', 'state': 'up', 'type': 'unknown'},
]
}
)
class TestRouteEntry(object):
@ -163,50 +141,66 @@ class TestRouteEntry(object):
def test_obj_unique(self):
route0 = _create_route('198.51.100.0/24', '192.0.2.1', 'eth1', 50, 103)
route1 = _create_route(
'2001:db8:a::/64', '2001:db8:1::a', 'eth2', 51, 104)
'2001:db8:a::/64', '2001:db8:1::a', 'eth2', 51, 104
)
route0_clone = _create_route(
'198.51.100.0/24', '192.0.2.1', 'eth1', 50, 103)
'198.51.100.0/24', '192.0.2.1', 'eth1', 50, 103
)
assert route0 == route0_clone
assert route0 != route1
def test_obj_unique_without_table_id(self):
route_with_default_table_id = _create_route(
'198.51.100.0/24', '192.0.2.1', 'eth1',
Route.USE_DEFAULT_ROUTE_TABLE, 103)
'198.51.100.0/24',
'192.0.2.1',
'eth1',
Route.USE_DEFAULT_ROUTE_TABLE,
103,
)
route_without_table_id = _create_route(
'198.51.100.0/24', '192.0.2.1', 'eth1', None, 103)
'198.51.100.0/24', '192.0.2.1', 'eth1', None, 103
)
assert route_without_table_id == route_with_default_table_id
def test_obj_unique_without_metric(self):
route_with_default_metric = _create_route(
'198.51.100.0/24', '192.0.2.1', 'eth1', 50,
Route.USE_DEFAULT_METRIC)
'198.51.100.0/24',
'192.0.2.1',
'eth1',
50,
Route.USE_DEFAULT_METRIC,
)
route_without_metric = _create_route(
'198.51.100.0/24', '192.0.2.1', 'eth1', 50, None)
'198.51.100.0/24', '192.0.2.1', 'eth1', 50, None
)
assert route_without_metric == route_with_default_metric
def test_obj_unique_without_next_hop(self):
route_with_default_next_hop = _create_route(
'198.51.100.0/24', '', 'eth1', 50, 103)
'198.51.100.0/24', '', 'eth1', 50, 103
)
route_without_next_hop = _create_route(
'198.51.100.0/24', None, 'eth1', 50, 103)
'198.51.100.0/24', None, 'eth1', 50, 103
)
assert route_without_next_hop == route_with_default_next_hop
def test_normal_route_object_as_dict(self):
route = _create_route_dict(
'198.51.100.0/24', '192.0.2.1', 'eth1', 50, 103)
'198.51.100.0/24', '192.0.2.1', 'eth1', 50, 103
)
route_obj = state.RouteEntry(route)
assert route_obj.to_dict() == route
def test_absent_route_object_as_dict(self):
route = _create_route_dict(
'198.51.100.0/24', '192.0.2.1', 'eth1', 50, 103)
'198.51.100.0/24', '192.0.2.1', 'eth1', 50, 103
)
route[Route.STATE] = Route.STATE_ABSENT
route_obj = state.RouteEntry(route)
assert route_obj.absent
@ -215,7 +209,8 @@ class TestRouteEntry(object):
@parametrize_route_property
def test_absent_route_with_missing_props_as_dict(self, route_property):
absent_route = _create_route_dict(
'198.51.100.0/24', '192.0.2.1', 'eth1', 50, 103)
'198.51.100.0/24', '192.0.2.1', 'eth1', 50, 103
)
absent_route[Route.STATE] = Route.STATE_ABSENT
del absent_route[route_property]
route_obj = state.RouteEntry(absent_route)
@ -225,12 +220,14 @@ class TestRouteEntry(object):
route0 = _create_route('198.51.100.0/24', '192.0.2.1', 'eth1', 50, 103)
absent_r0 = _create_route_dict(
'198.51.100.0/24', '192.0.2.1', 'eth1', 50, 103)
'198.51.100.0/24', '192.0.2.1', 'eth1', 50, 103
)
absent_r0[Route.STATE] = Route.STATE_ABSENT
absent_route0 = state.RouteEntry(absent_r0)
route1 = _create_route(
'2001:db8:a::/64', '2001:db8:1::a', 'eth2', 51, 104)
'2001:db8:a::/64', '2001:db8:1::a', 'eth2', 51, 104
)
assert absent_route0.match(route0)
assert absent_route0 == route0
@ -240,12 +237,15 @@ class TestRouteEntry(object):
@parametrize_route_property
def test_absent_route_wildcard_match(self, route_property):
original_route0 = _create_route(
'198.51.100.0/24', '192.0.2.1', 'eth1', 50, 103)
'198.51.100.0/24', '192.0.2.1', 'eth1', 50, 103
)
original_route1 = _create_route(
'2001:db8:a::/64', '2001:db8:1::a', 'eth2', 51, 104)
'2001:db8:a::/64', '2001:db8:1::a', 'eth2', 51, 104
)
absent_route0_state = _create_route_dict(
'198.51.100.0/24', '192.0.2.1', 'eth1', 50, 103)
'198.51.100.0/24', '192.0.2.1', 'eth1', 50, 103
)
absent_route0_state[Route.STATE] = Route.STATE_ABSENT
del absent_route0_state[route_property]
new_route0 = state.RouteEntry(absent_route0_state)
@ -255,7 +255,8 @@ class TestRouteEntry(object):
def test_absent_route_is_ignored_for_matching_and_equality(self):
route = _create_route_dict(
'198.51.100.0/24', '192.0.2.1', 'eth1', 50, 103)
'198.51.100.0/24', '192.0.2.1', 'eth1', 50, 103
)
route[Route.STATE] = Route.STATE_ABSENT
obj1 = state.RouteEntry(route)
obj2 = state.RouteEntry(route)
@ -277,8 +278,9 @@ class TestRouteEntry(object):
@parametrize_route_property
def test_sort_routes_with_absent_route(self, route_property):
absent_route = _create_route('198.51.100.0/24', '192.0.1.1', 'eth0',
9, 103).to_dict()
absent_route = _create_route(
'198.51.100.0/24', '192.0.1.1', 'eth0', 9, 103
).to_dict()
absent_route[Route.STATE] = Route.STATE_ABSENT
del absent_route[route_property]
absent_route = state.RouteEntry(absent_route)
@ -298,7 +300,6 @@ class TestRouteEntry(object):
class TestRouteStateMerge(object):
def test_merge_empty_states(self):
s0 = state.State({})
s1 = state.State({})
@ -342,52 +343,63 @@ class TestRouteStateMerge(object):
empty_state.merge_routes(state_with_route0)
assert ({'interfaces': [], 'routes': {'config': []}} ==
empty_state.state)
assert {
'interfaces': [],
'routes': {'config': []},
} == empty_state.state
assert {} == empty_state.config_iface_routes
def test_merge_iface_only_with_same_iface_routes_state(self):
route0_obj = self._create_route0()
route0 = route0_obj.to_dict()
iface_only_state = state.State({
Interface.KEY: [{Interface.NAME: route0_obj.next_hop_interface}]
})
iface_only_state = state.State(
{Interface.KEY: [{Interface.NAME: route0_obj.next_hop_interface}]}
)
state_with_route0 = state.State({Route.KEY: {Route.CONFIG: [route0]}})
iface_only_state.merge_routes(state_with_route0)
expected = {
Interface.KEY: [{
Interface.NAME: route0_obj.next_hop_interface,
Interface.IPV4: {},
Interface.IPV6: {},
}],
Interface.KEY: [
{
Interface.NAME: route0_obj.next_hop_interface,
Interface.IPV4: {},
Interface.IPV6: {},
}
],
Route.KEY: {Route.CONFIG: [route0]},
}
assert expected == iface_only_state.state
assert ({route0_obj.next_hop_interface: [route0_obj]} ==
iface_only_state.config_iface_routes)
assert {
route0_obj.next_hop_interface: [route0_obj]
} == iface_only_state.config_iface_routes
def test_merge_iface_down_with_same_iface_routes_state(self):
route0_obj = self._create_route0()
route0 = route0_obj.to_dict()
iface_down_state = state.State({
Interface.KEY: [{
Interface.NAME: route0_obj.next_hop_interface,
Interface.STATE: InterfaceState.DOWN,
}]
})
iface_down_state = state.State(
{
Interface.KEY: [
{
Interface.NAME: route0_obj.next_hop_interface,
Interface.STATE: InterfaceState.DOWN,
}
]
}
)
state_with_route0 = state.State({Route.KEY: {Route.CONFIG: [route0]}})
iface_down_state.merge_routes(state_with_route0)
expected = {
Interface.KEY: [{
Interface.NAME: route0_obj.next_hop_interface,
Interface.STATE: InterfaceState.DOWN,
Interface.IPV4: {},
Interface.IPV6: {},
}],
Interface.KEY: [
{
Interface.NAME: route0_obj.next_hop_interface,
Interface.STATE: InterfaceState.DOWN,
Interface.IPV4: {},
Interface.IPV6: {},
}
],
Route.KEY: {Route.CONFIG: []},
}
assert expected == iface_down_state.state
@ -396,24 +408,30 @@ class TestRouteStateMerge(object):
def test_merge_iface_ipv4_disabled_with_same_iface_routes_state(self):
route0_obj = self._create_route0()
route0 = route0_obj.to_dict()
iface_down_state = state.State({
Interface.KEY: [{
Interface.NAME: route0_obj.next_hop_interface,
Interface.STATE: InterfaceState.UP,
Interface.IPV4: {'enabled': False},
}]
})
iface_down_state = state.State(
{
Interface.KEY: [
{
Interface.NAME: route0_obj.next_hop_interface,
Interface.STATE: InterfaceState.UP,
Interface.IPV4: {'enabled': False},
}
]
}
)
state_with_route0 = state.State({Route.KEY: {Route.CONFIG: [route0]}})
iface_down_state.merge_routes(state_with_route0)
expected = {
Interface.KEY: [{
Interface.NAME: route0_obj.next_hop_interface,
Interface.STATE: InterfaceState.UP,
Interface.IPV4: {'enabled': False},
Interface.IPV6: {},
}],
Interface.KEY: [
{
Interface.NAME: route0_obj.next_hop_interface,
Interface.STATE: InterfaceState.UP,
Interface.IPV4: {'enabled': False},
Interface.IPV6: {},
}
],
Route.KEY: {Route.CONFIG: []},
}
assert expected == iface_down_state.state
@ -422,24 +440,30 @@ class TestRouteStateMerge(object):
def test_merge_iface_ipv6_disabled_with_same_iface_routes_state(self):
route1_obj = self._create_route1()
route1 = route1_obj.to_dict()
iface_down_state = state.State({
Interface.KEY: [{
Interface.NAME: route1_obj.next_hop_interface,
Interface.STATE: InterfaceState.UP,
Interface.IPV6: {'enabled': False},
}]
})
iface_down_state = state.State(
{
Interface.KEY: [
{
Interface.NAME: route1_obj.next_hop_interface,
Interface.STATE: InterfaceState.UP,
Interface.IPV6: {'enabled': False},
}
]
}
)
state_with_route1 = state.State({Route.KEY: {Route.CONFIG: [route1]}})
iface_down_state.merge_routes(state_with_route1)
expected = {
Interface.KEY: [{
Interface.NAME: route1_obj.next_hop_interface,
Interface.STATE: InterfaceState.UP,
Interface.IPV4: {},
Interface.IPV6: {'enabled': False},
}],
Interface.KEY: [
{
Interface.NAME: route1_obj.next_hop_interface,
Interface.STATE: InterfaceState.UP,
Interface.IPV4: {},
Interface.IPV6: {'enabled': False},
}
],
Route.KEY: {Route.CONFIG: []},
}
assert expected == iface_down_state.state
@ -448,29 +472,36 @@ class TestRouteStateMerge(object):
def test_merge_iface_ipv6_disabled_with_same_iface_ipv4_routes_state(self):
route0_obj = self._create_route0()
route0 = route0_obj.to_dict()
iface_down_state = state.State({
Interface.KEY: [{
Interface.NAME: route0_obj.next_hop_interface,
Interface.STATE: InterfaceState.UP,
Interface.IPV6: {'enabled': False},
}]
})
iface_down_state = state.State(
{
Interface.KEY: [
{
Interface.NAME: route0_obj.next_hop_interface,
Interface.STATE: InterfaceState.UP,
Interface.IPV6: {'enabled': False},
}
]
}
)
state_with_route0 = state.State({Route.KEY: {Route.CONFIG: [route0]}})
iface_down_state.merge_routes(state_with_route0)
expected = {
Interface.KEY: [{
Interface.NAME: route0_obj.next_hop_interface,
Interface.STATE: InterfaceState.UP,
Interface.IPV4: {},
Interface.IPV6: {'enabled': False},
}],
Interface.KEY: [
{
Interface.NAME: route0_obj.next_hop_interface,
Interface.STATE: InterfaceState.UP,
Interface.IPV4: {},
Interface.IPV6: {'enabled': False},
}
],
Route.KEY: {Route.CONFIG: [route0]},
}
assert expected == iface_down_state.state
assert ({route0_obj.next_hop_interface: [route0_obj]} ==
iface_down_state.config_iface_routes)
assert {
route0_obj.next_hop_interface: [route0_obj]
} == iface_down_state.config_iface_routes
def test_merge_non_empty_with_empty_state(self):
route0_obj = self._create_route0()
@ -480,8 +511,10 @@ class TestRouteStateMerge(object):
state_with_route0.merge_routes(empty_state)
assert ({'interfaces': [], 'routes': {'config': [route0]}} ==
state_with_route0.state)
assert {
'interfaces': [],
'routes': {'config': [route0]},
} == state_with_route0.state
assert {'eth1': [route0_obj]} == state_with_route0.config_iface_routes
def test_merge_absent_routes_with_no_matching(self):
@ -495,8 +528,7 @@ class TestRouteStateMerge(object):
s0.merge_routes(s1)
expected_state = {
'interfaces': [], 'routes': {'config': []}}
expected_state = {'interfaces': [], 'routes': {'config': []}}
assert expected_state == s0.state
assert {} == s0.config_iface_routes
@ -519,12 +551,14 @@ class TestRouteStateMerge(object):
def _create_route1(self):
return _create_route(
'2001:db8:a::/64', '2001:db8:1::a', 'eth2', 51, 104)
'2001:db8:a::/64', '2001:db8:1::a', 'eth2', 51, 104
)
def _create_route(dest, via_addr, via_iface, table, metric):
return state.RouteEntry(
_create_route_dict(dest, via_addr, via_iface, table, metric))
_create_route_dict(dest, via_addr, via_iface, table, metric)
)
def _create_route_dict(dest, via_addr, via_iface, table, metric):
@ -533,36 +567,25 @@ def _create_route_dict(dest, via_addr, via_iface, table, metric):
Route.METRIC: metric,
Route.NEXT_HOP_ADDRESS: via_addr,
Route.NEXT_HOP_INTERFACE: via_iface,
Route.TABLE_ID: table
Route.TABLE_ID: table,
}
def test_state_empty_routes():
route_state = state.State(
{
Route.KEY: {
Route.CONFIG: []
}
}
)
route_state = state.State({Route.KEY: {Route.CONFIG: []}})
assert {} == route_state.config_iface_routes
def test_state_iface_routes_with_distinct_ifaces():
routes = _get_mixed_test_routes()
route_state = state.State(
{
Route.KEY: {
Route.CONFIG: routes
}
}
)
route_state = state.State({Route.KEY: {Route.CONFIG: routes}})
expected_indexed_route_state = defaultdict(list)
for route in routes:
iface_name = route[Route.NEXT_HOP_INTERFACE]
expected_indexed_route_state[iface_name].append(
state.RouteEntry(route))
state.RouteEntry(route)
)
# No need to sort the routes as there is only 1 route per interface.
assert expected_indexed_route_state == route_state.config_iface_routes
@ -572,13 +595,7 @@ def test_state_iface_routes_with_same_iface():
routes = _get_mixed_test_routes()
for route in routes:
route[Route.NEXT_HOP_INTERFACE] = 'eth1'
route_state = state.State(
{
Route.KEY: {
Route.CONFIG: routes
}
}
)
route_state = state.State({Route.KEY: {Route.CONFIG: routes}})
expected_indexed_route_state = {
'eth1': sorted([state.RouteEntry(r) for r in routes])
}
@ -593,51 +610,29 @@ def test_state_iface_routes_order():
route[Route.NEXT_HOP_INTERFACE] = 'eth1'
route_state = state.State(
{
Route.KEY: {
Route.CONFIG: [routes[0], routes[1]],
}
}
{Route.KEY: {Route.CONFIG: [routes[0], routes[1]]}}
)
reverse_route_state = state.State(
{
Route.KEY: {
Route.CONFIG: [routes[1], routes[0]],
}
}
{Route.KEY: {Route.CONFIG: [routes[1], routes[0]]}}
)
assert (route_state.config_iface_routes ==
reverse_route_state.config_iface_routes)
assert (
route_state.config_iface_routes
== reverse_route_state.config_iface_routes
)
def test_state_verify_route_same():
routes = _get_mixed_test_routes()
route_state = state.State({
Route.KEY: {
Route.CONFIG: routes
}
})
route_state_2 = state.State({
Route.KEY: {
Route.CONFIG: routes
}
})
route_state = state.State({Route.KEY: {Route.CONFIG: routes}})
route_state_2 = state.State({Route.KEY: {Route.CONFIG: routes}})
route_state.verify_routes(route_state_2)
def test_state_verify_route_diff_route_count():
routes = _get_mixed_test_routes()
route_state = state.State({
Route.KEY: {
Route.CONFIG: routes
}
})
route_state_2 = state.State({
Route.KEY: {
Route.CONFIG: routes[:1]
}
})
route_state = state.State({Route.KEY: {Route.CONFIG: routes}})
route_state_2 = state.State({Route.KEY: {Route.CONFIG: routes[:1]}})
with pytest.raises(NmstateVerificationError):
route_state.verify_routes(route_state_2)
@ -645,17 +640,9 @@ def test_state_verify_route_diff_route_count():
def test_state_verify_route_diff_route_prop():
routes = _get_mixed_test_routes()
route_state = state.State({
Route.KEY: {
Route.CONFIG: routes
}
})
route_state = state.State({Route.KEY: {Route.CONFIG: routes}})
routes[0][Route.NEXT_HOP_INTERFACE] = 'another_nic'
route_state_2 = state.State({
Route.KEY: {
Route.CONFIG: routes
}
})
route_state_2 = state.State({Route.KEY: {Route.CONFIG: routes}})
with pytest.raises(NmstateVerificationError):
route_state.verify_routes(route_state_2)
@ -679,21 +666,19 @@ def _gen_iface_states_for_routes(routes):
{
Interface.NAME: iface,
Interface.STATE: InterfaceState.UP,
Interface.IPV4: {
'enabled': True
},
Interface.IPV6: {
'enabled': True
}
Interface.IPV4: {'enabled': True},
Interface.IPV6: {'enabled': True},
}
for iface in ifaces
]
def _route_sort_key(route):
return (route.get(Route.TABLE_ID, Route.USE_DEFAULT_ROUTE_TABLE),
route.get(Route.NEXT_HOP_INTERFACE, ''),
route.get(Route.DESTINATION, ''))
return (
route.get(Route.TABLE_ID, Route.USE_DEFAULT_ROUTE_TABLE),
route.get(Route.NEXT_HOP_INTERFACE, ''),
route.get(Route.DESTINATION, ''),
)
class TestAssertDnsState(object):
@ -740,6 +725,6 @@ class TestAssertDnsState(object):
return {
DNS.CONFIG: {
DNS.SERVER: ['192.168.122.1', '2001:db8:a::1'],
DNS.SEARCH: ['example.com', 'example.org']
DNS.SEARCH: ['example.com', 'example.org'],
}
}

329
tests/lib/validator_test.py
View File

@ -28,131 +28,124 @@ from libnmstate.error import NmstateValueError
class TestLinkAggregationState(object):
def test_bonds_with_no_slaves(self):
desired_state = state.State({
schema.Interface.KEY: [
{
'name': 'bond0',
'link-aggregation': {
'slaves': []
},
},
{
'name': 'bond1',
'link-aggregation': {
'slaves': []
},
}
]
})
desired_state = state.State(
{
schema.Interface.KEY: [
{'name': 'bond0', 'link-aggregation': {'slaves': []}},
{'name': 'bond1', 'link-aggregation': {'slaves': []}},
]
}
)
libnmstate.validator.validate_link_aggregation_state(desired_state,
empty_state())
libnmstate.validator.validate_link_aggregation_state(
desired_state, empty_state()
)
def test_bonds_with_single_slave(self):
desired_state = state.State({
schema.Interface.KEY: [
{'name': 'slave0'},
{'name': 'slave1'},
{
'name': 'bond0',
'link-aggregation': {
'slaves': ['slave0']
desired_state = state.State(
{
schema.Interface.KEY: [
{'name': 'slave0'},
{'name': 'slave1'},
{
'name': 'bond0',
'link-aggregation': {'slaves': ['slave0']},
},
},
{
'name': 'bond1',
'link-aggregation': {
'slaves': ['slave1']
{
'name': 'bond1',
'link-aggregation': {'slaves': ['slave1']},
},
}
]
})
libnmstate.validator.validate_link_aggregation_state(desired_state,
empty_state())
]
}
)
libnmstate.validator.validate_link_aggregation_state(
desired_state, empty_state()
)
def test_bonds_with_multiple_slaves(self):
desired_state = state.State({
schema.Interface.KEY: [
{'name': 'slave0'},
{'name': 'slave1'},
{'name': 'slave00'},
{'name': 'slave11'},
{
'name': 'bond0',
'link-aggregation': {
'slaves': ['slave0', 'slave00']
desired_state = state.State(
{
schema.Interface.KEY: [
{'name': 'slave0'},
{'name': 'slave1'},
{'name': 'slave00'},
{'name': 'slave11'},
{
'name': 'bond0',
'link-aggregation': {'slaves': ['slave0', 'slave00']},
},
},
{
'name': 'bond1',
'link-aggregation': {
'slaves': ['slave1', 'slave11']
{
'name': 'bond1',
'link-aggregation': {'slaves': ['slave1', 'slave11']},
},
}
]
})
libnmstate.validator.validate_link_aggregation_state(desired_state,
empty_state())
]
}
)
libnmstate.validator.validate_link_aggregation_state(
desired_state, empty_state()
)
def test_bonds_with_multiple_slaves_reused(self):
desired_state = state.State({
schema.Interface.KEY: [
{'name': 'slave0'},
{'name': 'slave1'},
{'name': 'slave00'},
{
'name': 'bond0',
'link-aggregation': {
'slaves': ['slave0', 'slave00']
desired_state = state.State(
{
schema.Interface.KEY: [
{'name': 'slave0'},
{'name': 'slave1'},
{'name': 'slave00'},
{
'name': 'bond0',
'link-aggregation': {'slaves': ['slave0', 'slave00']},
},
},
{
'name': 'bond1',
'link-aggregation': {
'slaves': ['slave1', 'slave00']
{
'name': 'bond1',
'link-aggregation': {'slaves': ['slave1', 'slave00']},
},
}
]
})
]
}
)
with pytest.raises(NmstateValueError):
libnmstate.validator.validate_link_aggregation_state(desired_state,
empty_state())
libnmstate.validator.validate_link_aggregation_state(
desired_state, empty_state()
)
def test_bonds_with_missing_slaves(self):
desired_state = state.State({
schema.Interface.KEY: [
{'name': 'slave0'},
{'name': 'slave1'},
{
'name': 'bond0',
'link-aggregation': {
'slaves': ['slave0', 'slave00']
desired_state = state.State(
{
schema.Interface.KEY: [
{'name': 'slave0'},
{'name': 'slave1'},
{
'name': 'bond0',
'link-aggregation': {'slaves': ['slave0', 'slave00']},
},
},
{
'name': 'bond1',
'link-aggregation': {
'slaves': ['slave1', 'slave11']
{
'name': 'bond1',
'link-aggregation': {'slaves': ['slave1', 'slave11']},
},
}
]
})
]
}
)
with pytest.raises(NmstateValueError):
libnmstate.validator.validate_link_aggregation_state(desired_state,
empty_state())
libnmstate.validator.validate_link_aggregation_state(
desired_state, empty_state()
)
@pytest.mark.xfail(raises=NmstateNotImplementedError,
reason='https://nmstate.atlassian.net/browse/NMSTATE-220',
strict=True)
@pytest.mark.xfail(
raises=NmstateNotImplementedError,
reason='https://nmstate.atlassian.net/browse/NMSTATE-220',
strict=True,
)
def test_dns_three_nameservers():
libnmstate.validator.validate_dns({
DNS.KEY: {
DNS.CONFIG: {
DNS.SERVER: ['8.8.8.8', '2001:4860:4860::8888', '8.8.4.4']
libnmstate.validator.validate_dns(
{
DNS.KEY: {
DNS.CONFIG: {
DNS.SERVER: ['8.8.8.8', '2001:4860:4860::8888', '8.8.4.4']
}
}
}
})
)
def empty_state():
@ -160,126 +153,126 @@ def empty_state():
class TestRouteValidation(object):
def test_empty_states(self):
validator.validate_routes(state.State({}), state.State({}))
def test_valid_route_based_on_desired_state(self):
iface0 = _create_interface_state('eth1', ipv4=True)
route0 = self._create_route0()
desired_state = state.State({
desired_state = state.State(
{
schema.Interface.KEY: [iface0],
schema.Route.KEY: {
schema.Route.CONFIG: [route0],
}
})
schema.Route.KEY: {schema.Route.CONFIG: [route0]},
}
)
validator.validate_routes(desired_state, state.State({}))
def test_valid_route_based_on_current_state(self):
iface0 = _create_interface_state('eth1', ipv4=True)
route0 = self._create_route0()
desired_state = state.State({
desired_state = state.State(
{
schema.Interface.KEY: [],
schema.Route.KEY: {
schema.Route.CONFIG: [route0],
}
})
current_state = state.State({
schema.Route.KEY: {schema.Route.CONFIG: [route0]},
}
)
current_state = state.State(
{
schema.Interface.KEY: [iface0],
schema.Route.KEY: {
schema.Route.CONFIG: [],
}
})
schema.Route.KEY: {schema.Route.CONFIG: []},
}
)
validator.validate_routes(desired_state, current_state)
def test_invalid_route_due_to_missing_iface(self):
route0 = self._create_route0()
desired_state = state.State({
desired_state = state.State(
{
schema.Interface.KEY: [],
schema.Route.KEY: {
schema.Route.CONFIG: [route0],
}
})
schema.Route.KEY: {schema.Route.CONFIG: [route0]},
}
)
with pytest.raises(validator.NmstateRouteWithNoInterfaceError):
validator.validate_routes(desired_state, state.State({}))
def test_invalid_route_due_to_non_up_iface(self):
iface0 = _create_interface_state('eth1',
state=schema.InterfaceState.DOWN,
ipv4=True)
iface0 = _create_interface_state(
'eth1', state=schema.InterfaceState.DOWN, ipv4=True
)
route0 = self._create_route0()
desired_state = state.State({
schema.Interface.KEY: [iface0],
schema.Route.KEY: {
schema.Route.CONFIG: [route0],
desired_state = state.State(
{
schema.Interface.KEY: [iface0],
schema.Route.KEY: {schema.Route.CONFIG: [route0]},
}
})
)
with pytest.raises(validator.NmstateRouteWithNoUpInterfaceError):
validator.validate_routes(desired_state, state.State({}))
def test_invalid_route_due_to_missing_ipv4(self):
iface0 = _create_interface_state('eth1', ipv4=False)
route0 = self._create_route0()
desired_state = state.State({
schema.Interface.KEY: [iface0],
schema.Route.KEY: {
schema.Route.CONFIG: [route0],
desired_state = state.State(
{
schema.Interface.KEY: [iface0],
schema.Route.KEY: {schema.Route.CONFIG: [route0]},
}
})
)
with pytest.raises(validator.NmstateRouteWithNoIPInterfaceError):
validator.validate_routes(desired_state, state.State({}))
def test_invalid_route_due_to_missing_ipv6(self):
iface1 = _create_interface_state('eth2', ipv6=False)
route1 = self._create_route1()
desired_state = state.State({
schema.Interface.KEY: [iface1],
schema.Route.KEY: {
schema.Route.CONFIG: [route1],
desired_state = state.State(
{
schema.Interface.KEY: [iface1],
schema.Route.KEY: {schema.Route.CONFIG: [route1]},
}
})
)
with pytest.raises(validator.NmstateRouteWithNoIPInterfaceError):
validator.validate_routes(desired_state, state.State({}))
def test_valid_route_based_on_desired_state_but_not_current(self):
iface0 = _create_interface_state('eth1', ipv4=True)
route0 = self._create_route0()
desired_state = state.State({
schema.Interface.KEY: [iface0],
schema.Route.KEY: {
schema.Route.CONFIG: [route0],
desired_state = state.State(
{
schema.Interface.KEY: [iface0],
schema.Route.KEY: {schema.Route.CONFIG: [route0]},
}
})
iface0_down = _create_interface_state('eth1',
state=schema.InterfaceState.DOWN)
current_state = state.State({
schema.Interface.KEY: [iface0_down],
schema.Route.KEY: {
schema.Route.CONFIG: [],
)
iface0_down = _create_interface_state(
'eth1', state=schema.InterfaceState.DOWN
)
current_state = state.State(
{
schema.Interface.KEY: [iface0_down],
schema.Route.KEY: {schema.Route.CONFIG: []},
}
})
)
validator.validate_routes(desired_state, current_state)
def test_invalid_route_based_on_desired_state_but_not_current(self):
iface0_ipv4_disabled = _create_interface_state('eth1', ipv4=False)
route0 = self._create_route0()
desired_state = state.State({
schema.Interface.KEY: [iface0_ipv4_disabled],
schema.Route.KEY: {
schema.Route.CONFIG: [route0],
desired_state = state.State(
{
schema.Interface.KEY: [iface0_ipv4_disabled],
schema.Route.KEY: {schema.Route.CONFIG: [route0]},
}
})
)
iface0_ipv4_enabled = _create_interface_state('eth1', ipv4=True)
current_state = state.State({
schema.Interface.KEY: [iface0_ipv4_enabled],
schema.Route.KEY: {
schema.Route.CONFIG: [],
current_state = state.State(
{
schema.Interface.KEY: [iface0_ipv4_enabled],
schema.Route.KEY: {schema.Route.CONFIG: []},
}
})
)
with pytest.raises(validator.NmstateRouteWithNoIPInterfaceError):
validator.validate_routes(desired_state, current_state)
@ -289,13 +282,13 @@ class TestRouteValidation(object):
def _create_route1(self):
return _create_route(
'2001:db8:a::/64', '2001:db8:1::a', 'eth2', 51, 104)
'2001:db8:a::/64', '2001:db8:1::a', 'eth2', 51, 104
)
def _create_interface_state(iface_name,
state=schema.InterfaceState.UP,
ipv4=True,
ipv6=True):
def _create_interface_state(
iface_name, state=schema.InterfaceState.UP, ipv4=True, ipv6=True
):
return {
schema.Interface.NAME: iface_name,
schema.Interface.TYPE: schema.InterfaceType.ETHERNET,
@ -311,5 +304,5 @@ def _create_route(dest, via_addr, via_iface, table, metric):
schema.Route.METRIC: metric,
schema.Route.NEXT_HOP_ADDRESS: via_addr,
schema.Route.NEXT_HOP_INTERFACE: via_iface,
schema.Route.TABLE_ID: table
schema.Route.TABLE_ID: table,
}

3
tox.ini
View File

@ -88,7 +88,8 @@ commands =
-S \
--check \
{posargs} \
tests/ctl
tests/ctl \
tests/lib
[testenv:pylint]
skip_install = true