Skip to content

Commit

Permalink
Addressing prefect secret patch issue in pytest script
Browse files Browse the repository at this point in the history
  • Loading branch information
davramov committed Dec 14, 2024
1 parent c6601be commit 6057aae
Showing 1 changed file with 31 additions and 6 deletions.
37 changes: 31 additions & 6 deletions orchestration/_tests/test_sfapi_flow.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,34 @@
import pytest
from unittest.mock import MagicMock, PropertyMock, patch
from orchestration.flows.bl832.nersc import nersc_recon_flow, NERSCTomographyHPCController
from orchestration.flows.bl832.config import Config832
from orchestration.flows.bl832.job_controller import HPC
from orchestration.nersc import NerscClient
from uuid import uuid4

from prefect.blocks.system import Secret
from prefect.testing.utilities import prefect_test_harness

# Patch Secret.load globally before importing application code
with patch("prefect.blocks.system.Secret.load") as mock_secret_load:
mock_secret_load.return_value = Secret(value=str(uuid4()))
# Import application modules after patching Secret.load
from orchestration.flows.bl832.nersc import nersc_recon_flow, NERSCTomographyHPCController
from orchestration.flows.bl832.config import Config832
from orchestration.flows.bl832.job_controller import HPC
from orchestration.nersc import NerscClient


@pytest.fixture(autouse=True, scope="session")
def prefect_test_fixture():
"""
A pytest fixture that automatically sets up and tears down the Prefect test harness
for the entire test session. It creates and saves test secrets and configurations
required for Globus integration.
Yields:
None
"""
with prefect_test_harness():
globus_client_id = Secret(value=str(uuid4()))
globus_client_id.save(name="globus-client-id")
yield


@pytest.fixture
Expand Down Expand Up @@ -82,11 +107,11 @@ def test_job_submission(mock_controller):
"""Test job submission and status updates."""
job_script = "mock_job_script"
mock_job = mock_controller.client.perlmutter.submit_job.return_value
mock_job.state = "COMPLETED"
job_id = mock_job.jobid

mock_controller.client.perlmutter.submit_job(job_script)
mock_controller.client.perlmutter.submit_job.assert_called_once_with(job_script)
assert mock_job.jobid == "12345", "Job ID should match the mock job ID."
assert job_id == "12345", "Job ID should match the mock job ID."


def test_job_recovery(mock_controller):
Expand Down

0 comments on commit 6057aae

Please sign in to comment.