openinfra/python-tempestconf
Code Issues Proposed changes

Add support for remote octavia test_server.bin

Browse Source 
Add support to download octavia-tempest-plugin test_server.bin from a
remote location.
test_server.bin is a golang application that is provided and used by
octavia-tempest-plugin, however some FIPS-compliant checkers flag it as
non-compliant because it's a static binary.

Change-Id: I1246951bc686cbc815a6f0808471c95f5252898d
Signed-off-by: Gregory Thiemonge <gthiemon@redhat.com>
This commit is contained in:
Gregory Thiemonge
2026-01-19 10:51:34 +01:00
parent 8515371b7c
commit c7cbe5dca3
5 changed files with 165 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
config_tempest/constants.py
View File

@@ -41,6 +41,10 @@ DEFAULT_FLAVOR_RAM_ALT = 192
DEFAULT_FLAVOR_DISK = 1
DEFAULT_FLAVOR_VCPUS = 1
DEFAULT_OCTAVIA_TEST_SERVER_FILE = '/tmp/test_server.bin'
DEFAULT_OCTAVIA_COMPAT_TEST_SERVER_FILE = (
'/usr/libexec/octavia-tempest-plugin-tests-httpd')
# The dict holds the credentials, which are not supposed to be printed
# to a tempest.conf when --test-accounts CLI parameter is used.
ALL_CREDENTIALS_KEYS = {

4
config_tempest/main.py
View File

@@ -625,6 +625,10 @@ def config_tempest(**kwargs):
network = services.get_service("network")
network.create_tempest_networks(conf, kwargs.get('network'))
if services.is_service(**{"type": "load-balancer"}):
load_balancer = services.get_service("load-balancer")
load_balancer.get_test_server_application(conf)
services.post_configuration()
services.set_supported_api_versions()
services.set_service_extensions()

48
config_tempest/services/octavia.py
View File

@@ -13,6 +13,15 @@
# License for the specific language governing permissions and limitations
# under the License.
import os
import urllib
from tenacity import retry
from tenacity import retry_if_exception_type
from tenacity import stop_after_attempt
from tenacity import wait_exponential
from config_tempest import constants as C
from config_tempest.services.base import VersionedService
import json
@@ -26,6 +35,11 @@ class LoadBalancerService(VersionedService):
conf.set('load_balancer', 'admin_role', 'admin')
conf.set('load_balancer', 'RBAC_test_type', 'owner_or_admin')
conf.set('network-feature-enabled', 'port_security', 'True')
# TOOD(gthiemonge) This is a backward-compatible setting for jobs
# that haven't yet migrated to load_balancer.test_server_remote_url
# Remove it once all jobs have been migrated.
conf.set('load_balancer', 'test_server_path',
C.DEFAULT_OCTAVIA_COMPAT_TEST_SERVER_FILE)
@staticmethod
def get_service_type():
@@ -55,3 +69,37 @@ class LoadBalancerService(VersionedService):
conf.set('load_balancer',
'enabled_provider_drivers',
','.join(self.list_drivers()))
@retry(retry=retry_if_exception_type(urllib.error.URLError),
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=2, min=3, max=10))
def _download_file(self, url, destination):
"""Downloads a file specified by `url` to `destination`.
:type url: string
:type destination: string
"""
if os.path.exists(destination):
C.LOG.info("File '%s' already fetched to '%s'.", url, destination)
return
C.LOG.info("Downloading '%s' and saving as '%s'", url, destination)
f = urllib.request.urlopen(url)
data = f.read()
with open(destination, "wb") as dest:
dest.write(data)
def get_test_server_application(self, conf):
if not conf.has_option("load_balancer",
"test_server_remote_url"):
return
remote_url = conf.get("load_balancer",
"test_server_remote_url")
if remote_url:
self._download_file(remote_url,
C.DEFAULT_OCTAVIA_TEST_SERVER_FILE)
os.chmod(C.DEFAULT_OCTAVIA_TEST_SERVER_FILE, 0o755)
conf.set("load_balancer",
"test_server_path",
C.DEFAULT_OCTAVIA_TEST_SERVER_FILE)

101
config_tempest/tests/services/test_octavia.py
View File

@@ -14,7 +14,12 @@
# under the License.
from unittest import mock
import urllib
from tenacity import RetryError
from tenacity import wait_fixed
from config_tempest import constants as C
from config_tempest.services.octavia import LoadBalancerService
from config_tempest.tests.base import BaseServiceTest
@@ -56,3 +61,99 @@ class TestOctaviaService(BaseServiceTest):
("amphora:The Octavia Amphora driver.,"
"octavia:Deprecated alias of the Octavia driver."),
)
@mock.patch("urllib.request.urlopen")
@mock.patch("os.path.exists")
def test_octavia__download_file(self,
mock_path_exists,
mock_urlopen):
mock_url = mock.Mock()
mock_destination = mock.Mock()
mock_data = mock.Mock(name='Fake data')
# File already exists
mock_path_exists.return_value = True
self.Service._download_file(mock_url, mock_destination)
mock_urlopen.assert_not_called()
# File doesn't exist, normal path
mock_path_exists.return_value = False
mock_response = mock.MagicMock()
mock_response.read.return_value = mock_data
mock_urlopen.return_value = mock_response
mock_open = mock.mock_open()
with mock.patch("builtins.open", mock_open):
self.Service._download_file(mock_url, mock_destination)
mock_urlopen.assert_called_once_with(mock_url)
mock_open.assert_called_once_with(mock_destination, "wb")
handle = mock_open()
handle.write.assert_called_once_with(mock_data)
mock_urlopen.reset_mock()
# File doesn't exist, with 2 URLErrors then it passes
mock_path_exists.return_value = False
mock_response = mock.MagicMock()
mock_response.read.return_value = mock_data
mock_urlopen.side_effect = [
urllib.error.URLError(reason="reason1"),
urllib.error.URLError(reason="reason2"),
mock_response]
mock_open = mock.mock_open()
with mock.patch("builtins.open", mock_open):
# override tenacity.retry wait param
with mock.patch.object(self.Service._download_file.retry,
"wait", wait_fixed(0)):
self.Service._download_file(mock_url, mock_destination)
mock_urlopen.assert_called_with(mock_url)
mock_open.assert_called_once_with(mock_destination, "wb")
handle = mock_open()
handle.write.assert_called_once_with(mock_data)
# File doesn't exist, with URLErrors
mock_path_exists.return_value = False
mock_urlopen.side_effect = [
urllib.error.URLError(reason="reason1"),
urllib.error.URLError(reason="reason2"),
urllib.error.URLError(reason="reason3"),
urllib.error.URLError(reason="reason4")]
mock_open = mock.mock_open()
with mock.patch("builtins.open", mock_open):
# override tenacity.retry wait param
with mock.patch.object(self.Service._download_file.retry,
"wait", wait_fixed(0)):
self.assertRaises(RetryError,
self.Service._download_file,
mock_url,
mock_destination)
mock_urlopen.assert_called_with(mock_url)
mock_open.assert_not_called()
@mock.patch("config_tempest.services.octavia.LoadBalancerService."
"_download_file")
@mock.patch("os.chmod")
def test_octavia_get_test_server_application(self,
mock_chmod,
mock_download_file):
# test_server_remote_url not configured
self.Service.get_test_server_application(self.conf)
mock_download_file.assert_not_called()
# test_server_remote_url set
location = "dummy://location"
self.conf.set("load_balancer", "test_server_remote_url",
location)
self.Service.get_test_server_application(self.conf)
mock_download_file.assert_called_once_with(
location, C.DEFAULT_OCTAVIA_TEST_SERVER_FILE)
mock_chmod.assert_called_once_with(
C.DEFAULT_OCTAVIA_TEST_SERVER_FILE, 0o755)
self.assertEqual(self.conf.get("load_balancer", "test_server_path"),
C.DEFAULT_OCTAVIA_TEST_SERVER_FILE)

8
releasenotes/notes/octavia-remote-test-server-5db3bee0e80a2193.yaml Normal file
View File

@@ -0,0 +1,8 @@
---
features:
- |
Add support for remote Octavia test_server.bin. In case users need to
download octavia-tempest-plugin test_server.bin from a remote location,
they can set the ``[load_balancer].test_server_remote_url`` to the URL of
the application, python-tempestconf will download the binary and set the
``[load_balancer].test_server_path`` parameter accordingly.