tests/framework/hooks/test_context_hooks.py
def _create_context_with_hooks(tmp_path, mocker, logging_hooks):
    """Create a context with some Hooks registered.
    We do this in a function to support both calling it directly as well as as part of a fixture.
    """

    class DummyContextWithHooks(KedroContext):
        project_name = "test hooks"
        package_name = "test_hooks"
        project_version = __version__
        hooks = (logging_hooks,)

        def _get_run_id(self, *args, **kwargs) -> Union[None, str]:
            return "mocked context with hooks run id"

        def _get_pipelines(self) -> Dict[str, Pipeline]:
            pipeline = Pipeline(
                [
                    node(identity, "cars", "planes", name="node1"),
                    node(identity, "boats", "ships", name="node2"),
                ],
                tags="pipeline",
            )
            return {"__default__": pipeline}

    mocker.patch("logging.config.dictConfig")
    return DummyContextWithHooks(tmp_path, env="local")
Create a context with some Hooks registered.
    We do this in a function to support both calling it directly as well as as part of a fixture.
    
Create a context with some Hooks registered.
We do this in a function to support both calling it directly as well as as part of a fixture.
Create a context with some Hooks registered . We do this in a function to support both calling it directly as well as as part of a fixture .
def _create_context_with_hooks(tmp_path, mocker, logging_hooks):
    class DummyContextWithHooks(KedroContext):
        project_name = "test hooks"
        package_name = "test_hooks"
        project_version = __version__
        hooks = (logging_hooks,)
        def _get_run_id(self, *args, **kwargs) -> Union[None, str]:
            return "mocked context with hooks run id"
        def _get_pipelines(self) -> Dict[str, Pipeline]:
            pipeline = Pipeline(
                [
                    node(identity, "cars", "planes", name="node1"),
                    node(identity, "boats", "ships", name="node2"),
                ],
                tags="pipeline",
            )
            return {"__default__": pipeline}
    mocker.patch("logging.config.dictConfig")
    return DummyContextWithHooks(tmp_path, env="local")
def _create_context_with_hooks ( tmp_path , mocker , logging_hooks ) : class DummyContextWithHooks ( KedroContext ) : project_name = "test hooks" package_name = "test_hooks" project_version = __version__ hooks = ( logging_hooks , ) def _get_run_id ( self , * args , ** kwargs ) -> Union [ None , str ] : return "mocked context with hooks run id" def _get_pipelines ( self ) -> Dict [ str , Pipeline ] : pipeline = Pipeline ( [ node ( identity , "cars" , "planes" , name = "node1" ) , node ( identity , "boats" , "ships" , name = "node2" ) , ] , tags = "pipeline" , ) return { "__default__" : pipeline } mocker . patch ( "logging.config.dictConfig" ) return DummyContextWithHooks ( tmp_path , env = "local" )
Create a context with some Hooks registered.
Create a context with some Hooks registered .
"""Create a context with some Hooks registered.
    We do this in a function to support both calling it directly as well as as part of a fixture.
    """
tmp_path
mocker
logging_hooks