tests/functional: Move the code for testing HTTP downloads to a common function

We are going to use this code in other tests, too, so let's move it
to the qemu_test module to be able to re-use it more easily.

Message-ID: <20250227103915.19795-2-thuth@redhat.com>
Signed-off-by: Thomas Huth <thuth@redhat.com>
This commit is contained in:
Thomas Huth 2025-02-27 11:39:10 +01:00
parent 9813958892
commit 2c92ecb678
2 changed files with 26 additions and 22 deletions

View file

@ -3,8 +3,12 @@
# This work is licensed under the terms of the GNU GPL, version 2 or # This work is licensed under the terms of the GNU GPL, version 2 or
# later. See the COPYING file in the top-level directory. # later. See the COPYING file in the top-level directory.
import hashlib
import urllib.request
from .cmd import wait_for_console_pattern, exec_command_and_wait_for_pattern
from .testcase import QemuSystemTest from .testcase import QemuSystemTest
from .cmd import wait_for_console_pattern from .utils import get_usernet_hostfwd_port
class LinuxKernelTest(QemuSystemTest): class LinuxKernelTest(QemuSystemTest):
@ -26,3 +30,23 @@ class LinuxKernelTest(QemuSystemTest):
self.vm.launch() self.vm.launch()
if wait_for: if wait_for:
self.wait_for_console_pattern(wait_for) self.wait_for_console_pattern(wait_for)
def check_http_download(self, filename, hashsum, guestport=8080,
pythoncmd='python3 -m http.server'):
exec_command_and_wait_for_pattern(self,
f'{pythoncmd} {guestport} & sleep 1',
f'Serving HTTP on 0.0.0.0 port {guestport}')
hl = hashlib.sha256()
hostport = get_usernet_hostfwd_port(self.vm)
url = f'http://localhost:{hostport}{filename}'
self.log.info(f'Downloading {url} ...')
with urllib.request.urlopen(url) as response:
while True:
chunk = response.read(1 << 20)
if not chunk:
break
hl.update(chunk)
digest = hl.hexdigest()
self.log.info(f'sha256sum of download is {digest}.')
self.assertEqual(digest, hashsum)

View file

@ -10,11 +10,7 @@
# This work is licensed under the terms of the GNU GPL, version 2 or # This work is licensed under the terms of the GNU GPL, version 2 or
# later. See the COPYING file in the top-level directory. # later. See the COPYING file in the top-level directory.
import hashlib
import urllib.request
from qemu_test import LinuxKernelTest, Asset, exec_command_and_wait_for_pattern from qemu_test import LinuxKernelTest, Asset, exec_command_and_wait_for_pattern
from qemu_test.utils import get_usernet_hostfwd_port
class IntelIOMMU(LinuxKernelTest): class IntelIOMMU(LinuxKernelTest):
@ -125,23 +121,7 @@ class IntelIOMMU(LinuxKernelTest):
# Check virtio-net via HTTP: # Check virtio-net via HTTP:
exec_command_and_wait_for_pattern(self, 'dhclient eth0', prompt) exec_command_and_wait_for_pattern(self, 'dhclient eth0', prompt)
exec_command_and_wait_for_pattern(self, self.check_http_download(filename, hashsum, self.GUEST_PORT)
f'python3 -m http.server {self.GUEST_PORT} & sleep 1',
f'Serving HTTP on 0.0.0.0 port {self.GUEST_PORT}')
hl = hashlib.sha256()
hostport = get_usernet_hostfwd_port(self.vm)
url = f'http://localhost:{hostport}{filename}'
self.log.info(f'Downloading {url} ...')
with urllib.request.urlopen(url) as response:
while True:
chunk = response.read(1 << 20)
if not chunk:
break
hl.update(chunk)
digest = hl.hexdigest()
self.log.info(f'sha256sum of download is {digest}.')
self.assertEqual(digest, hashsum)
def test_intel_iommu(self): def test_intel_iommu(self):
self.common_vm_setup() self.common_vm_setup()