RemoveOldPamUserDB: implement old db removal

Check the databases reported by ScanPamUserDB and remove them. It also
includes the unit-tests.

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

repos/system_upgrade/el9toel10/actors/pamuserdb/removeoldpamuserdb/actor.py

@@ -0,0 +1,18 @@
+from leapp.actors import Actor
+from leapp.libraries.actor import removeoldpamuserdb
+from leapp.models import PamUserDbLocation
+from leapp.tags import ApplicationsPhaseTag, IPUWorkflowTag
+
+
+class RemoveOldPamUserDb(Actor):
+    """
+    Remove old pam_userdb databases
+    """
+
+    name = 'remove_old_pam_user_db'
+    consumes = (PamUserDbLocation,)
+    produces = ()
+    tags = (ApplicationsPhaseTag, IPUWorkflowTag)
+
+    def process(self):
+        removeoldpamuserdb.process()

repos/system_upgrade/el9toel10/actors/pamuserdb/removeoldpamuserdb/libraries/removeoldpamuserdb.py

@@ -0,0 +1,25 @@
+from leapp.models import PamUserDbLocation
+from leapp.exceptions import StopActorExecutionError
+from leapp.libraries.stdlib import api, CalledProcessError, run
+
+
+def _remove_db(db_path):
+    cmd = ['rm', '-f', f'{db_path}.db']
+    try:
+        run(cmd)
+    except (CalledProcessError, OSError) as e:
+        api.current_logger().error(
+            'Failed to remove {}.db: {}'.format(
+                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:
+            _remove_db(location)

repos/system_upgrade/el9toel10/actors/pamuserdb/removeoldpamuserdb/tests/test_removeoldpamuserdb.py

@@ -0,0 +1,38 @@
+import os
+
+from leapp.libraries.actor import removeoldpamuserdb
+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_remove_db_success(monkeypatch):
+    location = os.path.join(CUR_DIR, '/files/db1')
+
+    def run_mocked(cmd, **kwargs):
+        assert cmd == ['rm', '-f', f'{location}.db']
+
+    monkeypatch.setattr(api, 'current_logger', logger_mocked())
+    monkeypatch.setattr(removeoldpamuserdb, 'run', run_mocked)
+    removeoldpamuserdb._remove_db(location)
+    assert len(api.current_logger.errmsg) == 0
+
+
+def test_remove_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(removeoldpamuserdb, 'run', run_mocked)
+    removeoldpamuserdb._remove_db(location)
+    assert (
+        'Failed to remove /files/db1.db'
+        not in api.current_logger.errmsg
+    )