Expand source code
@pytest.mark.p0
@pytest.mark.networks
@pytest.mark.virtualmachines
class TestVMNetwork:
@pytest.mark.dependency(name="add_vlan")
def test_add_vlan(
self, api_client, ssh_keypair, vm_mgmt_static, vm_checker, vm_shell_from_host, vm_network,
minimal_vm, gen_ifconfig
):
# clean cloud-init for rerun, and get the correct ifname
(unique_vm_name, ssh_user), (_, pri_key) = minimal_vm, ssh_keypair
vm_got_ips, (code, data) = vm_checker.wait_ip_addresses(unique_vm_name, ['default'])
assert vm_got_ips, (
f"Failed to Start VM({unique_vm_name}) with errors:\n"
f"Status: {data.get('status')}\n"
f"API Status({code}): {data}"
)
vm_ip = next(iface['ipAddress'] for iface in data['status']['interfaces']
if iface['name'] == 'default')
code, data = api_client.hosts.get(data['status']['nodeName'])
host_ip = next(addr['address'] for addr in data['status']['addresses']
if addr['type'] == 'InternalIP')
with vm_shell_from_host(host_ip, vm_ip, ssh_user, pkey=pri_key) as sh:
cloud_inited, (out, err) = vm_checker.wait_cloudinit_done(sh)
assert cloud_inited, (out, err)
out, err = sh.exec_command("sudo cloud-init clean")
out, err = sh.exec_command("sudo cloud-init status")
assert "not run" in out, (out, err)
out, err = sh.exec_command("ip --json a s")
assert not err
ifname = next(i['ifname'] for i in json.loads(out) if i['link_type'] != 'loopback')
# https://cloudinit.readthedocs.io/en/22.4.2/topics/network-config-format-v1.html#subnet-ip
# and https://harvesterhci.io/kb/multiple-nics-vm-connectivity/#cloud-init-config
nic_config = [gen_ifconfig(ifname, idx=i) for i in range(2)]
nic_config[0]['subnets'] = [vm_mgmt_static]
# add vlan NIC and network data then restart VM
code, data = api_client.vms.get(unique_vm_name)
vm_spec = api_client.vms.Spec.from_dict(data)
vm_spec.add_network('nic-1', f"{vm_network['namespace']}/{vm_network['name']}")
vm_spec.network_data = "#cloud-config\n" + yaml.dump({
"version": 1,
"config": nic_config
})
code, data = api_client.vms.update(unique_vm_name, vm_spec)
assert 200 == code, (code, data)
vm_restarted, ctx = vm_checker.wait_restarted(unique_vm_name)
assert vm_restarted, (
f"Failed to Restart VM({unique_vm_name}),"
f" timed out while executing {ctx.callee!r}"
)
vm_got_ips, (code, data) = vm_checker.wait_ip_addresses(unique_vm_name, ['default'])
assert vm_got_ips, (
f"Failed to Start VM({unique_vm_name}) with errors:\n"
f"Status: {data.get('status')}\n"
f"API Status({code}): {data}"
)
vm_ip = next(iface['ipAddress'] for iface in data['status']['interfaces']
if iface['name'] == 'default')
code, data = api_client.hosts.get(data['status']['nodeName'])
host_ip = next(addr['address'] for addr in data['status']['addresses']
if addr['type'] == 'InternalIP')
with vm_shell_from_host(host_ip, vm_ip, ssh_user, pkey=pri_key) as sh:
cloud_inited, (out, err) = vm_checker.wait_cloudinit_done(sh)
assert cloud_inited and not err, (out, err)
out, err = sh.exec_command("ip --json -4 a s")
ips = [j['local'] for i in json.loads(out) for j in i['addr_info']]
vlan_ip_range = ip_network(vm_network['cidr'])
def get_vlan_ip(ctx):
if ctx.callee == 'vm.get_status':
return all(iface.get('ipAddress') for iface in ctx.data['status']['interfaces']
if iface['name'] != 'default')
return True
# ???: status data from API will have delay a bit
vm_got_ips, (code, data) = vm_checker.wait_interfaces(unique_vm_name, callback=get_vlan_ip)
assert vm_got_ips, (code, data)
vm_vlan_ip = next(iface['ipAddress'] for iface in data['status']['interfaces']
if iface['name'] != 'default')
assert ip_address(vm_vlan_ip) in vlan_ip_range and vm_vlan_ip in ips
@pytest.mark.dependency(depends=["add_vlan"])
def test_ssh_connection(
self, api_client, ssh_keypair, vm_checker, vm_network, minimal_vm
):
(unique_vm_name, ssh_user), (_, pri_key) = minimal_vm, ssh_keypair
vm_started, (code, data) = vm_checker.wait_interfaces(unique_vm_name)
assert vm_started, (
f"Failed to Start VM({unique_vm_name}) with errors:\n"
f"Status: {data.get('status')}\n"
f"API Status({code}): {data}"
)
vm_ip = next(iface['ipAddress'] for iface in data['status']['interfaces']
if iface['name'] != 'default')
try:
with vm_checker.wait_ssh_connected(vm_ip, ssh_user, pkey=pri_key) as sh:
out, err = sh.exec_command("ip -brief a s")
assert vm_ip in out and not err
except AssertionError as ex:
raise ex
except Exception as ex:
raise AssertionError(
f"Unable to login to VM via VLAN IP {vm_ip}"
) from ex
def test_vms_on_same_vlan(
self, api_client, ssh_keypair, vm_checker, vm_network, two_mirror_vms
):
_, pri_key = ssh_keypair
vm_names, ssh_user = two_mirror_vms
def get_vlan_ip(ctx):
if ctx.callee == 'vm.get_status':
return all(iface.get('ipAddress') for iface in ctx.data['status']['interfaces']
if iface['name'] != 'default')
return True
# Verify VM having IP which belongs to VLAN
vm_info, vlan_ip_range = [], ip_network(vm_network['cidr'])
for vm_name in vm_names:
vm_got_ips, (code, data) = vm_checker.wait_interfaces(vm_name, callback=get_vlan_ip)
assert vm_got_ips, (
f"Failed to Start VM({vm_name}) with errors:\n"
f"Status: {data.get('status')}\n"
f"API Status({code}): {data}"
)
vm_ip = next(iface['ipAddress'] for iface in data['status']['interfaces']
if iface['name'] != 'default')
assert ip_address(vm_ip) in vlan_ip_range
vm_info.append((vm_name, vm_ip))
# verify Ping from each
for (src_name, src_ip), (dst_name, dst_ip) in zip(vm_info, vm_info[::-1]):
try:
with vm_checker.wait_ssh_connected(src_ip, ssh_user, pkey=pri_key) as sh:
out, err = sh.exec_command(f"ping -c5 {dst_ip}")
assert '100% packet loss' not in out, (
f"Failed to ping VM({dst_name!r}, {dst_ip}) <- VM({src_name!r}, {src_ip})"
)
except AssertionError as ex:
raise ex
except Exception as ex:
raise AssertionError(
f"Unable to login to VM via VLAN IP {src_ip}"
) from ex