From 5201f8a360b6197262d7260e687cba62a742909a Mon Sep 17 00:00:00 2001 From: Kevin Date: Mon, 13 Jul 2026 15:15:00 -0400 Subject: [PATCH 1/4] Make Recipe job storage private --- nvflare/app_common/np/recipes/fedavg.py | 6 +- nvflare/app_opt/sklearn/recipes/kmeans.py | 2 +- nvflare/app_opt/sklearn/recipes/svm.py | 2 +- nvflare/app_opt/xgboost/recipes/bagging.py | 4 +- nvflare/app_opt/xgboost/recipes/histogram.py | 4 +- nvflare/app_opt/xgboost/recipes/vertical.py | 4 +- nvflare/edge/tools/edge_fed_buff_recipe.py | 4 +- nvflare/recipe/spec.py | 40 ++++----- nvflare/recipe/utils.py | 32 +++---- .../fast/study_session_test.py | 4 +- .../slow/experiment_tracking_recipes_test.py | 4 +- .../tools/export_recipe_job.py | 4 +- .../pt/recipes/fed_eval_recipe_test.py | 6 +- .../app_opt/sklearn/sklearn_recipe_test.py | 14 +-- .../app_opt/xgboost/xgboost_recipe_test.py | 2 +- tests/unit_test/recipe/cyclic_recipe_test.py | 12 +-- tests/unit_test/recipe/edge_recipe_test.py | 16 ++-- tests/unit_test/recipe/eval_recipe_test.py | 10 +-- .../unit_test/recipe/fed_task_recipe_test.py | 8 +- tests/unit_test/recipe/fedavg_recipe_test.py | 30 +++---- tests/unit_test/recipe/fedopt_recipe_test.py | 8 +- .../unit_test/recipe/fedstats_recipe_test.py | 2 +- tests/unit_test/recipe/flower_recipe_test.py | 8 +- .../unit_test/recipe/numpy_lr_recipe_test.py | 10 +-- .../unit_test/recipe/scaffold_recipe_test.py | 6 +- .../recipe/server_memory_gc_rounds_test.py | 2 +- tests/unit_test/recipe/spec_test.py | 72 ++++++++------- tests/unit_test/recipe/swarm_recipe_test.py | 32 +++---- tests/unit_test/recipe/utils_test.py | 88 +++++++++---------- 29 files changed, 223 insertions(+), 213 deletions(-) diff --git a/nvflare/app_common/np/recipes/fedavg.py b/nvflare/app_common/np/recipes/fedavg.py index 4508b8d7a2..03d2278cba 100644 --- a/nvflare/app_common/np/recipes/fedavg.py +++ b/nvflare/app_common/np/recipes/fedavg.py @@ -208,8 +208,8 @@ def add_cse_validator_if_needed(self): # Check if validation task is explicitly configured (not just via wildcard) has_explicit_validator = False - if hasattr(self.job, "_deploy_map"): - for target, app in self.job._deploy_map.items(): + if hasattr(self._job, "_deploy_map"): + for target, app in self._job._deploy_map.items(): if target == "server": continue @@ -229,4 +229,4 @@ def add_cse_validator_if_needed(self): if not has_explicit_validator: # No explicit validator found - add NPValidator for cross-site evaluation validator = NPValidator() - self.job.to_clients(validator, tasks=[AppConstants.TASK_VALIDATION]) + self._job.to_clients(validator, tasks=[AppConstants.TASK_VALIDATION]) diff --git a/nvflare/app_opt/sklearn/recipes/kmeans.py b/nvflare/app_opt/sklearn/recipes/kmeans.py index df937fa5a6..7b9da9d388 100644 --- a/nvflare/app_opt/sklearn/recipes/kmeans.py +++ b/nvflare/app_opt/sklearn/recipes/kmeans.py @@ -180,4 +180,4 @@ def __init__( per_site_config=per_site_config, key_metric=key_metric, ) - self.job.to_server(assembler, id=assembler_id) + self._job.to_server(assembler, id=assembler_id) diff --git a/nvflare/app_opt/sklearn/recipes/svm.py b/nvflare/app_opt/sklearn/recipes/svm.py index a60caa7be0..d9ea20495b 100644 --- a/nvflare/app_opt/sklearn/recipes/svm.py +++ b/nvflare/app_opt/sklearn/recipes/svm.py @@ -181,4 +181,4 @@ def __init__( # Add the SVMAssembler as a component to the job # CollectAndAssembleModelAggregator will fetch it by ID at runtime - self.job.to_server(self._svm_assembler, id="svm_assembler") + self._job.to_server(self._svm_assembler, id="svm_assembler") diff --git a/nvflare/app_opt/xgboost/recipes/bagging.py b/nvflare/app_opt/xgboost/recipes/bagging.py index 9026d4654f..0d4c0d6454 100644 --- a/nvflare/app_opt/xgboost/recipes/bagging.py +++ b/nvflare/app_opt/xgboost/recipes/bagging.py @@ -223,8 +223,8 @@ def __init__( ) # Configure the job - self.job = self.configure() - Recipe.__init__(self, self.job) + job = self.configure() + Recipe.__init__(self, job) def configure(self): """Configure the federated job for XGBoost tree-based training.""" diff --git a/nvflare/app_opt/xgboost/recipes/histogram.py b/nvflare/app_opt/xgboost/recipes/histogram.py index 795ee57773..b0043f70fb 100644 --- a/nvflare/app_opt/xgboost/recipes/histogram.py +++ b/nvflare/app_opt/xgboost/recipes/histogram.py @@ -184,8 +184,8 @@ def __init__( ) # Configure the job - self.job = self.configure() - Recipe.__init__(self, self.job) + job = self.configure() + Recipe.__init__(self, job) def configure(self): """Configure the federated job for XGBoost histogram-based training.""" diff --git a/nvflare/app_opt/xgboost/recipes/vertical.py b/nvflare/app_opt/xgboost/recipes/vertical.py index 9afe788cd5..8b85ee0958 100644 --- a/nvflare/app_opt/xgboost/recipes/vertical.py +++ b/nvflare/app_opt/xgboost/recipes/vertical.py @@ -255,8 +255,8 @@ def __init__( self.client_ranks = {client_name: rank for rank, client_name in enumerate(ranked_clients)} # Configure the job - self.job = self.configure() - Recipe.__init__(self, self.job) + job = self.configure() + Recipe.__init__(self, job) def configure(self): """Configure the federated job for vertical XGBoost training.""" diff --git a/nvflare/edge/tools/edge_fed_buff_recipe.py b/nvflare/edge/tools/edge_fed_buff_recipe.py index fd99a0e09a..7935e92b85 100644 --- a/nvflare/edge/tools/edge_fed_buff_recipe.py +++ b/nvflare/edge/tools/edge_fed_buff_recipe.py @@ -303,8 +303,8 @@ def process_env(self, env: ExecEnv): f"invalid {DEVICE_SIMULATION_ENV_KEY} in env: expect SimulationConfig but got {type(simulation_config)}" ) - assert isinstance(self.job, EdgeJob) - self._configure_simulation(self.job, simulation_config) + assert isinstance(self._job, EdgeJob) + self._configure_simulation(self._job, simulation_config) def create_job(self) -> EdgeJob: """Create a new EdgeJob instance for cross-edge federated learning. diff --git a/nvflare/recipe/spec.py b/nvflare/recipe/spec.py index a493c49a91..8a8b9adc31 100644 --- a/nvflare/recipe/spec.py +++ b/nvflare/recipe/spec.py @@ -251,7 +251,7 @@ def __init__(self, job: FedJob): Args: job: the job that implements the recipe. """ - self.job = job + self._job = job self._helper_per_site_config = None self._tensor_streaming_added = False self._cse_added = False @@ -335,7 +335,7 @@ def configured_sites(self) -> List[str]: def _snapshot_additional_params(self) -> Dict[str, Dict]: snapshot = {} - deploy_map = getattr(self.job, "_deploy_map", {}) + deploy_map = getattr(self._job, "_deploy_map", {}) for target, app in deploy_map.items(): app_config = getattr(app, "app_config", None) if app_config is None: @@ -346,7 +346,7 @@ def _snapshot_additional_params(self) -> Dict[str, Dict]: return snapshot def _restore_additional_params(self, snapshot: Dict[str, Dict]) -> None: - deploy_map = getattr(self.job, "_deploy_map", {}) + deploy_map = getattr(self._job, "_deploy_map", {}) for target, app in deploy_map.items(): app_config = getattr(app, "app_config", None) if app_config is None: @@ -358,7 +358,7 @@ def _restore_additional_params(self, snapshot: Dict[str, Dict]) -> None: params.update(original) def _replace_additional_params_for_targets(self, targets: List[str], new_params: dict) -> None: - deploy_map = getattr(self.job, "_deploy_map", {}) + deploy_map = getattr(self._job, "_deploy_map", {}) for target in targets: app = deploy_map.get(target) if app is None: @@ -398,7 +398,7 @@ def _temporary_exec_params( try: if server_exec_params is not None: if server_exec_params: - self.job.to_server(server_exec_params) + self._job.to_server(server_exec_params) else: # Preserve the long-standing "empty dict means temporarily clear params" # behavior rather than treating {} as a no-op. @@ -408,7 +408,7 @@ def _temporary_exec_params( if client_exec_params: self._add_to_client_apps(client_exec_params) else: - client_targets = [target for target in getattr(self.job, "_deploy_map", {}) if target != "server"] + client_targets = [target for target in getattr(self._job, "_deploy_map", {}) if target != "server"] self._replace_additional_params_for_targets(client_targets, {}) yield finally: @@ -437,7 +437,7 @@ def _add_to_client_apps(self, obj, clients: Optional[List[str]] = None, **kwargs # FedJob has no public API to list per-site deploy targets, so we inspect # private deploy map to preserve existing per-site client topology. - deploy_map = getattr(self.job, "_deploy_map", {}) + deploy_map = getattr(self._job, "_deploy_map", {}) existing_client_sites = [ target for target in deploy_map.keys() @@ -447,9 +447,9 @@ def _add_to_client_apps(self, obj, clients: Optional[List[str]] = None, **kwargs if clients is None: if existing_client_sites: for site in existing_client_sites: - self.job.to(obj, site, **kwargs) + self._job.to(obj, site, **kwargs) else: - self.job.to_clients(obj, **kwargs) + self._job.to_clients(obj, **kwargs) else: # A bare string would iterate per character; an empty list would be a silent no-op. if not isinstance(clients, list): @@ -480,7 +480,7 @@ def _add_to_client_apps(self, obj, clients: Optional[List[str]] = None, **kwargs f"only for {sorted(existing_client_sites)}" ) for client in clients: - self.job.to(obj, client, **kwargs) + self._job.to(obj, client, **kwargs) def add_client_input_filter( self, filter: Filter, tasks: Optional[List[str]] = None, clients: Optional[List[str]] = None @@ -569,7 +569,7 @@ def add_server_output_filter(self, filter: Filter, tasks: Optional[List[str]] = Returns: None """ - self.job.to_server(filter, filter_type=FilterType.TASK_DATA, tasks=tasks) + self._job.to_server(filter, filter_type=FilterType.TASK_DATA, tasks=tasks) def add_server_input_filter(self, filter: Filter, tasks: Optional[List[str]] = None): """Add a filter to server for incoming task result from clients. . @@ -581,7 +581,7 @@ def add_server_input_filter(self, filter: Filter, tasks: Optional[List[str]] = N Returns: None """ - self.job.to_server(filter, filter_type=FilterType.TASK_RESULT, tasks=tasks) + self._job.to_server(filter, filter_type=FilterType.TASK_RESULT, tasks=tasks) def add_server_config(self, config: Dict): """Add top-level configuration parameters to config_fed_server.json. @@ -602,7 +602,7 @@ def add_server_config(self, config: Dict): warn_on_potential_secrets(config, context="add_server_config config") warn_on_unsupported_secret_ref_keys(config, context="add_server_config config") - self.job.to_server(config) + self._job.to_server(config) def add_server_file(self, file_path: str): """Add a file or directory to server app. @@ -623,7 +623,7 @@ def add_server_file(self, file_path: str): if not isinstance(file_path, str): raise TypeError(f"file_path must be a str, got {type(file_path).__name__}") - self.job.to_server(file_path) + self._job.to_server(file_path) @staticmethod def _get_full_class_name(obj): @@ -662,7 +662,7 @@ def enable_log_streaming(self, *file_names: str) -> None: for name in file_names: self._add_to_client_apps(JobLogStreamer(log_file_name=name)) - self.job.to_server(JobLogReceiver()) + self._job.to_server(JobLogReceiver()) def enable_tensor_streaming( self, @@ -712,7 +712,7 @@ def enable_tensor_streaming( server_tasks = list(tasks) if tasks is not None else None client_tasks = list(tasks) if tasks is not None else None - self.job.to_server( + self._job.to_server( TensorServerStreamer( format=format, tasks=server_tasks, @@ -753,7 +753,7 @@ def add_decomposers(self, decomposers: List[Union[str, Decomposer]]): else: raise TypeError(f"decomposer must be str or Decomposer, got {type(d).__name__}") - self.job.to_server(DecomposerRegister(class_names), id="decomposer_reg") + self._job.to_server(DecomposerRegister(class_names), id="decomposer_reg") self._add_to_client_apps(DecomposerRegister(class_names), id="decomposer_reg") def _warn_potential_secrets_in_exported_job(self, job_dir: str) -> None: @@ -765,7 +765,7 @@ def _warn_potential_secrets_in_exported_job(self, job_dir: str) -> None: """ warn_on_potential_secrets_in_job_dir( job_dir=job_dir, - job_name=getattr(self.job, "name", None), + job_name=getattr(self._job, "name", None), context="exported job file", ) @@ -797,7 +797,7 @@ class docstring for how to pass references instead. with self._temporary_exec_params(server_exec_params=server_exec_params, client_exec_params=client_exec_params): if env is not None: self.process_env(env) - self.job.export_job(job_dir) + self._job.export_job(job_dir) self._warn_potential_secrets_in_exported_job(job_dir) def run( @@ -816,7 +816,7 @@ def run( self._warn_potential_secrets_in_params() with self._temporary_exec_params(server_exec_params=server_exec_params, client_exec_params=client_exec_params): self.process_env(env) - job_id = env.deploy(self.job) + job_id = env.deploy(self._job) from nvflare.recipe.run import Run return Run(env, job_id) diff --git a/nvflare/recipe/utils.py b/nvflare/recipe/utils.py index 0bb71e28d4..341cb9b154 100644 --- a/nvflare/recipe/utils.py +++ b/nvflare/recipe/utils.py @@ -211,10 +211,10 @@ def _normalize_recipe_meta_value(key: JobMetaKey, key_str: str, value: Any) -> A def _get_recipe_job_config(recipe: Recipe) -> FedJobConfig: - job = getattr(recipe, "job", None) + job = getattr(recipe, "_job", None) job_config = getattr(job, "job", None) if not isinstance(job_config, FedJobConfig): - raise TypeError("recipe must provide a FedJob through recipe.job") + raise TypeError("recipe must be backed by a FedJob") return job_config @@ -378,7 +378,7 @@ def add_experiment_tracking( tracking_config["kw_args"] = kw_args elif not isinstance(kw_args, dict): raise TypeError(f"MLflow kw_args must be a dict, got {type(kw_args).__name__}") - recipe_name = getattr(recipe, "name", None) or getattr(recipe.job, "name", None) or "nvflare" + recipe_name = getattr(recipe, "name", None) or getattr(recipe._job, "name", None) or "nvflare" kw_args.setdefault("experiment_name", f"{recipe_name}-experiment") if not server_side and not client_side: @@ -407,7 +407,7 @@ def add_experiment_tracking( if tracking_type == "mlflow": server_config["kw_args"].setdefault("run_name", f"{recipe_name}-Server") receiver = receiver_class(**server_config) - recipe.job.to_server(receiver, "receiver") + recipe._job.to_server(receiver, "receiver") # Add client-side tracking if client_side: @@ -455,7 +455,7 @@ def add_final_global_evaluation( from nvflare.app_opt.pt.file_model_locator import PTFileModelLocator from nvflare.job_config.script_runner import FrameworkType - if getattr(recipe, "_cse_added", False) or _has_cross_site_eval_workflow(recipe.job): + if getattr(recipe, "_cse_added", False) or _has_cross_site_eval_workflow(recipe._job): raise RuntimeError("a cross-site evaluation workflow is already configured for this recipe") if getattr(recipe, "framework", None) != FrameworkType.PYTORCH: @@ -469,7 +469,7 @@ def add_final_global_evaluation( if not participating_clients: raise ValueError("participating_clients must not be empty; use None to evaluate on all clients") - comp_ids = getattr(recipe.job, "comp_ids", None) + comp_ids = getattr(recipe._job, "comp_ids", None) if not isinstance(comp_ids, dict): raise ValueError("final global evaluation requires a recipe that tracks component IDs") @@ -478,15 +478,15 @@ def add_final_global_evaluation( persistor_id = comp_ids.get("persistor_id", "") if not persistor_id: raise ValueError("final global evaluation requires a PyTorch model persistor") - model_locator_id = recipe.job.to_server( + model_locator_id = recipe._job.to_server( PTFileModelLocator(pt_persistor_id=persistor_id), id="final_model_locator" ) if not isinstance(model_locator_id, str) or not model_locator_id: raise RuntimeError("failed to register the final global model locator") comp_ids["locator_id"] = model_locator_id - recipe.job.to_server(ValidationJsonGenerator()) - recipe.job.to_server( + recipe._job.to_server(ValidationJsonGenerator()) + recipe._job.to_server( CrossSiteModelEval( model_locator_id=model_locator_id, submit_model_task_name="", @@ -621,7 +621,7 @@ def add_cross_site_evaluation( # Idempotency check: prevent multiple calls on the same recipe. # Keep the explicit flag fast-path, but also verify server workflow state so # protection remains effective even if dynamic attributes are lost. - if getattr(recipe, "_cse_added", False) or _has_cross_site_eval_workflow(recipe.job): + if getattr(recipe, "_cse_added", False) or _has_cross_site_eval_workflow(recipe._job): name = recipe.name if hasattr(recipe, "name") else "cross-site-evaluation job" raise RuntimeError( f"Cross-site evaluation has already been added to recipe '{name}'. " @@ -672,12 +672,12 @@ def add_cross_site_evaluation( locator_kwargs = {} if locator_config["persistor_param"] is not None: # For frameworks requiring persistor_id (PyTorch, TensorFlow), get it from comp_ids - if hasattr(recipe.job, "comp_ids"): - persistor_id = recipe.job.comp_ids.get("persistor_id", "") + if hasattr(recipe._job, "comp_ids"): + persistor_id = recipe._job.comp_ids.get("persistor_id", "") if not persistor_id: raise ValueError( f"Cross-site evaluation requires a persistor for {framework_to_locator[framework]} recipes, " - f"but no persistor_id found in recipe.job.comp_ids. " + "but no persistor_id was found in the generated job. " f"Ensure your recipe includes a model to create a persistor." ) locator_kwargs[locator_config["persistor_param"]] = persistor_id @@ -688,10 +688,10 @@ def add_cross_site_evaluation( ) model_locator = locator_class(**locator_kwargs) - model_locator_id = recipe.job.to_server(model_locator) + model_locator_id = recipe._job.to_server(model_locator) # Add validation JSON generator - recipe.job.to_server(ValidationJsonGenerator()) + recipe._job.to_server(ValidationJsonGenerator()) # Create and add cross-site evaluation controller eval_controller = CrossSiteModelEval( @@ -700,7 +700,7 @@ def add_cross_site_evaluation( validation_timeout=validation_timeout, participating_clients=participating_clients, ) - recipe.job.to_server(eval_controller) + recipe._job.to_server(eval_controller) # Let recipe handle framework-specific validator setup if needed # NumPy recipes implement add_cse_validator_if_needed() to add NPValidator automatically diff --git a/tests/integration_test/fast/study_session_test.py b/tests/integration_test/fast/study_session_test.py index ab940693e4..c40cdae493 100644 --- a/tests/integration_test/fast/study_session_test.py +++ b/tests/integration_test/fast/study_session_test.py @@ -543,8 +543,8 @@ def test_multistudy_prod_env_propagates_study_and_runtime_meta(self, multi_study recipe = _make_numpy_recipe("prod-env-study-a") env = ProdEnv(startup_kit_location=admin_root, username=MAIN_ADMIN, study="study-a", login_timeout=2.0) - recipe.process_env(env) - job_id = env.deploy(recipe.job) + run = recipe.execute(env) + job_id = run.job_id study_a_session = new_secure_session(MAIN_ADMIN, admin_root, study="study-a") multi_study_system["sessions"].append(study_a_session) diff --git a/tests/integration_test/slow/experiment_tracking_recipes_test.py b/tests/integration_test/slow/experiment_tracking_recipes_test.py index 172dda3f7d..885b0d2c93 100644 --- a/tests/integration_test/slow/experiment_tracking_recipes_test.py +++ b/tests/integration_test/slow/experiment_tracking_recipes_test.py @@ -87,9 +87,9 @@ def mlflow_client_model_path(self): def _add_model_to_apps(self, recipe, model_path=None): model_path = model_path or self.model_path - recipe.job.add_file_to_server(model_path) + recipe._job.add_file_to_server(model_path) for site_name in ("site-1", "site-2"): - recipe.job.add_file_to(model_path, site_name) + recipe._job.add_file_to(model_path, site_name) def _assert_exported_client_configs_have_executors(self, recipe, export_root: str): recipe.export(export_root) diff --git a/tests/integration_test/tools/export_recipe_job.py b/tests/integration_test/tools/export_recipe_job.py index bda3c735ca..7ea265550f 100644 --- a/tests/integration_test/tools/export_recipe_job.py +++ b/tests/integration_test/tools/export_recipe_job.py @@ -129,7 +129,7 @@ def patched_execute(self, env, server_exec_params=None, client_exec_params=None) exported job matches what a real run would produce. """ if server_exec_params: - self.job.to_server(server_exec_params) + self._job.to_server(server_exec_params) if client_exec_params: self._add_to_client_apps(client_exec_params) # Match real Recipe.execute behavior so export matches runtime configuration. @@ -189,7 +189,7 @@ def abort(self): if src_dir is not None: # Copy CONTENTS of src/ directly into app/custom/ (not as a subdirectory) # This way imports like "from data.xxx" work with just /local/custom in PYTHONPATH - job_name = recipe.job.name + job_name = recipe.name job_custom_dir = os.path.join(output_abs_path, job_name, "app", "custom") if os.path.exists(job_custom_dir): for item in os.listdir(src_dir): diff --git a/tests/unit_test/app_opt/pt/recipes/fed_eval_recipe_test.py b/tests/unit_test/app_opt/pt/recipes/fed_eval_recipe_test.py index 5399f1d390..b1dd92b2b2 100644 --- a/tests/unit_test/app_opt/pt/recipes/fed_eval_recipe_test.py +++ b/tests/unit_test/app_opt/pt/recipes/fed_eval_recipe_test.py @@ -70,8 +70,8 @@ def assert_recipe_basics(recipe, expected_name, expected_params): assert recipe.eval_script == expected_params.get("eval_script", "mock_eval_script.py") assert recipe.eval_args == expected_params.get("eval_args", "--batch_size 32") assert recipe.min_clients == expected_params.get("min_clients", 2) - assert recipe.job is not None - assert recipe.job.name == expected_name + assert recipe._job is not None + assert recipe._job.name == expected_name class TestFedEvalRecipe: @@ -123,7 +123,7 @@ def test_default_job_name(self, mock_file_system, base_recipe_params, simple_mod recipe = FedEvalRecipe(model=model, **base_recipe_params) assert recipe.name == "eval" - assert recipe.job.name == "eval" + assert recipe._job.name == "eval" def test_custom_command(self, mock_file_system, base_recipe_params, simple_model): """Test FedEvalRecipe with custom command.""" diff --git a/tests/unit_test/app_opt/sklearn/sklearn_recipe_test.py b/tests/unit_test/app_opt/sklearn/sklearn_recipe_test.py index fb67583be5..4f8d8e115a 100644 --- a/tests/unit_test/app_opt/sklearn/sklearn_recipe_test.py +++ b/tests/unit_test/app_opt/sklearn/sklearn_recipe_test.py @@ -59,7 +59,7 @@ def test_basic_initialization(self, mock_file_system, base_recipe_params): ) assert recipe.name == "test_sklearn" - assert recipe.job is not None + assert recipe._job is not None def test_weight_diff_is_allowed_for_clients_that_return_diffs(self, mock_file_system, base_recipe_params): from nvflare.app_opt.sklearn.recipes.fedavg import SklearnFedAvgRecipe @@ -84,7 +84,7 @@ def test_model_path_parameter_accepted(self, mock_file_system, base_recipe_param **base_recipe_params, ) - assert recipe.job is not None + assert recipe._job is not None def test_model_path_only_without_model_params(self, mock_file_system, base_recipe_params): """Test that model_path alone (load from file) works.""" @@ -96,7 +96,7 @@ def test_model_path_only_without_model_params(self, mock_file_system, base_recip **base_recipe_params, ) - assert recipe.job is not None + assert recipe._job is not None def test_relative_path_rejected(self, mock_file_system, base_recipe_params): """Test that relative model_path is rejected at construction time.""" @@ -144,7 +144,7 @@ def test_basic_initialization(self, mock_file_system): ) assert recipe.name == "test_kmeans" - assert recipe.job is not None + assert recipe._job is not None def test_model_path_accepted(self, mock_file_system): """Test that model_path parameter is accepted.""" @@ -159,7 +159,7 @@ def test_model_path_accepted(self, mock_file_system): model_path="/abs/path/to/kmeans.joblib", ) - assert recipe.job is not None + assert recipe._job is not None def test_relative_path_rejected(self): """Test that relative model_path is rejected (all sklearn recipes require absolute path).""" @@ -191,7 +191,7 @@ def test_basic_initialization(self, mock_file_system): ) assert recipe.name == "test_svm" - assert recipe.job is not None + assert recipe._job is not None def test_model_path_accepted(self, mock_file_system): """Test that model_path parameter is accepted.""" @@ -205,7 +205,7 @@ def test_model_path_accepted(self, mock_file_system): model_path="/abs/path/to/svm.joblib", ) - assert recipe.job is not None + assert recipe._job is not None def test_relative_path_rejected(self): """Test that relative model_path is rejected (all sklearn recipes require absolute path).""" diff --git a/tests/unit_test/app_opt/xgboost/xgboost_recipe_test.py b/tests/unit_test/app_opt/xgboost/xgboost_recipe_test.py index 24099398b1..42dbaf1469 100644 --- a/tests/unit_test/app_opt/xgboost/xgboost_recipe_test.py +++ b/tests/unit_test/app_opt/xgboost/xgboost_recipe_test.py @@ -32,7 +32,7 @@ def _per_site_config(): def _get_metrics_writer(recipe): - server_app = recipe.job._deploy_map[SERVER_SITE_NAME] + server_app = recipe._job._deploy_map[SERVER_SITE_NAME] return server_app.app_config.components.get("metrics_artifact_writer") diff --git a/tests/unit_test/recipe/cyclic_recipe_test.py b/tests/unit_test/recipe/cyclic_recipe_test.py index e1b1359a78..b710248d37 100644 --- a/tests/unit_test/recipe/cyclic_recipe_test.py +++ b/tests/unit_test/recipe/cyclic_recipe_test.py @@ -158,7 +158,7 @@ def test_applies_initial_ckpt_to_wrapper_model(self, mock_file_system, base_reci **base_recipe_params, ) - server_app = recipe.job._deploy_map.get("server") + server_app = recipe._job._deploy_map.get("server") persistor = server_app.app_config.components.get("persistor") assert persistor is not None assert getattr(persistor, "source_ckpt_file_full_name", None) == "/abs/path/to/model.pt" @@ -205,8 +205,8 @@ def test_pt_recipe_parameters_and_override_precedence(self, mock_file_system, ba **base_recipe_params, ) - controller = recipe.job._deploy_map["server"].app_config.workflows[0].controller - launcher = recipe.job._deploy_map[ALL_SITES].app_config.components["launcher"] + controller = recipe._job._deploy_map["server"].app_config.workflows[0].controller + launcher = recipe._job._deploy_map[ALL_SITES].app_config.components["launcher"] assert controller.task_assignment_timeout == 60 assert controller._task_check_period == 2.0 assert launcher._shutdown_timeout == 90.0 @@ -236,7 +236,7 @@ def test_pt_cyclic_initial_ckpt(self, mock_file_system, base_recipe_params, simp ) assert recipe.name == "test_pt_cyclic" - assert recipe.job is not None + assert recipe._job is not None def test_pt_cyclic_with_ptmodel_wrapper_returns_persistor_id( self, mock_file_system, base_recipe_params, simple_model @@ -250,7 +250,7 @@ def test_pt_cyclic_with_ptmodel_wrapper_returns_persistor_id( **base_recipe_params, ) - server_app = recipe.job._deploy_map.get("server") + server_app = recipe._job._deploy_map.get("server") assert server_app is not None assert "persistor" in server_app.app_config.components @@ -271,4 +271,4 @@ def test_tf_cyclic_initial_ckpt(self, mock_file_system, base_recipe_params): ) assert recipe.name == "test_tf_cyclic" - assert recipe.job is not None + assert recipe._job is not None diff --git a/tests/unit_test/recipe/edge_recipe_test.py b/tests/unit_test/recipe/edge_recipe_test.py index 5489151f5f..c24c5fd85d 100644 --- a/tests/unit_test/recipe/edge_recipe_test.py +++ b/tests/unit_test/recipe/edge_recipe_test.py @@ -165,7 +165,7 @@ def test_basic_initialization(self, mock_file_system, simple_pt_model, model_man device_manager_config=device_manager_config, ) - assert recipe.job is not None + assert recipe._job is not None def test_initial_ckpt_accepted( self, mock_file_system, simple_pt_model, model_manager_config, device_manager_config @@ -181,7 +181,7 @@ def test_initial_ckpt_accepted( initial_ckpt="/abs/path/to/model.pt", ) - assert recipe.job is not None + assert recipe._job is not None assert recipe.initial_ckpt == "/abs/path/to/model.pt" def test_dict_model_config_accepted(self, mock_file_system, model_manager_config, device_manager_config): @@ -195,7 +195,7 @@ def test_dict_model_config_accepted(self, mock_file_system, model_manager_config device_manager_config=device_manager_config, ) - assert recipe.job is not None + assert recipe._job is not None def test_dict_model_config_with_evaluator(self, mock_file_system, model_manager_config, device_manager_config): """Test that dict model config works with evaluator_config. @@ -216,7 +216,7 @@ def test_dict_model_config_with_evaluator(self, mock_file_system, model_manager_ evaluator_config=evaluator_config, ) - assert recipe.job is not None + assert recipe._job is not None # Verify model is stored as dict assert isinstance(recipe.model, dict) assert recipe.model["path"] == "torch.nn.Linear" @@ -290,7 +290,7 @@ def test_metrics_artifact_writer_is_configured( device_manager_config=device_manager_config, ) - self._find_metrics_writer(recipe.job) + self._find_metrics_writer(recipe._job) def test_device_wait_timeout_default_is_none( self, mock_file_system, simple_pt_model, model_manager_config, device_manager_config @@ -306,7 +306,7 @@ def test_device_wait_timeout_default_is_none( ) assert recipe.device_wait_timeout is None - assessor = self._find_assessor(recipe.job) + assessor = self._find_assessor(recipe._job) assert assessor.device_wait_timeout is None def test_device_wait_timeout_explicit_value( @@ -324,7 +324,7 @@ def test_device_wait_timeout_explicit_value( ) assert recipe.device_wait_timeout == 120.0 - assessor = self._find_assessor(recipe.job) + assessor = self._find_assessor(recipe._job) assert assessor.device_wait_timeout == 120.0 @pytest.mark.parametrize("bad_value", [0, -1, -100.0]) @@ -380,7 +380,7 @@ def test_sim_basic(self, mock_file_system, simple_device_model, model_manager_co model_manager_config=model_manager_config, device_manager_config=device_manager_config, ) - assert recipe.job is not None + assert recipe._job is not None class TestETFedBuffRecipeWithoutExecutorch: diff --git a/tests/unit_test/recipe/eval_recipe_test.py b/tests/unit_test/recipe/eval_recipe_test.py index 72ac6d7c39..2115aa891a 100644 --- a/tests/unit_test/recipe/eval_recipe_test.py +++ b/tests/unit_test/recipe/eval_recipe_test.py @@ -72,7 +72,7 @@ def test_basic_initialization(self, mock_file_system): min_clients=2, ) - assert recipe.job is not None + assert recipe._job is not None def test_initial_ckpt_accepted(self, mock_file_system): """Test that initial_ckpt parameter is accepted.""" @@ -84,7 +84,7 @@ def test_initial_ckpt_accepted(self, mock_file_system): initial_ckpt="/abs/path/to/model.npy", ) - assert recipe.job is not None + assert recipe._job is not None def test_with_model_dir(self, mock_file_system): """Test with model_dir specified.""" @@ -97,7 +97,7 @@ def test_with_model_dir(self, mock_file_system): model_name={"server": "best_model.npy"}, ) - assert recipe.job is not None + assert recipe._job is not None def test_with_eval_script(self, mock_file_system): """Test with custom eval_script.""" @@ -111,7 +111,7 @@ def test_with_eval_script(self, mock_file_system): initial_ckpt="/abs/path/to/model.npy", ) - assert recipe.job is not None + assert recipe._job is not None def test_with_eval_script_external_process(self, mock_file_system): """Test with eval_script in external process mode.""" @@ -125,4 +125,4 @@ def test_with_eval_script_external_process(self, mock_file_system): command="python3 -u", ) - assert recipe.job is not None + assert recipe._job is not None diff --git a/tests/unit_test/recipe/fed_task_recipe_test.py b/tests/unit_test/recipe/fed_task_recipe_test.py index 12c0b3b839..5000143c5f 100644 --- a/tests/unit_test/recipe/fed_task_recipe_test.py +++ b/tests/unit_test/recipe/fed_task_recipe_test.py @@ -97,7 +97,7 @@ def test_initializes_model_free_one_round_task(self, temp_task_script): assert recipe.framework == FrameworkType.RAW assert recipe.server_expected_format == ExchangeFormat.RAW - server_app = recipe.job._deploy_map["server"] + server_app = recipe._job._deploy_map["server"] controller = server_app.app_config.workflows[0].controller assert isinstance(controller, CmdTaskController) assert controller.task_name == "embed" @@ -107,7 +107,7 @@ def test_initializes_model_free_one_round_task(self, temp_task_script): assert controller.min_responses == 1 assert controller.timeout == 10 - client_app = recipe.job._deploy_map[ALL_SITES] + client_app = recipe._job._deploy_map[ALL_SITES] executor_def = client_app.app_config.executors[0] assert executor_def.tasks == ["embed"] assert temp_task_script in client_app.app_config.ext_scripts @@ -120,7 +120,7 @@ def test_defaults_send_task_name_and_request_meta(self, temp_task_script): task_script=temp_task_script, ) - controller = recipe.job._deploy_map["server"].app_config.workflows[0].controller + controller = recipe._job._deploy_map["server"].app_config.workflows[0].controller assert controller.task_data == {"task_name": "preprocess"} assert controller.task_meta == {"status": "request"} @@ -138,7 +138,7 @@ def test_exported_config_has_no_model_persistor(self, temp_task_script): ) with tempfile.TemporaryDirectory() as tmpdir: - recipe.job.export_job(tmpdir) + recipe._job.export_job(tmpdir) job_dir = os.path.join(tmpdir, "config_task") with open(os.path.join(job_dir, "app", "config", "config_fed_server.json")) as f: diff --git a/tests/unit_test/recipe/fedavg_recipe_test.py b/tests/unit_test/recipe/fedavg_recipe_test.py index a2221e102c..19c9af57cb 100644 --- a/tests/unit_test/recipe/fedavg_recipe_test.py +++ b/tests/unit_test/recipe/fedavg_recipe_test.py @@ -159,17 +159,17 @@ def assert_recipe_basics(recipe, expected_name, expected_params): assert recipe.train_args == expected_params.get("train_args", "--epochs 10") assert recipe.min_clients == expected_params.get("min_clients", 2) assert recipe.num_rounds == expected_params.get("num_rounds", 5) - assert recipe.job is not None - assert recipe.job.name == expected_name + assert recipe._job is not None + assert recipe._job.name == expected_name def get_model_selector(recipe): - server_app = recipe.job._deploy_map[SERVER_SITE_NAME] + server_app = recipe._job._deploy_map[SERVER_SITE_NAME] return server_app.app_config.components.get("model_selector") def get_server_component(recipe, component_id): - server_app = recipe.job._deploy_map[SERVER_SITE_NAME] + server_app = recipe._job._deploy_map[SERVER_SITE_NAME] return server_app.app_config.components.get(component_id) @@ -179,7 +179,7 @@ def get_server_component_from_job(job, component_id): def get_server_controller(recipe): - server_app = recipe.job._deploy_map[SERVER_SITE_NAME] + server_app = recipe._job._deploy_map[SERVER_SITE_NAME] return server_app.app_config.workflows[0].controller @@ -440,7 +440,7 @@ def test_numpy_recipe_basic_initialization(self, mock_file_system): assert recipe.name == "test_numpy" assert recipe.min_clients == 2 assert recipe.num_rounds == 3 - assert recipe.job is not None + assert recipe._job is not None def test_numpy_recipe_with_early_stopping(self, mock_file_system): """Test NumpyFedAvgRecipe with early stopping configuration.""" @@ -723,7 +723,7 @@ def test_per_site_empty_command_override_is_preserved(self, mock_file_system, ba **base_recipe_params, ) - site_app = recipe.job._deploy_map.get("site-1") + site_app = recipe._job._deploy_map.get("site-1") assert site_app is not None launcher = site_app.app_config.components.get("launcher") assert launcher is not None @@ -739,10 +739,10 @@ def test_custom_model_persistor_tracks_persistor_id(self, mock_file_system, base **base_recipe_params, ) - persistor_id = recipe.job.comp_ids.get("persistor_id", "") + persistor_id = recipe._job.comp_ids.get("persistor_id", "") assert persistor_id - assert "locator_id" not in recipe.job.comp_ids - server_app = recipe.job._deploy_map.get(SERVER_SITE_NAME) + assert "locator_id" not in recipe._job.comp_ids + server_app = recipe._job._deploy_map.get(SERVER_SITE_NAME) assert server_app is not None assert persistor_id in server_app.app_config.components @@ -759,10 +759,10 @@ def test_custom_model_persistor_with_locator_registers_locator( **base_recipe_params, ) - assert recipe.job.comp_ids.get("persistor_id", "") - locator_id = recipe.job.comp_ids.get("locator_id", "") + assert recipe._job.comp_ids.get("persistor_id", "") + locator_id = recipe._job.comp_ids.get("locator_id", "") assert locator_id - server_app = recipe.job._deploy_map.get(SERVER_SITE_NAME) + server_app = recipe._job._deploy_map.get(SERVER_SITE_NAME) assert server_app is not None assert server_app.app_config.components.get(locator_id) is locator @@ -916,7 +916,7 @@ def test_unified_numpy_initial_ckpt_only(self, mock_file_system, base_recipe_par ) assert recipe.initial_ckpt == "/abs/path/to/model.npy" - server_app = recipe.job._deploy_map[SERVER_SITE_NAME] + server_app = recipe._job._deploy_map[SERVER_SITE_NAME] persistor = server_app.app_config.components.get("persistor") assert isinstance(persistor, NPModelPersistor) assert persistor.source_ckpt_file_full_name == "/abs/path/to/model.npy" @@ -938,7 +938,7 @@ def test_unified_numpy_array_model(self, mock_file_system, base_recipe_params): **base_recipe_params, ) - server_app = recipe.job._deploy_map[SERVER_SITE_NAME] + server_app = recipe._job._deploy_map[SERVER_SITE_NAME] persistor = server_app.app_config.components.get("persistor") assert isinstance(persistor, NPModelPersistor) assert persistor.model == [1.0, 2.0, 3.0] diff --git a/tests/unit_test/recipe/fedopt_recipe_test.py b/tests/unit_test/recipe/fedopt_recipe_test.py index 6d81257f4d..a786eb6bf7 100644 --- a/tests/unit_test/recipe/fedopt_recipe_test.py +++ b/tests/unit_test/recipe/fedopt_recipe_test.py @@ -90,7 +90,7 @@ def test_basic_initialization(self, mock_file_system, base_recipe_params, simple assert recipe.name == "test_fedopt" assert recipe.model == simple_model - assert recipe.job is not None + assert recipe._job is not None def test_custom_aggregator_must_support_weight_diff(self, mock_file_system, base_recipe_params, simple_model): from nvflare.apis.dxo import DataKind @@ -128,7 +128,7 @@ def test_enable_tensor_disk_offload_configures_controller(self, mock_file_system ) assert recipe.enable_tensor_disk_offload is True - server_app = recipe.job._deploy_map[SERVER_SITE_NAME] + server_app = recipe._job._deploy_map[SERVER_SITE_NAME] controller = server_app.app_config.workflows[0].controller assert isinstance(controller, ScatterAndGather) assert controller.enable_tensor_disk_offload is True @@ -299,7 +299,7 @@ def test_dict_config_instantiates_model(self, mock_file_system, base_recipe_para # Verify instantiate_class was called with correct arguments mock_instantiate.assert_called_once_with("mymodule.MyModel", {"input_size": 10}) - assert recipe.job is not None + assert recipe._job is not None def test_model_none_raises_error(self, mock_file_system, base_recipe_params): """Test that model=None raises ValueError.""" @@ -331,7 +331,7 @@ def test_basic_initialization(self, mock_file_system, base_recipe_params): ) assert recipe.name == "test_tf_fedopt" - assert recipe.job is not None + assert recipe._job is not None def test_initial_ckpt_parameter_accepted(self, mock_file_system, base_recipe_params): """Test that initial_ckpt parameter is accepted (TF can load without model).""" diff --git a/tests/unit_test/recipe/fedstats_recipe_test.py b/tests/unit_test/recipe/fedstats_recipe_test.py index 379e31c86a..f44524c581 100644 --- a/tests/unit_test/recipe/fedstats_recipe_test.py +++ b/tests/unit_test/recipe/fedstats_recipe_test.py @@ -33,7 +33,7 @@ def test_fedstats_warns_for_secret_like_statistic_config(): ) stats_job_cls.return_value.setup_clients.assert_called_once_with(["site-1"]) - assert recipe.job is stats_job_cls.return_value + assert recipe._job is stats_job_cls.return_value def test_fedstats_warns_for_unsupported_secret_ref(): diff --git a/tests/unit_test/recipe/flower_recipe_test.py b/tests/unit_test/recipe/flower_recipe_test.py index 68d5be6f9f..3a217cc7a9 100644 --- a/tests/unit_test/recipe/flower_recipe_test.py +++ b/tests/unit_test/recipe/flower_recipe_test.py @@ -52,7 +52,7 @@ def test_flower_recipe_accepts_compatible_flwr_version(flwr_version): with patch("nvflare.app_opt.flower.recipe._create_flower_job", return_value=fake_job) as mock_flower_job: recipe = FlowerRecipe(flower_content="mock_flower_content") - assert recipe.job is fake_job + assert recipe._job is fake_job kwargs = mock_flower_job.call_args.kwargs assert kwargs["extra_env"] == {CLIENT_API_TYPE_KEY: ClientAPIType.EX_PROCESS_API.value} @@ -65,7 +65,7 @@ def test_flower_recipe_forwards_run_config(): with patch("nvflare.app_opt.flower.recipe._create_flower_job", return_value=fake_job) as mock_flower_job: recipe = FlowerRecipe(flower_content="mock_flower_content", run_config=run_config) - assert recipe.job is fake_job + assert recipe._job is fake_job kwargs = mock_flower_job.call_args.kwargs assert kwargs["run_config"] == run_config @@ -106,7 +106,7 @@ def test_flower_recipe_merges_extra_env(flwr_version): with patch("nvflare.app_opt.flower.recipe._create_flower_job", return_value=fake_job) as mock_flower_job: recipe = FlowerRecipe(flower_content="mock_flower_content", extra_env=user_env) - assert recipe.job is fake_job + assert recipe._job is fake_job kwargs = mock_flower_job.call_args.kwargs assert kwargs["extra_env"]["MY_VAR"] == "123" assert kwargs["extra_env"][CLIENT_API_TYPE_KEY] == ClientAPIType.EX_PROCESS_API.value @@ -136,7 +136,7 @@ def test_flower_recipe_with_predeployed_path(flwr_version): with patch("nvflare.app_opt.flower.recipe._create_flower_job", return_value=fake_job) as mock_flower_job: recipe = FlowerRecipe(flower_app_path="/opt/flower_apps/my_app") - assert recipe.job is fake_job + assert recipe._job is fake_job kwargs = mock_flower_job.call_args.kwargs assert kwargs["flower_app_path"] == "/opt/flower_apps/my_app" assert kwargs["flower_content"] is None diff --git a/tests/unit_test/recipe/numpy_lr_recipe_test.py b/tests/unit_test/recipe/numpy_lr_recipe_test.py index 259b919b79..4e427e6ac0 100644 --- a/tests/unit_test/recipe/numpy_lr_recipe_test.py +++ b/tests/unit_test/recipe/numpy_lr_recipe_test.py @@ -45,7 +45,7 @@ def test_basic_initialization(self, mock_file_system): num_features=13, ) - assert recipe.job is not None + assert recipe._job is not None def test_initial_ckpt_accepted(self, mock_file_system): """Test that initial_ckpt parameter is accepted.""" @@ -60,7 +60,7 @@ def test_initial_ckpt_accepted(self, mock_file_system): initial_ckpt="/abs/path/to/lr_weights.npy", ) - assert recipe.job is not None + assert recipe._job is not None def test_relative_path_accepted_if_exists(self, mock_file_system): """Test that existing relative paths are accepted and bundled.""" @@ -90,7 +90,7 @@ def test_custom_damping_factor(self, mock_file_system): damping_factor=0.5, ) - assert recipe.job is not None + assert recipe._job is not None assert recipe.damping_factor == 0.5 def test_with_train_args(self, mock_file_system): @@ -106,7 +106,7 @@ def test_with_train_args(self, mock_file_system): train_args="--data_root /tmp/data --batch_size 32", ) - assert recipe.job is not None + assert recipe._job is not None assert recipe.train_args == "--data_root /tmp/data --batch_size 32" def test_with_external_process(self, mock_file_system): @@ -123,7 +123,7 @@ def test_with_external_process(self, mock_file_system): command="python3 -u", ) - assert recipe.job is not None + assert recipe._job is not None assert recipe.launch_external_process is True def test_invalid_min_clients_rejected(self, mock_file_system): diff --git a/tests/unit_test/recipe/scaffold_recipe_test.py b/tests/unit_test/recipe/scaffold_recipe_test.py index ae2ecf85fe..93a0bbe1e8 100644 --- a/tests/unit_test/recipe/scaffold_recipe_test.py +++ b/tests/unit_test/recipe/scaffold_recipe_test.py @@ -70,7 +70,7 @@ def test_basic_initialization(self, mock_file_system, base_recipe_params, simple assert recipe.name == "test_scaffold" assert recipe.model == simple_model - assert recipe.job is not None + assert recipe._job is not None def test_enable_tensor_disk_offload_configures_controller(self, mock_file_system, base_recipe_params, simple_model): """Test PT ScaffoldRecipe passes tensor disk offload settings to the Scaffold controller.""" @@ -88,7 +88,7 @@ def test_enable_tensor_disk_offload_configures_controller(self, mock_file_system ) assert recipe.enable_tensor_disk_offload is True - server_app = recipe.job._deploy_map[SERVER_SITE_NAME] + server_app = recipe._job._deploy_map[SERVER_SITE_NAME] controller = server_app.app_config.workflows[0].controller assert isinstance(controller, Scaffold) assert controller.enable_tensor_disk_offload is True @@ -190,7 +190,7 @@ def test_basic_initialization(self, mock_file_system, base_recipe_params): ) assert recipe.name == "test_tf_scaffold" - assert recipe.job is not None + assert recipe._job is not None def test_initial_ckpt_parameter_accepted(self, mock_file_system, base_recipe_params): """Test that initial_ckpt parameter is accepted (TF can load without model).""" diff --git a/tests/unit_test/recipe/server_memory_gc_rounds_test.py b/tests/unit_test/recipe/server_memory_gc_rounds_test.py index 8fb2e284d0..7df5208563 100644 --- a/tests/unit_test/recipe/server_memory_gc_rounds_test.py +++ b/tests/unit_test/recipe/server_memory_gc_rounds_test.py @@ -59,7 +59,7 @@ def base_recipe_params(): def get_controller(recipe): """Extract controller from recipe's job.""" - server_app = recipe.job._deploy_map[SERVER_SITE_NAME] + server_app = recipe._job._deploy_map[SERVER_SITE_NAME] # Controller is usually the workflow component for comp_id, comp in server_app.app_config.components.items(): if hasattr(comp, "_memory_gc_rounds") or hasattr(comp, "memory_gc_rounds"): diff --git a/tests/unit_test/recipe/spec_test.py b/tests/unit_test/recipe/spec_test.py index c3758fb35f..ee1b9c5278 100644 --- a/tests/unit_test/recipe/spec_test.py +++ b/tests/unit_test/recipe/spec_test.py @@ -161,6 +161,16 @@ def test_deploy_with_non_local_script_logs_warning(self, caplog): class TestRecipeConfigMethods: """Test add_server_config and add_client_config methods in Recipe.""" + def test_job_storage_is_private(self): + from nvflare.recipe.spec import Recipe + + job = FedJob(name="test_job", min_clients=1) + + recipe = Recipe(job) + + assert recipe._job is job + assert not hasattr(recipe, "job") + @pytest.fixture def temp_script(self): """Create a temporary script file for testing.""" @@ -185,7 +195,7 @@ def test_add_server_config(self, temp_script): config = {"np_download_chunk_size": 2097152} recipe.add_server_config(config) - server_app = recipe.job._deploy_map.get("server") + server_app = recipe._job._deploy_map.get("server") assert server_app is not None assert server_app.app_config.additional_params == config @@ -205,7 +215,7 @@ def test_add_client_config(self, temp_script): config = {"timeout": 600} recipe.add_client_config(config) - all_clients_app = recipe.job._deploy_map.get(ALL_SITES) + all_clients_app = recipe._job._deploy_map.get(ALL_SITES) assert all_clients_app is not None assert all_clients_app.app_config.additional_params == config @@ -226,7 +236,7 @@ def test_add_client_file_adds_to_ext_scripts_and_ext_dirs(self, temp_script): recipe.add_client_file(temp_script) recipe.add_client_file(temp_dir) - all_clients_app = recipe.job._deploy_map.get(ALL_SITES) + all_clients_app = recipe._job._deploy_map.get(ALL_SITES) assert all_clients_app is not None assert temp_script in all_clients_app.app_config.ext_scripts assert temp_dir in all_clients_app.app_config.ext_dirs @@ -247,9 +257,9 @@ def test_add_client_file_preserves_per_site_clients_without_all_sites(self, temp recipe.add_client_file(temp_script) - assert ALL_SITES not in recipe.job._deploy_map - site_1_app = recipe.job._deploy_map.get("site-1") - site_2_app = recipe.job._deploy_map.get("site-2") + assert ALL_SITES not in recipe._job._deploy_map + site_1_app = recipe._job._deploy_map.get("site-1") + site_2_app = recipe._job._deploy_map.get("site-2") assert site_1_app is not None assert site_2_app is not None assert temp_script in site_1_app.app_config.ext_scripts @@ -275,9 +285,9 @@ def test_add_client_file_with_specific_clients_only_updates_selected_sites(self, recipe.add_client_file(targeted_file, clients=["site-2"]) try: - site_1_app = recipe.job._deploy_map.get("site-1") - site_2_app = recipe.job._deploy_map.get("site-2") - site_3_app = recipe.job._deploy_map.get("site-3") + site_1_app = recipe._job._deploy_map.get("site-1") + site_2_app = recipe._job._deploy_map.get("site-2") + site_3_app = recipe._job._deploy_map.get("site-3") assert site_1_app is not None assert site_2_app is not None assert site_3_app is not None @@ -303,7 +313,7 @@ def test_add_server_file_adds_to_server_ext_scripts_and_ext_dirs(self, temp_scri recipe.add_server_file(temp_script) recipe.add_server_file(temp_dir) - server_app = recipe.job._deploy_map.get("server") + server_app = recipe._job._deploy_map.get("server") assert server_app is not None assert temp_script in server_app.app_config.ext_scripts assert temp_dir in server_app.app_config.ext_dirs @@ -324,7 +334,7 @@ def test_config_in_generated_json(self, temp_script): recipe.add_client_config({"client_param": 456}) with tempfile.TemporaryDirectory() as tmpdir: - recipe.job.export_job(tmpdir) + recipe._job.export_job(tmpdir) job_dir = os.path.join(tmpdir, "test_config_gen") # Verify server config @@ -396,7 +406,7 @@ def __init__(self, class_names): self.class_names = class_names monkeypatch.setattr(spec_module, "DecomposerRegister", _DummyRegister) - monkeypatch.setattr(recipe.job, "to_server", _capture_server) + monkeypatch.setattr(recipe._job, "to_server", _capture_server) monkeypatch.setattr(recipe, "_add_to_client_apps", _capture_clients) recipe.add_decomposers(["pkg.mod.Dec"]) @@ -429,8 +439,8 @@ def test_enable_tensor_streaming_adds_matching_components(self): wait_send_task_data_all_clients_timeout=600.0, ) - server_streamer = recipe.job._deploy_map["server"].app_config.components["tensor_server_streamer"] - client_streamer = recipe.job._deploy_map[ALL_SITES].app_config.components["tensor_client_streamer"] + server_streamer = recipe._job._deploy_map["server"].app_config.components["tensor_server_streamer"] + client_streamer = recipe._job._deploy_map[ALL_SITES].app_config.components["tensor_client_streamer"] assert isinstance(server_streamer, TensorServerStreamer) assert isinstance(client_streamer, TensorClientStreamer) assert server_streamer.format == client_streamer.format == ExchangeFormat.PYTORCH @@ -447,8 +457,8 @@ def test_enable_tensor_streaming_uses_default_train_task(self): recipe = self._make_recipe() recipe.enable_tensor_streaming() - server_streamer = recipe.job._deploy_map["server"].app_config.components["tensor_server_streamer"] - client_streamer = recipe.job._deploy_map[ALL_SITES].app_config.components["tensor_client_streamer"] + server_streamer = recipe._job._deploy_map["server"].app_config.components["tensor_server_streamer"] + client_streamer = recipe._job._deploy_map[ALL_SITES].app_config.components["tensor_client_streamer"] assert server_streamer.tasks == client_streamer.tasks == ["train"] @pytest.mark.parametrize( @@ -466,7 +476,7 @@ def test_enable_tensor_streaming_validates_tasks(self, tasks, error_type): with pytest.raises(error_type, match="tasks must"): recipe.enable_tensor_streaming(tasks=tasks) - assert recipe.job._deploy_map == {} + assert recipe._job._deploy_map == {} def test_enable_tensor_streaming_rejects_mismatched_format(self): from nvflare.client.config import ExchangeFormat @@ -477,7 +487,7 @@ def test_enable_tensor_streaming_rejects_mismatched_format(self): with pytest.raises(ValueError, match="must match server_expected_format"): recipe.enable_tensor_streaming(format=ExchangeFormat.PYTORCH) - assert recipe.job._deploy_map == {} + assert recipe._job._deploy_map == {} def test_enable_tensor_streaming_rejects_duplicate_call(self): recipe = self._make_recipe() @@ -486,7 +496,7 @@ def test_enable_tensor_streaming_rejects_duplicate_call(self): with pytest.raises(RuntimeError, match="already been enabled"): recipe.enable_tensor_streaming() - assert len(recipe.job._deploy_map["server"].app_config.components) == 1 + assert len(recipe._job._deploy_map["server"].app_config.components) == 1 class _DummyExecEnv: @@ -532,7 +542,7 @@ def test_execute_server_params_do_not_accumulate(self, temp_script): ) env = _DummyExecEnv() - server_app = recipe.job._deploy_map.get("server") + server_app = recipe._job._deploy_map.get("server") assert server_app is not None assert server_app.app_config.additional_params == {} recipe.execute(env, server_exec_params={"param_a": 1}) @@ -553,7 +563,7 @@ def test_execute_empty_server_params_temporarily_clear_then_restore_snapshot(sel ) env = _DummyExecEnv() - server_app = recipe.job._deploy_map.get("server") + server_app = recipe._job._deploy_map.get("server") assert server_app is not None server_app.app_config.additional_params.update({"persisted": 1}) @@ -597,7 +607,7 @@ def test_execute_then_export_no_cross_contamination(self, temp_script): assert "from_execute" not in server_cfg assert server_cfg.get("from_export") == 2 - server_app = recipe.job._deploy_map.get("server") + server_app = recipe._job._deploy_map.get("server") assert server_app is not None assert server_app.app_config.additional_params == {} @@ -770,7 +780,7 @@ def __init__(self): set_per_site_config(recipe, {"site-1": {}, "site-2": {}}) assert recipe.configured_sites() == ["site-1", "site-2"] - assert recipe.job._deploy_map == {} + assert recipe._job._deploy_map == {} def test_configured_sites_prefers_helper_config_over_legacy_constructor_config(self): from nvflare.recipe.spec import Recipe @@ -857,7 +867,7 @@ def _make_recipe(self, name="test_placement_hardening"): def test_targeted_config_rejects_all_clients_topology(self): recipe = self._make_recipe() # Default recipe topology: one client app for all clients. - recipe.job.to_clients({"executor_standin": True}) + recipe._job.to_clients({"executor_standin": True}) with pytest.raises(ValueError, match="applies to all clients"): recipe.add_client_config({"timeout": 600}, clients=["site-1"]) @@ -866,25 +876,25 @@ def test_targeted_file_rejects_all_clients_topology(self, tmp_path): src_file = tmp_path / "wrapper.sh" src_file.write_text("#!/bin/sh\n") recipe = self._make_recipe() - recipe.job.to_clients({"executor_standin": True}) + recipe._job.to_clients({"executor_standin": True}) with pytest.raises(ValueError, match="applies to all clients"): recipe.add_client_file(str(src_file), clients=["site-1"]) def test_targeted_config_works_with_per_site_topology(self): recipe = self._make_recipe() - recipe.job.to({"site_arg": 1}, "site-1") - recipe.job.to({"site_arg": 2}, "site-2") + recipe._job.to({"site_arg": 1}, "site-1") + recipe._job.to({"site_arg": 2}, "site-2") recipe.add_client_config({"timeout": 600}, clients=["site-1"]) - assert recipe.job._deploy_map["site-1"].app_config.additional_params["timeout"] == 600 - assert "timeout" not in recipe.job._deploy_map["site-2"].app_config.additional_params + assert recipe._job._deploy_map["site-1"].app_config.additional_params["timeout"] == 600 + assert "timeout" not in recipe._job._deploy_map["site-2"].app_config.additional_params def test_targeted_config_rejects_unknown_site_with_per_site_topology(self): recipe = self._make_recipe() - recipe.job.to({"site_arg": 1}, "site-1") - recipe.job.to({"site_arg": 2}, "site-2") + recipe._job.to({"site_arg": 1}, "site-1") + recipe._job.to({"site_arg": 2}, "site-2") with pytest.raises(ValueError, match=r"unknown client site.*site-3"): recipe.add_client_config({"timeout": 600}, clients=["site-3"]) diff --git a/tests/unit_test/recipe/swarm_recipe_test.py b/tests/unit_test/recipe/swarm_recipe_test.py index 432c6ebfed..d21c560945 100644 --- a/tests/unit_test/recipe/swarm_recipe_test.py +++ b/tests/unit_test/recipe/swarm_recipe_test.py @@ -90,7 +90,7 @@ def test_import_from_new_location(self, mock_file_system, simple_pt_model): min_clients=2, ) - assert recipe.job is not None + assert recipe._job is not None def test_weight_diff_with_default_transfer_is_valid(self, mock_file_system, simple_pt_model): from nvflare.app_opt.pt.recipes.swarm import SwarmLearningRecipe @@ -104,7 +104,7 @@ def test_weight_diff_with_default_transfer_is_valid(self, mock_file_system, simp expected_data_kind=DataKind.WEIGHT_DIFF, ) - assert recipe.job is not None + assert recipe._job is not None def test_import_from_old_location_backward_compat(self, mock_file_system, simple_pt_model): """Test importing from old location (backward compatibility).""" @@ -118,7 +118,7 @@ def test_import_from_old_location_backward_compat(self, mock_file_system, simple min_clients=2, ) - assert recipe.job is not None + assert recipe._job is not None def test_initial_ckpt_accepted(self, mock_file_system, simple_pt_model): """Test that initial_ckpt parameter is accepted.""" @@ -133,7 +133,7 @@ def test_initial_ckpt_accepted(self, mock_file_system, simple_pt_model): initial_ckpt="/abs/path/to/model.pt", ) - assert recipe.job is not None + assert recipe._job is not None def test_relative_path_accepted_if_exists(self, mock_file_system, simple_pt_model): """Test that existing relative paths are accepted and bundled.""" @@ -164,7 +164,7 @@ def test_cross_site_eval_option(self, mock_file_system, simple_pt_model): cross_site_eval_timeout=600, ) - assert recipe.job is not None + assert recipe._job is not None def test_dict_model_config_accepted(self, mock_file_system): """Test that dict model config is accepted.""" @@ -178,7 +178,7 @@ def test_dict_model_config_accepted(self, mock_file_system): min_clients=2, ) - assert recipe.job is not None + assert recipe._job is not None def test_dict_model_config_with_ckpt(self, mock_file_system): """Test dict model config with initial checkpoint.""" @@ -193,7 +193,7 @@ def test_dict_model_config_with_ckpt(self, mock_file_system): initial_ckpt="/abs/path/to/model.pt", ) - assert recipe.job is not None + assert recipe._job is not None def test_dict_model_missing_class_path_or_path_rejected(self, mock_file_system): """Test that dict model without 'class_path' or 'path' key is rejected.""" @@ -235,7 +235,7 @@ def test_train_args_valid_keys_accepted(self, mock_file_system, simple_pt_model) train_args={"script_args": "--batch_size 32"}, # valid key ) - assert recipe.job is not None + assert recipe._job is not None def test_min_clients_accepted(self, mock_file_system, simple_pt_model): """Test that min_clients is a required parameter and is passed to the job.""" @@ -255,7 +255,7 @@ def test_min_clients_accepted(self, mock_file_system, simple_pt_model): min_clients=3, ) - assert recipe.job is not None + assert recipe._job is not None def test_launch_external_process_accepted(self, mock_file_system, simple_pt_model): """Test that launch_external_process=True is accepted.""" @@ -270,7 +270,7 @@ def test_launch_external_process_accepted(self, mock_file_system, simple_pt_mode launch_external_process=True, ) - assert recipe.job is not None + assert recipe._job is not None def test_command_accepted(self, mock_file_system, simple_pt_model): """Test that command is accepted alongside launch_external_process.""" @@ -286,7 +286,7 @@ def test_command_accepted(self, mock_file_system, simple_pt_model): command="python3 -u", ) - assert recipe.job is not None + assert recipe._job is not None class TestSwarmLearningRecipeControllerConfig: @@ -315,8 +315,8 @@ def test_parameters_and_override_precedence(self, mock_file_system, simple_pt_mo }, ) - server_controller = recipe.job._deploy_map["server"].app_config.workflows[0].controller - client_app = recipe.job._deploy_map[ALL_SITES] + server_controller = recipe._job._deploy_map["server"].app_config.workflows[0].controller + client_app = recipe._job._deploy_map[ALL_SITES] client_controller = next(item.executor for item in client_app.app_config.executors if item.tasks == ["swarm_*"]) assert server_controller.progress_timeout == 9000 assert client_controller.learn_task_timeout == 2400 @@ -394,7 +394,7 @@ def test_memory_gc_rounds_custom_accepted(self, mock_file_system, simple_pt_mode min_clients=2, memory_gc_rounds=2, ) - assert recipe.job is not None + assert recipe._job is not None def test_memory_gc_disabled_accepted(self, mock_file_system, simple_pt_model): """memory_gc_rounds=0 disables GC.""" @@ -408,7 +408,7 @@ def test_memory_gc_disabled_accepted(self, mock_file_system, simple_pt_model): min_clients=2, memory_gc_rounds=0, ) - assert recipe.job is not None + assert recipe._job is not None def test_cuda_empty_cache_accepted(self, mock_file_system, simple_pt_model): """cuda_empty_cache=True is accepted and wired through.""" @@ -422,7 +422,7 @@ def test_cuda_empty_cache_accepted(self, mock_file_system, simple_pt_model): min_clients=2, cuda_empty_cache=True, ) - assert recipe.job is not None + assert recipe._job is not None class TestSwarmLearningRecipePipeType: diff --git a/tests/unit_test/recipe/utils_test.py b/tests/unit_test/recipe/utils_test.py index f412446316..06f488498e 100644 --- a/tests/unit_test/recipe/utils_test.py +++ b/tests/unit_test/recipe/utils_test.py @@ -332,13 +332,13 @@ def _make_recipe(self, name, **fed_job_kwargs): return Recipe(FedJob(name=name, **fed_job_kwargs)) def _export_meta(self, recipe, tmp_path): - recipe.job.export_job(str(tmp_path)) - with open(tmp_path / recipe.job.name / "meta.json") as f: + recipe._job.export_job(str(tmp_path)) + with open(tmp_path / recipe._job.name / "meta.json") as f: return json.load(f) def test_set_recipe_meta_sets_recognized_top_level_meta_props(self, tmp_path): recipe = self._make_recipe("test_recipe_meta", min_clients=1, meta_props={"owner": "alice"}) - recipe.job.to_server({"server_arg": True}) + recipe._job.to_server({"server_arg": True}) resource_spec = { "site-1": {"num_of_gpus": 1, "mem_per_gpu_in_GiB": 4}, @@ -354,7 +354,7 @@ def test_set_recipe_meta_sets_recognized_top_level_meta_props(self, tmp_path): set_recipe_meta(recipe, JobMetaKey.SCOPE, "private") set_recipe_meta(recipe, JobMetaKey.CUSTOM_PROPS, {"team": "research"}) - job_config = recipe.job.job + job_config = recipe._job.job # Dedicated FedJobConfig fields are untouched by the helper. assert job_config.min_clients == 1 assert job_config.mandatory_clients is None @@ -384,16 +384,16 @@ def test_set_recipe_meta_warns_for_secret_and_unsupported_ref(self): def test_set_recipe_meta_warns_and_overrides_registered_resource_specs(self, tmp_path): recipe = self._make_recipe("test_recipe_meta_resource_conflict", min_clients=2) - recipe.job.job.add_resource_spec("base-site", {"num_of_gpus": 0}) - recipe.job.to_server({"server_arg": True}) + recipe._job.job.add_resource_spec("base-site", {"num_of_gpus": 0}) + recipe._job.to_server({"server_arg": True}) resource_spec = {"site-1": {"num_of_gpus": 1}} with pytest.warns(UserWarning, match="overrides the per-site resource specs"): set_recipe_meta(recipe, JobMetaKey.RESOURCE_SPEC, resource_spec) - assert recipe.job.job.meta_props["resource_spec"] == resource_spec + assert recipe._job.job.meta_props["resource_spec"] == resource_spec # The dedicated FedJobConfig field is never mutated by the helper. - assert recipe.job.job.resource_specs == {"base-site": {"num_of_gpus": 0}} + assert recipe._job.job.resource_specs == {"base-site": {"num_of_gpus": 0}} # meta_props is merged last, so the registered per-site spec is overridden. exported_meta = self._export_meta(recipe, tmp_path) @@ -401,18 +401,18 @@ def test_set_recipe_meta_warns_and_overrides_registered_resource_specs(self, tmp def test_set_recipe_meta_replaces_different_value_from_existing_meta_props(self): recipe = self._make_recipe("test_recipe_meta_props_conflict", min_clients=1, meta_props={"scope": "a"}) - original_meta_props = recipe.job.job.meta_props + original_meta_props = recipe._job.job.meta_props set_recipe_meta(recipe, JobMetaKey.SCOPE, "b") - assert recipe.job.job.meta_props is original_meta_props - assert recipe.job.job.meta_props["scope"] == "b" + assert recipe._job.job.meta_props is original_meta_props + assert recipe._job.job.meta_props["scope"] == "b" def test_set_recipe_meta_allows_existing_meta_props_that_differ_from_generated_values(self): recipe = self._make_recipe("test_recipe_meta_generated_conflict", min_clients=2, meta_props={"min_clients": 5}) set_recipe_meta(recipe, JobMetaKey.SCOPE, "private") - assert recipe.job.job.meta_props["min_clients"] == 5 - assert recipe.job.job.meta_props["scope"] == "private" + assert recipe._job.job.meta_props["min_clients"] == 5 + assert recipe._job.job.meta_props["scope"] == "private" @pytest.mark.parametrize( "key", @@ -443,14 +443,14 @@ def test_set_recipe_meta_stores_caller_independent_copy(self): launcher_spec["site-1"]["docker"]["image"] = "changed" custom_props["clients"].append("site-2") - assert recipe.job.job.meta_props["launcher_spec"] == {"site-1": {"docker": {"image": "nvflare:latest"}}} - assert recipe.job.job.meta_props["custom_props"] == {"clients": ["site-1"]} + assert recipe._job.job.meta_props["launcher_spec"] == {"site-1": {"docker": {"image": "nvflare:latest"}}} + assert recipe._job.job.meta_props["custom_props"] == {"clients": ["site-1"]} def test_set_recipe_meta_accepts_nested_booleans(self): recipe = self._make_recipe("test_recipe_meta_nested_bool", min_clients=1) set_recipe_meta(recipe, JobMetaKey.CUSTOM_PROPS, {"enabled": True, "flags": [False, True]}) - assert recipe.job.job.meta_props["custom_props"] == {"enabled": True, "flags": [False, True]} + assert recipe._job.job.meta_props["custom_props"] == {"enabled": True, "flags": [False, True]} def test_set_recipe_meta_normalizes_non_string_dict_keys(self): recipe = self._make_recipe("test_recipe_meta_key_coercion", min_clients=1) @@ -458,17 +458,17 @@ def test_set_recipe_meta_normalizes_non_string_dict_keys(self): # Stored value matches what meta.json will contain (keys coerced to strings), # so in-process consumers and reloaded-meta consumers agree. - assert recipe.job.job.meta_props["custom_props"] == {"1": "a"} + assert recipe._job.job.meta_props["custom_props"] == {"1": "a"} def test_set_recipe_meta_requires_recipe_with_fed_job(self): - with pytest.raises(TypeError, match="recipe must provide a FedJob through recipe.job"): + with pytest.raises(TypeError, match="recipe must be backed by a FedJob"): set_recipe_meta(object(), JobMetaKey.SCOPE, "private") def test_set_recipe_meta_rejects_wrong_typed_job_config(self): from types import SimpleNamespace - recipe = SimpleNamespace(job=SimpleNamespace(job=object())) - with pytest.raises(TypeError, match="recipe must provide a FedJob through recipe.job"): + recipe = SimpleNamespace(_job=SimpleNamespace(job=object())) + with pytest.raises(TypeError, match="recipe must be backed by a FedJob"): set_recipe_meta(recipe, JobMetaKey.SCOPE, "private") @pytest.mark.parametrize( @@ -608,7 +608,7 @@ def _make_recipe(self, comp_ids=None, framework=None): job = MagicMock() job._deploy_map = {} job.comp_ids = comp_ids - return SimpleNamespace(job=job, framework=framework or FrameworkType.PYTORCH) + return SimpleNamespace(_job=job, framework=framework or FrameworkType.PYTORCH) @pytest.fixture def recording_cross_site_eval(self, monkeypatch): @@ -642,11 +642,11 @@ def test_adds_locator_generator_and_final_eval_controller(self, recording_cross_ captured_kwargs, controller_type = recording_cross_site_eval recipe = self._make_recipe({"persistor_id": "persistor"}) - recipe.job.to_server.side_effect = ["final_model_locator", None, None] + recipe._job.to_server.side_effect = ["final_model_locator", None, None] add_final_global_evaluation(recipe, participating_clients=["site-1"], validation_timeout=42) - calls = recipe.job.to_server.call_args_list + calls = recipe._job.to_server.call_args_list assert isinstance(calls[0].args[0], PTFileModelLocator) assert calls[0].kwargs == {"id": "final_model_locator"} assert isinstance(calls[1].args[0], ValidationJsonGenerator) @@ -660,7 +660,7 @@ def test_adds_locator_generator_and_final_eval_controller(self, recording_cross_ "participating_clients": ["site-1"], } ] - assert recipe.job.comp_ids["locator_id"] == "final_model_locator" + assert recipe._job.comp_ids["locator_id"] == "final_model_locator" assert recipe._cse_added is True def test_reuses_existing_model_locator(self, recording_cross_site_eval): @@ -671,8 +671,8 @@ def test_reuses_existing_model_locator(self, recording_cross_site_eval): add_final_global_evaluation(recipe) - assert recipe.job.to_server.call_count == 2 - controller = recipe.job.to_server.call_args_list[-1].args[0] + assert recipe._job.to_server.call_count == 2 + controller = recipe._job.to_server.call_args_list[-1].args[0] assert isinstance(controller, controller_type) assert captured_kwargs == [ { @@ -700,7 +700,7 @@ def test_validates_participating_clients(self, participating_clients, error_type with pytest.raises(error_type, match="participating_clients must"): add_final_global_evaluation(recipe, participating_clients=participating_clients) - recipe.job.to_server.assert_not_called() + recipe._job.to_server.assert_not_called() def test_rejects_duplicate_configuration(self): from nvflare.recipe import add_final_global_evaluation @@ -741,7 +741,7 @@ def test_rejects_failed_model_locator_registration(self): from nvflare.recipe import add_final_global_evaluation recipe = self._make_recipe({"persistor_id": "persistor"}) - recipe.job.to_server.return_value = None + recipe._job.to_server.return_value = None with pytest.raises(RuntimeError, match="failed to register"): add_final_global_evaluation(recipe) @@ -785,7 +785,7 @@ def test_mlflow_defaults_derive_from_recipe_name(self, dummy_mlflow): add_experiment_tracking(recipe, "mlflow") - receiver = recipe.job._deploy_map["server"].app_config.components["receiver"] + receiver = recipe._job._deploy_map["server"].app_config.components["receiver"] assert receiver.kw_args == { "experiment_name": "named_job-experiment", "run_name": "named_job-Server", @@ -800,7 +800,7 @@ def test_mlflow_client_tracking_can_omit_config(self, dummy_mlflow): add_experiment_tracking(recipe, "mlflow", client_side=True, server_side=False) - receiver = recipe.job._deploy_map[ALL_SITES].app_config.components["client_receiver"] + receiver = recipe._job._deploy_map[ALL_SITES].app_config.components["client_receiver"] assert receiver.kw_args == { "experiment_name": "client_job-experiment", "run_name": "client_job-Client", @@ -815,7 +815,7 @@ def test_mlflow_defaults_preserve_explicit_values_and_input(self, dummy_mlflow): add_experiment_tracking(recipe, "mlflow", config) - receiver = recipe.job._deploy_map["server"].app_config.components["receiver"] + receiver = recipe._job._deploy_map["server"].app_config.components["receiver"] assert receiver.kw_args == { "experiment_name": "custom-experiment", "run_name": "named_job-Server", @@ -830,8 +830,8 @@ def test_mlflow_both_sides_use_side_specific_default_run_names(self, dummy_mlflo add_experiment_tracking(recipe, "mlflow", client_side=True) - server_receiver = recipe.job._deploy_map["server"].app_config.components["receiver"] - client_receiver = recipe.job._deploy_map[ALL_SITES].app_config.components["client_receiver"] + server_receiver = recipe._job._deploy_map["server"].app_config.components["receiver"] + client_receiver = recipe._job._deploy_map[ALL_SITES].app_config.components["client_receiver"] assert server_receiver.kw_args["run_name"] == "both_sides-Server" assert client_receiver.kw_args["run_name"] == "both_sides-Client" @@ -861,11 +861,11 @@ def test_client_side_tracking_specific_clients(self, dummy_tracking): clients=["site-1"], ) - receiver = recipe.job._deploy_map["site-1"].app_config.components["client_receiver"] + receiver = recipe._job._deploy_map["site-1"].app_config.components["client_receiver"] assert receiver.tracking_uri == "file:///tmp/site-1/mlruns" # Local (non-federated) analytics events are configured by default. assert receiver.events == [ANALYTIC_EVENT_TYPE] - assert ALL_SITES not in recipe.job._deploy_map + assert ALL_SITES not in recipe._job._deploy_map def test_client_side_tracking_per_site_configs(self, dummy_tracking): from nvflare.recipe.utils import add_experiment_tracking @@ -874,7 +874,7 @@ def test_client_side_tracking_per_site_configs(self, dummy_tracking): # Real recipes have per-site client apps (e.g. from per_site_config) before # tracking is added; targeting only existing sites is enforced. for site in ("site-1", "site-2"): - recipe.job.to({"site_arg": site}, site) + recipe._job.to({"site_arg": site}, site) for site in ("site-1", "site-2"): add_experiment_tracking( recipe, @@ -886,7 +886,7 @@ def test_client_side_tracking_per_site_configs(self, dummy_tracking): ) for site in ("site-1", "site-2"): - receiver = recipe.job._deploy_map[site].app_config.components["client_receiver"] + receiver = recipe._job._deploy_map[site].app_config.components["client_receiver"] assert receiver.tracking_uri == f"file:///tmp/{site}/mlruns" def test_client_side_tracking_all_clients_by_default(self, dummy_tracking): @@ -896,7 +896,7 @@ def test_client_side_tracking_all_clients_by_default(self, dummy_tracking): recipe = self._make_recipe() add_experiment_tracking(recipe, dummy_tracking, {"tracking_uri": "u"}, client_side=True, server_side=False) - receiver = recipe.job._deploy_map[ALL_SITES].app_config.components["client_receiver"] + receiver = recipe._job._deploy_map[ALL_SITES].app_config.components["client_receiver"] assert receiver.tracking_uri == "u" def test_server_side_tracking_unaffected_by_clients_feature(self, dummy_tracking): @@ -905,7 +905,7 @@ def test_server_side_tracking_unaffected_by_clients_feature(self, dummy_tracking recipe = self._make_recipe() add_experiment_tracking(recipe, dummy_tracking, {"tracking_uri": "u"}) - components = recipe.job._deploy_map["server"].app_config.components + components = recipe._job._deploy_map["server"].app_config.components assert components["receiver"].tracking_uri == "u" # Server receiver keeps federated (default) events. assert not hasattr(components["receiver"], "events") @@ -938,8 +938,8 @@ def test_clients_targeting_rejects_unknown_site(self, dummy_tracking): from nvflare.recipe.utils import add_experiment_tracking recipe = self._make_recipe() - recipe.job.to({"site_arg": 1}, "site-1") - recipe.job.to({"site_arg": 2}, "site-2") + recipe._job.to({"site_arg": 1}, "site-1") + recipe._job.to({"site_arg": 2}, "site-2") with pytest.raises(ValueError, match="unknown client site"): add_experiment_tracking( @@ -956,7 +956,7 @@ def test_clients_targeting_rejects_all_sites_topology(self, dummy_tracking): recipe = self._make_recipe() # Default recipe topology: one client app for all clients. - recipe.job.to_clients({"executor_standin": True}) + recipe._job.to_clients({"executor_standin": True}) with pytest.raises(ValueError, match="applies to all clients"): add_experiment_tracking( @@ -972,7 +972,7 @@ def test_per_site_client_receiver_survives_export(self, dummy_tracking, tmp_path from nvflare.recipe.utils import add_experiment_tracking recipe = self._make_recipe("test_tracking_export") - recipe.job.to_server({"server_arg": True}) + recipe._job.to_server({"server_arg": True}) add_experiment_tracking( recipe, dummy_tracking, @@ -982,7 +982,7 @@ def test_per_site_client_receiver_survives_export(self, dummy_tracking, tmp_path clients=["site-1"], ) - recipe.job.export_job(str(tmp_path)) + recipe._job.export_job(str(tmp_path)) with open(tmp_path / "test_tracking_export" / "app_site-1" / "config" / "config_fed_client.json") as f: client_cfg = json.load(f) From 461671da8cb8a0d92dfa74ff9063c64d57cb1210 Mon Sep 17 00:00:00 2001 From: Kevin Date: Tue, 14 Jul 2026 16:31:06 -0400 Subject: [PATCH 2/4] Address Recipe privacy review feedback --- nvflare/edge/tools/edge_fed_buff_recipe.py | 3 +- nvflare/recipe/spec.py | 1 + .../tools/export_recipe_job.py | 36 +++++++++++-------- tests/unit_test/recipe/flower_recipe_test.py | 13 +++---- tests/unit_test/recipe/spec_test.py | 1 + 5 files changed, 32 insertions(+), 22 deletions(-) diff --git a/nvflare/edge/tools/edge_fed_buff_recipe.py b/nvflare/edge/tools/edge_fed_buff_recipe.py index 7935e92b85..3be309fe92 100644 --- a/nvflare/edge/tools/edge_fed_buff_recipe.py +++ b/nvflare/edge/tools/edge_fed_buff_recipe.py @@ -303,7 +303,8 @@ def process_env(self, env: ExecEnv): f"invalid {DEVICE_SIMULATION_ENV_KEY} in env: expect SimulationConfig but got {type(simulation_config)}" ) - assert isinstance(self._job, EdgeJob) + if not isinstance(self._job, EdgeJob): + raise TypeError(f"expected recipe job to be an EdgeJob, got {type(self._job).__name__}") self._configure_simulation(self._job, simulation_config) def create_job(self) -> EdgeJob: diff --git a/nvflare/recipe/spec.py b/nvflare/recipe/spec.py index 8a8b9adc31..e0f00b0dcf 100644 --- a/nvflare/recipe/spec.py +++ b/nvflare/recipe/spec.py @@ -252,6 +252,7 @@ def __init__(self, job: FedJob): job: the job that implements the recipe. """ self._job = job + self.name = job.name self._helper_per_site_config = None self._tensor_streaming_added = False self._cse_added = False diff --git a/tests/integration_test/tools/export_recipe_job.py b/tests/integration_test/tools/export_recipe_job.py index 7ea265550f..a67e5fe6bb 100644 --- a/tests/integration_test/tools/export_recipe_job.py +++ b/tests/integration_test/tools/export_recipe_job.py @@ -118,24 +118,25 @@ def export_recipe_from_job_py(recipe_dir: str, output_dir: str, recipe_args: Opt # Change to recipe directory so relative imports work os.chdir(recipe_abs_path) - # Storage for captured recipe - captured_recipe = {"recipe": None} + # Storage for the captured execute call. The public Recipe.export API later + # applies these arguments with the same semantics as Recipe.execute. + captured_execute = { + "recipe": None, + "env": None, + "server_exec_params": None, + "client_exec_params": None, + } def patched_execute(self, env, server_exec_params=None, client_exec_params=None): """Capture the recipe instead of executing. - Applies server/client exec params exactly as the real Recipe.execute does - (including per-site topology preservation via _add_to_client_apps) so the - exported job matches what a real run would produce. + The original arguments are forwarded to Recipe.export after job.py has + finished loading so the exported job matches a real execution. """ - if server_exec_params: - self._job.to_server(server_exec_params) - if client_exec_params: - self._add_to_client_apps(client_exec_params) - # Match real Recipe.execute behavior so export matches runtime configuration. - self.process_env(env) - - captured_recipe["recipe"] = self + captured_execute["recipe"] = self + captured_execute["env"] = env + captured_execute["server_exec_params"] = server_exec_params + captured_execute["client_exec_params"] = client_exec_params class MockRun: """Mimics the real Run interface so post-execute code in job.py doesn't break.""" @@ -173,13 +174,18 @@ def abort(self): if e.code not in (None, 0): raise - recipe = captured_recipe["recipe"] + recipe = captured_execute["recipe"] if recipe is None: raise RuntimeError("Failed to capture recipe - job.py did not call recipe.execute()") # Export the recipe to job folder os.makedirs(output_abs_path, exist_ok=True) - recipe.export(job_dir=output_abs_path) + recipe.export( + job_dir=output_abs_path, + server_exec_params=captured_execute["server_exec_params"], + client_exec_params=captured_execute["client_exec_params"], + env=captured_execute["env"], + ) print(f"Exported recipe to: {output_abs_path}") # Copy the src/ folder if it exists (for model imports). diff --git a/tests/unit_test/recipe/flower_recipe_test.py b/tests/unit_test/recipe/flower_recipe_test.py index 3a217cc7a9..7599123b5a 100644 --- a/tests/unit_test/recipe/flower_recipe_test.py +++ b/tests/unit_test/recipe/flower_recipe_test.py @@ -23,6 +23,7 @@ from nvflare.client.api import ClientAPIType from nvflare.client.api_spec import CLIENT_API_TYPE_KEY from nvflare.fuel.utils.secret_utils import PotentialSecretWarning, UnsupportedSecretRefWarning +from nvflare.job_config.api import FedJob from nvflare.recipe import secret_ref @@ -47,7 +48,7 @@ def test_flower_recipe_rejects_missing_flwr_package(): @pytest.mark.parametrize("flwr_version", ["1.26.0", "1.26.1", "1.27.5"]) def test_flower_recipe_accepts_compatible_flwr_version(flwr_version): - fake_job = object() + fake_job = FedJob(name="test_flower", min_clients=1) with patch("nvflare.app_opt.flower.recipe.get_package_version", return_value=flwr_version): with patch("nvflare.app_opt.flower.recipe._create_flower_job", return_value=fake_job) as mock_flower_job: recipe = FlowerRecipe(flower_content="mock_flower_content") @@ -58,7 +59,7 @@ def test_flower_recipe_accepts_compatible_flwr_version(flwr_version): def test_flower_recipe_forwards_run_config(): - fake_job = object() + fake_job = FedJob(name="test_flower", min_clients=1) run_config = {"learning-rate": 0.01, "momentum": 0.9} with patch("nvflare.app_opt.flower.recipe.get_package_version", return_value="1.26.0"): @@ -78,7 +79,7 @@ def test_flower_recipe_forwards_run_config(): ], ) def test_flower_recipe_warns_on_secret_parameters(parameter, value): - fake_job = object() + fake_job = FedJob(name="test_flower", min_clients=1) with patch("nvflare.app_opt.flower.recipe.get_package_version", return_value="1.26.0"): with patch("nvflare.app_opt.flower.recipe._create_flower_job", return_value=fake_job): @@ -88,7 +89,7 @@ def test_flower_recipe_warns_on_secret_parameters(parameter, value): @pytest.mark.parametrize("parameter", ["extra_env", "run_config"]) def test_flower_recipe_warns_on_unsupported_secret_refs(parameter): - fake_job = object() + fake_job = FedJob(name="test_flower", min_clients=1) value = {"api_token": secret_ref("API_TOKEN")} with patch("nvflare.app_opt.flower.recipe.get_package_version", return_value="1.26.0"): @@ -99,7 +100,7 @@ def test_flower_recipe_warns_on_unsupported_secret_refs(parameter): @pytest.mark.parametrize("flwr_version", ["1.26.0", "1.26.1", "1.27.5"]) def test_flower_recipe_merges_extra_env(flwr_version): - fake_job = object() + fake_job = FedJob(name="test_flower", min_clients=1) user_env = {"MY_VAR": "123"} with patch("nvflare.app_opt.flower.recipe.get_package_version", return_value=flwr_version): @@ -131,7 +132,7 @@ def test_flower_recipe_rejects_extra_env_with_wrong_client_api_type(flwr_version @pytest.mark.parametrize("flwr_version", ["1.26.0", "1.26.1", "1.27.5"]) def test_flower_recipe_with_predeployed_path(flwr_version): - fake_job = object() + fake_job = FedJob(name="test_flower", min_clients=1) with patch("nvflare.app_opt.flower.recipe.get_package_version", return_value=flwr_version): with patch("nvflare.app_opt.flower.recipe._create_flower_job", return_value=fake_job) as mock_flower_job: recipe = FlowerRecipe(flower_app_path="/opt/flower_apps/my_app") diff --git a/tests/unit_test/recipe/spec_test.py b/tests/unit_test/recipe/spec_test.py index ee1b9c5278..f0b2574f66 100644 --- a/tests/unit_test/recipe/spec_test.py +++ b/tests/unit_test/recipe/spec_test.py @@ -169,6 +169,7 @@ def test_job_storage_is_private(self): recipe = Recipe(job) assert recipe._job is job + assert recipe.name == job.name assert not hasattr(recipe, "job") @pytest.fixture From de642fa33bc7e17d14260d8e3d5860afa451d079 Mon Sep 17 00:00:00 2001 From: Kevin Date: Wed, 15 Jul 2026 09:22:43 -0400 Subject: [PATCH 3/4] Simplify NumPy CSE validator setup --- nvflare/app_common/np/recipes/fedavg.py | 31 ++--------- nvflare/recipe/utils.py | 69 ++----------------------- 2 files changed, 9 insertions(+), 91 deletions(-) diff --git a/nvflare/app_common/np/recipes/fedavg.py b/nvflare/app_common/np/recipes/fedavg.py index 03d2278cba..3380f35f77 100644 --- a/nvflare/app_common/np/recipes/fedavg.py +++ b/nvflare/app_common/np/recipes/fedavg.py @@ -193,40 +193,17 @@ def _setup_model_and_persistor(self, job) -> str: return "" def add_cse_validator_if_needed(self): - """Add NPValidator for cross-site evaluation if not already configured. + """Add NPValidator for cross-site evaluation. NumPy recipes need specialized NPValidator because: - NumPy training scripts typically only handle training tasks - Wildcard executors (tasks=["*"]) don't actually implement validation - Cross-site evaluation requires dedicated validation component - This method checks if a dedicated validator is already configured. - If only wildcard executors exist, adds NPValidator. + ``add_cross_site_evaluation()`` invokes this hook only after its + idempotency check, so each successful CSE augmentation adds one validator. """ from nvflare.app_common.app_constant import AppConstants from nvflare.app_common.np.np_validator import NPValidator - # Check if validation task is explicitly configured (not just via wildcard) - has_explicit_validator = False - if hasattr(self._job, "_deploy_map"): - for target, app in self._job._deploy_map.items(): - if target == "server": - continue - - if hasattr(app, "app_config") and hasattr(app.app_config, "executors"): - for executor_def in app.app_config.executors: - if hasattr(executor_def, "tasks"): - try: - # Check if validation is explicitly listed (not just wildcard) - if AppConstants.TASK_VALIDATION in executor_def.tasks: - has_explicit_validator = True - break - except (TypeError, AttributeError): - continue - if has_explicit_validator: - break - - if not has_explicit_validator: - # No explicit validator found - add NPValidator for cross-site evaluation - validator = NPValidator() - self._job.to_clients(validator, tasks=[AppConstants.TASK_VALIDATION]) + self._job.to_clients(NPValidator(), tasks=[AppConstants.TASK_VALIDATION]) diff --git a/nvflare/recipe/utils.py b/nvflare/recipe/utils.py index 341cb9b154..501533f7b7 100644 --- a/nvflare/recipe/utils.py +++ b/nvflare/recipe/utils.py @@ -514,9 +514,9 @@ def add_cross_site_evaluation( **For standalone CSE without training**, use `NumpyCrossSiteEvalRecipe` instead. - **Note**: This utility is designed for adding CSE to training recipes. If you call it on - a CSE-only recipe (e.g., `NumpyCrossSiteEvalRecipe`), it will detect this and skip - adding duplicate validators automatically. + **Note**: This utility is designed for adding CSE to training recipes. Standalone CSE + recipes such as `NumpyCrossSiteEvalRecipe` already configure their CSE workflow; + calling this utility on them raises `RuntimeError` through the idempotency check. **WARNING**: Do not call this function multiple times on the same recipe instance. This function is idempotent and will raise a RuntimeError if called more than once @@ -602,9 +602,8 @@ def add_cross_site_evaluation( Note: - Currently supports PyTorch, NumPy, and TensorFlow frameworks. - **NumPy recipes using `NumpyFedAvgRecipe`**: Validators (NPValidator) are automatically - added to clients to handle validation tasks. The function intelligently detects if validators - are already configured by checking for executors handling TASK_VALIDATION, avoiding duplicates - for CSE-only recipes (like `NumpyCrossSiteEvalRecipe`). + added to clients to handle validation tasks. The idempotency check prevents duplicate + CSE augmentation and validator registration. - **Unified `FedAvgRecipe` with `framework=FrameworkType.NUMPY`**: Uses the same Client API validation pattern as PyTorch and TensorFlow. Your client script should handle `flare.is_evaluate()` and return metrics for validation tasks. @@ -713,64 +712,6 @@ def add_cross_site_evaluation( recipe._cse_added = True -def _has_task_executor(job, task_name: str) -> bool: - """Check if any executor is already configured for the specified task. - - This function inspects the job's internal structure to determine if a validator - or executor is already handling the specified task. It uses defensive programming - to handle potential variations in the internal API structure. - - IMPORTANT: This function accesses the private attribute job._deploy_map because: - 1. No public API exists in FedJob to query configured executors - 2. This check is necessary to avoid adding duplicate validators for CSE - 3. Without this, we'd rely on fragile string matching on recipe class names - - The implementation uses defensive programming (hasattr checks, try-except) to - minimize fragility. If FedJob's internal structure changes, this function will - gracefully return False rather than crashing. - - Future improvement: FedJob could provide a public method like get_executors(target) - to make this check safer and more maintainable. - - Args: - job: FedJob instance to check - task_name: Task name to check for (e.g., AppConstants.TASK_VALIDATION) - - Returns: - True if an executor is already configured for this task, False otherwise - """ - # Access _deploy_map (private attribute) - see docstring for justification - # Defensive check: ensure _deploy_map exists before accessing - if not hasattr(job, "_deploy_map"): - return False - - for target, app in job._deploy_map.items(): - # Skip server apps, only check client apps - if target == "server": - continue - - # Get the client app configuration - if hasattr(app, "app_config"): - app_config = app.app_config - # Check if it's a ClientAppConfig with executors - if hasattr(app_config, "executors"): - for executor_def in app_config.executors: - # Defensive check: ensure executor_def has tasks attribute - if not hasattr(executor_def, "tasks"): - continue - - try: - # Check if this executor handles the task - # Wildcard executors (["*"]) can handle any task - if "*" in executor_def.tasks or task_name in executor_def.tasks: - return True - except (TypeError, AttributeError): - # Handle case where tasks is not iterable or comparable - # This could happen if tasks has an unexpected type - continue - return False - - def collect_non_local_scripts(job: FedJob) -> List[str]: """Collect scripts that don't exist locally. From 38269b2c1bbc8e8d22a4bafe662221e602b87157 Mon Sep 17 00:00:00 2001 From: Kevin Date: Wed, 15 Jul 2026 17:45:24 -0400 Subject: [PATCH 4/4] Preserve per-site apps when adding NumPy CSE --- nvflare/app_common/np/recipes/fedavg.py | 2 +- tests/unit_test/recipe/fedavg_recipe_test.py | 32 ++++++++++++++++++++ 2 files changed, 33 insertions(+), 1 deletion(-) diff --git a/nvflare/app_common/np/recipes/fedavg.py b/nvflare/app_common/np/recipes/fedavg.py index 3380f35f77..32789fe5b8 100644 --- a/nvflare/app_common/np/recipes/fedavg.py +++ b/nvflare/app_common/np/recipes/fedavg.py @@ -206,4 +206,4 @@ def add_cse_validator_if_needed(self): from nvflare.app_common.app_constant import AppConstants from nvflare.app_common.np.np_validator import NPValidator - self._job.to_clients(NPValidator(), tasks=[AppConstants.TASK_VALIDATION]) + self._add_to_client_apps(NPValidator(), tasks=[AppConstants.TASK_VALIDATION]) diff --git a/tests/unit_test/recipe/fedavg_recipe_test.py b/tests/unit_test/recipe/fedavg_recipe_test.py index 19c9af57cb..268ce617e4 100644 --- a/tests/unit_test/recipe/fedavg_recipe_test.py +++ b/tests/unit_test/recipe/fedavg_recipe_test.py @@ -534,6 +534,38 @@ def test_numpy_recipe_with_per_site_config(self, mock_file_system): assert recipe.per_site_config == per_site_config + def test_numpy_cse_export_preserves_per_site_training_apps(self, tmp_path): + """Adding CSE must retain each per-site training executor alongside NPValidator.""" + from nvflare.recipe.utils import add_cross_site_evaluation + + train_script = tmp_path / "client.py" + train_script.write_text("# test client script\n") + sites = ["site-1", "site-2"] + recipe = NumpyFedAvgRecipe( + name="test_numpy_per_site_cse", + model=[1.0, 2.0], + min_clients=2, + train_script=str(train_script), + per_site_config={site: {"train_args": f"--site {site}"} for site in sites}, + ) + + add_cross_site_evaluation(recipe) + export_dir = tmp_path / "export" + recipe.export(job_dir=str(export_dir)) + + job_dir = export_dir / recipe.name + assert not (job_dir / "app").exists() + for site in sites: + with open(job_dir / f"app_{site}" / "config" / "config_fed_client.json") as f: + client_config = json.load(f) + + executors = client_config["executors"] + assert any("*" in executor["tasks"] for executor in executors) + assert any( + "validate" in executor["tasks"] and executor["executor"]["path"].endswith(".NPValidator") + for executor in executors + ) + def test_numpy_recipe_with_none_model_raises_error(self, mock_file_system): """Test NumpyFedAvgRecipe with no model raises error.""" with pytest.raises(ValueError, match="Must provide either model"):