Tests for workflow federated evaluation

.github/workflows/pq_pipeline.yml

@@ -157,7 +157,18 @@ jobs:
     uses: ./.github/workflows/task_runner_flower_e2e.yml
     with:
       commit_id: ${{ needs.set_commit_id_for_all_jobs.outputs.commit_id }}
-
+  # wf federated evaluation
+  wf_federated_evaluation:
+    if: |
+      (github.event_name == 'schedule' && github.repository_owner == 'securefederatedai') ||
+      (github.event_name == 'workflow_dispatch')
+    name: Workflow Federated Evaluation E2E
+    needs: wf_secagg_e2e
+    uses: ./.github/workflows/wf_federated_evaluation.yml
+    with:
+      commit_id: ${{ needs.set_commit_id_for_all_jobs.outputs.commit_id }}
+    secrets: inherit
+  # run code scanning tools
   run_trivy:
     if: |
       (github.event_name == 'schedule' && github.repository_owner == 'securefederatedai') ||
@@ -206,6 +217,8 @@ jobs:
       task_runner_dockerized_e2e,
       task_runner_secret_ssl_e2e,
       task_runner_flower_app_pytorch,
+      task_runner_connectivity_e2e,
+      wf_federated_evaluation,
       run_trivy,
       run_bandit
     ]

.github/workflows/wf_federated_evaluation.yml

@@ -0,0 +1,66 @@
+# --------------------------------------------------------
+# It runs the federated evaluation tests
+# --------------------------------------------------------
+name: Federated Evaluation E2E
+
+on:
+  workflow_call:
+    inputs:
+      commit_id:
+        required: false
+        type: string
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+env:
+  COMMIT_ID: ${{ inputs.commit_id || github.sha }} # use commit_id from the calling workflow
+
+jobs:
+  test_federated_evaluation_notebook:
+    name: WF Federated Evaluation Without TLS
+    if: |
+      (github.event_name == 'schedule' && github.repository_owner == 'securefederatedai') ||
+      (github.event_name == 'workflow_dispatch') ||
+      (github.event.pull_request.draft == false)
+    runs-on: ubuntu-22.04
+    timeout-minutes: 20
+    env:
+      PYTHON_VERSION: '3.11'
+    steps:
+    - name: Checkout OpenFL repository
+      uses: actions/checkout@v4
+      with:
+        ref: ${{ env.COMMIT_ID }}  # use commit_id from the calling workflow
+
+    - name: Pre test run
+      uses: ./.github/actions/wf_pre_test_run
+      if: ${{ always() }}
+
+    - name: Run Federated Runtime 301 MNIST Watermarking via pytest
+      id: run_tests
+      run: |
+        python -m pytest -s tests/end_to_end/test_suites/wf_federated_runtime_tests.py -k test_federated_evaluation
+        echo "Federated Runtime 301 MNIST Watermarking test run completed"
+
+    - name: Print test summary
+      id: print_test_summary
+      if: ${{ always() }}
+      run: |
+        export PYTHONPATH="$PYTHONPATH:."
+        python tests/end_to_end/utils/summary_helper.py --func_name "print_federated_runtime_score" --nb_name "wf_federated_evaluation"
+        echo "Test summary printed"
+
+    - name: Tar files
+      if: ${{ always() }}   # collect artifacts regardless of failures
+      run: |
+        tar -cvf notebook.tar --exclude="__pycache__" $HOME/results --ignore-failed-read
+        echo "TAR file created"
+
+    - name: Upload Artifacts
+      uses: actions/upload-artifact@v4
+      if: ${{ always() }}   # collect artifacts regardless of failures
+      with:
+        name: federated_evaluation_${{ github.run_id }}
+        path: notebook.tar

tests/end_to_end/test_suites/wf_federated_runtime_tests.py

@@ -12,6 +12,33 @@
 log = logging.getLogger(__name__)
 
 
+def activate_experimental_feature(workspace_path):
+    """
+    Activate the experimental feature.
+    Args:
+        workspace_path (str): Path to the workspace
+    """
+    # Activate the experimental feature
+    cmd = f"fx experimental activate"
+    error_msg = "Failed to activate the experimental feature"
+    return_code, output, error = fh.run_command(
+        cmd,
+        workspace_path=workspace_path,
+        error_msg=error_msg,
+        return_error=True,
+    )
+
+    if error:
+        # Check if the experimental feature is already activated
+        if [err for err in error if "No such command 'activate'" in err]:
+            log.info("Experimental feature already activated. Ignore the error.")
+        else:
+            log.error(f"{error_msg}: {error}")
+            raise Exception(error)
+
+    log.info(f"Activated the experimental feature.")
+
+
 @pytest.mark.federated_runtime_301_watermarking
 def test_federated_runtime_301_watermarking(request):
     """
@@ -141,28 +168,65 @@ def test_federated_runtime_secure_aggregation(request):
     log.info("Experiment completed successfully")
 
 
-def activate_experimental_feature(workspace_path):
+def test_federated_evaluation(request):
     """
-    Activate the experimental feature.
+    Test federated evaluation.
     Args:
-        workspace_path (str): Path to the workspace
+        request (Fixture): Pytest fixture
     """
+    envoys = ["Bengaluru", "Portland"]
+    workspace_path = os.path.join(
+        os.getcwd(),
+        "openfl-tutorials/experimental/workflow/FederatedEvaluation",
+    )
     # Activate the experimental feature
-    cmd = f"fx experimental activate"
-    error_msg = "Failed to activate the experimental feature"
-    return_code, output, error = fh.run_command(
-        cmd,
-        workspace_path=workspace_path,
-        error_msg=error_msg,
-        return_error=True,
+    activate_experimental_feature(workspace_path)
+
+    # Create result log files for the director and envoys
+    result_path, participant_res_files = fh.create_federated_runtime_participant_res_files(
+        request.config.results_dir, envoys, model_name="wf_federated_evaluation"
     )
 
-    if error:
-        # Check if the experimental feature is already activated
-        if [err for err in error if "No such command 'activate'" in err]:
-            log.info("Experimental feature already activated. Ignore the error.")
-        else:
-            log.error(f"{error_msg}: {error}")
-            raise Exception(error)
+    # Start the director
+    fh.start_director(workspace_path, participant_res_files["director"])
 
-    log.info(f"Activated the experimental feature.")
+    # Start envoys Bangalore and Chandler and connect them to the director
+    executor = concurrent.futures.ThreadPoolExecutor()
+    results = [
+        executor.submit(
+            fh.start_envoy,
+            envoy_name=envoy,
+            workspace_path=workspace_path,
+            res_file=participant_res_files[envoy.lower()],
+        )
+        for envoy in envoys
+    ]
+    assert all([f.result() for f in results]), "Failed to start one or more envoys"
+
+    # Based on the pattern, the envoys take time to connect to the director
+    # Hence, adding a sleep of 10 seconds anyways.
+    time.sleep(10)
+    nb_workspace_path = os.path.join(workspace_path, "workspace")
+    notebook_path = nb_workspace_path + "/" + "MNIST_FederatedEvaluation.ipynb"
+
+    assert fh.check_envoys_director_conn_federated_runtime(
+        notebook_path=notebook_path, expected_envoys=envoys
+    ), "Envoys are not connected to the director"
+
+    # IMP - Notebook MNIST_Federated_Evaluation.ipynb has hard coded notebook path set, hence changing the directory
+    # This might not be true for all notebooks, thus keeping it as a separate step
+    os.chdir(nb_workspace_path)
+
+    assert fh.run_notebook(
+        notebook_path=notebook_path,
+        output_notebook_path=result_path + "/" + "MNIST_Federated_Evaluation_output.ipynb"
+    ), "Notebook run failed"
+
+    # Change the directory back to the original directory
+    os.chdir(os.getcwd())
+
+    assert fh.verify_federated_runtime_experiment_completion(
+        participant_res_files ,
+        expected_envoys=envoys
+    ), "Experiment failed"
+    log.info("Experiment completed successfully")