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,24 @@
+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 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:
+        _convert_db(msg.locations)

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

@@ -0,0 +1,48 @@
+import os
+
+from leapp.libraries.actor import convertpamuserdb
+from leapp.libraries.stdlib import api, CalledProcessError
+
+CUR_DIR = os.path.dirname(os.path.abspath(__file__))
+
+
+class LoggerMocked(object):
+    def __init__(self):
+        self.errormsg = None
+
+    def error(self, *args):
+        self.errormsg = args
+
+    def __call__(self):
+        return self
+
+
+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', LoggerMocked())
+    monkeypatch.setattr(convertpamuserdb, 'run', run_mocked)
+    convertpamuserdb._convert_db(location)
+    assert api.current_logger.errormsg is None
+
+
+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={'signal': None, 'exit_code': 1, 'pid': 0, 'stdout': 'fake', 'stderr': 'fake'}
+        )
+
+    monkeypatch.setattr(api, 'current_logger', LoggerMocked())
+    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.errormsg
+    )