tests: ensure proper cleanup of auxiliary process

Various qrexec tests create auxiliary process (service_proc) as a local
variable. In case of test failure, process cleanup isn't called and may
lead to FD leaks and breaking subsequent tests.

Fix this by always saving such process instance in self.service_proc and
cleaning it up in self.tearDown() (this code is already there).

Add also waiting (and in case of timeout - killing) of a service call
process too.

qubes/tests/integ/vm_qrexec_gui.py

@@ -422,7 +422,7 @@ class TC_00_AppVMMixin(object):
                     "dd of=/dev/null bs=993 count=10000 iflag=fullblock; "
                     "wait", stdin=pipe1_r, stdout=pipe2_w))
 
-            service_proc = self.loop.run_until_complete(self.testvm2.run_service(
+            self.service_proc = self.loop.run_until_complete(self.testvm2.run_service(
                 "test.write", stdin=pipe2_r, stdout=pipe1_w))
         finally:
             os.close(pipe1_r)
@@ -432,17 +432,12 @@ class TC_00_AppVMMixin(object):
 
         try:
             self.loop.run_until_complete(
-                asyncio.wait_for(service_proc.wait(), timeout=10))
+                asyncio.wait_for(self.service_proc.wait(), timeout=10))
         except asyncio.TimeoutError:
             self.fail("Timeout, probably deadlock")
         else:
-            self.assertEqual(service_proc.returncode, 0,
+            self.assertEqual(self.service_proc.returncode, 0,
                 "Service call failed")
-        finally:
-            try:
-                service_proc.terminate()
-            except ProcessLookupError:
-                pass
 
     def test_072_qrexec_to_dom0_simultaneous_write(self):
         """Test for simultaneous write in dom0(src)<-VM(dst) connection
@@ -474,8 +469,9 @@ class TC_00_AppVMMixin(object):
                     "dd of=/dev/null bs=993 count=10000 iflag=fullblock; ",
                     stdin=pipe1_r, stdout=pipe2_w))
 
-            service_proc = self.loop.run_until_complete(self.testvm2.run_service(
-                "test.write", stdin=pipe2_r, stdout=pipe1_w))
+            self.service_proc = self.loop.run_until_complete(
+                self.testvm2.run_service(
+                    "test.write", stdin=pipe2_r, stdout=pipe1_w))
         finally:
             os.close(pipe1_r)
             os.close(pipe1_w)
@@ -484,17 +480,12 @@ class TC_00_AppVMMixin(object):
 
         try:
             self.loop.run_until_complete(
-                asyncio.wait_for(service_proc.wait(), timeout=10))
+                asyncio.wait_for(self.service_proc.wait(), timeout=10))
         except asyncio.TimeoutError:
             self.fail("Timeout, probably deadlock")
         else:
-            self.assertEqual(service_proc.returncode, 0,
+            self.assertEqual(self.service_proc.returncode, 0,
                 "Service call failed")
-        finally:
-            try:
-                service_proc.terminate()
-            except ProcessLookupError:
-                pass
 
     def test_080_qrexec_service_argument_allow_default(self):
         """Qrexec service call with argument"""
@@ -595,7 +586,7 @@ class TC_00_AppVMMixin(object):
         """Basic test socket services (dom0) - data receive"""
         self.loop.run_until_complete(self.testvm1.start())
 
-        service_proc = self.loop.run_until_complete(
+        self.service_proc = self.loop.run_until_complete(
             asyncio.create_subprocess_shell(
                 'socat -u UNIX-LISTEN:/etc/qubes-rpc/test.Socket,mode=666 -',
                 stdout=subprocess.PIPE, stdin=subprocess.PIPE))
@@ -616,7 +607,7 @@ class TC_00_AppVMMixin(object):
         try:
             (service_stdout, service_stderr) = self.loop.run_until_complete(
                 asyncio.wait_for(
-                    service_proc.communicate(),
+                    self.service_proc.communicate(),
                     timeout=10))
         except asyncio.TimeoutError:
             self.fail(
@@ -636,8 +627,9 @@ class TC_00_AppVMMixin(object):
 
         self.create_local_file('/tmp/service-input', TEST_DATA.decode())
 
-        service_proc = self.loop.run_until_complete(asyncio.create_subprocess_shell(
-            'socat -u OPEN:/tmp/service-input UNIX-LISTEN:/etc/qubes-rpc/test.Socket,mode=666'))
+        self.service_proc = self.loop.run_until_complete(
+            asyncio.create_subprocess_shell(
+                'socat -u OPEN:/tmp/service-input UNIX-LISTEN:/etc/qubes-rpc/test.Socket,mode=666'))
 
         try:
             with self.qrexec_policy('test.Socket', self.testvm1, '@adminvm'):
@@ -655,7 +647,7 @@ class TC_00_AppVMMixin(object):
         try:
             (service_stdout, service_stderr) = self.loop.run_until_complete(
                 asyncio.wait_for(
-                    service_proc.communicate(),
+                    self.service_proc.communicate(),
                     timeout=10))
         except asyncio.TimeoutError:
             self.fail(
@@ -814,7 +806,7 @@ class TC_00_AppVMMixin(object):
             '/tmp/service-input',
             TEST_DATA.decode())
 
-        service_proc = self.loop.run_until_complete(self.testvm1.run(
+        self.service_proc = self.loop.run_until_complete(self.testvm1.run(
             'socat -u OPEN:/tmp/service-input UNIX-LISTEN:/etc/qubes-rpc/test.Socket,mode=666',
             user='root'))
 
@@ -834,7 +826,7 @@ class TC_00_AppVMMixin(object):
         try:
             (service_stdout, service_stderr) = self.loop.run_until_complete(
                 asyncio.wait_for(
-                    service_proc.communicate(),
+                    self.service_proc.communicate(),
                     timeout=10))
         except asyncio.TimeoutError:
             self.fail(
@@ -867,7 +859,7 @@ class TC_00_AppVMMixin(object):
             'time.sleep(15)\n'
         )
 
-        service_proc = self.loop.run_until_complete(self.testvm1.run(
+        self.service_proc = self.loop.run_until_complete(self.testvm1.run(
             'python3 /tmp/service_script',
             stdout=subprocess.PIPE, stdin=subprocess.PIPE,
             user='root'))
@@ -881,8 +873,11 @@ class TC_00_AppVMMixin(object):
             stdout = self.loop.run_until_complete(asyncio.wait_for(p.stdout.read(),
                 timeout=10))
         except asyncio.TimeoutError:
+            p.terminate()
             self.fail(
                 "service timeout, probably EOF wasn't transferred from the VM process")
+        finally:
+            self.loop.run_until_complete(p.wait())
 
         self.assertEqual(stdout,
                 b'test\n',
@@ -933,8 +928,11 @@ class TC_00_AppVMMixin(object):
                 self.service_proc.stdout.read(),
                 timeout=10))
         except asyncio.TimeoutError:
+            p.terminate()
             self.fail(
                 "service timeout, probably EOF wasn't transferred to the VM process")
+        finally:
+            self.loop.run_until_complete(p.wait())
 
         service_descriptor = b'test.Socket+ dom0\0'
         self.assertEqual(service_stdout, service_descriptor + b'test1test2',