Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Rewriting the IPython test #2046

Merged
merged 8 commits into from
Dec 7, 2022
Merged

Rewriting the IPython test #2046

Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
4aec5dc
Add a new test for loading entry points
noklam Nov 21, 2022
3ae9d78
Remove deprecated test and create new tests for lazy pipelines
noklam Nov 25, 2022
d5945ba
Merge branch 'main' into fix/ipython_test
noklam Nov 29, 2022
b2e8760
Mock the global pipelines object instead of class
noklam Nov 30, 2022
b40b00c
Merge branch 'main' into fix/ipython_test
noklam Nov 30, 2022
e413c06
Organise tests in class
noklam Nov 30, 2022
4d362ce
Please the linter - attempt 1
noklam Nov 30, 2022
8bb97c2
Merge branch 'main' into fix/ipython_test
AhdraMeraliQB Dec 7, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
207 changes: 100 additions & 107 deletions tests/ipython/test_ipython.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,166 +5,159 @@
from IPython.core.error import UsageError
from IPython.testing.globalipapp import get_ipython

from kedro.framework.project import pipelines
from kedro.framework.startup import ProjectMetadata
from kedro.ipython import _resolve_project_path, load_ipython_extension, reload_kedro
from kedro.pipeline import Pipeline

PACKAGE_NAME = "fake_package_name"
PROJECT_NAME = "fake_project_name"
PROJECT_VERSION = "0.1"


@pytest.fixture(autouse=True)
def cleanup_pipeline():
yield
from kedro.framework.project import pipelines
from kedro.framework.project import pipelines # pylint: disable=reimported

pipelines.configure()


@pytest.fixture(scope="module") # get_ipython() twice will result in None
@pytest.fixture(scope="module", autouse=True) # get_ipython() twice will result in None
def ipython():
ipython = get_ipython()
load_ipython_extension(ipython)
return ipython


PACKAGE_NAME = "fake_package_name"
PROJECT_NAME = "fake_project_name"
PROJECT_VERSION = "0.1"
@pytest.fixture(autouse=True)
def fake_metadata(tmp_path):
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice cleanup here in making this a fixture 🧹

metadata = ProjectMetadata(
source_dir=tmp_path / "src", # default
config_file=tmp_path / "pyproject.toml",
package_name=PACKAGE_NAME,
project_name=PROJECT_NAME,
project_version=PROJECT_VERSION,
project_path=tmp_path,
)
return metadata


@pytest.fixture(autouse=True)
def mock_kedro_project(mocker, fake_metadata):
mocker.patch("kedro.ipython.bootstrap_project", return_value=fake_metadata)
mocker.patch("kedro.ipython.configure_project")
mocker.patch("kedro.ipython.KedroSession.create")


@pytest.mark.skip()
class TestLoadKedroObjects:
def test_load_kedro_objects(self, tmp_path, mocker, caplog):
kedro_path = tmp_path / "here"

fake_metadata = ProjectMetadata(
source_dir=kedro_path / "src", # default
config_file=kedro_path / "pyproject.toml",
package_name=PACKAGE_NAME,
project_name=PROJECT_NAME,
project_version=PROJECT_VERSION,
project_path=kedro_path,
)
def test_ipython_load_entry_points(
self,
mocker,
fake_metadata,
caplog,
):
mock_line_magic = mocker.MagicMock()
mock_line_magic_name = "abc"
mock_line_magic.__name__ = mock_line_magic_name
mock_line_magic.__qualname__ = mock_line_magic_name # Required by IPython

mocker.patch("kedro.ipython.load_entry_points", return_value=[mock_line_magic])
expected_message = f"Registered line magic '{mock_line_magic_name}'"

reload_kedro(fake_metadata.project_path)

log_messages = [record.getMessage() for record in caplog.records]
assert expected_message in log_messages

def test_ipython_lazy_load_pipeline(
self,
mocker,
):
pipelines.configure("dummy_pipeline") # Setup the pipelines

my_pipelines = {"ds": Pipeline([])}

def my_register_pipeline():
return my_pipelines

mocker.patch(
"kedro.framework.project._ProjectPipelines._get_pipelines_registry_callable",
mocker.patch.object(
pipelines,
"_get_pipelines_registry_callable",
return_value=my_register_pipeline,
)
mocker.patch("kedro.framework.startup.configure_project")
mocker.patch("kedro.ipython.bootstrap_project", return_value=fake_metadata)
mock_line_magic = mocker.Mock()
mock_line_magic.__name__ = "abc"
mocker.patch("kedro.ipython.load_entry_points", return_value=[mock_line_magic])
mock_register_line_magic = mocker.patch("kedro.ipython.register_line_magic")
mock_session_create = mocker.patch("kedro.ipython.KedroSession.create")
mock_ipython = mocker.patch("kedro.ipython.get_ipython")

reload_kedro(kedro_path)

mock_session_create.assert_called_once_with(
PACKAGE_NAME, kedro_path, env=None, extra_params=None
)
mock_ipython().push.assert_called_once_with(
variables={
"context": mock_session_create().load_context(),
"catalog": mock_session_create().load_context().catalog,
"session": mock_session_create(),
"pipelines": my_pipelines,
}
)
mock_register_line_magic.assert_called_once()
reload_kedro()

expected_path = kedro_path.expanduser().resolve()
expected_message = f"Updated path to Kedro project: {expected_path}"
assert pipelines._content == {} # Check if it is lazy loaded
pipelines._load_data() # Trigger data load
assert pipelines._content == my_pipelines

log_messages = [record.getMessage() for record in caplog.records]
assert expected_message in log_messages

def test_load_kedro_objects_extra_args(self, tmp_path, mocker):
fake_metadata = ProjectMetadata(
source_dir=tmp_path / "src", # default
config_file=tmp_path / "pyproject.toml",
package_name=PACKAGE_NAME,
project_name=PROJECT_NAME,
project_version=PROJECT_VERSION,
project_path=tmp_path,
)
def test_ipython_load_objects(
self,
mocker,
ipython,
):
mock_session_create = mocker.patch("kedro.ipython.KedroSession.create")
pipelines.configure("dummy_pipeline") # Setup the pipelines

my_pipelines = {"ds": Pipeline([])}

def my_register_pipeline():
return my_pipelines

mocker.patch(
"kedro.framework.project._ProjectPipelines._get_pipelines_registry_callable",
mocker.patch.object(
pipelines,
"_get_pipelines_registry_callable",
return_value=my_register_pipeline,
)
mocker.patch("kedro.ipython.configure_project")
mocker.patch("kedro.ipython.bootstrap_project", return_value=fake_metadata)
mock_line_magic = mocker.Mock()
mock_line_magic.__name__ = "abc"
mocker.patch("kedro.ipython.load_entry_points", return_value=[mock_line_magic])
mock_register_line_magic = mocker.patch("kedro.ipython.register_line_magic")
mock_session_create = mocker.patch("kedro.ipython.KedroSession.create")
mock_ipython = mocker.patch("kedro.ipython.get_ipython")
ipython_spy = mocker.spy(ipython, "push")

reload_kedro(tmp_path, env="env1", extra_params={"key": "val"})
reload_kedro()

mock_session_create.assert_called_once_with(
PACKAGE_NAME, tmp_path, env="env1", extra_params={"key": "val"}
)
mock_ipython().push.assert_called_once_with(
variables={
"context": mock_session_create().load_context(),
"catalog": mock_session_create().load_context().catalog,
"session": mock_session_create(),
"pipelines": {},
}
)
assert mock_register_line_magic.call_count == 1

def test_load_kedro_objects_no_path(
self, tmp_path, caplog, mocker, ipython
): # pylint: disable=unused-argument
fake_metadata = ProjectMetadata(
source_dir=tmp_path / "src", # default
config_file=tmp_path / "pyproject.toml",
package_name=PACKAGE_NAME,
project_name=PROJECT_NAME,
project_version=PROJECT_VERSION,
project_path=tmp_path,
PACKAGE_NAME, None, env=None, extra_params=None
)
_, kwargs = ipython_spy.call_args_list[0]
variables = kwargs["variables"]

assert variables["context"] == mock_session_create().load_context()
assert variables["catalog"] == mock_session_create().load_context().catalog
assert variables["session"] == mock_session_create()
assert variables["pipelines"] == my_pipelines

def test_ipython_load_objects_with_args(self, mocker, fake_metadata, ipython):
mock_session_create = mocker.patch("kedro.ipython.KedroSession.create")
pipelines.configure("dummy_pipeline") # Setup the pipelines

my_pipelines = {"ds": Pipeline([])}

def my_register_pipeline():
return my_pipelines

mocker.patch(
"kedro.framework.project._ProjectPipelines._get_pipelines_registry_callable",
mocker.patch.object(
pipelines,
"_get_pipelines_registry_callable",
return_value=my_register_pipeline,
)
ipython_spy = mocker.spy(ipython, "push")
dummy_env = "env"
dummy_dict = {"key": "value"}

mocker.patch("kedro.ipython.configure_project")
mocker.patch("kedro.ipython.bootstrap_project", return_value=fake_metadata)
mock_line_magic = mocker.Mock()
mock_line_magic.__name__ = "abc"
mocker.patch("kedro.ipython.load_entry_points", return_value=[mock_line_magic])
mocker.patch("kedro.ipython.register_line_magic")
mocker.patch("kedro.ipython.KedroSession.create")
mocker.patch("kedro.ipython.get_ipython")
mocker.patch("kedro.ipython._find_kedro_project", return_value=tmp_path)
reload_kedro(fake_metadata.project_path, "env", {"key": "value"})

reload_kedro()

expected_message = (
f"Resolved project path as: {tmp_path}.\nTo set a different path, run "
"'%reload_kedro <project_root>'"
mock_session_create.assert_called_once_with(
PACKAGE_NAME,
fake_metadata.project_path,
env=dummy_env,
extra_params=dummy_dict,
)
log_messages = [record.getMessage() for record in caplog.records]
assert expected_message in log_messages
_, kwargs = ipython_spy.call_args_list[0]
variables = kwargs["variables"]

assert variables["context"] == mock_session_create().load_context()
assert variables["catalog"] == mock_session_create().load_context().catalog
assert variables["session"] == mock_session_create()
assert variables["pipelines"] == my_pipelines


class TestLoadIPythonExtension:
Expand Down