[FIX]: Allow to write export sh scripts

bosd · bosd · commit 8eeb6c078a92 · 2025-07-02T11:32:44.000+02:00
diff --git a/src/odoo_data_flow/lib/internal/io.py b/src/odoo_data_flow/lib/internal/io.py
@@ -35,87 +35,110 @@ def write_csv(
         log.error(f"Failed to write to file {filename}: {e}")
 
 
+def _build_import_command(
+    filename: str, model: str, worker: int, batch_size: int, **kwargs: Any
+) -> list[str]:
+    """Builds the command parts for an 'import' shell command."""
+    model_name = (
+        os.path.basename(filename).replace(".csv", "").replace("_", ".")
+        if model == "auto"
+        else model
+    )
+    command_parts = [
+        "odoo-data-flow",
+        "import",
+        "--config",
+        shlex.quote(kwargs.get("conf_file", "conf/connection.conf")),
+        "--file",
+        shlex.quote(filename),
+        "--model",
+        shlex.quote(model_name),
+        "--encoding",
+        shlex.quote(kwargs.get("encoding", "utf-8")),
+        "--worker",
+        str(worker),
+        "--size",
+        str(batch_size),
+        "--sep",
+        shlex.quote(kwargs.get("sep", ";")),
+    ]
+    if kwargs.get("groupby"):
+        command_parts.extend(["--groupby", shlex.quote(kwargs["groupby"])])
+    if kwargs.get("ignore"):
+        command_parts.extend(["--ignore", shlex.quote(kwargs["ignore"])])
+    if kwargs.get("context"):
+        command_parts.extend(["--context", shlex.quote(str(kwargs["context"]))])
+    return command_parts
+
+
+def _build_export_command(filename: str, model: str, **kwargs: Any) -> list[str]:
+    """Builds the command parts for an 'export' shell command."""
+    return [
+        "odoo-data-flow",
+        "export",
+        "--config",
+        shlex.quote(kwargs.get("conf_file", "conf/connection.conf")),
+        "--file",
+        shlex.quote(filename),
+        "--model",
+        shlex.quote(model),
+        "--fields",
+        shlex.quote(kwargs.get("fields", "")),
+        "--domain",
+        shlex.quote(kwargs.get("domain", "[]")),
+        "--sep",
+        shlex.quote(kwargs.get("sep", ";")),
+        "--encoding",
+        shlex.quote(kwargs.get("encoding", "utf-8")),
+    ]
+
+
 def write_file(
     filename: Optional[str] = None,
     header: Optional[list[str]] = None,
     data: Optional[list[list[Any]]] = None,
     fail: bool = False,
     model: str = "auto",
     launchfile: str = "import_auto.sh",
-    worker: int = 1,
-    batch_size: int = 10,
-    init: bool = False,
-    encoding: str = "utf-8",
-    groupby: str = "",
-    sep: str = ";",
-    context: Optional[dict[str, Any]] = None,
-    ignore: str = "",
-    **kwargs: Any,  # to catch other unused params
+    command: str = "import",
+    **kwargs: Any,
 ) -> None:
-    """Filewriter.
+    """Writes data to a CSV and generates a corresponding shell script.
+
+    This function can generate scripts for both `import` and `export` commands
+    based on the `command` parameter.
 
-    Writes data to a CSV file and generates a corresponding shell script
-    to import that file using the odoo-data-flow CLI.
+    Args:
+        filename: The path to the data file to be written or referenced.
+        header: A list of strings for the header row.
+        data: A list of lists representing the data rows.
+        fail: If True (and command is 'import'), includes a second command
+            with the --fail flag.
+        model: The technical name of the Odoo model.
+        launchfile: The path where the shell script will be saved.
+        command: The command to generate in the script ('import' or 'export').
+        **kwargs: Catches other command-specific params like 'worker', 'fields', etc.
     """
-    # Step 1: Write the actual data file
     if filename and header is not None and data is not None:
-        write_csv(filename, header, data, encoding=encoding)
+        write_csv(filename, header, data, encoding=kwargs.get("encoding", "utf-8"))
+
+    if not launchfile or not filename:
+        return
 
-    # Step 2: If no launchfile is specified, we are done.
-    if not launchfile:
+    command_parts: list[str]
+    if command == "import":
+        command_parts = _build_import_command(filename, model, **kwargs)
+    elif command == "export":
+        command_parts = _build_export_command(filename, model, **kwargs)
+    else:
+        log.error(f"Invalid command type '{command}' provided to write_file.")
         return
 
-    # Step 3: Only generate the import script if a filename was provided.
-    if filename:
-        # Determine the target model name
-        if model == "auto":
-            model_name = (
-                os.path.basename(filename).replace(".csv", "").replace("_", ".")
-            )
-        else:
-            model_name = model
-
-        # Build the base command with its arguments
-        # We use shlex.quote to ensure all arguments
-        # are safely escaped for the shell.
-        command_parts = [
-            "odoo-data-flow",
-            "import",
-            "--config",
-            shlex.quote(kwargs.get("conf_file", "conf/connection.conf")),
-            "--file",
-            shlex.quote(filename),
-            "--model",
-            shlex.quote(model_name),
-            "--encoding",
-            shlex.quote(encoding),
-            "--worker",
-            str(worker),
-            "--size",
-            str(batch_size),
-            "--sep",
-            shlex.quote(sep),
-        ]
-
-        # Add optional arguments if they have a value
-        if groupby:
-            command_parts.extend(["--groupby", shlex.quote(groupby)])
-        if ignore:
-            command_parts.extend(["--ignore", shlex.quote(ignore)])
-        if context:
-            command_parts.extend(["--context", shlex.quote(str(context))])
-
-        # Write the command(s) to the shell script
-        mode = "w" if init else "a"
-        try:
-            with open(launchfile, mode, encoding="utf-8") as f:
-                # Write the main import command
-                f.write(" ".join(command_parts) + "\n")
-
-                # If fail mode is enabled,
-                # write the second command with the --fail flag
-                if fail:
-                    fail_command_parts = [*command_parts, "--fail"]
-                    f.write(" ".join(fail_command_parts) + "\n")
-        except OSError as e:
-            log.error(f"Failed to write to launch file {launchfile}: {e}")
+    mode = "w" if kwargs.get("init") else "a"
+    try:
+        with open(launchfile, mode, encoding="utf-8") as f:
+            f.write(" ".join(command_parts) + "\n")
+            if fail and command == "import":
+                f.write(" ".join([*command_parts, "--fail"]) + "\n")
+    except OSError as e:
+        log.error(f"Failed to write to launch file {launchfile}: {e}")
diff --git a/tests/test_io.py b/tests/test_io.py
@@ -1,7 +1,5 @@
 """Test the IO Handling functionalities."""
 
-# tests/test_io.py
-
 import shlex
 from pathlib import Path
 from unittest.mock import MagicMock, patch
@@ -38,61 +36,52 @@ def test_write_file_writes_csv_data(tmp_path: Path) -> None:
             filename=str(data_file),
             header=["id", "name"],
             data=[["1", "test"]],
-            launchfile="",  # Correctly pass an empty string instead of None
+            launchfile="",
         )
         mock_write_csv.assert_called_once_with(
             str(data_file), ["id", "name"], [["1", "test"]], encoding="utf-8"
         )
 
 
-@patch("odoo_data_flow.lib.internal.io.write_csv")  # Mock the CSV writing part
+@patch("odoo_data_flow.lib.internal.io.write_csv")
 @patch("odoo_data_flow.lib.internal.io.open")
 def test_write_file_no_launchfile(
     mock_open: MagicMock, mock_write_csv: MagicMock, tmp_path: Path
 ) -> None:
     """Tests that write_file exits early if no launchfile is specified."""
     data_file = tmp_path / "data.csv"
-
     write_file(
         filename=str(data_file),
         header=["id"],
         data=[["1"]],
-        launchfile="",  # Empty string means no script
+        launchfile="",
     )
-
-    # Assert that write_csv was called, but open was not (for the launchfile)
     mock_write_csv.assert_called_once()
     mock_open.assert_not_called()
 
 
-def test_write_file_full_script_generation(tmp_path: Path) -> None:
-    """Tests that write_file generates a complete shell script with all options."""
-    # 1. Setup
+def test_write_file_import_command(tmp_path: Path) -> None:
+    """Tests that write_file generates a complete import shell script."""
     script_file = tmp_path / "load.sh"
     data_file = tmp_path / "my_model.csv"
-
-    # 2. Action
     write_file(
         filename=str(data_file),
         header=["id", "name"],
         data=[["1", "test"]],
         launchfile=str(script_file),
+        command="import",
         model="my.model",
         fail=True,
         init=True,
         worker=4,
         batch_size=50,
         groupby="parent_id/id",
         ignore="field_to_ignore",
-        context={"active_test": False},  # Correctly pass a dict instead of a string
+        context={"active_test": False},
         conf_file="conf/custom.conf",
     )
-
-    # 3. Assertions
     assert script_file.exists()
     content = script_file.read_text()
-
-    # Check for the main command
     assert "odoo-data-flow import" in content
     assert f"--config {shlex.quote('conf/custom.conf')}" in content
     assert f"--file {shlex.quote(str(data_file))}" in content
@@ -102,33 +91,55 @@ def test_write_file_full_script_generation(tmp_path: Path) -> None:
     assert f"--groupby {shlex.quote('parent_id/id')}" in content
     assert f"--ignore {shlex.quote('field_to_ignore')}" in content
     assert f"--context {shlex.quote(str({'active_test': False}))}" in content
-
-    # Check for the second command with the --fail flag
     assert "--fail" in content
-    # Count occurrences to ensure both commands are present
     assert content.count("odoo-data-flow import") == 2
 
 
+def test_write_file_export_command(tmp_path: Path) -> None:
+    """Tests that write_file generates a correct export shell script."""
+    script_file = tmp_path / "export.sh"
+    data_file = tmp_path / "partner_export.csv"
+    domain_str = "[('is_company', '=', True)]"
+    write_file(
+        filename=str(data_file),
+        launchfile=str(script_file),
+        command="export",
+        model="res.partner",
+        fields="id,name",
+        domain=domain_str,
+        init=True,
+    )
+    assert script_file.exists()
+    content = script_file.read_text()
+    assert "odoo-data-flow export" in content
+    assert f"--model {shlex.quote('res.partner')}" in content
+    assert f"--fields {shlex.quote('id,name')}" in content
+    expected_domain_str = f"--domain {shlex.quote(domain_str)}"
+    assert expected_domain_str in content
+    assert "--fail" not in content
+
+
 def test_write_file_auto_model_name(tmp_path: Path) -> None:
     """Tests that the model name is correctly inferred when model='auto'."""
     script_file = tmp_path / "load_auto.sh"
     data_file = tmp_path / "res.partner.csv"
-
     write_file(
         filename=str(data_file),
         header=["id"],
         data=[["1"]],
         launchfile=str(script_file),
         model="auto",
         init=True,
+        # Provide default worker and batch_size to avoid TypeError
+        worker=1,
+        batch_size=10,
     )
-
     content = script_file.read_text()
-    # The model name should be inferred from 'res.partner.csv' -> 'res.partner'
+    # The model name should be inferred from 'res_partner.csv' -> 'res.partner'
     assert f"--model {shlex.quote('res.partner')}" in content
 
 
-@patch("odoo_data_flow.lib.internal.io.write_csv")  # Mock the CSV part
+@patch("odoo_data_flow.lib.internal.io.write_csv")
 @patch("odoo_data_flow.lib.internal.io.open")
 @patch("odoo_data_flow.lib.internal.io.log.error")
 def test_write_file_oserror(
@@ -138,19 +149,30 @@ def test_write_file_oserror(
 
     Tests that write_file logs an error if an OSError occurs during script writing.
     """
-    # 1. Setup: This time, the 'open' for the launchfile will fail
     mock_open.side_effect = OSError("Permission denied on script file")
 
-    # 2. Action
     write_file(
         filename="data.csv",
         header=["id"],
         data=[["1"]],
         launchfile="protected/load.sh",
         init=True,
+        # Provide default worker and batch_size to avoid TypeError
+        worker=1,
+        batch_size=10,
     )
-
-    # 3. Assertions
-    mock_write_csv.assert_called_once()  # Ensure the CSV part was attempted
+    mock_write_csv.assert_called_once()
     mock_log_error.assert_called_once()
     assert "Failed to write to launch file" in mock_log_error.call_args[0][0]
+
+
+@patch("odoo_data_flow.lib.internal.io.log.error")
+def test_write_file_invalid_command(mock_log_error: MagicMock) -> None:
+    """Tests that an error is logged for an invalid command type."""
+    write_file(
+        filename="dummy.csv",
+        launchfile="dummy.sh",
+        command="invalid-command",
+    )
+    mock_log_error.assert_called_once()
+    assert "Invalid command type" in mock_log_error.call_args[0][0]
