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

black: Cover all tests

Browse Source 
Signed-off-by: Edward Haas <edwardh@redhat.com>
This commit is contained in:
Edward Haas 2019-06-16 10:16:02 +03:00
parent 01519c12f0
commit 339fd1b754
17 changed files with 572 additions and 678 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

125
tests/integration/bond_test.py
View File

@ -49,13 +49,7 @@ interfaces:
def setup_remove_bond99():
yield
remove_bond = {
INTERFACES: [
{
'name': 'bond99',
'type': 'bond',
'state': 'absent'
}
]
INTERFACES: [{'name': 'bond99', 'type': 'bond', 'state': 'absent'}]
}
libnmstate.apply(remove_bond)
@ -68,27 +62,16 @@ def bond_interface(name, slaves):
'name': name,
'type': 'bond',
'state': 'up',
'link-aggregation': {
'mode': 'balance-rr',
'slaves': slaves
}
'link-aggregation': {'mode': 'balance-rr', 'slaves': slaves},
}
]
}
libnmstate.apply(desired_state)
try:
yield desired_state
finally:
libnmstate.apply({
INTERFACES: [
{
'name': name,
'type': 'bond',
'state': 'absent'
}
]
}
libnmstate.apply(
{INTERFACES: [{'name': name, 'type': 'bond', 'state': 'absent'}]}
)
@ -111,13 +94,7 @@ def test_remove_bond_with_minimum_desired_state(eth1_up, eth2_up):
libnmstate.apply(state)
remove_bond_state = {
INTERFACES: [
{
'name': 'bond99',
'type': 'bond',
'state': 'absent'
}
]
INTERFACES: [{'name': 'bond99', 'type': 'bond', 'state': 'absent'}]
}
libnmstate.apply(remove_bond_state)
state = statelib.show_only((state[INTERFACES][0]['name'],))
@ -132,26 +109,25 @@ def test_add_bond_without_slaves():
def test_add_bond_with_slaves_and_ipv4(eth1_up, eth2_up, setup_remove_bond99):
desired_bond_state = {
INTERFACES: [
{
'name': 'bond99',
'type': 'bond',
'state': 'up',
'ipv4': {
'enabled': True,
'address': [
{'ip': '192.168.122.250', 'prefix-length': 24}
]
},
'link-aggregation': {
'mode': 'balance-rr',
'slaves': ['eth1', 'eth2'],
'options':
{'miimon': '140'}
},
}
]
INTERFACES: [
{
'name': 'bond99',
'type': 'bond',
'state': 'up',
'ipv4': {
'enabled': True,
'address': [
{'ip': '192.168.122.250', 'prefix-length': 24}
],
},
'link-aggregation': {
'mode': 'balance-rr',
'slaves': ['eth1', 'eth2'],
'options': {'miimon': '140'},
},
}
]
}
libnmstate.apply(desired_bond_state)
@ -161,26 +137,25 @@ def test_add_bond_with_slaves_and_ipv4(eth1_up, eth2_up, setup_remove_bond99):
def test_rollback_for_bond(eth1_up, eth2_up):
current_state = libnmstate.show()
desired_state = {
INTERFACES: [
{
'name': 'bond99',
'type': 'bond',
'state': 'up',
'ipv4': {
'enabled': True,
'address': [
{'ip': '192.168.122.250', 'prefix-length': 24}
]
},
'link-aggregation': {
'mode': 'balance-rr',
'slaves': ['eth1', 'eth2'],
'options':
{'miimon': '140'}
},
}
]
INTERFACES: [
{
'name': 'bond99',
'type': 'bond',
'state': 'up',
'ipv4': {
'enabled': True,
'address': [
{'ip': '192.168.122.250', 'prefix-length': 24}
],
},
'link-aggregation': {
'mode': 'balance-rr',
'slaves': ['eth1', 'eth2'],
'options': {'miimon': '140'},
},
}
]
}
desired_state[INTERFACES][0]['invalid_key'] = 'foo'
@ -279,9 +254,11 @@ def test_reordering_the_slaves_does_not_change_the_mac(bond99):
eth1_state = state[Interface.KEY][1]
eth2_state = state[Interface.KEY][2]
assert (bond99_state[Interface.MAC] ==
eth1_state[Interface.MAC] ==
eth2_state[Interface.MAC])
assert (
bond99_state[Interface.MAC]
== eth1_state[Interface.MAC]
== eth2_state[Interface.MAC]
)
bond99[INTERFACES][0]['link-aggregation']['slaves'].reverse()
libnmstate.apply(bond99)
@ -291,7 +268,9 @@ def test_reordering_the_slaves_does_not_change_the_mac(bond99):
eth1_modified_state = modified_state[Interface.KEY][1]
eth2_modified_state = modified_state[Interface.KEY][2]
assert (bond99_modified_state[Interface.MAC] ==
eth1_modified_state[Interface.MAC] ==
eth2_modified_state[Interface.MAC] ==
bond99_state[Interface.MAC])
assert (
bond99_modified_state[Interface.MAC]
== eth1_modified_state[Interface.MAC]
== eth2_modified_state[Interface.MAC]
== bond99_state[Interface.MAC]
)

15
tests/integration/conftest.py
View File

@ -29,7 +29,8 @@ from .testlib.statelib import INTERFACES
def logging_setup():
logging.basicConfig(
format='%(asctime)s %(name)-12s %(levelname)-8s %(message)s',
level=logging.DEBUG)
level=logging.DEBUG,
)
@pytest.fixture(scope='session', autouse=True)
@ -61,9 +62,15 @@ def _set_eth_admin_state(ifname, state):
current_state = statelib.show_only((ifname,))
iface_current_state, = current_state[INTERFACES]
if iface_current_state['state'] != state or state == 'down':
desired_state = {INTERFACES: [{'name': iface_current_state['name'],
'type': iface_current_state['type'],
'state': state}]}
desired_state = {
INTERFACES: [
{
'name': iface_current_state['name'],
'type': iface_current_state['type'],
'state': state,
}
]
}
# FIXME: On most systems, IPv6 cannot be disabled by Nmstate/NM.
if state == 'up':
desired_state[INTERFACES][0].update({'ipv6': {'enabled': True}})

171
tests/integration/dhcp_test.py
View File

@ -58,14 +58,16 @@ enable-ra
dhcp-range={ipv6_prefix}::100,{ipv6_prefix}::fff,ra-names,slaac,64,480h
dhcp-option=option:classless-static-route,{classless_rt},{classless_rt_dst}
dhcp-option=option:dns-server,{v4_dns_server}
""".format(**{
""".format(
**{
'iface': DHCP_SRV_NIC,
'ipv4_prefix': DHCP_SRV_IP4_PREFIX,
'ipv6_prefix': DHCP_SRV_IP6_PREFIX,
'classless_rt': IPV4_CLASSLESS_ROUTE_DST_NET1,
'classless_rt_dst': IPV4_CLASSLESS_ROUTE_NEXT_HOP1,
'v4_dns_server': DHCP_SRV_IP4
})
'v4_dns_server': DHCP_SRV_IP4,
}
)
RADVD_CONF_STR = """
interface {}
@ -81,7 +83,9 @@ interface {}
route {} {{
}};
}};
""".format(DHCP_SRV_NIC, DHCP_SRV_IP6_NETWORK, IPV6_CLASSLESS_ROUTE_DST_NET1)
""".format(
DHCP_SRV_NIC, DHCP_SRV_IP6_NETWORK, IPV6_CLASSLESS_ROUTE_DST_NET1
)
RADVD_CONF_PATH = '/etc/radvd.conf'
DNSMASQ_CONF_PATH = '/etc/dnsmasq.d/nmstate.conf'
@ -90,9 +94,11 @@ DNSMASQ_CONF_PATH = '/etc/dnsmasq.d/nmstate.conf'
RESOLV_CONF_PATH = '/var/run/NetworkManager/resolv.conf'
SYSFS_DISABLE_IPV6_FILE = '/proc/sys/net/ipv6/conf/{}/disable_ipv6'.format(
DHCP_SRV_NIC)
DHCP_SRV_NIC
)
SYSFS_DISABLE_RA_SRV = '/proc/sys/net/ipv6/conf/{}/accept_ra'.format(
DHCP_SRV_NIC)
DHCP_SRV_NIC
)
# Python 2 does not have FileNotFoundError and treat file not exist as IOError
try:
@ -123,13 +129,7 @@ def dhcp_env():
def setup_remove_bond99():
yield
remove_bond = {
INTERFACES: [
{
'name': 'bond99',
'type': 'bond',
'state': 'absent'
}
]
INTERFACES: [{'name': 'bond99', 'type': 'bond', 'state': 'absent'}]
}
libnmstate.apply(remove_bond)
@ -139,11 +139,7 @@ def setup_remove_dhcpcli():
yield
remove_bond = {
INTERFACES: [
{
'name': 'dhcpcli',
'type': 'ethernet',
'state': 'absent'
}
{'name': 'dhcpcli', 'type': 'ethernet', 'state': 'absent'}
]
}
libnmstate.apply(remove_bond)
@ -191,7 +187,7 @@ def test_ipv6_dhcp_only(dhcp_env):
has_dhcp_ip_addr = True
break
assert has_dhcp_ip_addr
assert not _has_ipv6_auto_gateway() # DHCPv6 does not provide routes
assert not _has_ipv6_auto_gateway() # DHCPv6 does not provide routes
assert not _has_ipv6_auto_extra_route() # DHCPv6 does not provide routes
assert _has_ipv6_auto_nameserver()
@ -238,8 +234,8 @@ def test_dhcp_with_addresses(dhcp_env):
'dhcp': True,
'address': [
{'ip': IPV4_ADDRESS1, 'prefix-length': 24},
{'ip': IPV4_ADDRESS2, 'prefix-length': 24}
]
{'ip': IPV4_ADDRESS2, 'prefix-length': 24},
],
},
'ipv6': {
'enabled': True,
@ -247,9 +243,9 @@ def test_dhcp_with_addresses(dhcp_env):
'autoconf': True,
'address': [
{'ip': IPV6_ADDRESS1, 'prefix-length': 64},
{'ip': IPV6_ADDRESS2, 'prefix-length': 64}
]
}
{'ip': IPV6_ADDRESS2, 'prefix-length': 64},
],
},
}
]
}
@ -259,26 +255,21 @@ def test_dhcp_with_addresses(dhcp_env):
assertlib.assert_state(desired_state)
def test_dhcp_for_bond_with_ip_address_and_slave(dhcp_env,
setup_remove_dhcpcli,
setup_remove_bond99):
def test_dhcp_for_bond_with_ip_address_and_slave(
dhcp_env, setup_remove_dhcpcli, setup_remove_bond99
):
desired_state = {
INTERFACES: [
{
'name': 'bond99',
'type': 'bond',
'state': 'up',
'ipv4': {
'enabled': True,
'dhcp': False,
},
'ipv4': {'enabled': True, 'dhcp': False},
'link-aggregation': {
'mode': 'balance-rr',
'slaves': [DHCP_CLI_NIC],
'options':
{'miimon': '140'}
}
'options': {'miimon': '140'},
},
}
]
}
@ -530,8 +521,22 @@ def test_ipv6_dhcp_switch_on_to_off(dhcp_env):
def _create_veth_pair():
assert libcmd.exec_cmd(['ip', 'link', 'add', DHCP_SRV_NIC, 'type', 'veth',
'peer', 'name', DHCP_CLI_NIC])[0] == 0
assert (
libcmd.exec_cmd(
[
'ip',
'link',
'add',
DHCP_SRV_NIC,
'type',
'veth',
'peer',
'name',
DHCP_CLI_NIC,
]
)[0]
== 0
)
def _remove_veth_pair():
@ -541,8 +546,19 @@ def _remove_veth_pair():
def _setup_dhcp_nics():
assert libcmd.exec_cmd(['ip', 'link', 'set', DHCP_SRV_NIC, 'up'])[0] == 0
assert libcmd.exec_cmd(['ip', 'link', 'set', DHCP_CLI_NIC, 'up'])[0] == 0
assert libcmd.exec_cmd(['ip', 'addr', 'add', "{}/24".format(DHCP_SRV_IP4),
'dev', DHCP_SRV_NIC])[0] == 0
assert (
libcmd.exec_cmd(
[
'ip',
'addr',
'add',
"{}/24".format(DHCP_SRV_IP4),
'dev',
DHCP_SRV_NIC,
]
)[0]
== 0
)
with open(SYSFS_DISABLE_RA_SRV, 'w') as fd:
fd.write('0')
@ -550,8 +566,19 @@ def _setup_dhcp_nics():
with open(SYSFS_DISABLE_IPV6_FILE, 'w') as fd:
fd.write('0')
assert libcmd.exec_cmd(['ip', 'addr', 'add', "{}/64".format(DHCP_SRV_IP6),
'dev', DHCP_SRV_NIC])[0] == 0
assert (
libcmd.exec_cmd(
[
'ip',
'addr',
'add',
"{}/64".format(DHCP_SRV_IP6),
'dev',
DHCP_SRV_NIC,
]
)[0]
== 0
)
def _clean_up():
@ -569,17 +596,15 @@ def _clean_up():
def test_slave_ipaddr_learned_via_dhcp_added_as_static_to_linux_bridge(
dhcp_env, setup_remove_dhcpcli):
dhcp_env, setup_remove_dhcpcli
):
desired_state = {
INTERFACES: [
{
'name': 'dhcpcli',
'type': 'ethernet',
'state': 'up',
'ipv4': {
'enabled': True,
'dhcp': True
},
'ipv4': {'enabled': True, 'dhcp': True},
}
]
}
@ -599,7 +624,7 @@ def test_slave_ipaddr_learned_via_dhcp_added_as_static_to_linux_bridge(
'ipv4': {
'enabled': True,
'dhcp': False,
'address': dhcpcli_ip
'address': dhcpcli_ip,
},
'bridge': {
'options': {},
@ -608,24 +633,18 @@ def test_slave_ipaddr_learned_via_dhcp_added_as_static_to_linux_bridge(
'name': 'dhcpcli',
'stp-hairpin-mode': False,
'stp-path-cost': 100,
'stp-priority': 32
'stp-priority': 32,
}
]
}
],
},
},
{
'name': 'dhcpcli',
'type': 'ethernet',
'state': 'up',
'ipv4': {
'enabled': False,
'dhcp': False
},
'ipv6': {
'enabled': False,
'dhcp': False
}
}
'ipv4': {'enabled': False, 'dhcp': False},
'ipv6': {'enabled': False, 'dhcp': False},
},
]
}
@ -637,8 +656,12 @@ def _get_nameservers():
"""
Return a list of name server string configured in RESOLV_CONF_PATH.
"""
return libnmstate.show().get(
Constants.DNS, {}).get(DNS.RUNNING, {}).get(DNS.SERVER, [])
return (
libnmstate.show()
.get(Constants.DNS, {})
.get(DNS.RUNNING, {})
.get(DNS.SERVER, [])
)
def _get_running_routes():
@ -651,8 +674,10 @@ def _get_running_routes():
def _has_ipv6_auto_gateway():
routes = _get_running_routes()
for route in routes:
if route[RT.DESTINATION] == IPV6_DEFAULT_GATEWAY and \
route[RT.NEXT_HOP_INTERFACE] == DHCP_CLI_NIC:
if (
route[RT.DESTINATION] == IPV6_DEFAULT_GATEWAY
and route[RT.NEXT_HOP_INTERFACE] == DHCP_CLI_NIC
):
return True
return False
@ -660,8 +685,10 @@ def _has_ipv6_auto_gateway():
def _has_ipv6_auto_extra_route():
routes = _get_running_routes()
for route in routes:
if route[RT.DESTINATION] == IPV6_CLASSLESS_ROUTE_DST_NET1 and \
route[RT.NEXT_HOP_INTERFACE] == DHCP_CLI_NIC:
if (
route[RT.DESTINATION] == IPV6_CLASSLESS_ROUTE_DST_NET1
and route[RT.NEXT_HOP_INTERFACE] == DHCP_CLI_NIC
):
return True
return False
@ -677,8 +704,10 @@ def _has_ipv4_dhcp_nameserver():
def _has_ipv4_dhcp_gateway():
routes = _get_running_routes()
for route in routes:
if route[RT.DESTINATION] == IPV4_DEFAULT_GATEWAY and \
route[RT.NEXT_HOP_INTERFACE] == DHCP_CLI_NIC:
if (
route[RT.DESTINATION] == IPV4_DEFAULT_GATEWAY
and route[RT.NEXT_HOP_INTERFACE] == DHCP_CLI_NIC
):
return True
return False
@ -686,8 +715,10 @@ def _has_ipv4_dhcp_gateway():
def _has_ipv4_classless_route():
routes = _get_running_routes()
for route in routes:
if route[RT.DESTINATION] == IPV4_CLASSLESS_ROUTE_DST_NET1 and \
route[RT.NEXT_HOP_ADDRESS] == IPV4_CLASSLESS_ROUTE_NEXT_HOP1 and \
route[RT.NEXT_HOP_INTERFACE] == DHCP_CLI_NIC:
if (
route[RT.DESTINATION] == IPV4_CLASSLESS_ROUTE_DST_NET1
and route[RT.NEXT_HOP_ADDRESS] == IPV4_CLASSLESS_ROUTE_NEXT_HOP1
and route[RT.NEXT_HOP_INTERFACE] == DHCP_CLI_NIC
):
return True
return False

185
tests/integration/dns_test.py
View File

@ -33,9 +33,12 @@ EXAMPLE_SEARCHES = ['example.org', 'example.com']
parametrize_ip_ver = pytest.mark.parametrize(
'dns_config',
[({DNS.SERVER: IPV4_DNS_NAMESERVERS, DNS.SEARCH: EXAMPLE_SEARCHES}),
({DNS.SERVER: IPV6_DNS_NAMESERVERS, DNS.SEARCH: EXAMPLE_SEARCHES})],
ids=['ipv4', 'ipv6'])
[
({DNS.SERVER: IPV4_DNS_NAMESERVERS, DNS.SEARCH: EXAMPLE_SEARCHES}),
({DNS.SERVER: IPV6_DNS_NAMESERVERS, DNS.SEARCH: EXAMPLE_SEARCHES}),
],
ids=['ipv4', 'ipv6'],
)
@pytest.fixture(scope='function', autouse=True)
@ -45,12 +48,7 @@ def dns_test_env(eth1_up, eth2_up):
# failure when bring eth1/eth2 down.
desired_state = {
Interface.KEY: _get_test_iface_states(),
DNS.KEY: {
DNS.CONFIG: {
DNS.SERVER: [],
DNS.SEARCH: []
}
}
DNS.KEY: {DNS.CONFIG: {DNS.SERVER: [], DNS.SEARCH: []}},
}
netapplier.apply(desired_state)
@ -59,12 +57,8 @@ def dns_test_env(eth1_up, eth2_up):
def test_dns_edit_nameserver_with_static_gateway(dns_config):
desired_state = {
Interface.KEY: _get_test_iface_states(),
Route.KEY: {
Route.CONFIG: _gen_default_gateway_route()
},
DNS.KEY: {
DNS.CONFIG: dns_config
}
Route.KEY: {Route.CONFIG: _gen_default_gateway_route()},
DNS.KEY: {DNS.CONFIG: dns_config},
}
netapplier.apply(desired_state)
current_state = netinfo.show()
@ -74,16 +68,12 @@ def test_dns_edit_nameserver_with_static_gateway(dns_config):
def test_dns_edit_ipv4_nameserver_before_ipv6():
dns_config = {
DNS.SERVER: [IPV4_DNS_NAMESERVERS[0], IPV6_DNS_NAMESERVERS[0]],
DNS.SEARCH: []
DNS.SEARCH: [],
}
desired_state = {
Interface.KEY: _get_test_iface_states(),
Route.KEY: {
Route.CONFIG: _gen_default_gateway_route()
},
DNS.KEY: {
DNS.CONFIG: dns_config
}
Route.KEY: {Route.CONFIG: _gen_default_gateway_route()},
DNS.KEY: {DNS.CONFIG: dns_config},
}
netapplier.apply(desired_state)
current_state = netinfo.show()
@ -93,38 +83,32 @@ def test_dns_edit_ipv4_nameserver_before_ipv6():
def test_dns_edit_ipv6_nameserver_before_ipv4():
dns_config = {
DNS.SERVER: [IPV6_DNS_NAMESERVERS[0], IPV4_DNS_NAMESERVERS[0]],
DNS.SEARCH: []
DNS.SEARCH: [],
}
desired_state = {
Interface.KEY: _get_test_iface_states(),
Route.KEY: {
Route.CONFIG: _gen_default_gateway_route()
},
DNS.KEY: {
DNS.CONFIG: dns_config
}
Route.KEY: {Route.CONFIG: _gen_default_gateway_route()},
DNS.KEY: {DNS.CONFIG: dns_config},
}
netapplier.apply(desired_state)
current_state = netinfo.show()
assert dns_config == current_state[DNS.KEY][DNS.CONFIG]
@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_edit_three_nameservers():
dns_config = {
DNS.SERVER: IPV6_DNS_NAMESERVERS + [IPV4_DNS_NAMESERVERS[0]],
DNS.SEARCH: []
DNS.SEARCH: [],
}
desired_state = {
Interface.KEY: _get_test_iface_states(),
Route.KEY: {
Route.CONFIG: _gen_default_gateway_route()
},
DNS.KEY: {
DNS.CONFIG: dns_config
}
Route.KEY: {Route.CONFIG: _gen_default_gateway_route()},
DNS.KEY: {DNS.CONFIG: dns_config},
}
netapplier.apply(desired_state)
current_state = netinfo.show()
@ -134,72 +118,55 @@ def test_dns_edit_three_nameservers():
def test_remove_dns_config():
dns_config = {
DNS.SERVER: [IPV6_DNS_NAMESERVERS[0], IPV4_DNS_NAMESERVERS[0]],
DNS.SEARCH: []
DNS.SEARCH: [],
}
desired_state = {
Interface.KEY: _get_test_iface_states(),
Route.KEY: {
Route.CONFIG: _gen_default_gateway_route()
},
DNS.KEY: {
DNS.CONFIG: dns_config
}
Route.KEY: {Route.CONFIG: _gen_default_gateway_route()},
DNS.KEY: {DNS.CONFIG: dns_config},
}
netapplier.apply(desired_state)
netapplier.apply({
Interface.KEY: [],
DNS.KEY: {
DNS.CONFIG: {}
}
})
netapplier.apply({Interface.KEY: [], DNS.KEY: {DNS.CONFIG: {}}})
current_state = netinfo.show()
dns_config = {
DNS.SERVER: [],
DNS.SEARCH: []
}
dns_config = {DNS.SERVER: [], DNS.SEARCH: []}
assert dns_config == current_state[DNS.KEY][DNS.CONFIG]
def test_preserve_dns_config():
dns_config = {
DNS.SERVER: [IPV6_DNS_NAMESERVERS[0], IPV4_DNS_NAMESERVERS[0]],
DNS.SEARCH: []
DNS.SEARCH: [],
}
desired_state = {
Interface.KEY: _get_test_iface_states(),
Route.KEY: {
Route.CONFIG: _gen_default_gateway_route()
},
DNS.KEY: {
DNS.CONFIG: dns_config
}
Route.KEY: {Route.CONFIG: _gen_default_gateway_route()},
DNS.KEY: {DNS.CONFIG: dns_config},
}
netapplier.apply(desired_state)
current_state = netinfo.show()
# Remove default gateways, so that if nmstate try to find new interface
# for DNS profile, it will fail.
netapplier.apply({
Interface.KEY: _get_test_iface_states(),
Route.KEY: {
Route.CONFIG: [
{
Route.DESTINATION: '0.0.0.0/0',
Route.STATE: Route.STATE_ABSENT
},
{
Route.DESTINATION: '::/0',
Route.STATE: Route.STATE_ABSENT
},
]
},
})
netapplier.apply(
{
Interface.KEY: _get_test_iface_states(),
Route.KEY: {
Route.CONFIG: [
{
Route.DESTINATION: '0.0.0.0/0',
Route.STATE: Route.STATE_ABSENT,
},
{
Route.DESTINATION: '::/0',
Route.STATE: Route.STATE_ABSENT,
},
]
},
}
)
netapplier.apply({
Interface.KEY: [],
DNS.KEY: dns_config
})
netapplier.apply({Interface.KEY: [], DNS.KEY: dns_config})
assert dns_config == current_state[DNS.KEY][DNS.CONFIG]
@ -208,15 +175,13 @@ def test_preserve_dns_config():
def setup_ipv4_ipv6_name_server():
desired_state = {
Interface.KEY: _get_test_iface_states(),
Route.KEY: {
Route.CONFIG: _gen_default_gateway_route()
},
Route.KEY: {Route.CONFIG: _gen_default_gateway_route()},
DNS.KEY: {
DNS.CONFIG: {
DNS.SERVER: [IPV6_DNS_NAMESERVERS[0], IPV4_DNS_NAMESERVERS[0]],
DNS.SEARCH: []
DNS.SEARCH: [],
}
}
},
}
netapplier.apply(desired_state)
yield desired_state
@ -225,9 +190,7 @@ def setup_ipv4_ipv6_name_server():
def test_preserve_dns_config_with_empty_state(setup_ipv4_ipv6_name_server):
old_state = setup_ipv4_ipv6_name_server
netapplier.apply({
Interface.KEY: [],
})
netapplier.apply({Interface.KEY: []})
current_state = netinfo.show()
assert old_state[DNS.KEY][DNS.CONFIG] == current_state[DNS.KEY][DNS.CONFIG]
@ -240,52 +203,32 @@ def _get_test_iface_states():
Interface.STATE: InterfaceState.UP,
Interface.TYPE: InterfaceType.ETHERNET,
Interface.IPV4: {
'address': [
{
'ip': '192.0.2.251',
'prefix-length': 24
}
],
'address': [{'ip': '192.0.2.251', 'prefix-length': 24}],
'dhcp': False,
'enabled': True
'enabled': True,
},
Interface.IPV6: {
'address': [
{
'ip': '2001:db8:1::1',
'prefix-length': 64
}
],
'address': [{'ip': '2001:db8:1::1', 'prefix-length': 64}],
'dhcp': False,
'autoconf': False,
'enabled': True
}
'enabled': True,
},
},
{
Interface.NAME: 'eth2',
Interface.STATE: InterfaceState.UP,
Interface.TYPE: InterfaceType.ETHERNET,
Interface.IPV4: {
'address': [
{
'ip': '198.51.100.1',
'prefix-length': 24
}
],
'address': [{'ip': '198.51.100.1', 'prefix-length': 24}],
'dhcp': False,
'enabled': True
'enabled': True,
},
Interface.IPV6: {
'address': [
{
'ip': '2001:db8:2::1',
'prefix-length': 64
}
],
'address': [{'ip': '2001:db8:2::1', 'prefix-length': 64}],
'dhcp': False,
'autoconf': False,
'enabled': True
}
'enabled': True,
},
},
]
@ -303,5 +246,5 @@ def _gen_default_gateway_route():
Route.METRIC: 201,
Route.NEXT_HOP_ADDRESS: '2001:db8:2::f',
Route.NEXT_HOP_INTERFACE: 'eth1',
}
},
]

29
tests/integration/examples_test.py
View File

@ -29,8 +29,9 @@ def test_add_down_remove_vlan(eth1_up):
"""
vlan_ifname = 'eth1.101'
with example_state('vlan101_eth1_up.yml',
cleanup='vlan101_eth1_absent.yml') as desired_state:
with example_state(
'vlan101_eth1_up.yml', cleanup='vlan101_eth1_absent.yml'
) as desired_state:
assertlib.assert_state(desired_state)
with example_state('vlan101_eth1_down.yml') as desired_state:
assertlib.assert_absent(vlan_ifname)
@ -39,29 +40,31 @@ def test_add_down_remove_vlan(eth1_up):
def test_add_remove_ovs_bridge(eth1_up):
with example_state('ovsbridge_create.yml',
cleanup='ovsbridge_delete.yml') as desired_state:
with example_state(
'ovsbridge_create.yml', cleanup='ovsbridge_delete.yml'
) as desired_state:
assertlib.assert_state(desired_state)
assertlib.assert_absent('ovs-br0')
def test_add_remove_linux_bridge(eth1_up):
with example_state('linuxbrige_eth1_up.yml',
cleanup='linuxbrige_eth1_absent.yml') as desired_state:
with example_state(
'linuxbrige_eth1_up.yml', cleanup='linuxbrige_eth1_absent.yml'
) as desired_state:
assertlib.assert_state(desired_state)
assertlib.assert_absent('linux-br0')
def test_dns_edit(eth1_up):
with example_state('dns_edit_eth1.yml',
cleanup='dns_remove.yml') as desired_state:
with example_state(
'dns_edit_eth1.yml', cleanup='dns_remove.yml'
) as desired_state:
assertlib.assert_state(desired_state)
current_state = netinfo.show()
assert (current_state.get(DNS.KEY, {}).get(DNS.CONFIG, {}) ==
{
DNS.SERVER: [],
DNS.SEARCH: []
})
assert current_state.get(DNS.KEY, {}).get(DNS.CONFIG, {}) == {
DNS.SERVER: [],
DNS.SEARCH: [],
}

16
tests/integration/iface_admin_state_test.py
View File

@ -25,13 +25,7 @@ from .testlib.statelib import INTERFACES
def test_set_a_down_iface_down(eth1_up):
desired_state = {
INTERFACES: [
{
'name': 'eth1',
'type': 'ethernet',
'state': 'down',
}
]
INTERFACES: [{'name': 'eth1', 'type': 'ethernet', 'state': 'down'}]
}
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
@ -44,13 +38,7 @@ def test_set_a_down_iface_down(eth1_up):
@pytest.mark.xfail(reason='Some ifaces cannot be removed', strict=True)
def test_removing_a_non_removable_iface(eth1_up):
desired_state = {
INTERFACES: [
{
'name': 'eth1',
'type': 'ethernet',
'state': 'absent',
}
]
INTERFACES: [{'name': 'eth1', 'type': 'ethernet', 'state': 'absent'}]
}
libnmstate.apply(desired_state)

27
tests/integration/linux_bridge_test.py
View File

@ -93,8 +93,9 @@ def test_create_and_remove_linux_bridge_with_one_port(eth1_up):
def test_create_and_remove_linux_bridge_with_two_ports(eth1_up, eth2_up):
bridge_name = TEST_BRIDGE0
bridge_state = _create_bridge_subtree_config((TEST_BRIDGE0_PORT0,
TEST_BRIDGE0_PORT1))
bridge_state = _create_bridge_subtree_config(
(TEST_BRIDGE0_PORT0, TEST_BRIDGE0_PORT1)
)
with _linux_bridge(bridge_name, bridge_state) as desired_state:
assertlib.assert_state(desired_state)
@ -149,7 +150,7 @@ def _linux_bridge(name, bridge_state):
{
Interface.NAME: name,
Interface.TYPE: InterfaceType.LINUX_BRIDGE,
Interface.STATE: InterfaceState.UP
Interface.STATE: InterfaceState.UP,
}
]
}
@ -161,12 +162,14 @@ def _linux_bridge(name, bridge_state):
try:
yield desired_state
finally:
libnmstate.apply({
INTERFACES: [
{
Interface.NAME: name,
Interface.TYPE: InterfaceType.LINUX_BRIDGE,
Interface.STATE: InterfaceState.ABSENT
}
]
})
libnmstate.apply(
{
INTERFACES: [
{
Interface.NAME: name,
Interface.TYPE: InterfaceType.LINUX_BRIDGE,
Interface.STATE: InterfaceState.ABSENT,
}
]
}
)

30
tests/integration/nmstatectl_test.py
View File

@ -78,12 +78,17 @@ CONFIRMATION_INTERFACE = 'eth1.101'
CONFIRMATION_CLEAN = 'vlan101_eth1_absent.yml'
CONFIRMATION_TEST = 'vlan101_eth1_up.yml'
CONFIRMATION_TEST_STATE = load_example(CONFIRMATION_TEST)
CONFIRMATION_SET = SET_CMD + ['--no-commit',
os.path.join(EXAMPLES, CONFIRMATION_TEST)]
CONFIRMATION_SET = SET_CMD + [
'--no-commit',
os.path.join(EXAMPLES, CONFIRMATION_TEST),
]
CONFIRMATION_TIMEOUT = 5
CONFIRMATION_TIMOUT_COMMAND = SET_CMD + \
['--no-commit', '--timeout', str(CONFIRMATION_TIMEOUT),
os.path.join(EXAMPLES, CONFIRMATION_TEST)]
CONFIRMATION_TIMOUT_COMMAND = SET_CMD + [
'--no-commit',
'--timeout',
str(CONFIRMATION_TIMEOUT),
os.path.join(EXAMPLES, CONFIRMATION_TEST),
]
def test_missing_operation():
@ -144,8 +149,10 @@ def test_set_command_with_yaml_format():
def test_set_command_with_two_states():
examples = find_examples_dir()
cmd = SET_CMD + [os.path.join(examples, 'linuxbrige_eth1_up.yml'),
os.path.join(examples, 'linuxbrige_eth1_absent.yml')]
cmd = SET_CMD + [
os.path.join(examples, 'linuxbrige_eth1_up.yml'),
os.path.join(examples, 'linuxbrige_eth1_absent.yml'),
]
ret = libcmd.exec_cmd(cmd)
rc = ret[0]
@ -166,8 +173,7 @@ def test_manual_confirmation(eth1_up):
def test_manual_rollback(eth1_up):
""" I can manually roll back a state. """
with example_state(CONFIRMATION_CLEAN,
CONFIRMATION_CLEAN) as clean_state:
with example_state(CONFIRMATION_CLEAN, CONFIRMATION_CLEAN) as clean_state:
assert_command(CONFIRMATION_SET)
assertlib.assert_state(CONFIRMATION_TEST_STATE)
@ -179,8 +185,7 @@ def test_dual_change(eth1_up):
""" I cannot set a state without confirming/rolling back the state change.
"""
with example_state(CONFIRMATION_CLEAN,
CONFIRMATION_CLEAN) as clean_state:
with example_state(CONFIRMATION_CLEAN, CONFIRMATION_CLEAN) as clean_state:
assert_command(CONFIRMATION_SET)
assertlib.assert_state(CONFIRMATION_TEST_STATE)
@ -193,8 +198,7 @@ def test_dual_change(eth1_up):
def test_automatic_rollback(eth1_up):
""" If I do not confirm the state, it is automatically rolled back. """
with example_state(CONFIRMATION_CLEAN,
CONFIRMATION_CLEAN) as clean_state:
with example_state(CONFIRMATION_CLEAN, CONFIRMATION_CLEAN) as clean_state:
assert_command(CONFIRMATION_TIMOUT_COMMAND)
assertlib.assert_state(CONFIRMATION_TEST_STATE)

52
tests/integration/ovs_test.py
View File

@ -40,20 +40,13 @@ interfaces:
def test_create_and_remove_ovs_bridge_with_a_system_port(eth1_up):
state = yaml.load(OVS_BRIDGE_YAML_BASE, Loader=yaml.SafeLoader)
state[INTERFACES][0]['bridge']['port'] = [
{
'name': 'eth1',
'type': 'system'
}
{'name': 'eth1', 'type': 'system'}
]
libnmstate.apply(state)
setup_remove_ovs_bridge_state = {
INTERFACES: [
{
'name': 'ovs-br0',
'type': 'ovs-bridge',
'state': 'absent'
}
{'name': 'ovs-br0', 'type': 'ovs-bridge', 'state': 'absent'}
]
}
libnmstate.apply(setup_remove_ovs_bridge_state)
@ -63,23 +56,13 @@ def test_create_and_remove_ovs_bridge_with_a_system_port(eth1_up):
def test_create_and_remove_ovs_bridge_with_min_desired_state():
desired_state = {
INTERFACES: [
{
'name': 'ovs-br0',
'type': 'ovs-bridge',
'state': 'up'
}
]
INTERFACES: [{'name': 'ovs-br0', 'type': 'ovs-bridge', 'state': 'up'}]
}
libnmstate.apply(desired_state)
setup_remove_ovs_bridge_state = {
INTERFACES: [
{
'name': 'ovs-br0',
'type': 'ovs-bridge',
'state': 'absent'
}
{'name': 'ovs-br0', 'type': 'ovs-bridge', 'state': 'absent'}
]
}
libnmstate.apply(setup_remove_ovs_bridge_state)
@ -90,10 +73,7 @@ def test_create_and_remove_ovs_bridge_with_min_desired_state():
def test_create_and_remove_ovs_bridge_with_an_internal_port():
state = yaml.load(OVS_BRIDGE_YAML_BASE, Loader=yaml.SafeLoader)
state[INTERFACES][0]['bridge']['port'] = [
{
'name': 'ovs0',
'type': 'internal'
}
{'name': 'ovs0', 'type': 'internal'}
]
ovs_internal_interface_state = {
'name': 'ovs0',
@ -102,12 +82,7 @@ def test_create_and_remove_ovs_bridge_with_an_internal_port():
'mtu': 1500,
'ipv4': {
'enabled': True,
'address': [
{
'ip': '192.0.2.1',
'prefix-length': 24
}
]
'address': [{'ip': '192.0.2.1', 'prefix-length': 24}],
},
}
state[INTERFACES].append(ovs_internal_interface_state)
@ -115,19 +90,12 @@ def test_create_and_remove_ovs_bridge_with_an_internal_port():
setup_remove_ovs_bridge_state_and_port = {
INTERFACES: [
{
'name': 'ovs-br0',
'type': 'ovs-bridge',
'state': 'absent'
},
{
'name': 'ovs',
'type': 'ovs-interface',
'state': 'absent'
}
{'name': 'ovs-br0', 'type': 'ovs-bridge', 'state': 'absent'},
{'name': 'ovs', 'type': 'ovs-interface', 'state': 'absent'},
]
}
libnmstate.apply(setup_remove_ovs_bridge_state_and_port)
state = statelib.show_only(
(state[INTERFACES][0]['name'], state[INTERFACES][1]['name']))
(state[INTERFACES][0]['name'], state[INTERFACES][1]['name'])
)
assert not state[INTERFACES]

38
tests/integration/preserve_ip_config_test.py
View File

@ -43,33 +43,30 @@ def test_reapply_preserve_ip_config(eth1_up):
'state': 'up',
'ipv4': {
'address': [
{
'ip': IPV4_ADDRESS1,
'prefix-length': 24
}
{'ip': IPV4_ADDRESS1, 'prefix-length': 24}
],
'enabled': True
'enabled': True,
},
'ipv6': {
'address': [
{
'ip': IPV6_ADDRESS1,
'prefix-length': 64
}
{'ip': IPV6_ADDRESS1, 'prefix-length': 64}
],
'enabled': True
'enabled': True,
},
'mtu': 1500
},
'mtu': 1500,
}
]
})
}
)
cur_state = statelib.show_only(('eth1',))
iface_name = cur_state[INTERFACES][0]['name']
uuid = _get_nm_profile_uuid(iface_name)
for key, value in ((_IPV4_EXTRA_CONFIG, _IPV4_EXTRA_VALUE),
(_IPV6_EXTRA_CONFIG, _IPV6_EXTRA_VALUE)):
for key, value in (
(_IPV4_EXTRA_CONFIG, _IPV4_EXTRA_VALUE),
(_IPV6_EXTRA_CONFIG, _IPV6_EXTRA_VALUE),
):
with _extra_ip_config(uuid, key, value):
libnmstate.apply(cur_state)
_assert_extra_ip_config(uuid, key, value)
@ -88,8 +85,9 @@ def _get_nm_profile_uuid(iface_name):
def _get_cur_extra_ip_config(uuid, key):
rc, output, _ = libcmd.exec_cmd(['nmcli', '--get-values', key,
'connection', 'show', uuid])
rc, output, _ = libcmd.exec_cmd(
['nmcli', '--get-values', key, 'connection', 'show', uuid]
)
assert rc == 0
return output.split('\n')[0]
@ -105,8 +103,10 @@ def _extra_ip_config(uuid, key, value):
def _apply_extra_ip_config(uuid, key, value):
assert libcmd.exec_cmd(['nmcli', 'connection', 'modify', uuid,
key, value])[0] == 0
assert (
libcmd.exec_cmd(['nmcli', 'connection', 'modify', uuid, key, value])[0]
== 0
)
def _assert_extra_ip_config(uuid, key, value):

289
tests/integration/route_test.py
View File

@ -34,49 +34,39 @@ ETH1_INTERFACE_STATE = {
Interface.STATE: InterfaceState.UP,
Interface.TYPE: InterfaceType.ETHERNET,
Interface.IPV4: {
'address': [
{
'ip': IPV4_ADDRESS1,
'prefix-length': 24
}
],
'address': [{'ip': IPV4_ADDRESS1, 'prefix-length': 24}],
'dhcp': False,
'enabled': True
'enabled': True,
},
Interface.IPV6: {
'address': [
{
'ip': IPV6_ADDRESS1,
'prefix-length': 64
}
],
'address': [{'ip': IPV6_ADDRESS1, 'prefix-length': 64}],
'dhcp': False,
'autoconf': False,
'enabled': True
}
'enabled': True,
},
}
def test_add_static_routes(eth1_up):
routes = _get_ipv4_test_routes() + _get_ipv6_test_routes()
libnmstate.apply({
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: routes
libnmstate.apply(
{
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {Route.CONFIG: routes},
}
})
)
cur_state = libnmstate.show()
_assert_routes(routes, cur_state)
def test_add_gateway(eth1_up):
routes = [_get_ipv4_gateways()[0], _get_ipv6_test_routes()[0]]
libnmstate.apply({
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: routes
libnmstate.apply(
{
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {Route.CONFIG: routes},
}
})
)
cur_state = libnmstate.show()
_assert_routes(routes, cur_state)
@ -86,12 +76,12 @@ def test_add_route_without_metric(eth1_up):
for route in routes:
del route[Route.METRIC]
libnmstate.apply({
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: routes
libnmstate.apply(
{
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {Route.CONFIG: routes},
}
})
)
cur_state = libnmstate.show()
_assert_routes(routes, cur_state)
@ -102,56 +92,65 @@ def test_add_route_without_table_id(eth1_up):
for route in routes:
del route[Route.TABLE_ID]
libnmstate.apply({
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: routes
},
})
libnmstate.apply(
{
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {Route.CONFIG: routes},
}
)
cur_state = libnmstate.show()
_assert_routes(routes, cur_state)
@pytest.mark.xfail(raises=NmstateNotImplementedError,
reason="Red Hat Bug 1707396")
@pytest.mark.xfail(
raises=NmstateNotImplementedError, reason="Red Hat Bug 1707396"
)
def test_multiple_gateway(eth1_up):
libnmstate.apply(
{
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: _get_ipv4_gateways() + _get_ipv6_gateways()
}
})
},
}
)
def test_change_gateway(eth1_up):
libnmstate.apply({
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: [_get_ipv4_gateways()[0],
_get_ipv6_gateways()[0]]
libnmstate.apply(
{
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: [
_get_ipv4_gateways()[0],
_get_ipv6_gateways()[0],
]
},
}
})
)
routes = [_get_ipv4_gateways()[1], _get_ipv6_gateways()[1]]
libnmstate.apply({
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: [
{
Route.STATE: Route.STATE_ABSENT,
Route.NEXT_HOP_INTERFACE: 'eth1',
Route.DESTINATION: '0.0.0.0/0'
},
{
Route.STATE: Route.STATE_ABSENT,
Route.NEXT_HOP_INTERFACE: 'eth1',
Route.DESTINATION: '::/0'
}
] + routes,
},
})
libnmstate.apply(
{
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: [
{
Route.STATE: Route.STATE_ABSENT,
Route.NEXT_HOP_INTERFACE: 'eth1',
Route.DESTINATION: '0.0.0.0/0',
},
{
Route.STATE: Route.STATE_ABSENT,
Route.NEXT_HOP_INTERFACE: 'eth1',
Route.DESTINATION: '::/0',
},
]
+ routes
},
}
)
cur_state = libnmstate.show()
_assert_routes(routes, cur_state)
@ -209,15 +208,15 @@ def _get_ipv4_test_routes():
Route.METRIC: 103,
Route.NEXT_HOP_ADDRESS: '192.0.2.1',
Route.NEXT_HOP_INTERFACE: 'eth1',
Route.TABLE_ID: 50
Route.TABLE_ID: 50,
},
{
Route.DESTINATION: '203.0.113.0/24',
Route.METRIC: 103,
Route.NEXT_HOP_ADDRESS: '192.0.2.1',
Route.NEXT_HOP_INTERFACE: 'eth1',
Route.TABLE_ID: 51
}
Route.TABLE_ID: 51,
},
]
@ -228,15 +227,15 @@ def _get_ipv4_gateways():
Route.METRIC: 103,
Route.NEXT_HOP_ADDRESS: '192.0.2.1',
Route.NEXT_HOP_INTERFACE: 'eth1',
Route.TABLE_ID: 254
Route.TABLE_ID: 254,
},
{
Route.DESTINATION: '0.0.0.0/0',
Route.METRIC: 101,
Route.NEXT_HOP_ADDRESS: '192.0.2.2',
Route.NEXT_HOP_INTERFACE: 'eth1',
Route.TABLE_ID: 254
}
Route.TABLE_ID: 254,
},
]
@ -247,15 +246,15 @@ def _get_ipv6_test_routes():
Route.METRIC: 103,
Route.NEXT_HOP_ADDRESS: '2001:db8:1::a',
Route.NEXT_HOP_INTERFACE: 'eth1',
Route.TABLE_ID: 50
Route.TABLE_ID: 50,
},
{
Route.DESTINATION: '2001:db8:b::/64',
Route.METRIC: 103,
Route.NEXT_HOP_ADDRESS: '2001:db8:1::b',
Route.NEXT_HOP_INTERFACE: 'eth1',
Route.TABLE_ID: 50
}
Route.TABLE_ID: 50,
},
]
@ -266,51 +265,53 @@ def _get_ipv6_gateways():
Route.METRIC: 103,
Route.NEXT_HOP_ADDRESS: '2001:db8:1::f',
Route.NEXT_HOP_INTERFACE: 'eth1',
Route.TABLE_ID: 254
Route.TABLE_ID: 254,
},
{
Route.DESTINATION: '::/0',
Route.METRIC: 101,
Route.NEXT_HOP_ADDRESS: '2001:db8:1::e',
Route.NEXT_HOP_INTERFACE: 'eth1',
Route.TABLE_ID: 254
}
Route.TABLE_ID: 254,
},
]
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, ''),
)
parametrize_ip_ver_routes = pytest.mark.parametrize(
'get_routes_func',
[(_get_ipv4_test_routes),
(_get_ipv6_test_routes)],
ids=['ipv4', 'ipv6'])
[(_get_ipv4_test_routes), (_get_ipv6_test_routes)],
ids=['ipv4', 'ipv6'],
)
@parametrize_ip_ver_routes
def test_remove_specific_route(eth1_up, get_routes_func):
routes = get_routes_func()
libnmstate.apply({
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: routes
},
})
libnmstate.apply(
{
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {Route.CONFIG: routes},
}
)
cur_state = libnmstate.show()
_assert_routes(routes, cur_state)
absent_route = routes[0]
absent_route[Route.STATE] = Route.STATE_ABSENT
libnmstate.apply({
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: [absent_route]
},
})
libnmstate.apply(
{
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {Route.CONFIG: [absent_route]},
}
)
expected_routes = routes[1:]
@ -321,25 +322,25 @@ def test_remove_specific_route(eth1_up, get_routes_func):
@parametrize_ip_ver_routes
def test_remove_wildcast_route_with_iface(eth1_up, get_routes_func):
routes = get_routes_func()
libnmstate.apply({
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: routes
},
})
libnmstate.apply(
{
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {Route.CONFIG: routes},
}
)
cur_state = libnmstate.show()
_assert_routes(routes, cur_state)
absent_route = {
Route.STATE: Route.STATE_ABSENT,
Route.NEXT_HOP_INTERFACE: 'eth1'
Route.NEXT_HOP_INTERFACE: 'eth1',
}
libnmstate.apply({
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: [absent_route]
},
})
libnmstate.apply(
{
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {Route.CONFIG: [absent_route]},
}
)
expected_routes = []
@ -350,27 +351,29 @@ def test_remove_wildcast_route_with_iface(eth1_up, get_routes_func):
@parametrize_ip_ver_routes
def test_remove_wildcast_route_without_iface(eth1_up, get_routes_func):
routes = get_routes_func()
libnmstate.apply({
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: routes
},
})
libnmstate.apply(
{
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {Route.CONFIG: routes},
}
)
cur_state = libnmstate.show()
_assert_routes(routes, cur_state)
absent_routes = []
for route in routes:
absent_routes.append({
Route.STATE: Route.STATE_ABSENT,
Route.DESTINATION: route[Route.DESTINATION]
})
libnmstate.apply({
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: absent_routes
},
})
absent_routes.append(
{
Route.STATE: Route.STATE_ABSENT,
Route.DESTINATION: route[Route.DESTINATION],
}
)
libnmstate.apply(
{
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {Route.CONFIG: absent_routes},
}
)
expected_routes = []
@ -380,19 +383,17 @@ def test_remove_wildcast_route_without_iface(eth1_up, get_routes_func):
# TODO: Once we can disable IPv6, we should add an IPv6 test case here
def test_disable_ipv4_with_routes_in_current(eth1_up):
libnmstate.apply({
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: _get_ipv4_test_routes(),
libnmstate.apply(
{
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {Route.CONFIG: _get_ipv4_test_routes()},
}
})
)
eth1_state = copy.deepcopy(ETH1_INTERFACE_STATE)
eth1_state[Interface.IPV4] = {'enabled': False}
libnmstate.apply({
Interface.KEY: [eth1_state],
})
libnmstate.apply({Interface.KEY: [eth1_state]})
cur_state = libnmstate.show()
_assert_routes([], cur_state)
@ -400,20 +401,24 @@ def test_disable_ipv4_with_routes_in_current(eth1_up):
@parametrize_ip_ver_routes
def test_iface_down_with_routes_in_current(eth1_up, get_routes_func):
libnmstate.apply({
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: get_routes_func()
libnmstate.apply(
{
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {Route.CONFIG: get_routes_func()},
}
})
)
libnmstate.apply({
Interface.KEY: [{
Interface.NAME: 'eth1',
Interface.TYPE: InterfaceType.ETHERNET,
Interface.STATE: InterfaceState.DOWN
}]
})
libnmstate.apply(
{
Interface.KEY: [
{
Interface.NAME: 'eth1',
Interface.TYPE: InterfaceType.ETHERNET,
Interface.STATE: InterfaceState.DOWN,
}
]
}
)
cur_state = libnmstate.show()
_assert_routes([], cur_state)

150
tests/integration/static_ip_address_test.py
View File

@ -46,10 +46,8 @@ def setup_eth1_ipv4(eth1_up):
'state': 'up',
'ipv4': {
'enabled': True,
'address': [
{'ip': IPV4_ADDRESS1, 'prefix-length': 24}
]
}
'address': [{'ip': IPV4_ADDRESS1, 'prefix-length': 24}],
},
}
]
}
@ -66,10 +64,8 @@ def setup_eth1_ipv6(eth1_up):
'state': 'up',
'ipv6': {
'enabled': True,
'address': [
{'ip': IPV6_ADDRESS1, 'prefix-length': 64}
]
}
'address': [{'ip': IPV6_ADDRESS1, 'prefix-length': 64}],
},
}
]
}
@ -101,10 +97,8 @@ def test_add_static_ipv4_with_min_state(eth2_up):
'state': 'up',
'ipv4': {
'enabled': True,
'address': [
{'ip': IPV4_ADDRESS4, 'prefix-length': 24}
]
}
'address': [{'ip': IPV4_ADDRESS4, 'prefix-length': 24}],
},
}
]
}
@ -116,13 +110,7 @@ def test_add_static_ipv4_with_min_state(eth2_up):
def test_remove_static_ipv4(setup_eth1_ipv4):
desired_state = {
INTERFACES: [
{
'name': 'eth1',
'type': 'ethernet',
'ipv4': {
'enabled': False
}
}
{'name': 'eth1', 'type': 'ethernet', 'ipv4': {'enabled': False}}
]
}
@ -140,10 +128,8 @@ def test_edit_static_ipv4_address_and_prefix(setup_eth1_ipv4):
'state': 'up',
'ipv4': {
'enabled': True,
'address': [
{'ip': IPV4_ADDRESS2, 'prefix-length': 30}
]
}
'address': [{'ip': IPV4_ADDRESS2, 'prefix-length': 30}],
},
}
]
}
@ -153,8 +139,9 @@ def test_edit_static_ipv4_address_and_prefix(setup_eth1_ipv4):
assertlib.assert_state(desired_state)
def test_add_ifaces_with_same_static_ipv4_address_in_one_transaction(eth1_up,
eth2_up):
def test_add_ifaces_with_same_static_ipv4_address_in_one_transaction(
eth1_up, eth2_up
):
desired_state = {
INTERFACES: [
{
@ -163,10 +150,8 @@ def test_add_ifaces_with_same_static_ipv4_address_in_one_transaction(eth1_up,
'state': 'up',
'ipv4': {
'enabled': True,
'address': [
{'ip': IPV4_ADDRESS1, 'prefix-length': 24}
]
}
'address': [{'ip': IPV4_ADDRESS1, 'prefix-length': 24}],
},
},
{
'name': 'eth2',
@ -174,11 +159,9 @@ def test_add_ifaces_with_same_static_ipv4_address_in_one_transaction(eth1_up,
'state': 'up',
'ipv4': {
'enabled': True,
'address': [
{'ip': IPV4_ADDRESS1, 'prefix-length': 24}
]
}
}
'address': [{'ip': IPV4_ADDRESS1, 'prefix-length': 24}],
},
},
]
}
@ -187,8 +170,9 @@ def test_add_ifaces_with_same_static_ipv4_address_in_one_transaction(eth1_up,
assertlib.assert_state(desired_state)
def test_add_iface_with_same_static_ipv4_address_to_existing(setup_eth1_ipv4,
eth2_up):
def test_add_iface_with_same_static_ipv4_address_to_existing(
setup_eth1_ipv4, eth2_up
):
desired_state = {
INTERFACES: [
{
@ -197,10 +181,8 @@ def test_add_iface_with_same_static_ipv4_address_to_existing(setup_eth1_ipv4,
'state': 'up',
'ipv4': {
'enabled': True,
'address': [
{'ip': IPV4_ADDRESS1, 'prefix-length': 24}
]
}
'address': [{'ip': IPV4_ADDRESS1, 'prefix-length': 24}],
},
}
]
}
@ -230,7 +212,7 @@ def test_add_static_ipv6_with_link_local(eth1_up):
eth1_desired_state['ipv6']['enabled'] = True
eth1_desired_state['ipv6']['address'] = [
{'ip': IPV6_LINK_LOCAL_ADDRESS1, 'prefix-length': 64},
{'ip': IPV6_ADDRESS1, 'prefix-length': 64}
{'ip': IPV6_ADDRESS1, 'prefix-length': 64},
]
libnmstate.apply(desired_state)
@ -238,10 +220,14 @@ def test_add_static_ipv6_with_link_local(eth1_up):
# Make sure only the link local address got ignored.
cur_state = statelib.show_only(('eth1',))
eth1_cur_state = cur_state[INTERFACES][0]
assert (eth1_desired_state['ipv6']['address'][0] not in
eth1_cur_state['ipv6']['address'])
assert (eth1_desired_state['ipv6']['address'][1] in
eth1_cur_state['ipv6']['address'])
assert (
eth1_desired_state['ipv6']['address'][0]
not in eth1_cur_state['ipv6']['address']
)
assert (
eth1_desired_state['ipv6']['address'][1]
in eth1_cur_state['ipv6']['address']
)
def test_add_static_ipv6_with_link_local_only(eth1_up):
@ -259,10 +245,14 @@ def test_add_static_ipv6_with_link_local_only(eth1_up):
# Make sure the link local address got ignored.
cur_state = statelib.show_only(('eth1',))
eth1_cur_state = cur_state[INTERFACES][0]
assert (eth1_desired_state['ipv6']['address'][0] not in
eth1_cur_state['ipv6']['address'])
assert (eth1_desired_state['ipv6']['address'][1] not in
eth1_cur_state['ipv6']['address'])
assert (
eth1_desired_state['ipv6']['address'][0]
not in eth1_cur_state['ipv6']['address']
)
assert (
eth1_desired_state['ipv6']['address'][1]
not in eth1_cur_state['ipv6']['address']
)
def test_add_static_ipv6_with_no_address(eth1_up):
@ -288,10 +278,8 @@ def test_add_static_ipv6_with_min_state(eth2_up):
'state': 'up',
'ipv6': {
'enabled': True,
'address': [
{'ip': IPV6_ADDRESS1, 'prefix-length': 24}
]
}
'address': [{'ip': IPV6_ADDRESS1, 'prefix-length': 24}],
},
}
]
}
@ -304,13 +292,7 @@ def test_add_static_ipv6_with_min_state(eth2_up):
def test_disable_static_ipv6(setup_eth1_ipv6):
desired_state = {
INTERFACES: [
{
'name': 'eth1',
'type': 'ethernet',
'ipv6': {
'enabled': False
}
}
{'name': 'eth1', 'type': 'ethernet', 'ipv6': {'enabled': False}}
]
}
@ -329,10 +311,8 @@ def test_edit_static_ipv6_address_and_prefix(setup_eth1_ipv6):
'state': 'up',
'ipv6': {
'enabled': True,
'address': [
{'ip': IPV6_ADDRESS2, 'prefix-length': 24}
]
}
'address': [{'ip': IPV6_ADDRESS2, 'prefix-length': 24}],
},
}
]
}
@ -343,15 +323,20 @@ def test_edit_static_ipv6_address_and_prefix(setup_eth1_ipv6):
eth1_current_state = current_state[INTERFACES][0]
assert (eth1_desired_state['ipv6']['address'][0] in
eth1_current_state['ipv6']['address'])
assert (
eth1_desired_state['ipv6']['address'][0]
in eth1_current_state['ipv6']['address']
)
assert (eth1_setup['ipv6']['address'][0] not in
eth1_current_state['ipv6']['address'])
assert (
eth1_setup['ipv6']['address'][0]
not in eth1_current_state['ipv6']['address']
)
def test_add_ifaces_with_same_static_ipv6_address_in_one_transaction(eth1_up,
eth2_up):
def test_add_ifaces_with_same_static_ipv6_address_in_one_transaction(
eth1_up, eth2_up
):
desired_state = {
INTERFACES: [
{
@ -360,10 +345,8 @@ def test_add_ifaces_with_same_static_ipv6_address_in_one_transaction(eth1_up,
'state': 'up',
'ipv6': {
'enabled': True,
'address': [
{'ip': IPV6_ADDRESS1, 'prefix-length': 64}
]
}
'address': [{'ip': IPV6_ADDRESS1, 'prefix-length': 64}],
},
},
{
'name': 'eth2',
@ -371,11 +354,9 @@ def test_add_ifaces_with_same_static_ipv6_address_in_one_transaction(eth1_up,
'state': 'up',
'ipv6': {
'enabled': True,
'address': [
{'ip': IPV6_ADDRESS1, 'prefix-length': 64}
]
}
}
'address': [{'ip': IPV6_ADDRESS1, 'prefix-length': 64}],
},
},
]
}
@ -384,8 +365,9 @@ def test_add_ifaces_with_same_static_ipv6_address_in_one_transaction(eth1_up,
assertlib.assert_state(desired_state)
def test_add_iface_with_same_static_ipv6_address_to_existing(setup_eth1_ipv6,
eth2_up):
def test_add_iface_with_same_static_ipv6_address_to_existing(
setup_eth1_ipv6, eth2_up
):
desired_state = {
INTERFACES: [
{
@ -394,10 +376,8 @@ def test_add_iface_with_same_static_ipv6_address_to_existing(setup_eth1_ipv6,
'state': 'up',
'ipv6': {
'enabled': True,
'address': [
{'ip': IPV6_ADDRESS1, 'prefix-length': 64}
]
}
'address': [{'ip': IPV6_ADDRESS1, 'prefix-length': 64}],
},
}
]
}

2
tests/integration/testlib/cmd.py
View File

@ -40,7 +40,7 @@ def exec_cmd(cmd, env=None, stdin=None):
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=env
env=env,
)
out, err = p.communicate(stdin)

14
tests/integration/testlib/iproutelib.py
View File

@ -39,9 +39,12 @@ def ip_monitor_assert_stable_link_up(dev, timeout=10):
def wrapper_ip_monitor(*args, **kwargs):
with ip_monitor('link', dev, timeout) as result:
func(*args, **kwargs)
assert len(get_non_up_events(result, dev)) == 0, ('result: ' +
result.out)
assert len(get_non_up_events(result, dev)) == 0, (
'result: ' + result.out
)
return wrapper_ip_monitor
return decorator
@ -58,7 +61,7 @@ def ip_monitor(object_type, dev, timeout=10):
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=None
env=None,
)
result.out, result.err = result.popen.communicate(None)
result.out = result.out.decode('utf-8')
@ -81,8 +84,9 @@ def get_non_up_events(result, dev):
:param result: IpMonitorResult
:return: List of non UP events
"""
return [l for l in result.out.split('\n')
if 'state UP' not in l and dev in l]
return [
l for l in result.out.split('\n') if 'state UP' not in l and dev in l
]
@contextmanager

47
tests/integration/testlib/statelib.py
View File

@ -31,9 +31,7 @@ def show_only(ifnames):
"""
Report the current state, filtering based on the given interface names.
"""
base_filter_state = {
INTERFACES: [{'name': ifname} for ifname in ifnames]
}
base_filter_state = {INTERFACES: [{'name': ifname} for ifname in ifnames]}
current_state = State(libnmstate.show())
current_state.filter(base_filter_state)
return current_state.state
@ -66,8 +64,9 @@ class State(object):
mentioned in the based_on_state.
In case there are no entities for filtering, all are reported.
"""
base_iface_names = {ifstate['name']
for ifstate in based_on_state[INTERFACES]}
base_iface_names = {
ifstate['name'] for ifstate in based_on_state[INTERFACES]
}
if not base_iface_names:
return
@ -89,7 +88,8 @@ class State(object):
for base_iface_state in self._state[INTERFACES]:
ifname = base_iface_state['name']
other_iface_state = _lookup_iface_state_by_name(
other_interfaces_state, ifname)
other_interfaces_state, ifname
)
if other_iface_state is not None:
iface_state = _dict_update(base_iface_state, other_iface_state)
other_iface_state.update(iface_state)
@ -112,7 +112,8 @@ class State(object):
def remove_absent_entries(self):
self._state[INTERFACES] = [
ifstate for ifstate in self._state[INTERFACES]
ifstate
for ifstate in self._state[INTERFACES]
if ifstate.get('state') != 'absent'
]
@ -123,7 +124,8 @@ class State(object):
def _sort_iface_bridge_ports(self):
for ifstate in self._state[INTERFACES]:
ifstate.get('bridge', {}).get('port', []).sort(
key=itemgetter('name'))
key=itemgetter('name')
)
def _ipv6_skeleton_canonicalization(self):
for iface_state in self._state.get(INTERFACES, []):
@ -143,15 +145,17 @@ class State(object):
def _ignore_ipv6_link_local(self):
for iface_state in self._state.get(INTERFACES, []):
iface_state['ipv6']['address'] = list(
addr for addr in iface_state['ipv6']['address']
if not _is_ipv6_link_local(addr['ip'],
addr['prefix-length']))
addr
for addr in iface_state['ipv6']['address']
if not _is_ipv6_link_local(addr['ip'], addr['prefix-length'])
)
def _sort_ip_addresses(self):
for iface_state in self._state.get(INTERFACES, []):
for family in ('ipv4', 'ipv6'):
iface_state.get(family, {}).get('address', []).sort(
key=itemgetter('ip'))
key=itemgetter('ip')
)
def _ignore_dhcp_manual_addr(self):
for iface_state in self._state.get(INTERFACES, []):
@ -163,11 +167,15 @@ class State(object):
for iface_state in self._state.get(INTERFACES, []):
for family in ('ipv4', 'ipv6'):
ip = iface_state.get(family, {})
if not (ip.get('enabled') and
(ip.get('dhcp') or ip.get('autoconf'))):
for dhcp_option in ('auto-routes',
'auto-gateway',
'auto-dns'):
if not (
ip.get('enabled')
and (ip.get('dhcp') or ip.get('autoconf'))
):
for dhcp_option in (
'auto-routes',
'auto-gateway',
'auto-dns',
):
ip.pop(dhcp_option, None)
@ -201,8 +209,9 @@ def filter_current_state(desired_state):
In case there are no entities for filtering, all are reported.
"""
current_state = libnmstate.show()
desired_iface_names = {ifstate['name']
for ifstate in desired_state[INTERFACES]}
desired_iface_names = {
ifstate['name'] for ifstate in desired_state[INTERFACES]
}
if not desired_iface_names:
return current_state

56
tests/integration/vlan_test.py
View File

@ -37,21 +37,14 @@ TWO_VLANS_STATE = {
'name': VLAN_IFNAME,
'type': 'vlan',
'state': 'up',
'vlan': {
'id': 101,
'base-iface': 'eth1'
}
'vlan': {'id': 101, 'base-iface': 'eth1'},
},
{
'name': VLAN2_IFNAME,
'type': 'vlan',
'state': 'up',
'vlan': {
'id': 102,
'base-iface': 'eth1'
}
}
'vlan': {'id': 102, 'base-iface': 'eth1'},
},
]
}
@ -97,20 +90,17 @@ def test_rollback_for_vlans(eth1_up):
with pytest.raises(NmstateVerificationError):
libnmstate.apply(desired_state)
time.sleep(5) # Give some time for NetworkManager to rollback
time.sleep(5) # Give some time for NetworkManager to rollback
current_state_after_apply = libnmstate.show()
assert current_state == current_state_after_apply
def test_set_vlan_iface_down(eth1_up):
with vlan_interface(VLAN_IFNAME, 101):
libnmstate.apply({
libnmstate.apply(
{
INTERFACES: [
{
'name': VLAN_IFNAME,
'type': 'vlan',
'state': 'down'
}
{'name': VLAN_IFNAME, 'type': 'vlan', 'state': 'down'}
]
}
)
@ -127,10 +117,7 @@ def vlan_interface(ifname, vlan_id):
'name': ifname,
'type': 'vlan',
'state': 'up',
'vlan': {
'id': vlan_id,
'base-iface': 'eth1'
}
'vlan': {'id': vlan_id, 'base-iface': 'eth1'},
}
]
}
@ -138,15 +125,8 @@ def vlan_interface(ifname, vlan_id):
try:
yield desired_state
finally:
libnmstate.apply({
INTERFACES: [
{
'name': ifname,
'type': 'vlan',
'state': 'absent'
}
]
}
libnmstate.apply(
{INTERFACES: [{'name': ifname, 'type': 'vlan', 'state': 'absent'}]}
)
@ -157,19 +137,11 @@ def two_vlans_on_eth1():
try:
yield desired_state
finally:
libnmstate.apply({
libnmstate.apply(
{
INTERFACES: [
{
'name': VLAN_IFNAME,
'type': 'vlan',
'state': 'absent'
},
{
'name': VLAN2_IFNAME,
'type': 'vlan',
'state': 'absent'
}
{'name': VLAN_IFNAME, 'type': 'vlan', 'state': 'absent'},
{'name': VLAN2_IFNAME, 'type': 'vlan', 'state': 'absent'},
]
}
)

4
tox.ini
View File

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