Refactor/integration suite

.github/workflows/push-pr_workflow.yml

@@ -10,7 +10,7 @@ jobs:
 
     steps:
     - name: Checkout code
-      uses: actions/checkout@v2
+      uses: actions/checkout@v4
       with:
         fetch-depth: 0 # Checkout the whole history, in case the target is way far behind
 
@@ -40,14 +40,14 @@ jobs:
       MAX_COMPLEXITY: 15
 
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v4
     - name: Set up Python
-      uses: actions/setup-python@v2
+      uses: actions/setup-python@v5
       with:
         python-version: '3.x'
 
     - name: Check cache
-      uses: actions/cache@v2
+      uses: actions/cache@v4
       with:
         path: ~/.cache/pip
         key: ${{ hashFiles('requirements/release.txt') }}-${{ hashFiles('requirements/dev.txt') }}
@@ -95,14 +95,14 @@ jobs:
         python-version: ['3.7', '3.8', '3.9', '3.10', '3.11']
 
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v4
     - name: Set up Python ${{ matrix.python-version }}
-      uses: actions/setup-python@v2
+      uses: actions/setup-python@v5
       with:
         python-version: ${{ matrix.python-version }}
 
     - name: Check cache
-      uses: actions/cache@v2
+      uses: actions/cache@v4
       with:
         path: ${{ env.pythonLocation }}
         key: ${{ env.pythonLocation }}-${{ hashFiles('requirements/release.txt') }}-${{ hashFiles('requirements/dev.txt') }}
@@ -112,8 +112,7 @@ jobs:
         python3 -m pip install --upgrade pip
         if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
         pip3 install -r requirements/dev.txt
-        pip freeze
-
+
     - name: Install singularity
       run: |
         sudo apt-get update && sudo apt-get install -y \
@@ -153,6 +152,73 @@ jobs:
       run: |
         python3 tests/integration/run_tests.py --verbose --local
 
+  Integration-tests:
+    runs-on: ubuntu-latest
+    env: 
+      GO_VERSION: 1.18.1
+      SINGULARITY_VERSION: 3.9.9
+      OS: linux
+      ARCH: amd64
+
+    strategy:
+      matrix:
+        python-version: ['3.7', '3.8', '3.9', '3.10', '3.11']
+
+    steps:
+    - uses: actions/checkout@v4
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v5
+      with:
+        python-version: ${{ matrix.python-version }}
+
+    - name: Check cache
+      uses: actions/cache@v4
+      with:
+        path: ${{ env.pythonLocation }}
+        key: ${{ env.pythonLocation }}-${{ hashFiles('requirements/release.txt') }}-${{ hashFiles('requirements/dev.txt') }}
+
+    - name: Install dependencies
+      run: |
+        python3 -m pip install --upgrade pip
+        if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+        pip3 install -r requirements/dev.txt
+
+    - name: Install merlin
+      run: |
+        pip3 install -e .
+        merlin config
+
+    - name: Install singularity
+      run: |
+        sudo apt-get update && sudo apt-get install -y \
+          build-essential \
+          libssl-dev \
+          uuid-dev \
+          libgpgme11-dev \
+          squashfs-tools \
+          libseccomp-dev \
+          pkg-config
+        wget https://go.dev/dl/go$GO_VERSION.$OS-$ARCH.tar.gz
+        sudo tar -C /usr/local -xzf go$GO_VERSION.$OS-$ARCH.tar.gz
+        rm go$GO_VERSION.$OS-$ARCH.tar.gz
+        export PATH=$PATH:/usr/local/go/bin
+        wget https://github.com/sylabs/singularity/releases/download/v$SINGULARITY_VERSION/singularity-ce-$SINGULARITY_VERSION.tar.gz
+        tar -xzf singularity-ce-$SINGULARITY_VERSION.tar.gz
+        cd singularity-ce-$SINGULARITY_VERSION
+        ./mconfig && \
+          make -C ./builddir && \
+          sudo make -C ./builddir install
+
+    - name: Install CLI task dependencies generated from the 'feature demo' workflow
+      run: |
+        merlin example feature_demo
+        pip3 install -r feature_demo/requirements.txt
+
+    # TODO remove the --ignore statement once those tests are fixed
+    - name: Run integration test suite for distributed tests
+      run: |
+        pytest --ignore tests/integration/test_celeryadapter.py tests/integration/
+
   Distributed-test-suite:
     runs-on: ubuntu-latest
     services:
@@ -179,14 +245,14 @@ jobs:
         python-version: ['3.7', '3.8', '3.9', '3.10', '3.11']
 
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v4
     - name: Set up Python ${{ matrix.python-version }}
-      uses: actions/setup-python@v2
+      uses: actions/setup-python@v5
       with:
         python-version: ${{ matrix.python-version }}
 
     - name: Check cache
-      uses: actions/cache@v2
+      uses: actions/cache@v4
       with:
         path: ${{ env.pythonLocation }}
         key: ${{ env.pythonLocation }}-${{ hashFiles('requirements/release.txt') }}-${{ hashFiles('requirements/dev.txt') }}

CHANGELOG.md

@@ -21,6 +21,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Split the `start_server` and `config_server` functions of `merlin/server/server_commands.py` into multiple functions to make testing easier
 - Split the `create_server_config` function of `merlin/server/server_config.py` into two functions to make testing easier
 - Combined `set_snapshot_seconds` and `set_snapshot_changes` methods of `RedisConfig` into one method `set_snapshot`
+- Moved stop-workers and query-workers integration tests to pytest tests
 
 ## [1.12.2b1]
 ### Added

merlin/celery.py

@@ -114,9 +114,11 @@ def route_for_task(name, args, kwargs, options, task=None, **kw):  # pylint: dis
     BROKER_URI = None
     RESULTS_BACKEND_URI = None
 
+app_name = "merlin_test_app" if os.getenv("CELERY_ENV") == "test" else "merlin"
+
 # initialize app with essential properties
 app: Celery = patch_celery().Celery(
-    "merlin",
+    app_name,
     broker=BROKER_URI,
     backend=RESULTS_BACKEND_URI,
     broker_use_ssl=BROKER_SSL,

merlin/examples/dev_workflows/multiple_workers.yaml

@@ -46,11 +46,11 @@ merlin:
     resources:
         workers:
             step_1_merlin_test_worker:
-                args: -l INFO
+                args: -l INFO --concurrency 1
                 steps: [step_1]
             step_2_merlin_test_worker:
-                args: -l INFO
+                args: -l INFO --concurrency 1
                 steps: [step_2]
             other_merlin_test_worker:
-                args: -l INFO
+                args: -l INFO --concurrency 1
                 steps: [step_3, step_4]

tests/conftest.py

@@ -99,6 +99,22 @@ def create_encryption_file(key_filepath: str, encryption_key: bytes, app_yaml_fi
 #######################################
 
 
+@pytest.fixture(scope="session")
+def path_to_test_specs() -> str:
+    """
+    Fixture to provide the path to the directory containing test specifications.
+
+    This fixture returns the absolute path to the 'test_specs' directory
+    within the 'integration' folder of the test directory. It expands
+    environment variables and user home directory as necessary.
+
+    Returns:
+        The absolute path to the 'test_specs' directory.
+    """
+    path_to_test_dir = os.path.abspath(os.path.expandvars(os.path.expanduser(os.path.dirname(__file__))))
+    return os.path.join(path_to_test_dir, os.path.join("integration", "test_specs"))
+
+
 @pytest.fixture(scope="session")
 def temp_output_dir(tmp_path_factory: TempPathFactory) -> str:
     """
@@ -130,7 +146,7 @@ def merlin_server_dir(temp_output_dir: str) -> str:
     :param temp_output_dir: The path to the temporary output directory we'll be using for this test run
     :returns: The path to the merlin_server directory that will be created by the `redis_server` fixture
     """
-    server_dir = f"{temp_output_dir}/merlin_server"
+    server_dir = os.path.join(temp_output_dir, "merlin_server")
     if not os.path.exists(server_dir):
         os.mkdir(server_dir)
     return server_dir
@@ -146,11 +162,14 @@ def redis_server(merlin_server_dir: str, test_encryption_key: bytes) -> str:
     :param test_encryption_key: An encryption key to be used for testing
     :yields: The local redis server uri
     """
+    os.environ["CELERY_ENV"] = "test"
     with RedisServerManager(merlin_server_dir, SERVER_PASS) as redis_server_manager:
         redis_server_manager.initialize_server()
         redis_server_manager.start_server()
         create_encryption_file(
-            f"{merlin_server_dir}/encrypt_data_key", test_encryption_key, app_yaml_filepath=f"{merlin_server_dir}/app.yaml"
+            os.path.join(merlin_server_dir, "encrypt_data_key"),
+            test_encryption_key,
+            app_yaml_filepath=os.path.join(merlin_server_dir, "app.yaml"),
         )
         # Yield the redis_server uri to any fixtures/tests that may need it
         yield redis_server_manager.redis_server_uri
@@ -165,6 +184,7 @@ def celery_app(redis_server: str) -> Celery:
     :param redis_server: The redis server uri we'll use to connect to redis
     :returns: The celery app object we'll use for testing
     """
+    os.environ["CELERY_ENV"] = "test"
     return Celery("merlin_test_app", broker=redis_server, backend=redis_server)
 
 
@@ -258,7 +278,7 @@ def config(merlin_server_dir: str, test_encryption_key: bytes):
     orig_config = copy(CONFIG)
 
     # Create an encryption key file (if it doesn't already exist)
-    key_file = f"{merlin_server_dir}/encrypt_data_key"
+    key_file = os.path.join(merlin_server_dir, "encrypt_data_key")
     create_encryption_file(key_file, test_encryption_key)
 
     # Set the broker configuration for testing
@@ -305,7 +325,7 @@ def redis_broker_config(
     :param merlin_server_dir: The directory to the merlin test server configuration
     :param config: The fixture that sets up most of the CONFIG object for testing
     """
-    pass_file = f"{merlin_server_dir}/redis.pass"
+    pass_file = os.path.join(merlin_server_dir, "redis.pass")
     create_pass_file(pass_file)
 
     CONFIG.broker.password = pass_file
@@ -326,7 +346,7 @@ def redis_results_backend_config(
     :param merlin_server_dir: The directory to the merlin test server configuration
     :param config: The fixture that sets up most of the CONFIG object for testing
     """
-    pass_file = f"{merlin_server_dir}/redis.pass"
+    pass_file = os.path.join(merlin_server_dir, "redis.pass")
     create_pass_file(pass_file)
 
     CONFIG.results_backend.password = pass_file
@@ -347,7 +367,7 @@ def rabbit_broker_config(
     :param merlin_server_dir: The directory to the merlin test server configuration
     :param config: The fixture that sets up most of the CONFIG object for testing
     """
-    pass_file = f"{merlin_server_dir}/rabbit.pass"
+    pass_file = os.path.join(merlin_server_dir, "rabbit.pass")
     create_pass_file(pass_file)
 
     CONFIG.broker.password = pass_file
@@ -368,7 +388,7 @@ def mysql_results_backend_config(
     :param merlin_server_dir: The directory to the merlin test server configuration
     :param config: The fixture that sets up most of the CONFIG object for testing
     """
-    pass_file = f"{merlin_server_dir}/mysql.pass"
+    pass_file = os.path.join(merlin_server_dir, "mysql.pass")
     create_pass_file(pass_file)
 
     create_cert_files(merlin_server_dir, CERT_FILES)

tests/context_managers/celery_workers_manager.py

@@ -80,8 +80,9 @@ def __exit__(self, exc_type: Type[Exception], exc_value: Exception, traceback: T
             try:
                 if str(pid) in ps_proc.stdout:
                     os.kill(pid, signal.SIGKILL)
-            except ProcessLookupError as exc:
-                raise ProcessLookupError(f"PID {pid} not found. Output of 'ps ux':\n{ps_proc.stdout}") from exc
+            # If the process can't be found then it doesn't exist anymore
+            except ProcessLookupError:
+                pass
 
     def _is_worker_ready(self, worker_name: str, verbose: bool = False) -> bool:
         """
@@ -158,6 +159,8 @@ def launch_worker(self, worker_name: str, queues: List[str], concurrency: int =
             self.stop_all_workers()
             raise ValueError(f"The worker {worker_name} is already running. Choose a different name.")
 
+        queues = [f"[merlin]_{queue}" for queue in queues]
+
         # Create the launch command for this worker
         worker_launch_cmd = [
             "worker",
@@ -174,7 +177,7 @@ def launch_worker(self, worker_name: str, queues: List[str], concurrency: int =
         # Create an echo command to simulate a running celery worker since our celery worker will be spun up in
         # a different process and we won't be able to see it with 'ps ux' like we normally would
         echo_process = subprocess.Popen(  # pylint: disable=consider-using-with
-            f"echo 'celery merlin_test_app {' '.join(worker_launch_cmd)}'; sleep inf",
+            f"echo 'celery -A merlin_test_app {' '.join(worker_launch_cmd)}'; sleep inf",
             shell=True,
             preexec_fn=os.setpgrp,  # Make this the parent of the group so we can kill the 'sleep inf' that's spun up
         )

tests/context_managers/server_manager.py

@@ -91,7 +91,7 @@ def stop_server(self):
         if "Merlin server terminated." not in kill_process.stderr:
             # If it wasn't, try to kill the process by using the pid stored in a file created by `merlin server`
             try:
-                with open(f"{self.server_dir}/merlin_server.pf", "r") as process_file:
+                with open(os.path.join(self.server_dir, "merlin_server.pf"), "r") as process_file:
                     server_process_info = yaml.load(process_file, yaml.Loader)
                     os.kill(int(server_process_info["image_pid"]), signal.SIGKILL)
             # If the file can't be found then let's make sure there's even a redis-server process running