feat(mssql): add native bak backup format

engines/mssql.py

@@ -2,13 +2,38 @@
 
 from typing import Any
 
+from core.errors import ValidationError
 from core.fields import Field
 from core.specs import DatabaseSpec
 from core.utils import generate_password
 from engines.base import StandardSqlEngine
+from engines.files import host_dir
 from services.ports import PortAllocator
 
 
+def bak_only(answers: dict[str, Any]) -> bool:
+    return answers.get("backup_format") == "bak"
+
+
+def server_folder(value: str) -> str:
+    folder = value.strip()
+    if not folder:
+        raise ValidationError(
+            "bak_server_path is required: the shared folder as SQL Server sees it.",
+            hint=r"-o bak_server_path=D:\PortabaseBackups",
+        )
+    return folder
+
+
+def agent_folder(value: str) -> str:
+    if not value.strip():
+        raise ValidationError(
+            "bak_agent_path is required: the same folder on the agent host.",
+            hint="-o bak_agent_path=/mnt/sql-backups",
+        )
+    return host_dir(value)
+
+
 class MssqlEngine(StandardSqlEngine):
     key, display, default_port = "mssql", "Microsoft SQL Server", 1433
     template, slug, db_prefix = "engines/mssql.yml.j2", "mssql", "master"
@@ -20,20 +45,66 @@ def option_fields(self) -> list[Field]:
                 "Backup format",
                 "choice",
                 default="dacpac",
-                choices=("dacpac", "bacpac"),
+                choices=("dacpac", "bacpac", "bak"),
                 help=(
                     "dacpac: sqlpackage Extract with all table data (default, works "
                     "with Windows principals). bacpac: sqlpackage Export, validated "
                     "against Azure SQL rules; fails with SQL71627 on databases holding "
                     "Windows principals such as NT AUTHORITY\\SYSTEM. Requires agent "
-                    "1.22.0 or later."
+                    "1.22.0 or later. bak: native BACKUP DATABASE into a folder shared "
+                    "with SQL Server; for applications that reference other databases "
+                    "or are written to during backups. Existing servers only, not "
+                    "Azure SQL Database. Requires agent 1.26.0 or later."
                 ),
             ),
+            Field(
+                "bak_server_path",
+                "Shared folder as SQL Server sees it",
+                "text",
+                default="",
+                validator=server_folder,
+                when=bak_only,
+                help=(
+                    r"e.g. D:\PortabaseBackups or \\nas\sql-backups. The SQL Server "
+                    "service account must be able to write there."
+                ),
+            ),
+            Field(
+                "bak_agent_path",
+                "Same folder on the agent host (absolute)",
+                "text",
+                default="",
+                validator=agent_folder,
+                when=bak_only,
+                help=(
+                    "e.g. /mnt/sql-backups. Mounted into the agent at the same path. "
+                    "The agent must be able to read, write and delete there."
+                ),
+            ),
+            Field(
+                "bak_verify",
+                "Verify each backup with RESTORE VERIFYONLY?",
+                "bool",
+                default=False,
+                when=bak_only,
+                help="Reads the whole backup again on SQL Server after each run.",
+            ),
         ]
 
+    @staticmethod
+    def mount_for(spec: DatabaseSpec) -> tuple[str, str] | None:
+        path = spec.options.get("bak_agent_path")
+        return (path, path) if path else None
+
     def generate(
         self, *, auth: bool, ports: PortAllocator, answers: dict[str, Any]
     ) -> DatabaseSpec:
+        if bak_only(answers.get("options", {})):
+            raise ValidationError(
+                "bak needs a folder shared with SQL Server, so it only works on an "
+                "existing database.",
+                hint="Use --mode existing, or dacpac for a new database.",
+            )
         return DatabaseSpec(
             id=self.new_id(),
             engine=self.key,

tests/engines/mssql.py

@@ -1,11 +1,25 @@
 import re
+from typing import cast
 
+import pytest
+import yaml
+
+from core.errors import ValidationError
 from core.specs import DatabaseSpec
 from engines import registry
 from engines.mssql import MssqlEngine
-from tests.support import EXISTING_ANSWERS, field_specs
+from services.project import AgentProject
+from tests.support import AGENT_ENV, EXISTING_ANSWERS, agent_service, field_specs
+from ui.components.prompt import Prompt
+from ui.form import Form
 
 MSSQL = registry.get("mssql")
+FORM = Form(cast(Prompt, None), non_interactive=True)
+BAK_OPTIONS = {
+    "backup_format": "bak",
+    "bak_server_path": r"D:\PortabaseBackups",
+    "bak_agent_path": "/mnt/sql-backups",
+}
 
 
 def attributes():
@@ -72,8 +86,11 @@ def fields():
     assert MSSQL.fields_new() == []
     assert field_specs(MSSQL.option_fields()) == [
         ("backup_format", "choice", "dacpac"),
+        ("bak_server_path", "text", ""),
+        ("bak_agent_path", "text", ""),
+        ("bak_verify", "bool", False),
     ]
-    assert MSSQL.option_fields()[0].choices == ("dacpac", "bacpac")
+    assert MSSQL.option_fields()[0].choices == ("dacpac", "bacpac", "bak")
 
 
 def from_existing():
@@ -125,3 +142,69 @@ def only_non_default_options_reach_the_agent(ports):
     assert MSSQL.agent_entry(bacpac)["options"] == {"backup_format": "bacpac"}
     dacpac = spec.with_options({"backup_format": "dacpac"})
     assert "options" not in MSSQL.agent_entry(dacpac)
+
+
+def bak_options_reach_the_agent():
+    options = FORM.collect(MSSQL.option_fields(), {**BAK_OPTIONS, "bak_verify": "yes"})
+    spec = MSSQL.from_existing(EXISTING_ANSWERS).with_options(options)
+    assert MSSQL.agent_entry(spec)["options"] == {**BAK_OPTIONS, "bak_verify": True}
+
+
+def bak_verify_off_is_not_written():
+    options = FORM.collect(MSSQL.option_fields(), BAK_OPTIONS)
+    spec = MSSQL.from_existing(EXISTING_ANSWERS).with_options(options)
+    assert MSSQL.agent_entry(spec)["options"] == BAK_OPTIONS
+
+
+def bak_mounts_the_folder_on_the_agent():
+    spec = MSSQL.from_existing(EXISTING_ANSWERS)
+    assert MSSQL.mount_for(spec) is None
+    assert MSSQL.mount_for(spec.with_options({"backup_format": "bacpac"})) is None
+    assert MSSQL.mount_for(spec.with_options(BAK_OPTIONS)) == (
+        "/mnt/sql-backups",
+        "/mnt/sql-backups",
+    )
+
+
+@pytest.mark.parametrize("backup_format", ["dacpac", "bacpac"])
+def bak_options_only_apply_to_bak(backup_format):
+    options = FORM.collect(MSSQL.option_fields(), {"backup_format": backup_format})
+    assert options == {"backup_format": backup_format}
+    with pytest.raises(ValidationError, match="--bak-server-path does not apply"):
+        FORM.collect(
+            MSSQL.option_fields(),
+            {"backup_format": backup_format, "bak_server_path": "D:\\x"},
+        )
+
+
+@pytest.mark.parametrize(
+    ("overrides", "message"),
+    [
+        ({"bak_server_path": "  "}, "bak_server_path"),
+        ({"bak_agent_path": ""}, "bak_agent_path"),
+        ({"bak_agent_path": "mnt/sql-backups"}, "absolute"),
+        ({"bak_agent_path": "/"}, "absolute"),
+        ({"bak_agent_path": "/mnt/a:b"}, "cannot contain"),
+    ],
+)
+def bak_paths_are_required_and_checked(overrides, message):
+    with pytest.raises(ValidationError, match=message):
+        FORM.collect(MSSQL.option_fields(), {**BAK_OPTIONS, **overrides})
+
+
+def bak_is_refused_on_a_new_database(ports):
+    with pytest.raises(ValidationError, match="existing"):
+        MSSQL.generate(auth=True, ports=ports, answers={"options": BAK_OPTIONS})
+
+
+def bak_compose_mounts_the_folder(tmp_path, renderer):
+    project = AgentProject.create(tmp_path, dict(AGENT_ENV), host_gateway=False)
+    spec = MSSQL.from_existing(EXISTING_ANSWERS).with_options(BAK_OPTIONS)
+    project.add(spec, MSSQL)
+    result = renderer.render_agent(project, inline=False)
+    result.validate()
+    doc = yaml.safe_load(result.compose)
+    assert doc["services"] == {
+        "agent": agent_service("/mnt/sql-backups:/mnt/sql-backups")
+    }
+    assert result.databases == [MSSQL.agent_entry(spec)]