Add an action to pre-check if no MariaDB repository is enabled

pleskdistup/actions/mariadb.py

@@ -5,7 +5,7 @@
 import typing
 from abc import ABC, abstractmethod
 
-from pleskdistup.common import action, dpkg, files, log, mariadb, motd, packages, systemd
+from pleskdistup.common import action, dpkg, files, log, mariadb, motd, packages, rpm, systemd
 
 
 MARIADB_VERSION_ON_UBUNTU_20 = mariadb.MariaDBVersion("10.3.38")
@@ -29,6 +29,31 @@ def _revert_action(self) -> action.ActionResult:
         return action.ActionResult()
 
 
+class AssertMariadbRepoEnabled(action.CheckAction):
+    def __init__(
+        self,
+        mariadb_version_on_target: mariadb.MariaDBVersion,
+        known_repo_files: typing.List[str],
+        repo_dir: str = "/etc/yum.repos.d",
+    ):
+        self.name = "check an enabled mariadb repository is available"
+        self.mariadb_version_on_target = mariadb_version_on_target
+        self.known_repo_files = known_repo_files
+        self.repo_dir = repo_dir
+        self.description = """MariaDB is installed from a disabled or missing repository since there are
+\tno enabled repositories found in any of the known repository files:
+\t- {files}
+\tTo proceed with the conversion, enable a MariaDB repository in one of these files.
+""".format(files="\n\t- ".join([repo_dir + '/' + repo_file for repo_file in known_repo_files]))
+
+    def _do_check(self) -> bool:
+        if not mariadb.is_mariadb_installed() or mariadb.get_installed_mariadb_version() <= self.mariadb_version_on_target:
+            return True
+
+        repofiles = files.find_files_case_insensitive(self.repo_dir, self.known_repo_files)
+        return any(rpm.has_enabled_repository(repofile) for repofile in repofiles)
+
+
 def get_db_server_config_file():
     return mariadb.get_mysql_config_file_path() if mariadb.is_mysql_installed() else mariadb.get_mariadb_config_file_path()
 

pleskdistup/common/src/rpm.py

@@ -525,6 +525,20 @@ def is_repository_url_enabled(repository_url: str, sources_dir: str = "/etc/yum.
     return False
 
 
+def has_enabled_repository(repofile: str) -> bool:
+    """
+    Check whether the given .repo file contains at least one enabled repository.
+    A repository without an explicit ``enabled=`` line is treated as enabled (yum default).
+    A non-existent file is treated as having no enabled repository.
+    """
+    if not os.path.exists(repofile):
+        return False
+    for repo in extract_repodata(repofile):
+        if repo.enabled is None or repo.enabled.strip() != "0":
+            return True
+    return False
+
+
 def get_package_installed_version(package_name: str) -> typing.Optional[str]:
     """
     Get the installed version of the package.

pleskdistup/common/tests/rpmtests.py

@@ -1814,3 +1814,71 @@ def test_with_exclude_line_partial_leapp(self):
             rpm.yum_conf_get_exclude_list())
         rpm.yum_conf_rm_leapp_disablement()
         self.assertEqual(['test52', 'test22'], rpm.yum_conf_get_exclude_list())
+
+
+class HasEnabledRepositoryTests(unittest.TestCase):
+    TEST_DIR = "has_enabled_repository_test_dir"
+
+    def setUp(self):
+        os.makedirs(self.TEST_DIR, exist_ok=True)
+
+    def tearDown(self):
+        if os.path.exists(self.TEST_DIR):
+            shutil.rmtree(self.TEST_DIR)
+
+    def _write_repofile(self, name, content):
+        path = os.path.join(self.TEST_DIR, name)
+        with open(path, "w") as f:
+            f.write(content)
+        return path
+
+    def test_enabled_repository(self):
+        path = self._write_repofile("enabled.repo", """[repo1]
+name=repo1
+baseurl=http://repo1
+enabled=1
+gpgcheck=0
+""")
+        self.assertTrue(rpm.has_enabled_repository(path))
+
+    def test_all_disabled_repositories(self):
+        path = self._write_repofile("disabled.repo", """[repo1]
+name=repo1
+baseurl=http://repo1
+enabled=0
+gpgcheck=0
+
+[repo2]
+name=repo2
+baseurl=http://repo2
+enabled=0
+gpgcheck=0
+""")
+        self.assertFalse(rpm.has_enabled_repository(path))
+
+    def test_missing_enabled_line_is_enabled(self):
+        path = self._write_repofile("no_enabled.repo", """[repo1]
+name=repo1
+baseurl=http://repo1
+gpgcheck=0
+""")
+        self.assertTrue(rpm.has_enabled_repository(path))
+
+    def test_one_enabled_among_disabled(self):
+        path = self._write_repofile("mixed.repo", """[repo1]
+name=repo1
+baseurl=http://repo1
+enabled=0
+gpgcheck=0
+
+[repo2]
+name=repo2
+baseurl=http://repo2
+enabled=1
+gpgcheck=0
+""")
+        self.assertTrue(rpm.has_enabled_repository(path))
+
+    def test_non_existent_file(self):
+        self.assertFalse(rpm.has_enabled_repository(
+            os.path.join(self.TEST_DIR, "does_not_exist.repo")))