Fix query db in containerized envs (#22140)

* Update query_db function to support containerized envs

* update test_health.py

* ruff format

* move Install method import to top level

* remove local imports

* add db_user arg to query_db()

* replace with self.install_method with settings.server.install_method

Author: aidenfine

robottelo/hosts.py

@@ -54,7 +54,7 @@
     RHSSO_USER_UPDATE,
     SATELLITE_VERSION,
 )
-from robottelo.enums import NetworkType
+from robottelo.enums import InstallMethod, NetworkType
 from robottelo.exceptions import (
     CapsuleHostError,
     CLIFactoryError,
@@ -1805,7 +1805,6 @@ def detect_install_method(self):
         :return: InstallMethod enum value
         :rtype: InstallMethod
         """
-        from robottelo.enums import InstallMethod
 
         # Runtime override
         if hasattr(self, '_install_method_override'):
@@ -1855,7 +1854,6 @@ def get_service_names(self):
         :rtype: list
         """
         from robottelo.constants import InstallationServices
-        from robottelo.enums import InstallMethod
 
         if self.install_method == InstallMethod.FOREMANCTL:
             return InstallationServices.FOREMANCTL_SERVICES
@@ -2271,7 +2269,6 @@ def install_satellite(
         :param foremanctl_parameters: Parameters list for foremanctl deploy
         :return: Installation result
         """
-        from robottelo.enums import InstallMethod
         from robottelo.utils.installer import InstallerCommand
 
         # Determine method
@@ -2324,13 +2321,14 @@ def install_satellite(
 
         return result
 
-    def query_db(self, query, db='foreman', output_format='json'):
+    def query_db(self, query, db='foreman', output_format='json', db_user='foreman'):
         """Execute a PostgreSQL query and return the result.
 
         Args:
             query: SQL query to execute
             db: Database name (default: 'foreman')
             output_format: Output format - 'json' for JSON array, raw output otherwise
+            db_user: Database user (default: 'foreman')
 
         Returns:
             list of dicts if output_format='json', str otherwise
@@ -2345,7 +2343,10 @@ def _execute_db_query(cmd):
                 raise CLIReturnCodeError(result.status, result.stderr, f'"{cmd}" failed')
             return result
 
-        base_cmd = f'sudo -u postgres psql -d {db}'
+        if settings.server.install_method == InstallMethod.FOREMANCTL:
+            base_cmd = f'podman exec postgresql psql -U {db_user} -d {db}'
+        else:
+            base_cmd = f'sudo -u postgres psql -d {db}'
 
         if output_format == 'json':
             cmd = f'{base_cmd} -A -t -c "SELECT json_agg(row_to_json(t)) FROM ({query}) t"'

tests/foreman/api/test_contentview.py

@@ -2723,11 +2723,12 @@ def test_repository_rpms_id_type(target_sat):
 
     :CaseImportance: Medium
     """
-    db_out = target_sat.execute(
-        'sudo -u postgres psql -d foreman -c "select * from pg_sequences where sequencename=\'katello_repository_rpms_id_seq\';"'
+    db_out = target_sat.query_db(
+        "select * from pg_sequences where sequencename='katello_repository_rpms_id_seq'",
+        output_format='raw',
     )
-    assert 'bigint' in db_out.stdout
-    assert 'integer' not in db_out.stdout
+    assert 'bigint' in db_out
+    assert 'integer' not in db_out
 
 
 def test_negative_readonly_user_actions(

tests/foreman/api/test_notifications.py

@@ -233,15 +233,15 @@ def long_running_task(target_sat):
         },
     )
     sql_date_2_days_ago = "now() - INTERVAL '2 days'"
-    result = target_sat.execute(
-        "su - postgres -c \"psql foreman postgres <<EOF\n"
+    query = (
         "UPDATE foreman_tasks_tasks "
         f"SET start_at = {sql_date_2_days_ago}, "
-        f" started_at = {sql_date_2_days_ago}, "
-        f" state_updated_at = {sql_date_2_days_ago} "
-        f"WHERE id=\'{job['task']['id']}\';\nEOF\n\" "
-    )  # fmt: skip  # skip formatting to avoid breaking the SQL query
-    assert 'UPDATE 1' in result.stdout, f'Failed to age task {job["task"]["id"]}: {result.stderr}'
+        f"started_at = {sql_date_2_days_ago}, "
+        f"state_updated_at = {sql_date_2_days_ago} "
+        f"WHERE id='{job['task']['id']}'"
+    )
+    result = target_sat.query_db(query, output_format='raw')
+    assert 'UPDATE 1' in result, f'Failed to age task {job["task"]["id"]}'
 
     yield job
 

tests/foreman/api/test_repository.py

@@ -1377,13 +1377,14 @@ def test_positive_sync_repo_null_contents_changed(self, module_sca_manifest_org,
             releasever=None,
         )
         target_sat.api.Repository(id=repo_id).sync()
-        prod_log_out = target_sat.execute(
-            'sudo -u postgres psql -d foreman -c "select class,execution_plan_uuid,input '
-            'from dynflow_actions where input LIKE \'%"contents_changed":null%\''
-            ' AND class = \'Actions::Katello::Repository::Sync\';"'
+        assert (
+            target_sat.query_db(
+                "select class,execution_plan_uuid,input "
+                "from dynflow_actions where input LIKE '%\"contents_changed\":null%'"
+                " AND class = 'Actions::Katello::Repository::Sync'"
+            )
+            == []
         )
-        assert prod_log_out.status == 0
-        assert "(0 rows)" in prod_log_out.stdout
 
     def test_positive_validate_async_operation_response(self, module_sca_manifest_org, target_sat):
         """Verify that RefreshDistribution action properly tracks Pulp tasks via AsyncOperationResponse.

tests/foreman/cli/test_oscap.py

@@ -619,16 +619,8 @@ def test_positive_associate_scap_policy_with_hostgroups(self, scap_content, modu
             pytest.fail("failed to list policies")
         assert name in [policy['name'] for policy in result]
         # check for orphaned entries
-        db_out = module_target_sat.execute(
-            'sudo -u postgres psql -d foreman -c "select * from foreman_openscap_assets"'
-        )
-        assert db_out.status == 0
-        assert "(0 rows)" in db_out.stdout
-        db_out = module_target_sat.execute(
-            'sudo -u postgres psql -d foreman -c "select * from foreman_openscap_asset_policies"'
-        )
-        assert db_out.status == 0
-        assert "(0 rows)" in db_out.stdout
+        assert module_target_sat.query_db('select * from foreman_openscap_assets') == []
+        assert module_target_sat.query_db('select * from foreman_openscap_asset_policies') == []
 
     def test_positive_associate_scap_policy_with_hostgroup_via_ansible(
         self, scap_content, module_target_sat

tests/foreman/maintain/test_health.py

@@ -667,24 +667,20 @@ def test_positive_health_check_corrupted_roles(sat_maintain, request):
 
     @request.addfinalizer
     def _finalize():
-        resource_type = r"'\''Host'\''"
-        sat_maintain.execute(
-            f'''sudo su - postgres -c "psql -d foreman -c 'UPDATE permissions SET
-                        resource_type = {resource_type} WHERE name = {permission_name};'"'''
+        sat_maintain.query_db(
+            "UPDATE permissions SET resource_type = 'Host' WHERE name = 'console_hosts'",
+            output_format='raw',
         )
         sat_maintain.cli.Role.delete(options={'name': role_name})
 
     # Check the filter created to verify the role, resource type, and permissions assigned.
     sat_maintain.cli.Filter.create(
         options={'role': role_name, 'permissions': ['view_hosts', 'console_hosts']}
     )
-    permission_name = r"'\''console_hosts'\''"
-    resource_type = rf"'\''{resource_type}'\''"
-    setup = sat_maintain.execute(
-        f'''sudo su - postgres -c "psql -d foreman -c 'UPDATE permissions SET
-            resource_type = {resource_type} WHERE name = {permission_name};'"'''
+    sat_maintain.query_db(
+        f"UPDATE permissions SET resource_type = '{resource_type}' WHERE name = 'console_hosts'",
+        output_format='raw',
     )
-    assert setup.status == 0
     result = sat_maintain.cli.Filter.list(options={'search': role_name}, output_format='yaml')
     # Shows the filter id which comprises of role id hence asserting 2 here
     assert result.count('Id') == 2
@@ -758,13 +754,11 @@ def test_positive_health_check_duplicate_permissions(sat_maintain):
     :BZ: 1849110, 1884024
     """
     # Verify if check failed because of duplicate permissions
-    name = r"'\''view_ansible_variables'\''"
-    resource_type = r"'\''AnsibleVariable'\''"
-    result = sat_maintain.execute(
-        f'''sudo su - postgres -c "psql -d foreman -c 'INSERT INTO permissions(name, resource_type)
-            VALUES({name}, {resource_type});'"'''
+    sat_maintain.query_db(
+        "INSERT INTO permissions(name, resource_type) "
+        "VALUES('view_ansible_variables', 'AnsibleVariable')",
+        output_format='raw',
     )
-    assert result.status == 0
     result = sat_maintain.cli.Health.check({'label': 'duplicate-permissions', 'assumeyes': True})
     assert result.status == 0
     assert 'FAIL' in result.stdout

tests/foreman/ui/test_host.py

@@ -4531,19 +4531,19 @@ def test_positive_search_by_report_origin_shows_all_hosts(target_sat):
     timestamp = '2025-05-16 16:16:16'
     report_count = 25
 
-    target_sat.execute(
-        'su - postgres -c "psql -d foreman -c \\"'
+    target_sat.query_db(
         f"INSERT INTO reports (host_id, origin, reported_at) "
-        f"SELECT {host1.id}, 'Puppet', '{timestamp}'::timestamp + (i || ' seconds')::interval "
-        f"FROM generate_series(1, {report_count}) AS i"
-        '\\""'
+        f"SELECT {host1.id}, 'Puppet', '{timestamp}'::timestamp "
+        f"+ (i || ' seconds')::interval "
+        f"FROM generate_series(1, {report_count}) AS i",
+        output_format='raw',
     )
-    target_sat.execute(
-        'su - postgres -c "psql -d foreman -c \\"'
+    target_sat.query_db(
         f"INSERT INTO reports (host_id, origin, reported_at) "
-        f"SELECT {host2.id}, 'Puppet', '{timestamp}'::timestamp + (i || ' seconds')::interval "
-        f"FROM generate_series(1, 2) AS i"
-        '\\""'
+        f"SELECT {host2.id}, 'Puppet', '{timestamp}'::timestamp "
+        f"+ (i || ' seconds')::interval "
+        f"FROM generate_series(1, 2) AS i",
+        output_format='raw',
     )
 
     with target_sat.ui_session() as session: