tests: port network tests to core3 API
This commit is contained in:
		
							parent
							
								
									df5562d2c7
								
							
						
					
					
						commit
						422f30b969
					
				@ -797,7 +797,7 @@ def load_tests(loader, tests, pattern): # pylint: disable=unused-argument
 | 
				
			|||||||
            # integration tests
 | 
					            # integration tests
 | 
				
			||||||
            'qubes.tests.int.basic',
 | 
					            'qubes.tests.int.basic',
 | 
				
			||||||
            'qubes.tests.int.dom0_update',
 | 
					            'qubes.tests.int.dom0_update',
 | 
				
			||||||
#           'qubes.tests.network',
 | 
					            'qubes.tests.int.network',
 | 
				
			||||||
#           'qubes.tests.vm_qrexec_gui',
 | 
					#           'qubes.tests.vm_qrexec_gui',
 | 
				
			||||||
#           'qubes.tests.backup',
 | 
					#           'qubes.tests.backup',
 | 
				
			||||||
#           'qubes.tests.backupcompatibility',
 | 
					#           'qubes.tests.backupcompatibility',
 | 
				
			||||||
 | 
				
			|||||||
@ -30,9 +30,8 @@ import subprocess
 | 
				
			|||||||
import unittest
 | 
					import unittest
 | 
				
			||||||
import time
 | 
					import time
 | 
				
			||||||
 | 
					
 | 
				
			||||||
from qubes.qubes import QubesVmCollection, defaults
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
import qubes.tests
 | 
					import qubes.tests
 | 
				
			||||||
 | 
					import qubes.vm.appvm
 | 
				
			||||||
 | 
					
 | 
				
			||||||
class NcVersion:
 | 
					class NcVersion:
 | 
				
			||||||
    Trad = 1
 | 
					    Trad = 1
 | 
				
			||||||
@ -54,16 +53,18 @@ class VmNetworkingMixin(qubes.tests.SystemTestsMixin):
 | 
				
			|||||||
 | 
					
 | 
				
			||||||
    def setUp(self):
 | 
					    def setUp(self):
 | 
				
			||||||
        super(VmNetworkingMixin, self).setUp()
 | 
					        super(VmNetworkingMixin, self).setUp()
 | 
				
			||||||
        self.testnetvm = self.qc.add_new_vm("QubesNetVm",
 | 
					        self.init_default_template(self.template)
 | 
				
			||||||
 | 
					        self.testnetvm = self.app.add_new_vm(qubes.vm.appvm.AppVM,
 | 
				
			||||||
            name=self.make_vm_name('netvm1'),
 | 
					            name=self.make_vm_name('netvm1'),
 | 
				
			||||||
            template=self.qc.get_vm_by_name(self.template))
 | 
					            label='red')
 | 
				
			||||||
        self.testnetvm.create_on_disk(verbose=False)
 | 
					        self.testnetvm.create_on_disk()
 | 
				
			||||||
        self.testvm1 = self.qc.add_new_vm("QubesAppVm",
 | 
					        self.testnetvm.provides_network = True
 | 
				
			||||||
 | 
					        self.testvm1 = self.app.add_new_vm(qubes.vm.appvm.AppVM,
 | 
				
			||||||
            name=self.make_vm_name('vm2'),
 | 
					            name=self.make_vm_name('vm2'),
 | 
				
			||||||
            template=self.qc.get_vm_by_name(self.template))
 | 
					            label='red')
 | 
				
			||||||
        self.testvm1.create_on_disk(verbose=False)
 | 
					        self.testvm1.create_on_disk()
 | 
				
			||||||
        self.testvm1.netvm = self.testnetvm
 | 
					        self.testvm1.netvm = self.testnetvm
 | 
				
			||||||
        self.qc.save()
 | 
					        self.app.save()
 | 
				
			||||||
 | 
					
 | 
				
			||||||
        self.configure_netvm()
 | 
					        self.configure_netvm()
 | 
				
			||||||
 | 
					
 | 
				
			||||||
@ -93,21 +94,19 @@ class VmNetworkingMixin(qubes.tests.SystemTestsMixin):
 | 
				
			|||||||
 | 
					
 | 
				
			||||||
 | 
					
 | 
				
			||||||
    def test_000_simple_networking(self):
 | 
					    def test_000_simple_networking(self):
 | 
				
			||||||
        self.qc.unlock_db()
 | 
					 | 
				
			||||||
        self.testvm1.start()
 | 
					        self.testvm1.start()
 | 
				
			||||||
        self.assertEqual(self.run_cmd(self.testvm1, self.ping_ip), 0)
 | 
					        self.assertEqual(self.run_cmd(self.testvm1, self.ping_ip), 0)
 | 
				
			||||||
        self.assertEqual(self.run_cmd(self.testvm1, self.ping_name), 0)
 | 
					        self.assertEqual(self.run_cmd(self.testvm1, self.ping_name), 0)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					
 | 
				
			||||||
    def test_010_simple_proxyvm(self):
 | 
					    def test_010_simple_proxyvm(self):
 | 
				
			||||||
        self.proxy = self.qc.add_new_vm("QubesProxyVm",
 | 
					        self.proxy = self.app.add_new_vm(qubes.vm.appvm.AppVM,
 | 
				
			||||||
            name=self.make_vm_name('proxy'),
 | 
					            name=self.make_vm_name('proxy'),
 | 
				
			||||||
            template=self.qc.get_vm_by_name(self.template))
 | 
					            label='red')
 | 
				
			||||||
        self.proxy.create_on_disk(verbose=False)
 | 
					        self.proxy.provides_network = True
 | 
				
			||||||
        self.proxy.netvm = self.testnetvm
 | 
					        self.proxy.netvm = self.testnetvm
 | 
				
			||||||
 | 
					        self.proxy.create_on_disk()
 | 
				
			||||||
        self.testvm1.netvm = self.proxy
 | 
					        self.testvm1.netvm = self.proxy
 | 
				
			||||||
        self.qc.save()
 | 
					 | 
				
			||||||
        self.qc.unlock_db()
 | 
					 | 
				
			||||||
 | 
					
 | 
				
			||||||
        self.testvm1.start()
 | 
					        self.testvm1.start()
 | 
				
			||||||
        self.assertTrue(self.proxy.is_running())
 | 
					        self.assertTrue(self.proxy.is_running())
 | 
				
			||||||
@ -124,15 +123,15 @@ class VmNetworkingMixin(qubes.tests.SystemTestsMixin):
 | 
				
			|||||||
    @unittest.skipUnless(spawn.find_executable('xdotool'),
 | 
					    @unittest.skipUnless(spawn.find_executable('xdotool'),
 | 
				
			||||||
                         "xdotool not installed")
 | 
					                         "xdotool not installed")
 | 
				
			||||||
    def test_020_simple_proxyvm_nm(self):
 | 
					    def test_020_simple_proxyvm_nm(self):
 | 
				
			||||||
        self.proxy = self.qc.add_new_vm("QubesProxyVm",
 | 
					        self.proxy = self.app.add_new_vm("QubesProxyVm",
 | 
				
			||||||
            name=self.make_vm_name('proxy'),
 | 
					            name=self.make_vm_name('proxy'),
 | 
				
			||||||
            template=self.qc.get_vm_by_name(self.template))
 | 
					            label='red')
 | 
				
			||||||
        self.proxy.create_on_disk(verbose=False)
 | 
					        self.proxy.provides_network = True
 | 
				
			||||||
 | 
					        self.proxy.create_on_disk()
 | 
				
			||||||
        self.proxy.netvm = self.testnetvm
 | 
					        self.proxy.netvm = self.testnetvm
 | 
				
			||||||
        self.proxy.services['network-manager'] = True
 | 
					        self.proxy.services['network-manager'] = True
 | 
				
			||||||
        self.testvm1.netvm = self.proxy
 | 
					        self.testvm1.netvm = self.proxy
 | 
				
			||||||
        self.qc.save()
 | 
					        self.app.save()
 | 
				
			||||||
        self.qc.unlock_db()
 | 
					 | 
				
			||||||
 | 
					
 | 
				
			||||||
        self.testvm1.start()
 | 
					        self.testvm1.start()
 | 
				
			||||||
        self.assertTrue(self.proxy.is_running())
 | 
					        self.assertTrue(self.proxy.is_running())
 | 
				
			||||||
@ -169,14 +168,14 @@ class VmNetworkingMixin(qubes.tests.SystemTestsMixin):
 | 
				
			|||||||
 | 
					
 | 
				
			||||||
 | 
					
 | 
				
			||||||
    def test_030_firewallvm_firewall(self):
 | 
					    def test_030_firewallvm_firewall(self):
 | 
				
			||||||
        self.proxy = self.qc.add_new_vm("QubesProxyVm",
 | 
					        self.proxy = self.app.add_new_vm(qubes.vm.appvm.AppVM,
 | 
				
			||||||
            name=self.make_vm_name('proxy'),
 | 
					            name=self.make_vm_name('proxy'),
 | 
				
			||||||
            template=self.qc.get_vm_by_name(self.template))
 | 
					            label='red')
 | 
				
			||||||
        self.proxy.create_on_disk(verbose=False)
 | 
					        self.proxy.provides_network = True
 | 
				
			||||||
 | 
					        self.proxy.create_on_disk()
 | 
				
			||||||
        self.proxy.netvm = self.testnetvm
 | 
					        self.proxy.netvm = self.testnetvm
 | 
				
			||||||
        self.testvm1.netvm = self.proxy
 | 
					        self.testvm1.netvm = self.proxy
 | 
				
			||||||
        self.qc.save()
 | 
					        self.app.save()
 | 
				
			||||||
        self.qc.unlock_db()
 | 
					 | 
				
			||||||
 | 
					
 | 
				
			||||||
        if self.run_cmd(self.testnetvm, 'nc -h 2>&1|grep -q nmap.org') == 0:
 | 
					        if self.run_cmd(self.testnetvm, 'nc -h 2>&1|grep -q nmap.org') == 0:
 | 
				
			||||||
            nc_version = NcVersion.Nmap
 | 
					            nc_version = NcVersion.Nmap
 | 
				
			||||||
@ -279,20 +278,19 @@ class VmNetworkingMixin(qubes.tests.SystemTestsMixin):
 | 
				
			|||||||
 | 
					
 | 
				
			||||||
 | 
					
 | 
				
			||||||
    def test_040_inter_vm(self):
 | 
					    def test_040_inter_vm(self):
 | 
				
			||||||
        self.proxy = self.qc.add_new_vm("QubesProxyVm",
 | 
					        self.proxy = self.app.add_new_vm(qubes.vm.appvm.AppVM,
 | 
				
			||||||
            name=self.make_vm_name('proxy'),
 | 
					            name=self.make_vm_name('proxy'),
 | 
				
			||||||
            template=self.qc.get_vm_by_name(self.template))
 | 
					            label='red')
 | 
				
			||||||
        self.proxy.create_on_disk(verbose=False)
 | 
					        self.proxy.create_on_disk()
 | 
				
			||||||
        self.proxy.netvm = self.testnetvm
 | 
					        self.proxy.netvm = self.testnetvm
 | 
				
			||||||
        self.testvm1.netvm = self.proxy
 | 
					        self.testvm1.netvm = self.proxy
 | 
				
			||||||
 | 
					
 | 
				
			||||||
        self.testvm2 = self.qc.add_new_vm("QubesAppVm",
 | 
					        self.testvm2 = self.app.add_new_vm(qubes.vm.appvm.AppVM,
 | 
				
			||||||
            name=self.make_vm_name('vm3'),
 | 
					            name=self.make_vm_name('vm3'),
 | 
				
			||||||
            template=self.qc.get_vm_by_name(self.template))
 | 
					            label='red')
 | 
				
			||||||
        self.testvm2.create_on_disk(verbose=False)
 | 
					        self.testvm2.create_on_disk()
 | 
				
			||||||
        self.testvm2.netvm = self.proxy
 | 
					        self.testvm2.netvm = self.proxy
 | 
				
			||||||
        self.qc.save()
 | 
					        self.app.save()
 | 
				
			||||||
        self.qc.unlock_db()
 | 
					 | 
				
			||||||
 | 
					
 | 
				
			||||||
        self.testvm1.start()
 | 
					        self.testvm1.start()
 | 
				
			||||||
        self.testvm2.start()
 | 
					        self.testvm2.start()
 | 
				
			||||||
@ -316,7 +314,6 @@ class VmNetworkingMixin(qubes.tests.SystemTestsMixin):
 | 
				
			|||||||
 | 
					
 | 
				
			||||||
    def test_050_spoof_ip(self):
 | 
					    def test_050_spoof_ip(self):
 | 
				
			||||||
        """Test if VM IP spoofing is blocked"""
 | 
					        """Test if VM IP spoofing is blocked"""
 | 
				
			||||||
        self.qc.unlock_db()
 | 
					 | 
				
			||||||
        self.testvm1.start()
 | 
					        self.testvm1.start()
 | 
				
			||||||
 | 
					
 | 
				
			||||||
        self.assertEqual(self.run_cmd(self.testvm1, self.ping_ip), 0)
 | 
					        self.assertEqual(self.run_cmd(self.testvm1, self.ping_ip), 0)
 | 
				
			||||||
@ -433,18 +430,18 @@ class VmUpdatesMixin(qubes.tests.SystemTestsMixin):
 | 
				
			|||||||
            self.skipTest("Template {} not supported by this test".format(
 | 
					            self.skipTest("Template {} not supported by this test".format(
 | 
				
			||||||
                self.template))
 | 
					                self.template))
 | 
				
			||||||
 | 
					
 | 
				
			||||||
        self.testvm1 = self.qc.add_new_vm(
 | 
					        self.init_default_template(self.template)
 | 
				
			||||||
            "QubesAppVm",
 | 
					        self.testvm1 = self.app.add_new_vm(
 | 
				
			||||||
 | 
					            qubes.vm.appvm.AppVM,
 | 
				
			||||||
            name=self.make_vm_name('vm1'),
 | 
					            name=self.make_vm_name('vm1'),
 | 
				
			||||||
            template=self.qc.get_vm_by_name(self.template))
 | 
					            label='red')
 | 
				
			||||||
        self.testvm1.create_on_disk(verbose=False)
 | 
					        self.testvm1.create_on_disk()
 | 
				
			||||||
 | 
					
 | 
				
			||||||
    def test_000_simple_update(self):
 | 
					    def test_000_simple_update(self):
 | 
				
			||||||
        self.save_and_reload_db()
 | 
					        self.save_and_reload_db()
 | 
				
			||||||
        self.qc.unlock_db()
 | 
					 | 
				
			||||||
        # reload the VM to have all the properties properly set (especially
 | 
					        # reload the VM to have all the properties properly set (especially
 | 
				
			||||||
        # default netvm)
 | 
					        # default netvm)
 | 
				
			||||||
        self.testvm1 = self.qc[self.testvm1.qid]
 | 
					        self.testvm1 = self.app.domains[self.testvm1.qid]
 | 
				
			||||||
        self.testvm1.start()
 | 
					        self.testvm1.start()
 | 
				
			||||||
        p = self.testvm1.run(self.update_cmd, wait=True, user="root",
 | 
					        p = self.testvm1.run(self.update_cmd, wait=True, user="root",
 | 
				
			||||||
                             passio_popen=True, passio_stderr=True)
 | 
					                             passio_popen=True, passio_stderr=True)
 | 
				
			||||||
@ -591,18 +588,18 @@ class VmUpdatesMixin(qubes.tests.SystemTestsMixin):
 | 
				
			|||||||
            self.skipTest("Template {} not supported by this test".format(
 | 
					            self.skipTest("Template {} not supported by this test".format(
 | 
				
			||||||
                self.template))
 | 
					                self.template))
 | 
				
			||||||
 | 
					
 | 
				
			||||||
        self.netvm_repo = self.qc.add_new_vm(
 | 
					        self.netvm_repo = self.app.add_new_vm(
 | 
				
			||||||
            "QubesNetVm",
 | 
					            qubes.vm.appvm.AppVM,
 | 
				
			||||||
            name=self.make_vm_name('net'),
 | 
					            name=self.make_vm_name('net'),
 | 
				
			||||||
            template=self.qc.get_vm_by_name(self.template))
 | 
					            label='red')
 | 
				
			||||||
        self.netvm_repo.create_on_disk(verbose=False)
 | 
					        self.netvm_repo.provides_network = True
 | 
				
			||||||
 | 
					        self.netvm_repo.create_on_disk()
 | 
				
			||||||
        self.testvm1.netvm = self.netvm_repo
 | 
					        self.testvm1.netvm = self.netvm_repo
 | 
				
			||||||
        # NetVM should have qubes-updates-proxy enabled by default
 | 
					        # NetVM should have qubes-updates-proxy enabled by default
 | 
				
			||||||
        #self.netvm_repo.services['qubes-updates-proxy'] = True
 | 
					        #self.netvm_repo.services['qubes-updates-proxy'] = True
 | 
				
			||||||
        # TODO: consider also adding a test for the template itself
 | 
					        # TODO: consider also adding a test for the template itself
 | 
				
			||||||
        self.testvm1.services['updates-proxy-setup'] = True
 | 
					        self.testvm1.services['updates-proxy-setup'] = True
 | 
				
			||||||
        self.qc.save()
 | 
					        self.app.save()
 | 
				
			||||||
        self.qc.unlock_db()
 | 
					 | 
				
			||||||
 | 
					
 | 
				
			||||||
        # Setup test repo
 | 
					        # Setup test repo
 | 
				
			||||||
        self.netvm_repo.start()
 | 
					        self.netvm_repo.start()
 | 
				
			||||||
@ -640,12 +637,9 @@ class VmUpdatesMixin(qubes.tests.SystemTestsMixin):
 | 
				
			|||||||
 | 
					
 | 
				
			||||||
def load_tests(loader, tests, pattern):
 | 
					def load_tests(loader, tests, pattern):
 | 
				
			||||||
    try:
 | 
					    try:
 | 
				
			||||||
        qc = qubes.qubes.QubesVmCollection()
 | 
					        app = qubes.Qubes()
 | 
				
			||||||
        qc.lock_db_for_reading()
 | 
					        templates = [vm.name for vm in app.domains if
 | 
				
			||||||
        qc.load()
 | 
					                     isinstance(vm, qubes.vm.templatevm.TemplateVM)]
 | 
				
			||||||
        qc.unlock_db()
 | 
					 | 
				
			||||||
        templates = [vm.name for vm in qc.values() if
 | 
					 | 
				
			||||||
                     isinstance(vm, qubes.qubes.QubesTemplateVm)]
 | 
					 | 
				
			||||||
    except OSError:
 | 
					    except OSError:
 | 
				
			||||||
        templates = []
 | 
					        templates = []
 | 
				
			||||||
    for template in templates:
 | 
					    for template in templates:
 | 
				
			||||||
@ -273,6 +273,7 @@ fi
 | 
				
			|||||||
%{python_sitelib}/qubes/tests/int/__init__.py*
 | 
					%{python_sitelib}/qubes/tests/int/__init__.py*
 | 
				
			||||||
%{python_sitelib}/qubes/tests/int/basic.py*
 | 
					%{python_sitelib}/qubes/tests/int/basic.py*
 | 
				
			||||||
%{python_sitelib}/qubes/tests/int/dom0_update.py*
 | 
					%{python_sitelib}/qubes/tests/int/dom0_update.py*
 | 
				
			||||||
 | 
					%{python_sitelib}/qubes/tests/int/network.py*
 | 
				
			||||||
 | 
					
 | 
				
			||||||
%dir %{python_sitelib}/qubes/tests/int/tools
 | 
					%dir %{python_sitelib}/qubes/tests/int/tools
 | 
				
			||||||
%{python_sitelib}/qubes/tests/int/tools/__init__.py*
 | 
					%{python_sitelib}/qubes/tests/int/tools/__init__.py*
 | 
				
			||||||
 | 
				
			|||||||
		Loading…
	
		Reference in New Issue
	
	Block a user