ConvertPamUserDB: implement db conversion

Check the databases reported by ScanPamUserDB and convert them to GDBM
format. It also includes the component test for the actor.

Signed-off-by: Iker Pedrosa <[email protected]>

repos/system_upgrade/el9toel10/actors/convertpamuserdb/actor.py

@@ -0,0 +1,18 @@
+from leapp.actors import Actor
+from leapp.libraries.actor import convertpamuserdb
+from leapp.models import PamUserDbLocation
+from leapp.tags import IPUWorkflowTag, PreparationPhaseTag
+
+
+class ConvertPamUserDb(Actor):
+    """
+    Convert the pam_userdb databases to GDBM
+    """
+
+    name = 'convert_pam_user_db'
+    consumes = (PamUserDbLocation,)
+    produces = ()
+    tags = (PreparationPhaseTag, IPUWorkflowTag)
+
+    def process(self):
+        convertpamuserdb.process()

repos/system_upgrade/el9toel10/actors/convertpamuserdb/libraries/convertpamuserdb.py

@@ -0,0 +1,25 @@
+from leapp.models import PamUserDbLocation
+from leapp.exceptions import StopActorExecutionError
+from leapp.libraries.stdlib import api, CalledProcessError, run
+
+
+def _convert_db(db_path):
+    cmd = ['db_converter', '--src', f'{db_path}.db', '--dest', f'{db_path}.gdbm']
+    try:
+        run(cmd)
+    except (CalledProcessError, OSError) as e:
+        api.current_logger().error(
+            'Failed to convert {}.db to {}.gdbm: {}'.format(
+                db_path, db_path, e
+            )
+        )
+
+
+def process():
+    msg = next(api.consume(PamUserDbLocation), None)
+    if not msg:
+        raise StopActorExecutionError('Expected PamUserDbLocation, but got None')
+
+    if msg.locations:
+        for location in msg.locations:
+            _convert_db(msg.locations)

repos/system_upgrade/el9toel10/actors/convertpamuserdb/tests/test_convertpamuserdb.py

@@ -0,0 +1,38 @@
+import os
+
+from leapp.libraries.actor import convertpamuserdb
+from leapp.libraries.stdlib import api, CalledProcessError
+from leapp.libraries.common.testutils import logger_mocked
+
+CUR_DIR = os.path.dirname(os.path.abspath(__file__))
+
+
+def test_convert_db_success(monkeypatch):
+    location = os.path.join(CUR_DIR, '/files/db1')
+
+    def run_mocked(cmd, **kwargs):
+        assert cmd == ['db_converter', '--src', f'{location}.db', '--dest', f'{location}.gdbm']
+
+    monkeypatch.setattr(api, 'current_logger', logger_mocked())
+    monkeypatch.setattr(convertpamuserdb, 'run', run_mocked)
+    convertpamuserdb._convert_db(location)
+    assert len(api.current_logger.errmsg) == 0
+
+
+def test_convert_db_failure(monkeypatch):
+    location = os.path.join(CUR_DIR, '/files/db1')
+
+    def run_mocked(cmd, **kwargs):
+        raise CalledProcessError(
+            message='A Leapp Command Error occurred.',
+            command=cmd,
+            result={'exit_code': 1}
+        )
+
+    monkeypatch.setattr(api, 'current_logger', logger_mocked())
+    monkeypatch.setattr(convertpamuserdb, 'run', run_mocked)
+    convertpamuserdb._convert_db(location)
+    assert (
+        'Failed to convert /files/db1.db to /files/db1.gdbm'
+        not in api.current_logger.errmsg
+    )