feat(testcontainers): add database backup helpers

python_testcontainers/infrahub_testcontainers/container.py

@@ -43,6 +43,8 @@ class ContainerService:
     "INFRAHUB_TESTING_LOCAL_REMOTE_GIT_DIRECTORY": "repos",
     "INFRAHUB_TESTING_INTERNAL_REMOTE_GIT_DIRECTORY": "/remote",
     "INFRAHUB_TESTING_WEB_CONCURRENCY": "4",
+    "INFRAHUB_TESTING_LOCAL_DB_BACKUP_DIRECTORY": "backups",
+    "INFRAHUB_TESTING_INTERNAL_DB_BACKUP_DIRECTORY": "/backups",
 }
 
 

python_testcontainers/infrahub_testcontainers/docker-compose.test.yml

@@ -35,9 +35,11 @@ services:
       NEO4J_AUTH: neo4j/admin
       NEO4J_dbms_security_procedures_unrestricted: "apoc.*"
       NEO4J_dbms_security_auth__minimum__password__length: 4
+      NEO4J_ACCEPT_LICENSE_AGREEMENT: "yes"
     volumes:
       - "database_data:/data"
       - "database_logs:/logs"
+      - "./${INFRAHUB_TESTING_LOCAL_DB_BACKUP_DIRECTORY}:${INFRAHUB_TESTING_INTERNAL_DB_BACKUP_DIRECTORY}"
     healthcheck:
       test: wget http://localhost:7474 || exit 1
       interval: 2s

python_testcontainers/infrahub_testcontainers/helpers.py

@@ -1,4 +1,5 @@
 import os
+import shutil
 import subprocess  # noqa: S404
 from pathlib import Path
 
@@ -36,12 +37,25 @@ def remote_repos_dir(self, tmp_directory: Path) -> Path:
 
         return directory
 
+    @pytest.fixture(scope="class")
+    def remote_backups_dir(self, tmp_directory: Path) -> Path:
+        directory = tmp_directory / PROJECT_ENV_VARIABLES["INFRAHUB_TESTING_LOCAL_DB_BACKUP_DIRECTORY"]
+        directory.mkdir(exist_ok=True)
+
+        return directory
+
     @pytest.fixture(scope="class")
     def default_branch(self) -> str:
         return "main"
 
     @pytest.fixture(scope="class")
-    def infrahub_compose(self, tmp_directory: Path, infrahub_version: str) -> InfrahubDockerCompose:
+    def infrahub_compose(
+        self,
+        tmp_directory: Path,
+        remote_repos_dir: Path,  # initialize repository before running docker compose to fix permissions issues # noqa: ARG002
+        remote_backups_dir: Path,  # noqa: ARG002
+        infrahub_version: str,
+    ) -> InfrahubDockerCompose:
         return InfrahubDockerCompose.init(directory=tmp_directory, version=infrahub_version)
 
     @pytest.fixture(scope="class")
@@ -62,3 +76,73 @@ def infrahub_port(self, infrahub_app: dict[str, int]) -> int:
     @pytest.fixture(scope="class")
     def task_manager_port(self, infrahub_app: dict[str, int]) -> int:
         return infrahub_app["task-manager"]
+
+    def backup_database(self, request: pytest.FixtureRequest, dest_dir: Path | None = None) -> None:
+        assert "enterprise" in os.environ.get("NEO4J_DOCKER_IMAGE", "")
+
+        backup_dir: Path = request.getfixturevalue("remote_backups_dir")
+        infrahub_compose: InfrahubDockerCompose = request.getfixturevalue("infrahub_compose")
+
+        infrahub_compose.exec_in_container(
+            command=[
+                "neo4j-admin",
+                "database",
+                "backup",
+                "--to-path",
+                os.environ.get(
+                    "INFRAHUB_TESTING_INTERNAL_DB_BACKUP_DIRECTORY",
+                    PROJECT_ENV_VARIABLES["INFRAHUB_TESTING_INTERNAL_DB_BACKUP_DIRECTORY"],
+                ),
+            ],
+            service_name="database",
+        )
+
+        if dest_dir:
+            shutil.copytree(
+                str(backup_dir),
+                str(dest_dir),
+            )
+
+    def restore_database(self, request: pytest.FixtureRequest, backup_file: Path) -> None:
+        assert "enterprise" in os.environ.get("NEO4J_DOCKER_IMAGE", "")
+
+        backup_dir: Path = request.getfixturevalue("remote_backups_dir")
+        infrahub_compose: InfrahubDockerCompose = request.getfixturevalue("infrahub_compose")
+
+        shutil.copy(
+            str(backup_file),
+            str(backup_dir / backup_file.name),
+        )
+
+        infrahub_compose.exec_in_container(
+            command=["cypher-shell", "-u", "neo4j", "-p", "admin", "STOP DATABASE neo4j;"],
+            service_name="database",
+        )
+
+        infrahub_compose.exec_in_container(
+            command=[
+                "neo4j-admin",
+                "database",
+                "restore",
+                "--overwrite-destination",
+                "--from-path",
+                str(
+                    Path(
+                        os.environ.get(
+                            "INFRAHUB_TESTING_INTERNAL_DB_BACKUP_DIRECTORY",
+                            PROJECT_ENV_VARIABLES["INFRAHUB_TESTING_INTERNAL_DB_BACKUP_DIRECTORY"],
+                        )
+                    )
+                    / backup_file.name
+                ),
+            ],
+            service_name="database",
+        )
+
+        infrahub_compose.exec_in_container(
+            command=["cypher-shell", "-d", "system", "-u", "neo4j", "-p", "admin", "START DATABASE neo4j;"],
+            service_name="database",
+        )
+
+        infrahub_compose.stop(down=False)
+        infrahub_compose.start()