CSV formatting fixes (#15689)

pytest_fixtures/component/satellite_auth.py

@@ -463,10 +463,10 @@ def configure_hammer_no_negotiate(parametrized_enrolled_sat):
 def hammer_logout(parametrized_enrolled_sat):
     """Logout in Hammer."""
     result = parametrized_enrolled_sat.cli.Auth.logout()
-    assert result.split("\n")[1] == LOGGEDOUT
+    assert result[0]['message'] == LOGGEDOUT
     yield
     result = parametrized_enrolled_sat.cli.Auth.logout()
-    assert result.split("\n")[1] == LOGGEDOUT
+    assert result[0]['message'] == LOGGEDOUT
 
 
 @pytest.fixture

robottelo/cli/hammer.py

@@ -4,6 +4,8 @@
 import json
 import re
 
+from robottelo.logging import logger
+
 
 def _normalize(header):
     """Replace empty spaces with '-' and lower all chars"""
@@ -35,31 +37,17 @@ def _normalize_obj(obj):
     return obj
 
 
-def is_csv(output):
-    """Verifies if the output string is eligible for converting into CSV"""
-    sniffer = csv.Sniffer()
-    try:
-        sniffer.sniff(output)
-        return True
-    except csv.Error:
-        return False
-
-
 def parse_csv(output):
-    """Parse CSV output from Hammer CLI and convert it to python dictionary."""
-    # ignore warning about puppet and ostree deprecation
-    output.replace('Puppet and OSTree will no longer be supported in Katello 3.16\n', '')
-    is_rex = 'Job invocation' in output
-    is_pkg_list = 'Nvra' in output
-    # Validate if the output is eligible for CSV conversions else return as it is
-    if not is_csv(output) and not is_rex and not is_pkg_list:
-        return output
-    output = output.splitlines()[0:2] if is_rex else output.splitlines()
-    reader = csv.reader(output)
-    # Generate the key names, spaces will be converted to dashes "-"
-    keys = [_normalize(header) for header in next(reader)]
-    # For each entry, create a dict mapping each key with each value
-    return [dict(zip(keys, values, strict=True)) for values in reader if len(values) > 0]
+    """Parse CSV output from Hammer CLI and return a Python dictionary."""
+    output = output.splitlines()
+
+    # Normalize the column names to use when generating the dictionary
+    try:
+        keys = [_normalize(header) for header in next(csv.reader(output))]
+        return [value for value in csv.DictReader(output[1:], fieldnames=keys)]
+    except csv.Error as err:
+        logger.error(f'Exception while parsing CSV output {output}: {err}')
+        raise
 
 
 def parse_help(output):

robottelo/cli/repository.py

@@ -62,7 +62,7 @@ def synchronize(cls, options, return_raw_response=None, timeout=3600000):
         cls.command_sub = 'synchronize'
         return cls.execute(
             cls._construct_command(options),
-            output_format='base',
+            output_format='csv',
             ignore_stderr=True,
             return_raw_response=return_raw_response,
             timeout=timeout,

tests/foreman/cli/test_ldapauthsource.py

@@ -102,7 +102,7 @@ def test_positive_refresh_usergroup_with_ad(self, member_group, ad_data, module_
             based on user-sync
         """
         ad_data = ad_data()
-        LOGEDIN_MSG = "Using configured credentials for user '{0}'."
+        LOGGEDIN_MSG = "Using configured credentials for user '{0}'."
         auth_source = module_target_sat.cli_factory.ldap_auth_source(
             {
                 'name': gen_string('alpha'),
@@ -134,7 +134,7 @@ def test_positive_refresh_usergroup_with_ad(self, member_group, ad_data, module_
         result = module_target_sat.cli.Auth.with_user(
             username=ad_data['ldap_user_name'], password=ad_data['ldap_user_passwd']
         ).status()
-        assert LOGEDIN_MSG.format(ad_data['ldap_user_name']) in result.split("\n")[1]
+        assert LOGGEDIN_MSG.format(ad_data['ldap_user_name']) in result[0]['message']
         module_target_sat.cli.UserGroupExternal.refresh(
             {'user-group-id': user_group['id'], 'name': member_group}
         )
@@ -221,7 +221,7 @@ def test_usergroup_sync_with_refresh(self, default_ipa_host, module_target_sat):
         ipa_group_base_dn = default_ipa_host.group_base_dn.replace('foobargroup', 'foreman_group')
         member_username = 'foreman_test'
         member_group = 'foreman_group'
-        LOGEDIN_MSG = "Using configured credentials for user '{0}'."
+        LOGGEDIN_MSG = "Using configured credentials for user '{0}'."
         auth_source_name = gen_string('alpha')
         auth_source = module_target_sat.cli_factory.ldap_auth_source(
             {
@@ -262,7 +262,7 @@ def test_usergroup_sync_with_refresh(self, default_ipa_host, module_target_sat):
         result = module_target_sat.cli.Auth.with_user(
             username=member_username, password=default_ipa_host.ldap_user_passwd
         ).status()
-        assert LOGEDIN_MSG.format(member_username) in result[0]['message']
+        assert LOGGEDIN_MSG.format(member_username) in result[0]['message']
         with pytest.raises(CLIReturnCodeError) as error:
             module_target_sat.cli.Role.with_user(
                 username=member_username, password=default_ipa_host.ldap_user_passwd
@@ -308,7 +308,7 @@ def test_usergroup_with_usergroup_sync(self, default_ipa_host, module_target_sat
         ipa_group_base_dn = default_ipa_host.group_base_dn.replace('foobargroup', 'foreman_group')
         member_username = 'foreman_test'
         member_group = 'foreman_group'
-        LOGEDIN_MSG = "Using configured credentials for user '{0}'."
+        LOGGEDIN_MSG = "Using configured credentials for user '{0}'."
         auth_source_name = gen_string('alpha')
         auth_source = module_target_sat.cli_factory.ldap_auth_source(
             {
@@ -349,7 +349,7 @@ def test_usergroup_with_usergroup_sync(self, default_ipa_host, module_target_sat
         result = module_target_sat.cli.Auth.with_user(
             username=member_username, password=default_ipa_host.ldap_user_passwd
         ).status()
-        assert LOGEDIN_MSG.format(member_username) in result[0]['message']
+        assert LOGGEDIN_MSG.format(member_username) in result[0]['message']
         list = module_target_sat.cli.Role.with_user(
             username=member_username, password=default_ipa_host.ldap_user_passwd
         ).list()

tests/foreman/destructive/test_auth.py

@@ -18,7 +18,7 @@
 from robottelo.config import settings
 from robottelo.constants import HAMMER_CONFIG
 
-LOGEDIN_MSG = "Session exists, currently logged in as '{0}'"
+LOGGEDIN_MSG = "Session exists, currently logged in as '{0}'"
 password = gen_string('alpha')
 pytestmark = pytest.mark.destructive
 
@@ -45,7 +45,7 @@ def test_positive_password_reset(target_sat):
         {'username': settings.server.admin_username, 'password': reset_password}
     )
     result = target_sat.cli.Auth.with_user().status()
-    assert LOGEDIN_MSG.format(settings.server.admin_username) in result.split("\n")[1]
+    assert LOGGEDIN_MSG.format(settings.server.admin_username) in result[0]['message']
     assert target_sat.cli.Org.with_user().list()
 
 
@@ -72,5 +72,5 @@ def test_positive_password_reset_chosen(target_sat):
         {'username': settings.server.admin_username, 'password': new_password}
     )
     result = target_sat.cli.Auth.with_user().status()
-    assert LOGEDIN_MSG.format(settings.server.admin_username) in result.split("\n")[1]
+    assert LOGGEDIN_MSG.format(settings.server.admin_username) in result[0]['message']
     assert target_sat.cli.Org.with_user().list()