Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 0 additions & 4 deletions tests/foreman/cli/test_rhcloud_inventory.py
Original file line number Diff line number Diff line change
Expand Up @@ -542,10 +542,6 @@ def generate_report(rhcloud_manifest_org, module_target_sat, disconnected=False)
)
assert task_output[0].result == "success"

report_log = module_target_sat.api.Organization(id=org.id).rh_cloud_fetch_last_report_log()
expected = 'Check the Uploading tab for report uploading status'
assert expected in report_log['output']


def test_positive_config_on_sat_without_network_protocol(module_target_sat, module_sca_manifest):
"""Test cloud connector configuration on Satellite without explicit network protocol.
Expand Down
47 changes: 37 additions & 10 deletions tests/foreman/ui/test_rhcloud_inventory.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,8 @@
"""

from datetime import UTC, datetime, timedelta
import os
import tempfile

import pytest
from wait_for import wait_for
Expand Down Expand Up @@ -57,22 +59,26 @@ def common_assertion(
):
"""Function to perform common assertions"""
local_file_data = get_local_file_data(report_path)
upload_success_msg = (
f'Done: /var/lib/foreman/red_hat_inventory/uploads/report_for_{org.id}.tar.xz'
)
upload_error_messages = ['NSS error', 'Permission denied']

assert (
'Check the Uploading tab for report uploading status'
in inventory_data['generating']['terminal']
)
if subscription_connection_enabled:
# Only check upload-related assertions when connection is enabled
upload_success_msg = 'Uploaded file moved to done/ folder'
assert (
'Check the Uploading tab for report uploading status'
in inventory_data['generating']['terminal']
)
assert upload_success_msg in inventory_data['uploading']['terminal']
assert 'x-rh-insights-request-id' in inventory_data['uploading']['terminal'].lower()
for error_msg in upload_error_messages:
assert error_msg not in inventory_data['uploading']['terminal']
# There is no uploaded report with subscription_connection_enabled set to false
# Verify uploaded report checksum matches
assert local_file_data['checksum'] == get_remote_report_checksum(satellite, org.id)
else:
# When connection disabled, just verify report was generated
assert (
f'Generated /var/lib/foreman/red_hat_inventory/generated_reports/report_for_{org.id}.tar.xz'
in inventory_data['generating']['terminal']
)

assert local_file_data['size'] > 0
assert local_file_data['extractable']
Expand Down Expand Up @@ -150,7 +156,28 @@ def test_rhcloud_inventory_e2e(
silent_failure=True,
handle_exception=True,
)
report_path = session.cloudinventory.download_report(org.name)
# Get report based on subscription_connection_enabled setting
if subscription_setting:
# When enabled, download via UI
report_path = session.cloudinventory.download_report(org.name)
else:
# When disabled, get from filesystem
remote_report_path = (
f'/var/lib/foreman/red_hat_inventory/generated_reports/report_for_{org.id}.tar.xz'
)

# Verify file exists on Satellite
result = module_target_sat.execute(f'test -f {remote_report_path}')
assert result.status == 0, f"Report file not found at {remote_report_path}"

# Copy report from Satellite to local temp location
temp_dir = tempfile.mkdtemp()
local_report_name = f'report_for_{org.id}.tar.xz'
report_path = os.path.join(temp_dir, local_report_name)

# Download the file from satellite
module_target_sat.get(remote_path=remote_report_path, local_path=report_path)
Comment on lines +174 to +179
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion (testing): Use a context manager for the temporary directory to avoid leftover files and improve test hygiene

This path uses tempfile.mkdtemp() without cleanup, which can leave stray directories across runs. Wrap it in with tempfile.TemporaryDirectory() as temp_dir: so the directory is cleaned up automatically after the test.


inventory_data = session.cloudinventory.read(org.name)
# Verify that generated archive is valid.
common_assertion(report_path, inventory_data, org, module_target_sat, subscription_setting)
Expand Down
Loading