e2e-test for %load_node magic

features/load_node.feature

@@ -0,0 +1,9 @@
+Feature: load_node in new project
+
+  Background:
+    Given I have prepared a config file
+    And I have run a non-interactive kedro new with starter "default"
+
+  Scenario: Execute ipython load_node magic
+    When I execute the load_node magic command
+    Then the logs should show that load_node executed successfully

features/steps/cli_steps.py

@@ -302,6 +302,7 @@ def create_project_with_starter(context, starter):
         env=context.env,
         cwd=context.temp_dir,
     )
+
     assert res.returncode == OK_EXIT_CODE, res
 
 
@@ -561,6 +562,17 @@ def check_correct_nodes_run(context, node):
     )
 
 
+@then("the logs should show that load_node executed successfully")
+def check_load_node_run(context):
+    expected_log_line = "load_node executed successfully"
+    stdout = context.result.stdout
+    clean_logs = util.clean_up_log(stdout)
+    assert expected_log_line in clean_logs, (
+        "Expected the following message segment to be printed on stdout: "
+        f"{expected_log_line},\nbut got {stdout}"
+    )
+
+
 @then("I should get a successful exit code")
 def check_status_code(context):
     if context.result.returncode != OK_EXIT_CODE:
@@ -724,6 +736,16 @@ def add_micropkg_to_pyproject_toml(context: behave.runner.Context):
         file.write(project_toml_str)
 
 
+@given("I have executed the load_node magic command")
+@when("I execute the load_node magic command")
+def exec_magic_command(context):
+    """Execute Kedro target."""
+    cmd = [context.python, "ipython_script.py"]
+    context.result = run(
+        cmd, env=context.env, cwd=str(context.root_project_dir), print_output=True
+    )
+
+
 @given('I have changed the current working directory to "{dir}"')
 def change_dir(context, dir):
     """Execute Kedro target."""

features/steps/test_starter/{{ cookiecutter.repo_name }}/ipython_script.py

@@ -0,0 +1,9 @@
+if __name__ == "__main__":
+    from IPython.testing.globalipapp import get_ipython
+    ip = get_ipython()
+    ip.run_line_magic("load_ext", "kedro.ipython")
+    # Assume cwd is project root
+    ip.run_line_magic("reload_kedro", "")
+    ip.run_line_magic("load_node", "split_data_node")
+    # ip.rl_next_input is what you see in the terminal input
+    ip.run_cell(ip.rl_next_input)

features/steps/test_starter/{{ cookiecutter.repo_name }}/src/{{ cookiecutter.python_package }}/pipelines/data_engineering/nodes.py

@@ -17,6 +17,7 @@ def split_data(data: pd.DataFrame, example_test_data_ratio: float) -> dict[str,
     The data and the parameters will be loaded and provided to your function
     automatically when the pipeline is executed and it is time to run this node.
     """
+    print("load_node executed successfully")
     data.columns = [
         "sepal_length",
         "sepal_width",

features/steps/test_starter/{{ cookiecutter.repo_name }}/src/{{ cookiecutter.python_package }}/pipelines/data_engineering/pipeline.py

@@ -21,6 +21,7 @@ def create_pipeline(**kwargs):
                     test_x="example_test_x",
                     test_y="example_test_y",
                 ),
+                name="split_data_node"
             )
         ]
     )

kedro/ipython/__init__.py

@@ -163,7 +163,11 @@ def _resolve_project_path(
     if path:
         project_path = Path(path).expanduser().resolve()
     else:
-        if local_namespace and "context" in local_namespace:
+        if (
+            local_namespace
+            and local_namespace.get("context")
+            and hasattr(local_namespace["context"], "project_path")
+        ):
             project_path = local_namespace["context"].project_path
         else:
             project_path = _find_kedro_project(Path.cwd())
@@ -177,7 +181,8 @@ def _resolve_project_path(
     if (
         project_path
         and local_namespace
-        and "context" in local_namespace
+        and local_namespace.get("context")
+        and hasattr(local_namespace["context"], "project_path")  # Avoid name collision
         and project_path != local_namespace["context"].project_path
     ):
         logger.info("Updating path to Kedro project: %s...", project_path)

tests/framework/cli/test_registry.py

@@ -10,13 +10,13 @@ def yaml_dump_mock(mocker):
 @pytest.fixture
 def pipelines_dict():
     pipelines = {
-        "data_engineering": ["split_data (split_data)"],
+        "data_engineering": ["split_data_node (split_data)"],
         "data_science": [
             "train_model (train_model)",
             "predict (predict)",
             "report_accuracy (report_accuracy)",
         ],
-        "data_processing": ["data_processing.split_data (split_data)"],
+        "data_processing": ["data_processing.split_data_node (split_data)"],
     }
     pipelines["__default__"] = pipelines["data_engineering"] + pipelines["data_science"]
     return pipelines