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

API: Promote the public API to libnmstate level

Browse Source 
Promoting these public classes and functions to libnmstate module top level:
 * The 'show()' function of 'netinfo.py'.
 * The 'apply()', 'commit', 'rollback' functions of 'netapplier.py'.
 * The exception classes defined in 'error.py' as exposed as
   'libnmstate.error'
 * The 'PrettyState' class defined in 'prettystate.py' which is used by
   nmstatectl.
 * The schema classes are exposed as 'libnmstate.schema'.

The integration test cases and nmstatctl have been updated to use this
new public interface.

Signed-off-by: Gris Ge <fge@redhat.com>
This commit is contained in:
Gris Ge 2019-06-13 20:39:42 +08:00
parent 4f11dccd4f
commit fa12bcd9a2
17 changed files with 166 additions and 162 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
libnmstate/__init__.py
View File

@ -16,5 +16,16 @@
#
from __future__ import absolute_import
from . import validator
validator
from . import error
from . import schema
from .netapplier import apply
from .netapplier import commit
from .netapplier import rollback
from .netinfo import show
from .prettystate import PrettyState
__all__ = ['show', 'apply', 'commit', 'rollback', 'error', 'schema',
'PrettyState']

37
nmstatectl/nmstatectl.py
View File

@ -29,13 +29,12 @@ import tempfile
from six.moves import input
import yaml
from libnmstate import netapplier
from libnmstate import netinfo
import libnmstate
from libnmstate import PrettyState
from libnmstate.error import NmstateConflictError
from libnmstate.error import NmstatePermissionError
from libnmstate.error import NmstateValueError
from libnmstate.prettystate import PrettyState
from libnmstate.schema import Constants
from libnmstate.schema import Interface
from libnmstate.schema import Route
@ -76,7 +75,7 @@ def setup_subcommand_edit(subparsers):
parser_edit.add_argument('--json', help='Edit as JSON', default=True,
action='store_false', dest='yaml')
parser_edit.add_argument(
'only', default='*', nargs='?', metavar=Constants.INTERFACES,
'only', default='*', nargs='?', metavar=Interface.KEY,
help='Edit only specified interfaces (comma-separated)'
)
parser_edit.add_argument(
@ -123,23 +122,23 @@ def setup_subcommand_show(subparsers):
parser_show.add_argument('--json', help='Edit as JSON', default=True,
action='store_false', dest='yaml')
parser_show.add_argument(
'only', default='*', nargs='?', metavar=Constants.INTERFACES,
'only', default='*', nargs='?', metavar=Interface.KEY,
help='Show only specified interfaces (comma-separated)'
)
def commit(args):
try:
netapplier.commit(args.checkpoint)
libnmstate.commit(args.checkpoint)
except NmstateValueError as e:
print("ERROR committing change: {}\n".format(str(e)))
return os.EX_DATAERR
def edit(args):
state = _filter_state(netinfo.show(), args.only)
state = _filter_state(libnmstate.show(), args.only)
if not state[Constants.INTERFACES]:
if not state[Interface.KEY]:
sys.stderr.write('ERROR: No such interface\n')
return os.EX_USAGE
@ -159,19 +158,19 @@ def edit(args):
print('Applying the following state: ')
print_state(new_state, use_yaml=args.yaml)
netapplier.apply(new_state, verify_change=args.verify)
libnmstate.apply(new_state, verify_change=args.verify)
def rollback(args):
try:
netapplier.rollback(args.checkpoint)
libnmstate.rollback(args.checkpoint)
except NmstateValueError as e:
print("ERROR rolling back change: {}\n".format(str(e)))
return os.EX_DATAERR
def show(args):
state = _filter_state(netinfo.show(), args.only)
state = _filter_state(libnmstate.show(), args.only)
print_state(state, use_yaml=args.yaml)
@ -204,7 +203,7 @@ def apply_state(statedata, verify_change, commit, timeout):
use_yaml = True
try:
checkpoint = netapplier.apply(state, verify_change, commit, timeout)
checkpoint = libnmstate.apply(state, verify_change, commit, timeout)
except NmstatePermissionError as e:
sys.stderr.write('ERROR: Missing permissions:{}\n'.format(str(e)))
return os.EX_NOPERM
@ -222,8 +221,8 @@ def apply_state(statedata, verify_change, commit, timeout):
def _filter_state(state, whitelist):
if whitelist != '*':
patterns = [p for p in whitelist.split(',')]
state[Constants.INTERFACES] = _filter_interfaces(state, patterns)
state[Constants.ROUTES] = _filter_routes(state, patterns)
state[Interface.KEY] = _filter_interfaces(state, patterns)
state[Route.KEY] = _filter_routes(state, patterns)
return state
@ -234,7 +233,7 @@ def _filter_interfaces(state, patterns):
"""
showinterfaces = []
for interface in state[Constants.INTERFACES]:
for interface in state[Interface.KEY]:
for pattern in patterns:
if fnmatch.fnmatch(interface['name'], pattern):
showinterfaces.append(interface)
@ -291,9 +290,9 @@ def _parse_state(txtstate, parse_yaml):
except ValueError as e:
error = 'Invalid JSON syntax: %s\n' % e
if not error and Constants.INTERFACES not in state:
if not error and Interface.KEY not in state:
# Allow editing routes only.
state[Constants.INTERFACES] = []
state[Interface.KEY] = []
return state, error
@ -334,7 +333,7 @@ def _filter_routes(state, patterns):
Route.RUNNING: []
}
for route_type in (Route.RUNNING, Route.CONFIG):
for route in state.get(Constants.ROUTES, {}).get(route_type, []):
for route in state.get(Route.KEY, {}).get(route_type, []):
for pattern in patterns:
if fnmatch.fnmatch(route[Route.NEXT_HOP_INTERFACE], pattern):
routes[route_type].append(route)

12
tests/ctl/nmstatectl_test.py
View File

@ -68,7 +68,7 @@ interfaces:
@mock.patch('sys.argv', ['nmstatectl', 'set', 'mystate.json'])
@mock.patch.object(nmstatectl.netapplier, 'apply',
@mock.patch.object(nmstatectl.libnmstate, 'apply',
lambda state, verify_change, commit, timeout: None)
@mock.patch.object(nmstatectl, 'open', mock.mock_open(read_data='{}'),
create=True)
@ -77,14 +77,14 @@ def test_run_ctl_directly_set():
@mock.patch('sys.argv', ['nmstatectl', 'show'])
@mock.patch.object(nmstatectl.netinfo, 'show', lambda: {})
@mock.patch.object(nmstatectl.libnmstate, 'show', lambda: {})
def test_run_ctl_directly_show_empty():
nmstatectl.main()
@mock.patch('sys.argv', ['nmstatectl', 'show', 'non_existing_interface'])
@mock.patch.object(
nmstatectl.netinfo, 'show', lambda: json.loads(LO_JSON_STATE))
nmstatectl.libnmstate, 'show', lambda: json.loads(LO_JSON_STATE))
@mock.patch('nmstatectl.nmstatectl.sys.stdout', new_callable=six.StringIO)
def test_run_ctl_directly_show_only_empty(mock_stdout):
nmstatectl.main()
@ -93,7 +93,7 @@ def test_run_ctl_directly_show_only_empty(mock_stdout):
@mock.patch('sys.argv', ['nmstatectl', 'show', 'lo'])
@mock.patch.object(
nmstatectl.netinfo, 'show', lambda: json.loads(LO_JSON_STATE))
nmstatectl.libnmstate, 'show', lambda: json.loads(LO_JSON_STATE))
@mock.patch('nmstatectl.nmstatectl.sys.stdout', new_callable=six.StringIO)
def test_run_ctl_directly_show_only(mock_stdout):
nmstatectl.main()
@ -103,7 +103,7 @@ def test_run_ctl_directly_show_only(mock_stdout):
@mock.patch('sys.argv', ['nmstatectl', 'show', '--json',
'non_existing_interface'])
@mock.patch.object(
nmstatectl.netinfo, 'show', lambda: json.loads(LO_JSON_STATE))
nmstatectl.libnmstate, 'show', lambda: json.loads(LO_JSON_STATE))
@mock.patch('nmstatectl.nmstatectl.sys.stdout', new_callable=six.StringIO)
def test_run_ctl_directly_show_json_only_empty(mock_stdout):
nmstatectl.main()
@ -112,7 +112,7 @@ def test_run_ctl_directly_show_json_only_empty(mock_stdout):
@mock.patch('sys.argv', ['nmstatectl', 'show', '--json', 'lo'])
@mock.patch.object(
nmstatectl.netinfo, 'show', lambda: json.loads(LO_JSON_STATE))
nmstatectl.libnmstate, 'show', lambda: json.loads(LO_JSON_STATE))
@mock.patch('nmstatectl.nmstatectl.sys.stdout', new_callable=six.StringIO)
def test_run_ctl_directly_show_json_only(mock_stdout):
nmstatectl.main()

37
tests/integration/bond_test.py
View File

@ -20,8 +20,7 @@ import pytest
import time
import yaml
from libnmstate import netapplier
from libnmstate import netinfo
import libnmstate
from libnmstate.error import NmstateVerificationError
from libnmstate.schema import Interface
@ -58,7 +57,7 @@ def setup_remove_bond99():
}
]
}
netapplier.apply(remove_bond)
libnmstate.apply(remove_bond)
@contextmanager
@ -77,11 +76,11 @@ def bond_interface(name, slaves):
]
}
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
try:
yield desired_state
finally:
netapplier.apply({
libnmstate.apply({
INTERFACES: [
{
'name': name,
@ -95,13 +94,13 @@ def bond_interface(name, slaves):
def test_add_and_remove_bond_with_two_slaves(eth1_up, eth2_up):
state = yaml.load(BOND99_YAML_BASE, Loader=yaml.SafeLoader)
netapplier.apply(state)
libnmstate.apply(state)
assertlib.assert_state(state)
state[INTERFACES][0]['state'] = 'absent'
netapplier.apply(state)
libnmstate.apply(state)
state = statelib.show_only((state[INTERFACES][0]['name'],))
assert not state[INTERFACES]
@ -109,7 +108,7 @@ def test_add_and_remove_bond_with_two_slaves(eth1_up, eth2_up):
def test_remove_bond_with_minimum_desired_state(eth1_up, eth2_up):
state = yaml.load(BOND99_YAML_BASE, Loader=yaml.SafeLoader)
netapplier.apply(state)
libnmstate.apply(state)
remove_bond_state = {
INTERFACES: [
@ -120,7 +119,7 @@ def test_remove_bond_with_minimum_desired_state(eth1_up, eth2_up):
}
]
}
netapplier.apply(remove_bond_state)
libnmstate.apply(remove_bond_state)
state = statelib.show_only((state[INTERFACES][0]['name'],))
assert not state[INTERFACES]
@ -154,13 +153,13 @@ def test_add_bond_with_slaves_and_ipv4(eth1_up, eth2_up, setup_remove_bond99):
]
}
netapplier.apply(desired_bond_state)
libnmstate.apply(desired_bond_state)
assertlib.assert_state(desired_bond_state)
def test_rollback_for_bond(eth1_up, eth2_up):
current_state = netinfo.show()
current_state = libnmstate.show()
desired_state = {
INTERFACES: [
{
@ -186,11 +185,11 @@ def test_rollback_for_bond(eth1_up, eth2_up):
desired_state[INTERFACES][0]['invalid_key'] = 'foo'
with pytest.raises(NmstateVerificationError):
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
time.sleep(5)
current_state_after_apply = netinfo.show()
current_state_after_apply = libnmstate.show()
assert current_state[INTERFACES] == current_state_after_apply[INTERFACES]
@ -199,7 +198,7 @@ def test_add_slave_to_bond_without_slaves(eth1_up):
with bond_interface(name='bond99', slaves=[]) as bond_state:
bond_state[INTERFACES][0]['link-aggregation']['slaves'] = ['eth1']
netapplier.apply(bond_state)
libnmstate.apply(bond_state)
current_state = statelib.show_only(('bond99',))
bond99_cur_state = current_state[INTERFACES][0]
@ -213,7 +212,7 @@ def test_remove_all_slaves_from_bond(eth1_up):
with bond_interface(name='bond99', slaves=['eth1']) as bond_state:
bond_state[INTERFACES][0]['link-aggregation']['slaves'] = []
netapplier.apply(bond_state)
libnmstate.apply(bond_state)
current_state = statelib.show_only(('bond99',))
bond99_cur_state = current_state[INTERFACES][0]
@ -226,7 +225,7 @@ def test_replace_bond_slave(eth1_up, eth2_up):
with bond_interface(name='bond99', slaves=['eth1']) as bond_state:
bond_state[INTERFACES][0]['link-aggregation']['slaves'] = ['eth2']
netapplier.apply(bond_state)
libnmstate.apply(bond_state)
current_state = statelib.show_only(('bond99',))
bond99_cur_state = current_state[INTERFACES][0]
@ -239,7 +238,7 @@ def test_remove_one_of_the_bond_slaves(eth1_up, eth2_up):
bond_state[INTERFACES][0]['link-aggregation']['slaves'] = ['eth2']
netapplier.apply(bond_state)
libnmstate.apply(bond_state)
current_state = statelib.show_only(('bond99',))
bond99_cur_state = current_state[INTERFACES][0]
@ -250,14 +249,14 @@ def test_remove_one_of_the_bond_slaves(eth1_up, eth2_up):
def test_set_bond_mac_address(eth1_up):
with bond_interface(name='bond99', slaves=['eth1']) as bond_state:
bond_state[Interface.KEY][0][Interface.MAC] = MAC0
netapplier.apply(bond_state)
libnmstate.apply(bond_state)
current_state = statelib.show_only(('bond99',))
bond99_cur_state = current_state[INTERFACES][0]
assert bond99_cur_state[Interface.MAC] == MAC0.upper()
bond_state[Interface.KEY][0][Interface.MAC] = MAC1
netapplier.apply(bond_state)
libnmstate.apply(bond_state)
current_state = statelib.show_only(('bond99',))
bond99_cur_state = current_state[INTERFACES][0]

4
tests/integration/conftest.py
View File

@ -19,7 +19,7 @@ import logging
import pytest
from libnmstate import netapplier
import libnmstate
from .testlib import statelib
from .testlib.statelib import INTERFACES
@ -67,4 +67,4 @@ def _set_eth_admin_state(ifname, state):
# FIXME: On most systems, IPv6 cannot be disabled by Nmstate/NM.
if state == 'up':
desired_state[INTERFACES][0].update({'ipv6': {'enabled': True}})
netapplier.apply(desired_state)
libnmstate.apply(desired_state)

53
tests/integration/dhcp_test.py
View File

@ -19,8 +19,7 @@ import time
import pytest
from libnmstate import netapplier
from libnmstate import netinfo
import libnmstate
from libnmstate.schema import Constants
from libnmstate.schema import DNS
from libnmstate.schema import Route as RT
@ -132,7 +131,7 @@ def setup_remove_bond99():
}
]
}
netapplier.apply(remove_bond)
libnmstate.apply(remove_bond)
@pytest.fixture
@ -147,7 +146,7 @@ def setup_remove_dhcpcli():
}
]
}
netapplier.apply(remove_bond)
libnmstate.apply(remove_bond)
def test_ipv4_dhcp(dhcp_env):
@ -161,7 +160,7 @@ def test_ipv4_dhcp(dhcp_env):
dhcp_cli_desired_state['ipv4']['auto-gateway'] = True
dhcp_cli_desired_state['ipv6']['enabled'] = True
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
time.sleep(5) # wait to get resolv.conf updated
assert _has_ipv4_dhcp_nameserver()
@ -180,7 +179,7 @@ def test_ipv6_dhcp_only(dhcp_env):
dhcp_cli_desired_state['ipv6']['auto-dns'] = True
dhcp_cli_desired_state['ipv6']['auto-gateway'] = True
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
time.sleep(5) # libnm does not wait on ipv6-ra or DHCPv6.
@ -208,7 +207,7 @@ def test_ipv6_dhcp_and_autoconf(dhcp_env):
dhcp_cli_desired_state['ipv6']['auto-gateway'] = True
dhcp_cli_desired_state['ipv6']['auto-routes'] = True
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
time.sleep(5) # libnm does not wait on ipv6-ra or DHCPv6.
@ -224,7 +223,7 @@ def test_ipv6_autoconf_only(dhcp_env):
dhcp_cli_desired_state['ipv6']['dhcp'] = False
dhcp_cli_desired_state['ipv6']['autoconf'] = True
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
def test_dhcp_with_addresses(dhcp_env):
@ -255,7 +254,7 @@ def test_dhcp_with_addresses(dhcp_env):
]
}
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
@ -284,7 +283,7 @@ def test_dhcp_for_bond_with_ip_address_and_slave(dhcp_env,
]
}
netapplier.apply(desired_state, verify_change=False)
libnmstate.apply(desired_state, verify_change=False)
# Long story for why we doing 'dhcp=False' with 'verify_change=False'
# above:
# For `dhcp=False`:
@ -298,7 +297,7 @@ def test_dhcp_for_bond_with_ip_address_and_slave(dhcp_env,
# interface will result in `state:down`. As a workaround the
# verification is ignored.
desired_state[INTERFACES][0]['ipv4']['dhcp'] = True
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
@ -314,7 +313,7 @@ def test_ipv4_dhcp_ignore_gateway(dhcp_env):
dhcp_cli_desired_state['ipv4']['auto-dns'] = True
dhcp_cli_desired_state['ipv6']['enabled'] = True
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
time.sleep(5) # wait to get resolv.conf updated
@ -333,7 +332,7 @@ def test_ipv4_dhcp_ignore_dns(dhcp_env):
dhcp_cli_desired_state['ipv4']['auto-gateway'] = True
dhcp_cli_desired_state['ipv6']['enabled'] = True
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
assert not _has_ipv4_dhcp_nameserver()
@ -352,7 +351,7 @@ def test_ipv4_dhcp_ignore_routes(dhcp_env):
dhcp_cli_desired_state['ipv4']['auto-dns'] = True
dhcp_cli_desired_state['ipv6']['enabled'] = True
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
time.sleep(5) # wait to get resolv.conf updated
@ -371,7 +370,7 @@ def test_ipv6_dhcp_and_autoconf_ignore_gateway(dhcp_env):
dhcp_cli_desired_state['ipv6']['auto-gateway'] = False
dhcp_cli_desired_state['ipv6']['auto-routes'] = True
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
time.sleep(5) # libnm does not wait on ipv6-ra or DHCPv6.
@ -390,7 +389,7 @@ def test_ipv6_dhcp_and_autoconf_ignore_dns(dhcp_env):
dhcp_cli_desired_state['ipv6']['auto-gateway'] = True
dhcp_cli_desired_state['ipv6']['auto-routes'] = True
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
time.sleep(5) # libnm does not wait on ipv6-ra or DHCPv6.
@ -409,7 +408,7 @@ def test_ipv6_dhcp_and_autoconf_ignore_routes(dhcp_env):
dhcp_cli_desired_state['ipv6']['auto-gateway'] = True
dhcp_cli_desired_state['ipv6']['auto-routes'] = False
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
time.sleep(5) # libnm does not wait on ipv6-ra or DHCPv6.
@ -429,7 +428,7 @@ def test_ipv4_dhcp_off_and_option_on(dhcp_env):
dhcp_cli_desired_state['ipv4']['auto-gateway'] = False
dhcp_cli_desired_state['ipv6']['enabled'] = True
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
dhcp_cli_current_state = statelib.show_only((DHCP_CLI_NIC,))[INTERFACES][0]
assert not dhcp_cli_current_state['ipv4']['dhcp']
@ -452,7 +451,7 @@ def test_ipv6_dhcp_off_and_option_on(dhcp_env):
dhcp_cli_desired_state['ipv6']['auto-dns'] = False
dhcp_cli_desired_state['ipv6']['auto-gateway'] = False
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
dhcp_cli_current_state = statelib.show_only((DHCP_CLI_NIC,))[INTERFACES][0]
assert not dhcp_cli_current_state['ipv6']['dhcp']
@ -475,7 +474,7 @@ def test_ipv4_dhcp_switch_on_to_off(dhcp_env):
dhcp_cli_desired_state['ipv4']['auto-routes'] = True
dhcp_cli_desired_state['ipv6']['enabled'] = True
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
time.sleep(5) # wait to get resolv.conf updated
assert _has_ipv4_dhcp_nameserver()
@ -490,7 +489,7 @@ def test_ipv4_dhcp_switch_on_to_off(dhcp_env):
dhcp_cli_desired_state['ipv4']['dhcp'] = False
dhcp_cli_desired_state['ipv6']['enabled'] = True
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
assert not _has_ipv4_dhcp_nameserver()
assert not _has_ipv4_dhcp_gateway()
@ -507,7 +506,7 @@ def test_ipv6_dhcp_switch_on_to_off(dhcp_env):
dhcp_cli_desired_state['ipv6']['auto-dns'] = True
dhcp_cli_desired_state['ipv6']['auto-routes'] = True
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
time.sleep(5) # libnm does not wait on ipv6-ra or DHCPv6.
@ -522,7 +521,7 @@ def test_ipv6_dhcp_switch_on_to_off(dhcp_env):
dhcp_cli_desired_state['ipv6']['dhcp'] = False
dhcp_cli_desired_state['ipv6']['autoconf'] = False
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
assert not _has_ipv6_auto_gateway()
@ -585,7 +584,7 @@ def test_slave_ipaddr_learned_via_dhcp_added_as_static_to_linux_bridge(
]
}
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
current_state = statelib.show_only(('dhcpcli',))
client_current_state = current_state[INTERFACES][0]
@ -630,7 +629,7 @@ def test_slave_ipaddr_learned_via_dhcp_added_as_static_to_linux_bridge(
]
}
netapplier.apply(bridge_desired_state)
libnmstate.apply(bridge_desired_state)
assertlib.assert_state(bridge_desired_state)
@ -638,7 +637,7 @@ def _get_nameservers():
"""
Return a list of name server string configured in RESOLV_CONF_PATH.
"""
return netinfo.show().get(
return libnmstate.show().get(
Constants.DNS, {}).get(DNS.RUNNING, {}).get(DNS.SERVER, [])
@ -646,7 +645,7 @@ def _get_running_routes():
"""
return a list of running routes
"""
return netinfo.show().get(Constants.ROUTES, {}).get(RT.RUNNING, [])
return libnmstate.show().get(Constants.ROUTES, {}).get(RT.RUNNING, [])
def _has_ipv6_auto_gateway():

18
tests/integration/ethernet_mtu_test.py
View File

@ -21,7 +21,7 @@ import time
import pytest
import jsonschema as js
from libnmstate import netapplier
import libnmstate
from libnmstate.error import NmstateVerificationError
from .testlib import assertlib
@ -39,7 +39,7 @@ def test_increase_iface_mtu():
eth1_desired_state = desired_state[INTERFACES][0]
eth1_desired_state['mtu'] = 1900
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
@ -49,7 +49,7 @@ def test_decrease_iface_mtu():
eth1_desired_state = desired_state[INTERFACES][0]
eth1_desired_state['mtu'] = 1400
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
@ -59,7 +59,7 @@ def test_upper_limit_jambo_iface_mtu():
eth1_desired_state = desired_state[INTERFACES][0]
eth1_desired_state['mtu'] = 9000
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
@ -69,7 +69,7 @@ def test_increase_more_than_jambo_iface_mtu():
eth1_desired_state = desired_state[INTERFACES][0]
eth1_desired_state['mtu'] = 10000
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
@ -81,7 +81,7 @@ def test_decrease_to_zero_iface_mtu():
eth1_desired_state['mtu'] = 0
with pytest.raises(NmstateVerificationError) as err:
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assert '-mtu: 0' in err.value.args[0]
# FIXME: Drop the sleep when the waiting logic is implemented.
time.sleep(2)
@ -95,7 +95,7 @@ def test_decrease_to_negative_iface_mtu():
eth1_desired_state['mtu'] = -1
with pytest.raises(js.ValidationError) as err:
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assert '-1' in err.value.args[0]
assertlib.assert_state(origin_desired_state)
@ -105,7 +105,7 @@ def test_decrease_to_ipv6_min_ethernet_frame_size_iface_mtu():
eth1_desired_state = desired_state[INTERFACES][0]
eth1_desired_state['mtu'] = 1280
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
@ -117,7 +117,7 @@ def test_decrease_to_lower_than_min_ipv6_iface_mtu():
eth1_desired_state['mtu'] = 1279
with pytest.raises(NmstateVerificationError) as err:
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assert '1279' in err.value.args[0]
# FIXME: Drop the sleep when the waiting logic is implemented.
time.sleep(2)

8
tests/integration/iface_admin_state_test.py
View File

@ -17,7 +17,7 @@
import pytest
from libnmstate import netapplier
import libnmstate
from .testlib import assertlib
from .testlib.statelib import INTERFACES
@ -33,10 +33,10 @@ def test_set_a_down_iface_down(eth1_up):
}
]
}
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
@ -53,6 +53,6 @@ def test_removing_a_non_removable_iface(eth1_up):
]
}
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)

6
tests/integration/linux_bridge_test.py
View File

@ -20,7 +20,7 @@ from contextlib import contextmanager
import pytest
import yaml
from libnmstate import netapplier
import libnmstate
from libnmstate.schema import Interface
from libnmstate.schema import InterfaceState
from libnmstate.schema import InterfaceType
@ -134,12 +134,12 @@ def _linux_bridge(name, bridge_state):
if bridge_state:
desired_state[INTERFACES][0][LinuxBridge.CONFIG_SUBTREE] = bridge_state
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
try:
yield desired_state
finally:
netapplier.apply({
libnmstate.apply({
INTERFACES: [
{
Interface.NAME: name,

14
tests/integration/ovs_test.py
View File

@ -17,7 +17,7 @@
import yaml
from libnmstate import netapplier
import libnmstate
from .testlib import statelib
from .testlib.statelib import INTERFACES
@ -45,7 +45,7 @@ def test_create_and_remove_ovs_bridge_with_a_system_port(eth1_up):
'type': 'system'
}
]
netapplier.apply(state)
libnmstate.apply(state)
setup_remove_ovs_bridge_state = {
INTERFACES: [
@ -56,7 +56,7 @@ def test_create_and_remove_ovs_bridge_with_a_system_port(eth1_up):
}
]
}
netapplier.apply(setup_remove_ovs_bridge_state)
libnmstate.apply(setup_remove_ovs_bridge_state)
state = statelib.show_only((state[INTERFACES][0]['name'],))
assert not state[INTERFACES]
@ -71,7 +71,7 @@ def test_create_and_remove_ovs_bridge_with_min_desired_state():
}
]
}
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
setup_remove_ovs_bridge_state = {
INTERFACES: [
@ -82,7 +82,7 @@ def test_create_and_remove_ovs_bridge_with_min_desired_state():
}
]
}
netapplier.apply(setup_remove_ovs_bridge_state)
libnmstate.apply(setup_remove_ovs_bridge_state)
state = statelib.show_only((desired_state[INTERFACES][0]['name'],))
assert not state[INTERFACES]
@ -111,7 +111,7 @@ def test_create_and_remove_ovs_bridge_with_an_internal_port():
},
}
state[INTERFACES].append(ovs_internal_interface_state)
netapplier.apply(state, verify_change=False)
libnmstate.apply(state, verify_change=False)
setup_remove_ovs_bridge_state_and_port = {
INTERFACES: [
@ -127,7 +127,7 @@ def test_create_and_remove_ovs_bridge_with_an_internal_port():
}
]
}
netapplier.apply(setup_remove_ovs_bridge_state_and_port)
libnmstate.apply(setup_remove_ovs_bridge_state_and_port)
state = statelib.show_only(
(state[INTERFACES][0]['name'], state[INTERFACES][1]['name']))
assert not state[INTERFACES]

6
tests/integration/preserve_ip_config_test.py
View File

@ -17,7 +17,7 @@
from contextlib import contextmanager
from libnmstate import netapplier
import libnmstate
from libnmstate.nm import nmclient
from .testlib import statelib
@ -34,7 +34,7 @@ IPV6_ADDRESS1 = '2001:db8:1::1'
def test_reapply_preserve_ip_config(eth1_up):
netapplier.apply(
libnmstate.apply(
{
'interfaces': [
{
@ -71,7 +71,7 @@ def test_reapply_preserve_ip_config(eth1_up):
for key, value in ((_IPV4_EXTRA_CONFIG, _IPV4_EXTRA_VALUE),
(_IPV6_EXTRA_CONFIG, _IPV6_EXTRA_VALUE)):
with _extra_ip_config(uuid, key, value):
netapplier.apply(cur_state)
libnmstate.apply(cur_state)
_assert_extra_ip_config(uuid, key, value)

45
tests/integration/route_test.py
View File

@ -18,8 +18,7 @@ import copy
import pytest
from libnmstate import netapplier
from libnmstate import netinfo
import libnmstate
from libnmstate.error import NmstateNotImplementedError
from libnmstate.schema import Interface
from libnmstate.schema import InterfaceState
@ -60,25 +59,25 @@ ETH1_INTERFACE_STATE = {
def test_add_static_routes(eth1_up):
routes = _get_ipv4_test_routes() + _get_ipv6_test_routes()
netapplier.apply({
libnmstate.apply({
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: routes
}
})
cur_state = netinfo.show()
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]]
netapplier.apply({
libnmstate.apply({
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: routes
}
})
cur_state = netinfo.show()
cur_state = libnmstate.show()
_assert_routes(routes, cur_state)
@ -87,14 +86,14 @@ def test_add_route_without_metric(eth1_up):
for route in routes:
del route[Route.METRIC]
netapplier.apply({
libnmstate.apply({
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: routes
}
})
cur_state = netinfo.show()
cur_state = libnmstate.show()
_assert_routes(routes, cur_state)
@ -103,21 +102,21 @@ def test_add_route_without_table_id(eth1_up):
for route in routes:
del route[Route.TABLE_ID]
netapplier.apply({
libnmstate.apply({
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: routes
},
})
cur_state = netinfo.show()
cur_state = libnmstate.show()
_assert_routes(routes, cur_state)
@pytest.mark.xfail(raises=NmstateNotImplementedError,
reason="Red Hat Bug 1707396")
def test_multiple_gateway(eth1_up):
netapplier.apply(
libnmstate.apply(
{
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
@ -263,18 +262,18 @@ parametrize_ip_ver_routes = pytest.mark.parametrize(
@parametrize_ip_ver_routes
def test_remove_specific_route(eth1_up, get_routes_func):
routes = get_routes_func()
netapplier.apply({
libnmstate.apply({
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: routes
},
})
cur_state = netinfo.show()
cur_state = libnmstate.show()
_assert_routes(routes, cur_state)
absent_route = routes[0]
absent_route[Route.STATE] = Route.STATE_ABSENT
netapplier.apply({
libnmstate.apply({
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: [absent_route]
@ -283,27 +282,27 @@ def test_remove_specific_route(eth1_up, get_routes_func):
expected_routes = routes[1:]
cur_state = netinfo.show()
cur_state = libnmstate.show()
_assert_routes(expected_routes, cur_state)
@parametrize_ip_ver_routes
def test_remove_wildcast_route_with_iface(eth1_up, get_routes_func):
routes = get_routes_func()
netapplier.apply({
libnmstate.apply({
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: routes
},
})
cur_state = netinfo.show()
cur_state = libnmstate.show()
_assert_routes(routes, cur_state)
absent_route = {
Route.STATE: Route.STATE_ABSENT,
Route.NEXT_HOP_INTERFACE: 'eth1'
}
netapplier.apply({
libnmstate.apply({
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: [absent_route]
@ -312,20 +311,20 @@ def test_remove_wildcast_route_with_iface(eth1_up, get_routes_func):
expected_routes = []
cur_state = netinfo.show()
cur_state = libnmstate.show()
_assert_routes(expected_routes, cur_state)
@parametrize_ip_ver_routes
def test_remove_wildcast_route_without_iface(eth1_up, get_routes_func):
routes = get_routes_func()
netapplier.apply({
libnmstate.apply({
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: routes
},
})
cur_state = netinfo.show()
cur_state = libnmstate.show()
_assert_routes(routes, cur_state)
absent_routes = []
@ -334,7 +333,7 @@ def test_remove_wildcast_route_without_iface(eth1_up, get_routes_func):
Route.STATE: Route.STATE_ABSENT,
Route.DESTINATION: route[Route.DESTINATION]
})
netapplier.apply({
libnmstate.apply({
Interface.KEY: [ETH1_INTERFACE_STATE],
Route.KEY: {
Route.CONFIG: absent_routes
@ -343,5 +342,5 @@ def test_remove_wildcast_route_without_iface(eth1_up, get_routes_func):
expected_routes = []
cur_state = netinfo.show()
cur_state = libnmstate.show()
_assert_routes(expected_routes, cur_state)

36
tests/integration/static_ip_address_test.py
View File

@ -17,7 +17,7 @@
import pytest
from libnmstate import netapplier
import libnmstate
from .testlib import assertlib
from .testlib import statelib
@ -53,7 +53,7 @@ def setup_eth1_ipv4(eth1_up):
}
]
}
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
@pytest.fixture
@ -73,7 +73,7 @@ def setup_eth1_ipv6(eth1_up):
}
]
}
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
return desired_state
@ -87,7 +87,7 @@ def test_add_static_ipv4_with_full_state(eth1_up):
eth1_desired_state['ipv4']['address'] = [
{'ip': IPV4_ADDRESS3, 'prefix-length': 24}
]
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
@ -108,7 +108,7 @@ def test_add_static_ipv4_with_min_state(eth2_up):
}
]
}
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
@ -126,7 +126,7 @@ def test_remove_static_ipv4(setup_eth1_ipv4):
]
}
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
@ -148,7 +148,7 @@ def test_edit_static_ipv4_address_and_prefix(setup_eth1_ipv4):
]
}
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
@ -182,7 +182,7 @@ def test_add_ifaces_with_same_static_ipv4_address_in_one_transaction(eth1_up,
]
}
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
@ -204,7 +204,7 @@ def test_add_iface_with_same_static_ipv4_address_to_existing(setup_eth1_ipv4,
}
]
}
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
@ -219,7 +219,7 @@ def test_add_static_ipv6_with_full_state(eth1_up):
# This sequence is intentionally made for IP address sorting.
{'ip': IPV6_ADDRESS1, 'prefix-length': 64},
]
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
@ -233,7 +233,7 @@ def test_add_static_ipv6_with_link_local(eth1_up):
{'ip': IPV6_ADDRESS1, 'prefix-length': 64}
]
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
# Make sure only the link local address got ignored.
cur_state = statelib.show_only(('eth1',))
@ -254,7 +254,7 @@ def test_add_static_ipv6_with_link_local_only(eth1_up):
{'ip': IPV6_LINK_LOCAL_ADDRESS2, 'prefix-length': 64},
]
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
# Make sure the link local address got ignored.
cur_state = statelib.show_only(('eth1',))
@ -271,7 +271,7 @@ def test_add_static_ipv6_with_no_address(eth1_up):
eth1_desired_state['state'] = 'up'
eth1_desired_state['ipv6']['enabled'] = True
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
cur_state = statelib.show_only(('eth1',))
eth1_cur_state = cur_state[INTERFACES][0]
@ -295,7 +295,7 @@ def test_add_static_ipv6_with_min_state(eth2_up):
}
]
}
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
@ -314,7 +314,7 @@ def test_disable_static_ipv6(setup_eth1_ipv6):
]
}
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
@ -337,7 +337,7 @@ def test_edit_static_ipv6_address_and_prefix(setup_eth1_ipv6):
]
}
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
eth1_desired_state = desired_state[INTERFACES][0]
current_state = statelib.show_only(('eth1',))
@ -379,7 +379,7 @@ def test_add_ifaces_with_same_static_ipv6_address_in_one_transaction(eth1_up,
]
}
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)
@ -401,6 +401,6 @@ def test_add_iface_with_same_static_ipv6_address_to_existing(setup_eth1_ipv6,
}
]
}
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
assertlib.assert_state(desired_state)

4
tests/integration/testlib/assertlib.py
View File

@ -16,7 +16,7 @@
#
import copy
from libnmstate import netinfo
import libnmstate
from libnmstate.schema import DNS
from libnmstate.schema import Route
@ -27,7 +27,7 @@ from .statelib import INTERFACES
def assert_state(desired_state_data):
"""Given a state, assert it against the current state."""
current_state_data = netinfo.show()
current_state_data = libnmstate.show()
# Ignore route and dns for assert check as the check are done in the test
# case code.
current_state_data.pop(Route.KEY, None)

8
tests/integration/testlib/examplelib.py
View File

@ -18,11 +18,9 @@
from contextlib import contextmanager
import os
import yaml
from libnmstate import netapplier
import libnmstate
PATH_MAX = 4096
@ -36,12 +34,12 @@ def example_state(initial, cleanup=None):
desired_state = load_example(initial)
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
try:
yield desired_state
finally:
if cleanup:
netapplier.apply(load_example(cleanup))
libnmstate.apply(load_example(cleanup))
def load_example(name):

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

@ -20,7 +20,7 @@ import copy
from operator import itemgetter
import six
from libnmstate import netinfo
import libnmstate
from libnmstate.schema import Constants
@ -34,7 +34,7 @@ def show_only(ifnames):
base_filter_state = {
INTERFACES: [{'name': ifname} for ifname in ifnames]
}
current_state = State(netinfo.show())
current_state = State(libnmstate.show())
current_state.filter(base_filter_state)
return current_state.state
@ -200,7 +200,7 @@ def filter_current_state(desired_state):
in the desired state.
In case there are no entities for filtering, all are reported.
"""
current_state = netinfo.show()
current_state = libnmstate.show()
desired_iface_names = {ifstate['name']
for ifstate in desired_state[INTERFACES]}

19
tests/integration/vlan_test.py
View File

@ -20,8 +20,7 @@ import time
import pytest
from libnmstate import netapplier
from libnmstate import netinfo
import libnmstate
from libnmstate.schema import Interface
from libnmstate.error import NmstateVerificationError
@ -91,21 +90,21 @@ def test_add_and_remove_two_vlans_on_same_iface(eth1_up):
def test_rollback_for_vlans(eth1_up):
current_state = netinfo.show()
current_state = libnmstate.show()
desired_state = TWO_VLANS_STATE
desired_state[INTERFACES][1]['invalid_key'] = 'foo'
with pytest.raises(NmstateVerificationError):
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
time.sleep(5) # Give some time for NetworkManager to rollback
current_state_after_apply = netinfo.show()
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):
netapplier.apply({
libnmstate.apply({
INTERFACES: [
{
'name': VLAN_IFNAME,
@ -135,11 +134,11 @@ def vlan_interface(ifname, vlan_id):
}
]
}
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
try:
yield desired_state
finally:
netapplier.apply({
libnmstate.apply({
INTERFACES: [
{
'name': ifname,
@ -154,11 +153,11 @@ def vlan_interface(ifname, vlan_id):
@contextmanager
def two_vlans_on_eth1():
desired_state = TWO_VLANS_STATE
netapplier.apply(desired_state)
libnmstate.apply(desired_state)
try:
yield desired_state
finally:
netapplier.apply({
libnmstate.apply({
INTERFACES: [
{
'name': VLAN_IFNAME,