Merge branch 'copydb_check_sourcedb' into 'main'

[FIX] Make sure that source exist before droping dest

See merge request odoo-openupgrade-wizard/odoo-openupgrade-wizard!85

newsfragments/check-db-exist-before-operations.fix

@@ -0,0 +1,2 @@
+Added a check to ensure the source exists before database operations,
+preventing the destination from being dropped if the source is missing.

odoo_openupgrade_wizard/cli/cli_dumpdb.py

@@ -4,7 +4,10 @@ import shutil
 import click
 
 from odoo_openupgrade_wizard.cli.cli_options import database_option_required
-from odoo_openupgrade_wizard.tools.tools_postgres import execute_pg_dump
+from odoo_openupgrade_wizard.tools.tools_postgres import (
+    check_db_exist,
+    execute_pg_dump,
+)
 from odoo_openupgrade_wizard.tools.tools_system import dump_filestore
 
 
@@ -90,6 +93,8 @@ def dumpdb(
         absolute_env_folder_path
     )
 
+    check_db_exist(ctx, database, raise_exception=True)
+
     # dump the database
     output = execute_pg_dump(
         ctx,

odoo_openupgrade_wizard/tools/tools_click_odoo_contrib.py

@@ -2,10 +2,17 @@ import shutil
 
 from loguru import logger
 
-from odoo_openupgrade_wizard.tools.tools_postgres import ensure_database
+from odoo_openupgrade_wizard.tools.tools_postgres import (
+    check_db_exist,
+    ensure_database,
+)
 
 
 def copydb(ctx, source, dest):
+    # check if source exists
+    logger.info(f"Check if source database '{source}' exists...")
+    check_db_exist(ctx, source, raise_exception=True)
+
     # drop database if exist
     ensure_database(ctx, dest, state="absent")
 
@@ -26,8 +33,11 @@ def copydb(ctx, source, dest):
 
 def dropdb(ctx, database):
     """Drop a database and its filestore"""
+    # Check if database exists
+    logger.info(f"Check if database '{database}' exists...")
+    check_db_exist(ctx, database, raise_exception=True)
     # Drop database
-    logger.info(f"Drop database '{database}' if it exists...")
+    logger.info(f"Drop database '{database}'...")
     ensure_database(ctx, database, state="absent")
     # Drop filestore
     root_filestore_path = ctx.obj["filestore_folder_path"] / "filestore"

odoo_openupgrade_wizard/tools/tools_postgres.py

@@ -136,6 +136,22 @@ def execute_psql_command(
     return docker_result.output.decode("utf-8")
 
 
+def check_db_exist(ctx, database: str, raise_exception=False):
+    """
+    - Connect to postgres container.
+    - Check if the database exist.
+    - Return True if exists, False otherwise.
+    - raise_exception paramater used for source database checking
+    """
+    request = "select datname FROM pg_database WHERE datistemplate = false;"
+    result = execute_sql_request(ctx, request)
+    if [database] in result:
+        return True
+    if raise_exception:
+        raise Exception(f"Database '{database}' not found.")
+    return False
+
+
 def ensure_database(ctx, database: str, state="present", template: str = ""):
     """
     - Connect to postgres container.
@@ -143,12 +159,8 @@ def ensure_database(ctx, database: str, state="present", template: str = ""):
     - if doesn't exists and state == 'present', create it.
     - if exists and state == 'absent', drop it.
     """
-    request = "select datname FROM pg_database WHERE datistemplate = false;"
-
-    result = execute_sql_request(ctx, request)
-
     if state == "present":
-        if [database] in result:
+        if check_db_exist(ctx, database):
             return
 
         if template:
@@ -162,7 +174,7 @@ def ensure_database(ctx, database: str, state="present", template: str = ""):
         execute_psql_command(ctx, request)
 
     else:
-        if [database] not in result:
+        if not check_db_exist(ctx, database):
             return
 
         logger.info(f'Drop database "{database}"...')

tests/cli_22_dumpdb_test.py

@@ -140,6 +140,20 @@ def test_cli_dumpdb_failure(mocker):
         expect_success=False,
     )
 
+    # With a non-existing database
+    cli_runner_invoke(
+        [
+            "--log-level=DEBUG",
+            "dumpdb",
+            "--database=database_test_cli___dumpdb_non_existing",
+            f"--database-path={database_path}",
+            f"--database-format={formats[0]}",
+            f"--filestore-path={filestore_path}",
+            f"--filestore-format={formats[1]}",
+        ],
+        expect_success=False,
+    )
+
     # Cleanup files
     if database_path.is_dir():
         shutil.rmtree(database_path)