feat(tests.integration): add test for each main version of applicatio…

…n (2.14 and 2.15), add shortcut test

src/antares_web_installer/app.py

@@ -1,5 +1,4 @@
 import dataclasses
-import logging
 import os
 import re
 import subprocess
@@ -130,8 +129,9 @@ def install_files(self):
 
             # update config file
             logger.info("Update configuration file...")
-            config_path = self.target_dir.joinpath("config.yaml")
-            update_config(config_path, config_path, version)
+            src_config_path = self.source_dir.joinpath("config.yaml")
+            target_config_path = self.target_dir.joinpath("config.yaml")
+            update_config(src_config_path, target_config_path, version)
             logger.info("Configuration file updated.")
             self.update_progress(50)
 
@@ -286,7 +286,10 @@ def start_server(self):
             finally:
                 nb_attempts += 1
                 if nb_attempts == max_attempts:
-                    raise InstallError(f"Impossible to launch Antares Web Server after {nb_attempts} attempts.")
+                    try:
+                        server_process.wait(timeout=5)
+                    except subprocess.TimeoutExpired as e:
+                        raise InstallError(f"Impossible to launch Antares Web Server after {nb_attempts} attempts: {e}")
                 time.sleep(5)
 
     def open_browser(self):

src/antares_web_installer/config/__init__.py

@@ -23,7 +23,11 @@ def update_config(source_path: Path, target_path: Path, version: str) -> None:
 
     version_info = tuple(map(int, version.split(".")))
     if version_info < (2, 15):
-        update_to_2_15(config)
+        try:
+            update_to_2_15(config)
+        except AttributeError:
+            with target_path.open("r") as f:
+                config = yaml.safe_load(f)
 
-    with target_path.open(mode="w") as f:
+    with source_path.open(mode="w") as f:
         yaml.dump(config, f)

tests/integration/conftest.py

@@ -1,86 +1,23 @@
 import os
-import socket
-import subprocess
-import time
-from pathlib import Path
-import typing as t
-
 import pytest
 
+from pathlib import Path
 from tests.integration.server_mock.builder import build
 
 EXE_NAME = "AntaresWebServer.exe" if os.name == "nt" else "AntaresWebServer"
 """Name of the executable file for the Antares web server."""
 
-SPAWN_TIMEOUT = 10
-"""Timeout in seconds to wait for the server process to start."""
-
-SERVER_TIMEOUT = 10
-"""Timeout in seconds to wait for the server to be ready."""
-
-
-@pytest.fixture(name="antares_web_server_path", scope="session", autouse=True)
-def antares_web_server_path_fixture(tmp_path_factory: pytest.TempPathFactory) -> Path:
-    """Fixture used to build the Antares web server application."""
-    target_dir = tmp_path_factory.mktemp("antares_web_server", numbered=False)
-    return build(target_dir, exe_name=EXE_NAME)
-
-
-def ping(host: str, port: int, timeout: float = 2) -> bool:
-    """
-    Checks if a host is reachable by attempting to connect to a specified port.
-
-    Args:
-        host: The host to connect to.
-        port: The port to connect on.
-        timeout: The time in seconds to wait for a connection.
-
-    Returns:
-        True if the host is reachable, False otherwise.
-    """
-    try:
-        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
-            sock.settimeout(timeout)
-            sock.connect((host, port))
-    except (socket.timeout, ConnectionError):
-        return False
-    else:
-        return True
-
-
-@pytest.fixture(name="antares_web_server")
-def antares_web_server_fixture(antares_web_server_path: Path) -> t.Generator[subprocess.Popen, None, None]:
-    """Fixture used to provide a running instance of the Antares web server."""
-    # Spawn the server process
-    assert antares_web_server_path.exists()
-    server = subprocess.Popen(
-        [str(antares_web_server_path)],
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
-        text=True,
-        shell=False,
-    )
 
-    try:
-        # Wait for the subprocess to start
-        timeout_time = time.time() + SPAWN_TIMEOUT
-        while time.time() < timeout_time:
-            if server.poll() is None:
-                break
-            time.sleep(0.1)
-        else:
-            raise RuntimeError("The process did not start in time.")
+@pytest.fixture(name="antares_web_server_paths", scope="session", autouse=True)
+def antares_web_server_paths_fixture(tmp_path_factory: pytest.TempPathFactory) -> [Path]:
+    """Fixture used to build both the Antares web server version (2.14.4 and 2.15.2)."""
+    target_dir = tmp_path_factory.mktemp("servers", numbered=False)
 
-        # Wait for the server to be ready
-        timeout_time = time.time() + SERVER_TIMEOUT
-        while time.time() < timeout_time:
-            if ping("localhost", 8080, timeout=0.1):
-                break
-            time.sleep(0.1)
-        else:
-            raise RuntimeError("The server did not start in time.")
-        yield server
+    apps = []
+    for version in ["2.14.4", "2.15.2"]:
+        import tests.integration.server_mock.server as server
+        server.__dict__["version"] = version
+        version_target_dir = target_dir.joinpath("AntaresWeb-"+version)
+        apps.append(build(version_target_dir, exe_name=EXE_NAME))
 
-    finally:
-        server.terminate()
-        server.wait()
+    return apps

tests/integration/server_mock/server.py

@@ -5,7 +5,6 @@
 
 This server will only run 10 seconds, so it is not suitable for long tests.
 """
-
 import argparse
 
 import uvicorn
@@ -15,7 +14,7 @@
 # Define the server configuration
 HOST = "127.0.0.1"
 PORT = 8080
-VERSION = "2.14.0"
+VERSION = "2.14.4"
 
 
 def main():

tests/integration/test_app.py

@@ -16,21 +16,20 @@
 
 
 @pytest.fixture(name="downloaded_dir")
-def downloaded_dir_fixture(antares_web_server_path, tmp_path: Path) -> Path:
+def downloaded_dir_fixture(antares_web_server_paths, tmp_path: Path) -> Path:
     # Prepare the temporary directory to store "download" files
-    downloaded_dir = tmp_path.joinpath(DOWNLOAD_FOLDER, "AntaresWeb-MyOS-vX.Y.Z")
+    downloaded_dir = tmp_path.joinpath(DOWNLOAD_FOLDER)
     downloaded_dir.mkdir(parents=True)
 
     # Copy sample data
     sample_dir = SAMPLES_DIR.joinpath(os.name)
     shutil.copytree(sample_dir, downloaded_dir, dirs_exist_ok=True)
 
-    # The sample directory may contain an extra scripts that must be removed
-    downloaded_dir.joinpath("AntaresWeb/cli.py").unlink(missing_ok=True)
-    downloaded_dir.joinpath("AntaresWeb/server.py").unlink(missing_ok=True)
-
-    # Patch the `AntaresWeb/AntaresWebServer.exe` file
-    shutil.copy2(antares_web_server_path, downloaded_dir.joinpath("AntaresWeb"))
+    for server_path in antares_web_server_paths:
+        target_dir = downloaded_dir.joinpath(server_path.parent.name)
+        shutil.copy2(server_path, target_dir.joinpath("AntaresWeb"))
+        target_dir.joinpath("AntaresWeb/cli.py").unlink(missing_ok=True)
+        target_dir.joinpath("AntaresWeb/server.py").unlink(missing_ok=True)
     return downloaded_dir
 
 
@@ -48,17 +47,30 @@ def program_dir_fixture(tmp_path: Path) -> Path:
     return program_dir
 
 
+@pytest.fixture()
+def settings(desktop_dir: Path, monkeypatch: MonkeyPatch):
+    # Patch the `get_desktop` function according to the current platform
+    platform = {"nt": "_win32_shell", "posix": "_linux_shell"}[os.name]
+    monkeypatch.setattr(f"antares_web_installer.shortcuts.{platform}.get_desktop", lambda: desktop_dir)
+    yield
+    # kill the running servers
+    for proc in psutil.process_iter():
+        matching_ratio = SequenceMatcher(None, "antareswebserver", proc.name().lower()).ratio()
+        if matching_ratio > 0.8:
+            proc.kill()
+            proc.wait(1)
+            break
+
+
 class TestApp:
     """
     Integration tests for the app
     """
-
     def test_run__empty_target(
         self,
         downloaded_dir: Path,
-        desktop_dir: Path,
         program_dir: Path,
-        monkeypatch: MonkeyPatch,
+        settings: None
     ) -> None:
         """
         The goal of this test is to verify the behavior of the application when:
@@ -70,28 +82,39 @@ def test_run__empty_target(
         - Shortcuts are correctly created on the desktop
         - The server is correctly started
         """
-        # Patch the `get_desktop` function according to the current platform
-        platform = {"nt": "_win32_shell", "posix": "_linux_shell"}[os.name]
-        monkeypatch.setattr(f"antares_web_installer.shortcuts.{platform}.get_desktop", lambda: desktop_dir)
-
         # For each application versions, check if everything is working
-        for application_dir in program_dir.iterdir():
+        for application_dir in downloaded_dir.iterdir():
+            # Run the application
+            app = App(source_dir=application_dir, target_dir=program_dir, shortcut=True, launch=True)
+            app.run()
+
+    def test_shortcut__created(
+            self,
+            downloaded_dir: Path,
+            program_dir: Path,
+            desktop_dir: Path,
+            settings: None
+    ):
+        for application_dir in downloaded_dir.iterdir():
             # Run the application
-            app = App(source_dir=downloaded_dir, target_dir=application_dir, shortcut=True, launch=True)
+            app = App(source_dir=application_dir, target_dir=program_dir, shortcut=True, launch=True)
             app.run()
 
-            # Check the target directory
-            assert application_dir.is_dir()
-            assert list(application_dir.iterdir())
-
-            # Check the desktop directory
-            assert desktop_dir.is_dir()
-            assert list(desktop_dir.iterdir())
-
-            # kill the running server
-            for proc in psutil.process_iter():
-                matching_ratio = SequenceMatcher(None, "antareswebserver", proc.name().lower()).ratio()
-                if matching_ratio > 0.8:
-                    proc.kill()
-                    proc.wait(1)
-                    break
+            desktop_files = [file_name.name for file_name in list(desktop_dir.iterdir())]
+
+            if os.name != "nt":
+                assert "AntaresWebServer.desktop" in desktop_files
+            else:
+                assert "Antares Web Server.lnk" in desktop_files
+
+    def test_shortcut__not_created(self):
+        """
+        Test if a shortcut was created on the desktop
+        @param downloaded_dir:
+        @param desktop_dir:
+        @param program_dir:
+        @return:
+        """
+        pass
+
+

tests/integration/test_run_server.py

@@ -1,11 +1,76 @@
 import httpx  # type: ignore
+import subprocess
+import socket
+import time
 
+SPAWN_TIMEOUT = 10
+"""Timeout in seconds to wait for the server process to start."""
 
-def test_server_health(antares_web_server):
+SERVER_TIMEOUT = 10
+"""Timeout in seconds to wait for the server to be ready."""
+
+
+def ping(host: str, port: int, timeout: float = 2) -> bool:
+    """
+    Checks if a host is reachable by attempting to connect to a specified port.
+
+    Args:
+        host: The host to connect to.
+        port: The port to connect on.
+        timeout: The time in seconds to wait for a connection.
+
+    Returns:
+        True if the host is reachable, False otherwise.
+    """
+    try:
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+            sock.settimeout(timeout)
+            sock.connect((host, port))
+    except (socket.timeout, ConnectionError):
+        return False
+    else:
+        return True
+
+
+def test_server_health(antares_web_server_paths):
     """
     Test the health endpoint of the Antares web server.
+    After retrieving newly built servers, run each, make a
+    simple get request and kill it.
     """
 
-    res = httpx.get("http://localhost:8080/health", timeout=0.25)
-    assert res.status_code == 200, res.json()
-    assert res.json() == {"status": "available"}
+    # Set up: run each server
+    for server_path in antares_web_server_paths:
+        server = subprocess.Popen(
+            [str(server_path)],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            shell=False,
+        )
+
+        # Wait for the subprocess to start
+        timeout_time = time.time() + SPAWN_TIMEOUT
+        while time.time() < timeout_time:
+            if server.poll() is None:
+                break
+            time.sleep(0.1)
+        else:
+            raise RuntimeError("The process did not start in time.")
+
+        # Wait for the server to be ready
+        timeout_time = time.time() + SERVER_TIMEOUT
+        while time.time() < timeout_time:
+            if ping("localhost", 8080, timeout=0.1):
+                break
+            time.sleep(0.1)
+        else:
+            raise RuntimeError("The server did not start in time.")
+
+        res = httpx.get("http://localhost:8080/health", timeout=0.25)
+        assert res.status_code == 200, res.json()
+        assert res.json() == {"status": "available"}
+
+        # Tear down: kill server and make sure it is terminated
+        server.terminate()
+        server.wait()

tests/samples/nt/AntaresWeb-2.14.4/config.yaml

@@ -0,0 +1,27 @@
+db:
+  url: sqlite:///database.db
+debug: false
+launcher:
+  local:
+    binaries:
+      880: ./AntaresWeb/antares_solver/antares-8.8-solver
+logging:
+  logfile: ./tmp/antarest.log
+root_path: api
+security:
+  disabled: true
+  jwt:
+    key: super-secret
+server:
+  services:
+  - watcher
+  worker_threadpool_size: 12
+storage:
+  archive_dir: ./examples/archives
+  matrixstore: ./matrices
+  tmp_dir: ./tmp
+  workspaces:
+    default:
+      path: ./examples/internal_studies/
+    studies:
+      path: ./examples/studies/

tests/samples/nt/AntaresWeb-2.15.2/AntaresWeb/__init__.py

@@ -1 +1 @@
-__version__ = "2.14.4"
+__version__ = "2.15.2"