fedora-image-tester: various fix-ups
Fixes up the Containerfile so it installs properly. This also ensures the resource groups LISA creates are tagged so they can be easily cleaned up asynchronously if LISA fails to clean up for whatever reason (e.g. the container is restarted during a run). Signed-off-by: Jeremy Cline <jeremycline@linux.microsoft.com>
This commit is contained in:
Jeremy Cline
parent
862a365251
commit
15b67101ef
10 changed files with 381 additions and 717 deletions
55
Containerfile.fit
Normal file
55
Containerfile.fit
Normal file
|
|
@ -0,0 +1,55 @@
|
|||
FROM quay.io/fedora/fedora:42 as builder
|
||||
|
||||
RUN dnf install -y \
|
||||
git \
|
||||
python3-pip \
|
||||
python3-build \
|
||||
python3-hatchling
|
||||
|
||||
RUN mkdir -p /srv/{fedora-image-tester,fedora-image-uploader-messages,dist}
|
||||
COPY fedora-image-tester /srv/fedora-image-tester
|
||||
COPY fedora-image-uploader-messages /srv/fedora-image-uploader-messages
|
||||
RUN cd /srv/fedora-image-tester && hatchling build -d /srv/dist/ --target=wheel
|
||||
RUN cd /srv/fedora-image-uploader-messages && hatchling build -d /srv/dist/ --target=wheel
|
||||
|
||||
# Currently LISA isn't on PyPI, but I'm working on it
|
||||
RUN git clone https://github.com/microsoft/lisa.git /srv/lisa
|
||||
WORKDIR /srv/lisa
|
||||
RUN git checkout 20251028.1 && python -m build
|
||||
|
||||
FROM quay.io/fedora/fedora:42
|
||||
|
||||
LABEL org.opencontainers.image.authors="Fedora Cloud SIG <cloud@lists.fedoraproject.org>"
|
||||
|
||||
RUN mkdir -p /srv/fedora-image-tester
|
||||
WORKDIR /srv/fedora-image-tester
|
||||
|
||||
COPY --from=builder /srv/dist /srv/dist
|
||||
COPY --from=builder /srv/lisa/dist /srv/lisa/dist
|
||||
|
||||
# Use the system-provided libraries as much as we can here.
|
||||
#
|
||||
# We do need to commit a small crime so the system-provided fedora-messaging
|
||||
# library uses our virtualenv
|
||||
RUN dnf install -y \
|
||||
python3-pip \
|
||||
fedora-messaging \
|
||||
python3-fedora-image-uploader-messages \
|
||||
python3-gobject \
|
||||
python3-paramiko \
|
||||
python3-pillow \
|
||||
python3-pyyaml \
|
||||
python3-retry \
|
||||
python3-requests
|
||||
RUN python3 -m venv --system-site-packages venv && \
|
||||
venv/bin/pip install --no-cache-dir '/srv/lisa/dist/mslisa-20251028.1-py3-none-any.whl' && \
|
||||
venv/bin/pip install --no-cache-dir /srv/dist/*.whl
|
||||
RUN cp /usr/bin/fedora-messaging /srv/fedora-image-tester/venv/bin/fedora-messaging && \
|
||||
sed -i 's|/usr/bin/python3|/srv/fedora-image-tester/venv/bin/python3|g' \
|
||||
/srv/fedora-image-tester/venv/bin/fedora-messaging
|
||||
|
||||
ENV PATH="/srv/fedora-image-tester/venv/bin:$PATH"
|
||||
ENV VIRTUAL_ENV="/srv/fedora-image-tester/venv"
|
||||
|
||||
ENTRYPOINT ["/srv/fedora-image-tester/venv/bin/fedora-messaging"]
|
||||
CMD ["consume"]
|
||||
|
|
@ -1,51 +0,0 @@
|
|||
FROM registry.fedoraproject.org/fedora:43 as builder
|
||||
|
||||
RUN dnf install -y \
|
||||
git \
|
||||
python3-pip \
|
||||
python3-build \
|
||||
python3-hatchling
|
||||
|
||||
RUN mkdir -p /srv/fedora-image-upload-tester
|
||||
COPY . /srv/fedora-image-upload-tester
|
||||
RUN cd /srv/fedora-image-upload-tester && hatchling build --target=wheel
|
||||
|
||||
# Currently LISA isn't on PyPI, but I'm working on it
|
||||
RUN git clone https://github.com/microsoft/lisa.git /srv/lisa
|
||||
WORKDIR /srv/lisa
|
||||
RUN git checkout 20251028.1 && python -m build
|
||||
|
||||
FROM registry.fedoraproject.org/fedora:43
|
||||
|
||||
LABEL org.opencontainers.image.authors="Fedora Cloud SIG <cloud@lists.fedoraproject.org>"
|
||||
|
||||
RUN mkdir -p /srv/fedora-image-upload-tester
|
||||
WORKDIR /srv/fedora-image-upload-tester
|
||||
|
||||
COPY --from=builder /srv/fedora-image-upload-tester/dist /srv/fedora-cloud-testing/dist
|
||||
COPY --from=builder /srv/lisa/dist/*whl /srv/fedora-image-upload-tester/dist/
|
||||
|
||||
# Use the system-provided libraries as much as we can here.
|
||||
#
|
||||
# We do need to commit a small crime so the system-provided fedora-messaging
|
||||
# library uses our virtualenv
|
||||
RUN dnf install -y \
|
||||
python3-pip \
|
||||
fedora-messaging \
|
||||
python3-fedora-image-uploader-messages \
|
||||
python3-gobject \
|
||||
python3-paramiko \
|
||||
python3-pillow \
|
||||
python3-pyyaml \
|
||||
python3-retry \
|
||||
python3-requests
|
||||
RUN python3 -m venv --system-site-packages venv && venv/bin/pip install --no-cache-dir dist/*
|
||||
RUN cp /usr/bin/fedora-messaging /srv/fedora-image-upload-tester/venv/bin/fedora-messaging && \
|
||||
sed -i 's|/usr/bin/python3|/srv/fedora-image-upload-tester/venv/bin/python3|g' \
|
||||
/srv/fedora-image-upload-tester/venv/bin/fedora-messaging
|
||||
|
||||
ENV PATH="/srv/fedora-image-upload-tester/venv/bin:$PATH"
|
||||
ENV VIRTUAL_ENV="/srv/fedora-image-upload-tester/venv"
|
||||
|
||||
ENTRYPOINT ["/srv/fedora-image-upload-tester/venv/bin/fedora-messaging"]
|
||||
CMD ["consume"]
|
||||
|
|
@ -0,0 +1,46 @@
|
|||
# fedora-image-tester
|
||||
|
||||
An AMQP consumer that runs the [LISA](https://github.com/microsoft/lisa) test suite against cloud images
|
||||
uploaded by fedora-image-uploader.
|
||||
|
||||
## Configuring
|
||||
|
||||
The `fedora-messaging` service is the container entrypoint and is also used to provide most of the configuration. An example configuration file, `fedora-messaging-fit.toml.example` is included and the full list of options are in [fedora-messaging's configuration documentation](https://fedora-messaging.readthedocs.io/en/stable/user-guide/configuration.html). The `FEDORA_MESSAGING_CONF` environment variable should be set to the configuration file's location.
|
||||
|
||||
## Run Locally
|
||||
|
||||
First, build the container:
|
||||
|
||||
```
|
||||
$ podman build -t fedora-image-tester:latest -f Containerfile.fit .
|
||||
```
|
||||
|
||||
Copy `fedora-messaging-fit.toml.example` to `fit.toml` and edit it to fit your needs.
|
||||
|
||||
Next, set up any credentials. For example, to authenticate to Azure using a client certificate:
|
||||
|
||||
```
|
||||
$ openssl req -x509 -new -nodes -sha256 -days 365 \
|
||||
-addext "extendedKeyUsage = clientAuth" \
|
||||
-subj "/CN=fedora-image-tester" \
|
||||
-newkey rsa:4096 \
|
||||
-keyout fedora-image-tester.key.pem \
|
||||
-out fedora-image-tester.cert.pem
|
||||
# Make note of the 'Id' and 'AppId' fields in the output as you'll need these later.
|
||||
$ az ad app create --display-name fedora-image-tester-dev
|
||||
# Add our certificate to use when authenticating
|
||||
$ az ad app credential reset --id $APP_ID --append \
|
||||
--display-name "Fedora Image Tester Certificate" \
|
||||
--cert "@./fedora-image-tester.cert.pem"
|
||||
$ az ad sp create --id $APP_ID
|
||||
# Note that this is an absurdly broad permission set; don't do this for production
|
||||
$ az role assignment create --assignee $APP_ID --role "Contributor" --scope "/subscriptions/<your sub>"
|
||||
$ cat fedora-image-tester.key.pem fedora-image-tester.cert.pem > azure-creds
|
||||
$ podman secret create fedora-image-tester-cert azure-creds
|
||||
```
|
||||
|
||||
Finally, run the consumer:
|
||||
|
||||
```
|
||||
podman run --rm -it -v "$(pwd)"/fit.toml:/etc/fedora-messaging/config.toml:ro,Z --secret source=fedora-image-tester-cert,type=mount --env 'AZURE_CLIENT_CERTIFICATE_PATH=/run/secrets/fedora-image-tester-cert' --env 'AZURE_TENANT_ID=cf237184-4b74-4d45-933d-e415c43f13d8' --env 'AZURE_CLIENT_ID=1571446d-5a0b-4784-83a3-cf4501afda69' --entrypoint=/bin/bash fedora-image-tester:latest
|
||||
```
|
||||
|
|
@ -9,37 +9,30 @@ image with the results.
|
|||
import asyncio
|
||||
import logging
|
||||
import os
|
||||
import subprocess
|
||||
import xml.etree.ElementTree as ET
|
||||
from datetime import datetime, timezone
|
||||
from tempfile import TemporaryDirectory
|
||||
|
||||
import lisa
|
||||
import yaml
|
||||
from azure import identity as az_identity
|
||||
from azure.core.exceptions import AzureError
|
||||
from azure.storage.blob import BlobServiceClient, ContentSettings
|
||||
from fedora_image_uploader_messages.publish import AzurePublishedV1
|
||||
from fedora_image_uploader_messages.test_results import AzureTestResults
|
||||
from fedora_messaging import api, config
|
||||
from fedora_messaging.exceptions import (
|
||||
ConnectionException,
|
||||
PublishTimeout,
|
||||
ValidationError,
|
||||
)
|
||||
from fedora_messaging import config
|
||||
|
||||
from .trigger_lisa import LisaRunner
|
||||
from .utils import trigger_lisa, fallible_publish
|
||||
|
||||
_log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class AzurePublishedConsumer:
|
||||
"""Consumer class for AzurePublishedV1 messages to trigger LISA tests."""
|
||||
FEDORA_GALLERY = "Fedora-5e266ba4-2250-406d-adad-5d73860d958f"
|
||||
RG_OWNER = "fedora-image-tester"
|
||||
|
||||
# Supported Fedora versions for testing
|
||||
SUPPORTED_FEDORA_VERSIONS = [
|
||||
"Fedora-Cloud-Rawhide-x64",
|
||||
"Fedora-Cloud-41-x64",
|
||||
"Fedora-Cloud-41-Arm64",
|
||||
"Fedora-Cloud-Rawhide-Arm64",
|
||||
"Fedora-Cloud-42-x64",
|
||||
"Fedora-Cloud-42-Arm64",
|
||||
]
|
||||
|
||||
class Consumer:
|
||||
"""Consumer class for AzurePublishedV1 messages to trigger LISA tests."""
|
||||
|
||||
def __init__(self):
|
||||
try:
|
||||
|
|
@ -48,171 +41,114 @@ class AzurePublishedConsumer:
|
|||
_log.error("The Azure consumer requires an 'azure' config section")
|
||||
raise
|
||||
|
||||
def __call__(self, message):
|
||||
self.azure_subscription = os.environ["AZURE_SUBSCRIPTION_ID"]
|
||||
self.azure_credentials = az_identity.DefaultAzureCredential()
|
||||
self.azure_blob_client = BlobServiceClient(
|
||||
account_url=(f"https://{self.conf['results_storage_account']}.blob.core.windows.net"),
|
||||
credential=self.azure_credentials,
|
||||
)
|
||||
|
||||
def __call__(self, message: AzurePublishedV1):
|
||||
"""Callback method to handle incoming messages."""
|
||||
_log.info("Received message: %s", message)
|
||||
self.azure_published_callback(message)
|
||||
image_definition_name = message.body["image_definition_name"]
|
||||
community_gallery_image = (
|
||||
f"{self.conf['region']}/{FEDORA_GALLERY}/"
|
||||
f"{image_definition_name}/{message.body['image_version_name']}"
|
||||
)
|
||||
|
||||
def _get_image_definition_name(self, message):
|
||||
"""Get image definition name from the message body.
|
||||
|
||||
Args:
|
||||
message (AzurePublishedV1): The message containing image details.
|
||||
|
||||
Returns:
|
||||
str: The image definition name if found, else None.
|
||||
Eg: "Fedora-Cloud-Rawhide-x64", "Fedora-Cloud-41-x64", etc.
|
||||
"""
|
||||
try:
|
||||
image_definition_name = message.body.get("image_definition_name")
|
||||
if not isinstance(image_definition_name, str):
|
||||
_log.error("image_definition_name is not a string: %s", image_definition_name)
|
||||
return None
|
||||
_log.info("Extracted image_definition_name: %s", image_definition_name)
|
||||
return image_definition_name
|
||||
except AttributeError:
|
||||
_log.error("Message body does not have 'image_definition_name' field.")
|
||||
return None
|
||||
|
||||
def get_community_gallery_image(self, message):
|
||||
"""Extract community gallery image from the messages."""
|
||||
_log.info("Extracting community gallery image from the message: %s", message.body)
|
||||
try:
|
||||
# Validate message.body is a dict
|
||||
if not isinstance(message.body, dict):
|
||||
_log.error("Message body is not a dictionary.")
|
||||
return None
|
||||
|
||||
image_definition_name = self._get_image_definition_name(message)
|
||||
# Run tests only for fedora rawhide, 41 and 42,
|
||||
# include your Fedora versions in SUPPORTED_FEDORA_VERSIONS
|
||||
if image_definition_name not in self.SUPPORTED_FEDORA_VERSIONS:
|
||||
_log.info(
|
||||
"image_definition_name '%s' not in supported Fedora" " versions, skipping.",
|
||||
image_definition_name,
|
||||
)
|
||||
return None
|
||||
image_version_name = message.body.get("image_version_name")
|
||||
image_resource_id = message.body.get("image_resource_id")
|
||||
|
||||
# Check for missing fields
|
||||
if not all([image_definition_name, image_version_name, image_resource_id]):
|
||||
_log.error("Missing required image fields in message body.")
|
||||
return None
|
||||
|
||||
# Defensive split and validation
|
||||
parts = image_resource_id.split("/")
|
||||
if len(parts) < 3:
|
||||
_log.error("image_resource_id format is invalid: %s", image_resource_id)
|
||||
return None
|
||||
resource_id = parts[2]
|
||||
|
||||
community_gallery_image = (
|
||||
f"{self.conf['region']}/{resource_id}/"
|
||||
f"{image_definition_name}/{image_version_name}"
|
||||
)
|
||||
_log.info("Constructed community gallery image: %s", community_gallery_image)
|
||||
return community_gallery_image
|
||||
|
||||
except AttributeError as e:
|
||||
_log.error("Failed to extract image details from the message: %s", str(e))
|
||||
return None
|
||||
|
||||
def azure_published_callback(self, message):
|
||||
"""Handle Azure published messages"""
|
||||
_log.info("Received message on topic: %s", message.topic)
|
||||
_log.info("Message %s", message.body)
|
||||
try:
|
||||
if isinstance(message, AzurePublishedV1):
|
||||
_log.info("Message properties match AzurePublishedV1 schema.")
|
||||
except TypeError as e:
|
||||
_log.error("Message properties do not match AzurePublishedV1 schema: %s", str(e))
|
||||
|
||||
community_gallery_image = self.get_community_gallery_image(message)
|
||||
|
||||
if not community_gallery_image:
|
||||
_log.error("Unsupported or No community gallery image found in the message.")
|
||||
return
|
||||
|
||||
image_definition_name = self._get_image_definition_name(message)
|
||||
|
||||
# Generate run name with UTC format
|
||||
run_name = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%MZ")
|
||||
run_name = f"{message.body['compose_id']}-{message.body['architecture']}"
|
||||
_log.info("Run name generated: %s", run_name)
|
||||
base_runbook = os.path.join(
|
||||
os.path.dirname(lisa.__file__), "microsoft/runbook/tiers/tier.yml"
|
||||
)
|
||||
runbook = {
|
||||
"name": "fedora azure",
|
||||
"include": [
|
||||
{"path": base_runbook},
|
||||
],
|
||||
"notifier": [
|
||||
{"type": "html"},
|
||||
{"type": "junit"},
|
||||
{"type": "env_stats"},
|
||||
],
|
||||
"variable": [
|
||||
{"name": "tier", "value": "1"},
|
||||
],
|
||||
"concurrency": 3,
|
||||
"platform": [
|
||||
{
|
||||
"type": "azure",
|
||||
"keep_environment": "no",
|
||||
"azure": {
|
||||
"deploy": True,
|
||||
"wait_delete": False,
|
||||
"subscription_id": self.azure_subscription,
|
||||
"use_public_address": True,
|
||||
"create_public_address": True,
|
||||
"resource_group_tags": {"owner": RG_OWNER},
|
||||
},
|
||||
"requirement": {
|
||||
"core_count": {"min": 2},
|
||||
"azure": {
|
||||
"community_gallery_image": community_gallery_image,
|
||||
"location": self.conf["region"],
|
||||
},
|
||||
},
|
||||
},
|
||||
],
|
||||
}
|
||||
|
||||
try:
|
||||
# Use TemporaryDirectory context manager for auto cleanup at the end of
|
||||
# the test run
|
||||
with TemporaryDirectory(
|
||||
prefix=f"lisa_results_{image_definition_name}_", suffix="_logs"
|
||||
) as log_path:
|
||||
with TemporaryDirectory(prefix=f"lisa_{image_definition_name}_") as workdir:
|
||||
runbook_path = os.path.join(workdir, "runbook.yml")
|
||||
log_path = os.path.join(workdir, "logs")
|
||||
os.mkdir(log_path)
|
||||
_log.info("Temporary log path created: %s", log_path)
|
||||
with open(runbook_path, "w") as f:
|
||||
yaml.safe_dump(runbook, f)
|
||||
|
||||
# Generate SSH key pair for authentication
|
||||
private_key = self._generate_ssh_key_pair(log_path)
|
||||
|
||||
config_params = {
|
||||
"subscription": self.conf["subscription_id"],
|
||||
"private_key": private_key,
|
||||
"log_path": log_path,
|
||||
"run_name": run_name,
|
||||
}
|
||||
_log.info("LISA config parameters: %s", config_params)
|
||||
_log.info("Triggering tests for image: %s", community_gallery_image)
|
||||
runner = LisaRunner()
|
||||
ret = asyncio.run(
|
||||
runner.trigger_lisa(
|
||||
region=self.conf["region"],
|
||||
community_gallery_image=community_gallery_image,
|
||||
config=config_params,
|
||||
asyncio.run(
|
||||
trigger_lisa(
|
||||
runbook_path=runbook_path,
|
||||
log_path=log_path,
|
||||
run_name=run_name,
|
||||
)
|
||||
)
|
||||
_log.info("LISA trigger completed with return code: %d", ret)
|
||||
if ret == 0:
|
||||
_log.info("LISA trigger executed successfully.")
|
||||
test_results = self._parse_test_results(log_path, run_name)
|
||||
if test_results is not None:
|
||||
_log.info("Test execution completed with results: %s", test_results)
|
||||
# To Do: Implement sending the results using publisher
|
||||
self.publish_test_results(message, test_results)
|
||||
else:
|
||||
_log.error("Failed to parse test results, skipping image")
|
||||
test_results = self._parse_test_results(log_path, run_name)
|
||||
_log.info("Test execution completed with results: %s", test_results)
|
||||
if test_results is not None:
|
||||
body = self._build_result_message_body(message, test_results)
|
||||
_log.debug("Publishing test results for image: %s", body["image_id"])
|
||||
fallible_publish(AzureTestResults(body=body))
|
||||
_log.info("Successfully published test results for %s", body["image_id"])
|
||||
else:
|
||||
_log.error("LISA trigger failed with return code: %d", ret)
|
||||
# TemporaryDirectory automatically cleans up when exiting the context
|
||||
_log.error("Failed to parse test results, skipping image")
|
||||
|
||||
html_results_path = os.path.join(log_path, run_name, "lisa.html")
|
||||
junit_results_path = os.path.join(log_path, run_name, "lisa.junit.xml")
|
||||
compose_id = message.body["compose_id"]
|
||||
architecture = message.body["architecture"]
|
||||
html_blob = self.azure_blob_client.get_blob_client(
|
||||
container="$web", blob=f"{compose_id}/{architecture}/index.html"
|
||||
)
|
||||
junit_blob = self.azure_blob_client.get_blob_client(
|
||||
container="$web", blob=f"{compose_id}/{architecture}/junit.xml"
|
||||
)
|
||||
|
||||
try:
|
||||
with open(html_results_path, "rb") as data:
|
||||
content_settings = ContentSettings(content_type="text/html; charset=utf-8")
|
||||
html_blob.upload_blob(data=data, content_settings=content_settings)
|
||||
with open(junit_results_path, "rb") as data:
|
||||
content_settings = ContentSettings(content_type="application/xml")
|
||||
junit_blob.upload_blob(data=data, content_settings=content_settings)
|
||||
except AzureError as e:
|
||||
_log.warning("Failed to upload results to storage account: %s", str(e))
|
||||
except OSError as e:
|
||||
_log.exception("Failed to trigger LISA: %s", str(e))
|
||||
|
||||
def publish_test_results(self, message, test_results):
|
||||
"""
|
||||
Publish the test results using AzureTestResults publisher.
|
||||
|
||||
Following fedora-image-uploader patterns for message publishing.
|
||||
"""
|
||||
try:
|
||||
# Extract metadata from original message
|
||||
body = self._build_result_message_body(message, test_results)
|
||||
|
||||
# Create message instance with body (following fedora-messaging patterns)
|
||||
result_message = AzureTestResults(body=body)
|
||||
|
||||
_log.info("Publishing test results for image: %s", body["image_id"])
|
||||
_log.debug("Full message body: %s", body)
|
||||
|
||||
# Publish message using fedora-messaging API
|
||||
api.publish(result_message)
|
||||
|
||||
_log.info("Successfully published test results for %s", body["image_id"])
|
||||
|
||||
except ValidationError as e:
|
||||
_log.error("Message validation failed: %s", str(e))
|
||||
_log.error("Invalid message body: %s", body)
|
||||
except (PublishTimeout, ConnectionException) as e:
|
||||
_log.error("Failed to publish test results due to connectivity: %s", str(e))
|
||||
except (OSError, KeyError, TypeError) as e:
|
||||
_log.error("Unexpected error during publishing: %s", str(e))
|
||||
|
||||
def _build_result_message_body(self, original_message, test_results):
|
||||
"""
|
||||
Build the message body for test results publication.
|
||||
|
|
@ -224,17 +160,12 @@ class AzurePublishedConsumer:
|
|||
Returns:
|
||||
dict: Message body for AzureTestResults
|
||||
"""
|
||||
# Extract image metadata from original message
|
||||
body = original_message.body
|
||||
|
||||
# Build the result message body following the schema
|
||||
result_body = {
|
||||
# Image identification
|
||||
"architecture": body["architecture"],
|
||||
"compose_id": body["compose_id"],
|
||||
"image_id": body["image_definition_name"], # Use definition name as image ID
|
||||
"image_id": body["image_definition_name"],
|
||||
"image_resource_id": body["image_resource_id"],
|
||||
# Detailed test lists
|
||||
"failed_tests": test_results.get("failed_tests", {"count": 0, "tests": {}}),
|
||||
"skipped_tests": test_results.get("skipped_tests", {"count": 0, "tests": {}}),
|
||||
"passed_tests": test_results.get("passed_tests", {"count": 0, "tests": {}}),
|
||||
|
|
@ -253,7 +184,6 @@ class AzurePublishedConsumer:
|
|||
'total_tests', 'passed', 'failed', 'skipped', 'errors'
|
||||
None: If parsing fails and results cannot be determined
|
||||
"""
|
||||
# Find and validate XML file
|
||||
xml_file = self._find_xml_file(log_path, run_name)
|
||||
if not xml_file or not os.path.exists(xml_file):
|
||||
_log.error("No XML file found in the log path: %s", log_path)
|
||||
|
|
@ -261,13 +191,11 @@ class AzurePublishedConsumer:
|
|||
|
||||
_log.info("Found XML file: %s", xml_file)
|
||||
|
||||
# Parse the XML file
|
||||
try:
|
||||
tree = ET.parse(xml_file)
|
||||
root = tree.getroot()
|
||||
_log.info("Parsing xml root element: %s", root.tag)
|
||||
|
||||
# Extract individual test details
|
||||
test_details = self._extract_test_details(root)
|
||||
results = self._format_for_schema(test_details)
|
||||
|
||||
|
|
@ -292,29 +220,23 @@ class AzurePublishedConsumer:
|
|||
|
||||
test_suites = root.findall("testsuite") if root.tag == "testsuites" else [root]
|
||||
|
||||
# Iterate through test suites and test cases
|
||||
for suite in test_suites:
|
||||
suite_name = suite.attrib.get("name")
|
||||
|
||||
for testcase in suite.findall("testcase"):
|
||||
test_name = testcase.attrib.get("name")
|
||||
|
||||
# Create a descriptive test identifier
|
||||
test_identifier = f"{suite_name}.{test_name}"
|
||||
test_time = testcase.attrib.get("time", "0.000")
|
||||
|
||||
# Check test status and extract the message if available
|
||||
failure_elem = testcase.find("failure")
|
||||
error_elem = testcase.find("error")
|
||||
skipped_elem = testcase.find("skipped")
|
||||
|
||||
# Log test details for failed, skipped and errored tests
|
||||
if failure_elem is not None:
|
||||
failure_msg = failure_elem.attrib.get("message", "Test case failed")
|
||||
failure_msg = self._remove_html_tags(failure_msg)
|
||||
traceback_msg = failure_elem.text or ""
|
||||
|
||||
# Combine failure_message and traceback if available
|
||||
if traceback_msg.strip():
|
||||
failure_msg = (
|
||||
f"Summary: {failure_msg}\n Traceback: \n{traceback_msg.strip()}"
|
||||
|
|
@ -332,10 +254,7 @@ class AzurePublishedConsumer:
|
|||
elif skipped_elem is not None:
|
||||
skip_msg = skipped_elem.attrib.get("message", "Test skipped")
|
||||
skip_msg = self._remove_html_tags(skip_msg)
|
||||
|
||||
# As there won't be any traceback will return the entire message
|
||||
test_details["skipped"].append((test_identifier, skip_msg))
|
||||
|
||||
else:
|
||||
passed_msg = f"Test passed in {test_time} seconds."
|
||||
test_details["passed"].append((test_identifier, passed_msg))
|
||||
|
|
@ -403,39 +322,3 @@ class AzurePublishedConsumer:
|
|||
|
||||
_log.warning("No XML file with suffix 'lisa.junit.xml' found in %s", xml_path)
|
||||
return None
|
||||
|
||||
def _generate_ssh_key_pair(self, temp_dir):
|
||||
"""
|
||||
Generate an SSH key pair for authentication.
|
||||
|
||||
Args:
|
||||
temp_dir (str): Directory to store the generated key pair.
|
||||
|
||||
Returns:
|
||||
str: Path to the private key file. or None if generation fails.
|
||||
"""
|
||||
|
||||
private_key_path = os.path.join(temp_dir, "id_ed25519")
|
||||
public_key_path = os.path.join(temp_dir, "id_ed25519.pub")
|
||||
|
||||
try:
|
||||
# Generate SSH key pair using ssh-keygen
|
||||
cmd = ["ssh-keygen", "-t", "ed25519", "-f", private_key_path, "-N", ""]
|
||||
ret = subprocess.run(cmd, check=True, capture_output=True, text=True, timeout=60)
|
||||
_log.info("SSH key pair generated at: %s and %s", private_key_path, public_key_path)
|
||||
_log.debug("ssh-keygen output: %s", ret.stdout)
|
||||
|
||||
# Verify the private key file is created
|
||||
if not os.path.exists(private_key_path):
|
||||
_log.error(
|
||||
"SSH key generation succeeded but private key file was not found at: %s",
|
||||
private_key_path,
|
||||
)
|
||||
return None
|
||||
|
||||
# Set the permissions for the file
|
||||
os.chmod(private_key_path, 0o600)
|
||||
return private_key_path
|
||||
except (subprocess.CalledProcessError, OSError) as e:
|
||||
_log.error("Failed to generate SSH key pair: %s", str(e))
|
||||
return None
|
||||
|
|
|
|||
|
|
@ -1,108 +0,0 @@
|
|||
"""Module to trigger LISA tests asynchronously."""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import subprocess
|
||||
|
||||
_log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# pylint: disable=too-few-public-methods
|
||||
class LisaRunner:
|
||||
"""Class to run LISA tests asynchronously"""
|
||||
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
async def trigger_lisa(self, region, community_gallery_image, config):
|
||||
# pylint: disable=too-many-return-statements,too-many-branches
|
||||
"""Trigger LISA tier 1 tests with the provided parameters.
|
||||
|
||||
Args:
|
||||
region (str): The Azure region to run the tests in.
|
||||
community_gallery_image (str): The community gallery image to use for testing.
|
||||
config (dict): A dictionary containing the configuration parameters.
|
||||
- subscription (str): The Azure subscription ID.
|
||||
- private_key (str): The path to the private key file for authentication.
|
||||
- log_path (str): The path to the log file for the LISA tests.
|
||||
- run_name (str): The name of the test run.
|
||||
|
||||
Returns:
|
||||
bool: True if the LISA test completed successfully (return code 0),
|
||||
False if the test failed, had errors, or if required parameters are missing.
|
||||
"""
|
||||
# Validate the input parameters
|
||||
if not region or not isinstance(region, str):
|
||||
_log.error("Invalid region parameter: must be a non-empty string")
|
||||
return False
|
||||
|
||||
if not community_gallery_image or not isinstance(community_gallery_image, str):
|
||||
_log.error("Invalid community_gallery_image parameter: must be a non-empty string")
|
||||
return False
|
||||
|
||||
if not isinstance(config, dict):
|
||||
_log.error("Invalid config parameter: must be a dictionary")
|
||||
return False
|
||||
|
||||
if not config.get("subscription"):
|
||||
_log.error("Missing required parameter: subscription")
|
||||
return False
|
||||
|
||||
if not config.get("private_key"):
|
||||
_log.error("Missing required parameter: private_key")
|
||||
return False
|
||||
|
||||
try:
|
||||
variables = [
|
||||
f"region:{region}",
|
||||
f"community_gallery_image:{community_gallery_image}",
|
||||
f"subscription_id:{config.get('subscription')}",
|
||||
f"admin_private_key_file:{config.get('private_key')}",
|
||||
]
|
||||
command = [
|
||||
"lisa",
|
||||
"-r",
|
||||
"microsoft/runbook/azure_fedora.yml",
|
||||
"-v",
|
||||
"tier:1",
|
||||
"-v",
|
||||
"test_case_name:verify_dhcp_file_configuration",
|
||||
]
|
||||
for var in variables:
|
||||
command.extend(["-v", var])
|
||||
|
||||
# Add optional parameters only if they are provided
|
||||
log_path = config.get("log_path")
|
||||
if log_path:
|
||||
command.extend(["-l", log_path])
|
||||
_log.debug("Added log path: %s", log_path)
|
||||
else:
|
||||
_log.debug("No log path provided, using LISA default")
|
||||
|
||||
run_name = config.get("run_name")
|
||||
if run_name:
|
||||
command.extend(["-i", run_name])
|
||||
_log.debug("Added run name: %s", run_name)
|
||||
else:
|
||||
_log.debug("No run name provided, using LISA default")
|
||||
|
||||
_log.info("Starting LISA test with command: %s", " ".join(command))
|
||||
process = await asyncio.create_subprocess_exec(
|
||||
*command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT
|
||||
)
|
||||
async for line in process.stdout:
|
||||
line_content = line.decode().strip()
|
||||
if line_content: # Only log non-empty lines
|
||||
_log.info("LISA OUTPUT: %s ", line_content)
|
||||
|
||||
await process.wait()
|
||||
# stderr = await process.communicate()
|
||||
|
||||
if process.returncode == 0:
|
||||
_log.info("LISA test completed successfully")
|
||||
return True
|
||||
_log.error("LISA test failed with return code: %d", process.returncode)
|
||||
return False
|
||||
except Exception as e: # pylint: disable=broad-except
|
||||
_log.error("An error occurred while running the tests: %s", str(e))
|
||||
return False
|
||||
67
fedora-image-tester/fedora_image_tester/utils.py
Normal file
67
fedora-image-tester/fedora_image_tester/utils.py
Normal file
|
|
@ -0,0 +1,67 @@
|
|||
"""Module to trigger LISA tests asynchronously."""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import subprocess
|
||||
|
||||
from fedora_messaging import api, exceptions as fm_exceptions
|
||||
|
||||
|
||||
_log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def trigger_lisa(runbook_path: str, log_path: str, run_name: str) -> None:
|
||||
"""
|
||||
Trigger a LISA test run.
|
||||
|
||||
LISA exits with the number of failed tests, so it makes it difficult to use return codes.
|
||||
Your runbook should, at a minimum, contains the "junit" notifier so you can examine results.
|
||||
|
||||
Args:
|
||||
runbook_path: The absolute path to a valid LISA runbook.
|
||||
log_path: The absolute path to a directory where LISA will store various logs and results.
|
||||
run_name: The test run name; LISA will create a directory inside log_path with this name.
|
||||
"""
|
||||
command = [
|
||||
"lisa",
|
||||
"-r",
|
||||
runbook_path,
|
||||
"-l",
|
||||
log_path,
|
||||
"-i",
|
||||
run_name,
|
||||
]
|
||||
_log.info("Starting LISA test with command: '%s'", " ".join(command))
|
||||
process = await asyncio.create_subprocess_exec(
|
||||
*command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT
|
||||
)
|
||||
async for line in process.stdout:
|
||||
line_content = line.decode().strip()
|
||||
if line_content:
|
||||
_log.info("LISA OUTPUT: %s ", line_content)
|
||||
|
||||
await process.wait()
|
||||
|
||||
if process.returncode != 0:
|
||||
_log.warning("LISA test failed with return code: %d", process.returncode)
|
||||
|
||||
|
||||
def fallible_publish(message):
|
||||
"""
|
||||
Helper to publish AMQP messages fallibly.
|
||||
|
||||
Rather than try really hard to publish every message, if the broker is unavailable it's
|
||||
reasonable to just wait until the next image (which happens daily) to get built and try
|
||||
again then.
|
||||
"""
|
||||
try:
|
||||
api.publish(message=message)
|
||||
_log.info("Published %s message to %s", message.__class__.__name__, message.topic)
|
||||
except (fm_exceptions.PublishTimeout, fm_exceptions.PublishReturned) as e:
|
||||
_log.warning("Unable to publish %s message: %s", message.__class__.__name__, str(e))
|
||||
except fm_exceptions.PublishForbidden as e:
|
||||
_log.error(
|
||||
"Unable to publish message to topic %s, permission denied: %s",
|
||||
message.topic,
|
||||
str(e),
|
||||
)
|
||||
|
|
@ -14,8 +14,7 @@ requires-python = ">=3.10"
|
|||
dependencies = [
|
||||
"fedora-messaging",
|
||||
"fedora-image-uploader-messages",
|
||||
# Until we get LISA on PyPI...
|
||||
"mslisa[azure] @ git+https://github.com/microsoft/lisa.git",
|
||||
"mslisa[azure,aws]",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
"""Unit tests for the AzurePublishedConsumer class in azure.py."""
|
||||
"""Unit tests for the Consumer class in azure.py."""
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
|
|
@ -9,7 +9,7 @@ import pytest
|
|||
from fedora_image_uploader_messages.publish import AzurePublishedV1
|
||||
from fedora_messaging import config as fm_config
|
||||
|
||||
from fedora_image_tester.azure import AzurePublishedConsumer
|
||||
from fedora_image_tester.azure import Consumer
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
|
|
@ -29,8 +29,8 @@ def azure_conf():
|
|||
|
||||
@pytest.fixture
|
||||
def consumer(azure_conf): # pylint: disable=unused-argument
|
||||
"""Create an AzurePublishedConsumer instance for testing."""
|
||||
return AzurePublishedConsumer()
|
||||
"""Create an Consumer instance for testing."""
|
||||
return Consumer()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
|
@ -49,22 +49,9 @@ def valid_message():
|
|||
return message
|
||||
|
||||
|
||||
class TestAzurePublishedConsumer:
|
||||
class TestConsumer:
|
||||
# pylint: disable=protected-access
|
||||
"""Test class for AzurePublishedConsumer."""
|
||||
|
||||
def test_supported_fedora_versions_constant(self):
|
||||
"""Test that SUPPORTED_FEDORA_VERSIONS contains expected versions."""
|
||||
# Test that the constant is defined and is a list
|
||||
assert hasattr(AzurePublishedConsumer, "SUPPORTED_FEDORA_VERSIONS")
|
||||
assert isinstance(AzurePublishedConsumer.SUPPORTED_FEDORA_VERSIONS, list)
|
||||
assert len(AzurePublishedConsumer.SUPPORTED_FEDORA_VERSIONS) > 0
|
||||
|
||||
# Test that all versions follow expected naming pattern
|
||||
for version in AzurePublishedConsumer.SUPPORTED_FEDORA_VERSIONS:
|
||||
assert isinstance(version, str)
|
||||
assert version.startswith("Fedora-Cloud-")
|
||||
assert version.endswith(("-x64", "-Arm64"))
|
||||
"""Test class for Consumer."""
|
||||
|
||||
def test_get_image_definition_name_success(self, consumer, valid_message):
|
||||
"""Test successful extraction of image definition name."""
|
||||
|
|
@ -177,9 +164,14 @@ class TestAzurePublishedConsumer:
|
|||
|
||||
@patch("fedora_cloud_tests.azure.asyncio.run")
|
||||
@patch("fedora_cloud_tests.azure.LisaRunner")
|
||||
@patch.object(AzurePublishedConsumer, "_generate_ssh_key_pair")
|
||||
@patch.object(Consumer, "_generate_ssh_key_pair")
|
||||
def test_azure_published_callback_success(
|
||||
self, mock_ssh_keygen, mock_lisa_runner, mock_asyncio_run, consumer, valid_message
|
||||
self,
|
||||
mock_ssh_keygen,
|
||||
mock_lisa_runner,
|
||||
mock_asyncio_run,
|
||||
consumer,
|
||||
valid_message,
|
||||
): # pylint: disable=R0913,R0917
|
||||
"""Test successful message processing and LISA trigger."""
|
||||
mock_runner_instance = MagicMock()
|
||||
|
|
@ -204,11 +196,19 @@ class TestAzurePublishedConsumer:
|
|||
mock_lisa_runner.assert_not_called()
|
||||
mock_asyncio_run.assert_not_called()
|
||||
|
||||
@patch("fedora_cloud_tests.azure.asyncio.run", side_effect=OSError("LISA execution failed"))
|
||||
@patch(
|
||||
"fedora_cloud_tests.azure.asyncio.run",
|
||||
side_effect=OSError("LISA execution failed"),
|
||||
)
|
||||
@patch("fedora_cloud_tests.azure.LisaRunner")
|
||||
@patch.object(AzurePublishedConsumer, "_generate_ssh_key_pair")
|
||||
@patch.object(Consumer, "_generate_ssh_key_pair")
|
||||
def test_azure_published_callback_lisa_exception(
|
||||
self, mock_ssh_keygen, mock_lisa_runner, mock_asyncio_run, consumer, valid_message
|
||||
self,
|
||||
mock_ssh_keygen,
|
||||
mock_lisa_runner,
|
||||
mock_asyncio_run,
|
||||
consumer,
|
||||
valid_message,
|
||||
): # pylint: disable=R0913,R0917
|
||||
"""Test exception handling when LISA execution fails."""
|
||||
mock_runner_instance = MagicMock()
|
||||
|
|
|
|||
|
|
@ -1,310 +0,0 @@
|
|||
"""Unit tests for the LisaRunner class in trigger_lisa.py."""
|
||||
|
||||
import subprocess
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from fedora_image_tester import trigger_lisa
|
||||
|
||||
|
||||
# pylint: disable=protected-access
|
||||
@pytest.fixture
|
||||
def runner():
|
||||
"""Create a LisaRunner instance for testing."""
|
||||
return trigger_lisa.LisaRunner()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_setup(runner, region, community_gallery_image, config_params):
|
||||
"""Create a test setup object combining common fixtures."""
|
||||
return {
|
||||
"runner": runner,
|
||||
"region": region,
|
||||
"community_gallery_image": community_gallery_image,
|
||||
"config_params": config_params,
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_process():
|
||||
"""Create a properly mocked async subprocess for testing."""
|
||||
process = MagicMock()
|
||||
process.returncode = 0
|
||||
process.wait = AsyncMock()
|
||||
|
||||
# Mock stdout as an async iterator
|
||||
async def mock_stdout_lines():
|
||||
lines = [b"LISA test output line 1\n", b"LISA test output line 2\n"]
|
||||
for line in lines:
|
||||
yield line
|
||||
|
||||
process.stdout = mock_stdout_lines()
|
||||
return process
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def config_params():
|
||||
"""Create test configuration parameters."""
|
||||
return {
|
||||
"subscription": "test-subscription-id",
|
||||
"private_key": "/path/to/private/key",
|
||||
"log_path": "/tmp/test_logs",
|
||||
"run_name": "test-run-name",
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def region():
|
||||
"""Create test region."""
|
||||
return "westus2"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def community_gallery_image():
|
||||
"""Create test community gallery image."""
|
||||
return "test/gallery/image"
|
||||
|
||||
|
||||
class TestLisaRunner:
|
||||
"""Test class for LisaRunner."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_trigger_lisa_success(self, test_setup, mock_process):
|
||||
"""Test successful execution of the trigger_lisa method."""
|
||||
with patch("asyncio.create_subprocess_exec") as mock_subproc_exec:
|
||||
mock_subproc_exec.return_value = mock_process
|
||||
|
||||
result = await test_setup["runner"].trigger_lisa(
|
||||
test_setup["region"],
|
||||
test_setup["community_gallery_image"],
|
||||
test_setup["config_params"],
|
||||
)
|
||||
|
||||
assert result is True
|
||||
mock_subproc_exec.assert_called_once()
|
||||
mock_process.wait.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_trigger_lisa_success_with_warnings(self, test_setup, mock_process):
|
||||
"""Test successful execution with output."""
|
||||
with patch("asyncio.create_subprocess_exec") as mock_subproc_exec:
|
||||
mock_subproc_exec.return_value = mock_process
|
||||
|
||||
with patch.object(trigger_lisa._log, "info") as mock_logger_info:
|
||||
result = await test_setup["runner"].trigger_lisa(
|
||||
test_setup["region"],
|
||||
test_setup["community_gallery_image"],
|
||||
test_setup["config_params"],
|
||||
)
|
||||
|
||||
assert result is True
|
||||
# Check that LISA output was logged
|
||||
mock_logger_info.assert_any_call("LISA OUTPUT: %s ", "LISA test output line 1")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_trigger_lisa_failure_non_zero_return_code(
|
||||
self, runner, region, community_gallery_image, config_params
|
||||
):
|
||||
"""Test failure when LISA returns non-zero exit code."""
|
||||
with patch("asyncio.create_subprocess_exec") as mock_subproc_exec:
|
||||
mock_process = MagicMock()
|
||||
mock_process.returncode = 1
|
||||
mock_process.wait = AsyncMock()
|
||||
|
||||
# Mock stdout as an async iterator with error output
|
||||
async def mock_stdout_lines():
|
||||
lines = [b"Error: LISA test failed\n", b"Additional error details\n"]
|
||||
for line in lines:
|
||||
yield line
|
||||
|
||||
mock_process.stdout = mock_stdout_lines()
|
||||
mock_subproc_exec.return_value = mock_process
|
||||
|
||||
with patch.object(trigger_lisa._log, "error") as mock_logger_error:
|
||||
result = await runner.trigger_lisa(region, community_gallery_image, config_params)
|
||||
|
||||
assert result is False
|
||||
mock_logger_error.assert_any_call("LISA test failed with return code: %d", 1)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_trigger_lisa_exception_handling(
|
||||
self, runner, region, community_gallery_image, config_params
|
||||
):
|
||||
"""Test error handling and logging when subprocess execution fails."""
|
||||
with patch("asyncio.create_subprocess_exec", side_effect=Exception("Process failed")):
|
||||
with patch.object(trigger_lisa._log, "error") as mock_logger_error:
|
||||
result = await runner.trigger_lisa(region, community_gallery_image, config_params)
|
||||
|
||||
assert result is False
|
||||
mock_logger_error.assert_called_with(
|
||||
"An error occurred while running the tests: %s", "Process failed"
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_trigger_lisa_missing_region(
|
||||
self, runner, community_gallery_image, config_params
|
||||
):
|
||||
"""Test validation failure when region is missing."""
|
||||
with patch.object(trigger_lisa._log, "error") as mock_logger_error:
|
||||
result = await runner.trigger_lisa("", community_gallery_image, config_params)
|
||||
|
||||
assert result is False
|
||||
mock_logger_error.assert_called_with(
|
||||
"Invalid region parameter: must be a non-empty string"
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_trigger_lisa_missing_community_gallery_image(
|
||||
self, runner, region, config_params
|
||||
):
|
||||
"""Test validation failure when community_gallery_image is missing."""
|
||||
with patch.object(trigger_lisa._log, "error") as mock_logger_error:
|
||||
result = await runner.trigger_lisa(region, "", config_params)
|
||||
|
||||
assert result is False
|
||||
mock_logger_error.assert_called_with(
|
||||
"Invalid community_gallery_image parameter: must be a non-empty string"
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_trigger_lisa_missing_subscription(
|
||||
self, runner, region, community_gallery_image, config_params
|
||||
):
|
||||
"""Test validation failure when subscription is missing."""
|
||||
config_without_subscription = config_params.copy()
|
||||
del config_without_subscription["subscription"]
|
||||
|
||||
with patch.object(trigger_lisa._log, "error") as mock_logger_error:
|
||||
result = await runner.trigger_lisa(
|
||||
region, community_gallery_image, config_without_subscription
|
||||
)
|
||||
|
||||
assert result is False
|
||||
mock_logger_error.assert_called_with("Missing required parameter: subscription")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_trigger_lisa_missing_private_key(
|
||||
self, runner, region, community_gallery_image, config_params
|
||||
):
|
||||
"""Test validation failure when private_key is missing."""
|
||||
config_without_private_key = config_params.copy()
|
||||
del config_without_private_key["private_key"]
|
||||
|
||||
with patch.object(trigger_lisa._log, "error") as mock_logger_error:
|
||||
result = await runner.trigger_lisa(
|
||||
region, community_gallery_image, config_without_private_key
|
||||
)
|
||||
|
||||
assert result is False
|
||||
mock_logger_error.assert_called_with("Missing required parameter: private_key")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_trigger_lisa_command_construction(self, test_setup, mock_process):
|
||||
"""Test that the LISA command is constructed correctly."""
|
||||
with patch("asyncio.create_subprocess_exec") as mock_subproc_exec:
|
||||
mock_subproc_exec.return_value = mock_process
|
||||
|
||||
await test_setup["runner"].trigger_lisa(
|
||||
test_setup["region"],
|
||||
test_setup["community_gallery_image"],
|
||||
test_setup["config_params"],
|
||||
)
|
||||
|
||||
# Verify the command was called with correct arguments
|
||||
expected_command = [
|
||||
"lisa",
|
||||
"-r",
|
||||
"microsoft/runbook/azure_fedora.yml",
|
||||
"-v",
|
||||
"tier:1",
|
||||
"-v",
|
||||
"test_case_name:verify_dhcp_file_configuration",
|
||||
"-v",
|
||||
f"region:{test_setup['region']}",
|
||||
"-v",
|
||||
f"community_gallery_image:{test_setup['community_gallery_image']}",
|
||||
"-v",
|
||||
f"subscription_id:{test_setup['config_params']['subscription']}",
|
||||
"-v",
|
||||
f"admin_private_key_file:{test_setup['config_params']['private_key']}",
|
||||
"-l",
|
||||
test_setup["config_params"]["log_path"],
|
||||
"-i",
|
||||
test_setup["config_params"]["run_name"],
|
||||
]
|
||||
|
||||
mock_subproc_exec.assert_called_once_with(
|
||||
*expected_command,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT,
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_trigger_lisa_missing_optional_config_parameters(
|
||||
self, runner, region, community_gallery_image, mock_process
|
||||
):
|
||||
"""Test successful execution when optional config parameters
|
||||
(log_path, run_name) are missing."""
|
||||
minimal_config = {
|
||||
"subscription": "test-subscription",
|
||||
"private_key": "/path/to/key",
|
||||
# log_path and run_name are missing
|
||||
}
|
||||
|
||||
with patch("asyncio.create_subprocess_exec") as mock_subproc_exec:
|
||||
mock_subproc_exec.return_value = mock_process
|
||||
|
||||
result = await runner.trigger_lisa(region, community_gallery_image, minimal_config)
|
||||
|
||||
# Now the implementation should handle missing optional parameters gracefully
|
||||
assert result is True
|
||||
|
||||
# Verify command is called but without the optional -l and -i flags
|
||||
args, _ = mock_subproc_exec.call_args
|
||||
command_list = list(args)
|
||||
assert "-l" not in command_list
|
||||
assert "-i" not in command_list
|
||||
# But should still have the required arguments
|
||||
assert "lisa" in command_list
|
||||
assert "-r" in command_list
|
||||
assert "microsoft/runbook/azure_fedora.yml" in command_list
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_trigger_lisa_with_optional_config_parameters(
|
||||
self, runner, region, community_gallery_image, mock_process
|
||||
):
|
||||
"""Test successful execution when optional config parameters are provided."""
|
||||
config_with_optionals = {
|
||||
"subscription": "test-subscription",
|
||||
"private_key": "/path/to/key",
|
||||
"log_path": "/custom/log/path",
|
||||
"run_name": "custom-run-name",
|
||||
}
|
||||
|
||||
with patch("asyncio.create_subprocess_exec") as mock_subproc_exec:
|
||||
mock_subproc_exec.return_value = mock_process
|
||||
|
||||
result = await runner.trigger_lisa(
|
||||
region, community_gallery_image, config_with_optionals
|
||||
)
|
||||
|
||||
assert result is True
|
||||
# Verify command includes the provided optional parameters
|
||||
args, _ = mock_subproc_exec.call_args
|
||||
command_list = list(args)
|
||||
assert "-l" in command_list
|
||||
assert "/custom/log/path" in command_list
|
||||
assert "-i" in command_list
|
||||
assert "custom-run-name" in command_list
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_trigger_lisa_invalid_config_type(self, runner, region, community_gallery_image):
|
||||
"""Test validation failure when config is not a dictionary."""
|
||||
with patch.object(trigger_lisa._log, "error") as mock_logger_error:
|
||||
result = await runner.trigger_lisa(
|
||||
region, community_gallery_image, "not a dict" # Invalid type
|
||||
)
|
||||
|
||||
assert result is False
|
||||
mock_logger_error.assert_called_with("Invalid config parameter: must be a dictionary")
|
||||
83
fedora-messaging-fit.toml.example
Normal file
83
fedora-messaging-fit.toml.example
Normal file
|
|
@ -0,0 +1,83 @@
|
|||
# A sample configuration for fedora-messaging. This file is in the TOML format.
|
||||
amqp_url = "amqps://fedora:@rabbitmq.fedoraproject.org/%2Fpublic_pubsub"
|
||||
callback = "fedora_image_tester.azure:Consumer"
|
||||
|
||||
[tls]
|
||||
ca_cert = "/etc/fedora-messaging/cacert.pem"
|
||||
keyfile = "/etc/fedora-messaging/fedora-key.pem"
|
||||
certfile = "/etc/fedora-messaging/fedora-cert.pem"
|
||||
|
||||
# Queue names *must* be in the normal UUID format: run "uuidgen" and use the
|
||||
# output as your queue name. If you don't define a queue here, the server will
|
||||
# generate a queue name for you. This queue will be non-durable, auto-deleted and
|
||||
# exclusive.
|
||||
# If your queue is not exclusive, anyone can connect and consume from it, causing
|
||||
# you to miss messages, so do not share your queue name. Any queues that are not
|
||||
# auto-deleted on disconnect are garbage-collected after approximately one hour.
|
||||
#
|
||||
# If you require a stronger guarantee about delivery, please talk to Fedora's
|
||||
# Infrastructure team.
|
||||
#
|
||||
# If you use the server-generated queue names, you can leave out the "queue"
|
||||
# parameter in the bindings definition.
|
||||
[[bindings]]
|
||||
# queue = "00000000-0000-0000-0000-000000000000"
|
||||
exchange = "amq.topic"
|
||||
routing_keys = ["org.fedoraproject.*.fedora_image_uploader.published.v1.azure.*"]
|
||||
|
||||
[qos]
|
||||
prefetch_size = 0
|
||||
prefetch_count = 25
|
||||
|
||||
|
||||
[client_properties]
|
||||
app = "Fedora Image Tester"
|
||||
app_url = "https://pagure.io/cloud-image-uploader"
|
||||
app_contacts_email = "cloud@lists.fedoraproject.org"
|
||||
|
||||
# The Azure consumer namespaces its configuration under the "azure" key.
|
||||
[consumer_config.azure]
|
||||
# The region to use when creating test resources
|
||||
region = "westus3"
|
||||
results_storage_account = "fedoratestresults"
|
||||
|
||||
|
||||
[log_config]
|
||||
version = 1
|
||||
disable_existing_loggers = true
|
||||
|
||||
[log_config.formatters.simple]
|
||||
format = "%(asctime)s [%(name)s - %(levelname)s] - %(message)s"
|
||||
|
||||
[log_config.handlers.console]
|
||||
class = "logging.StreamHandler"
|
||||
formatter = "simple"
|
||||
stream = "ext://sys.stdout"
|
||||
|
||||
[log_config.loggers.fedora_image_tester]
|
||||
level = "INFO"
|
||||
propagate = false
|
||||
handlers = ["console"]
|
||||
|
||||
[log_config.loggers.fedora_messaging]
|
||||
level = "INFO"
|
||||
propagate = false
|
||||
handlers = ["console"]
|
||||
|
||||
# Twisted is the asynchronous framework that manages the TCP/TLS connection, as well
|
||||
# as the consumer event loop. When debugging you may want to lower this log level.
|
||||
[log_config.loggers.twisted]
|
||||
level = "INFO"
|
||||
propagate = false
|
||||
handlers = ["console"]
|
||||
|
||||
# Pika is the underlying AMQP client library. When debugging you may want to
|
||||
# lower this log level.
|
||||
[log_config.loggers.pika]
|
||||
level = "WARNING"
|
||||
propagate = false
|
||||
handlers = ["console"]
|
||||
|
||||
[log_config.root]
|
||||
level = "ERROR"
|
||||
handlers = ["console"]
|
||||
Loading…
Add table
Add a link
Reference in a new issue