logger = get_logger(__name__)


class EnvHook:
    @hook_impl
    def after_catalog_created(self, *args, **kwargs) -> None:
        """ User credentials.yml file over .env. If not found, fall back to .env """
        creds = kwargs.get("conf_creds")

        if creds is None:
            load_dotenv()
        else:
            env = creds["envs"]
            for k, v in env.items():
                k = k.upper()   # Convert envs to upper case if they are not already.

                if k in os.environ:
                    logger.warn(f"Environment variable {k} is already set to {v}")
                    continue  # In case env is already set don't touch it
                if v is not None:
                    os.environ[k] = v

from projx.hooks import EnvHook
# Hooks are executed in a Last-In-First-Out (LIFO) order.
HOOKS = (EnvHook(),)

conf_cred

conftest.py

@fixture(scope='session')
def kedro_context(config_loader):
    return KedroContext(
        package_name="projx",
        project_path=PROJECT_PATH,
        config_loader=config_loader,
        hook_manager=_create_hook_manager(),
        env="test"
    )

test_run.py

def test_catalog_hook(kedro_context):
        # Invoke catalog loading
        catalog = kedro_context.catalog.list()
        
        # By this time we should be running the catalog hook
        print("stop")

hooks.py

class EnvHook:
    @hook_impl
    def after_catalog_created(self, *args, **kwargs) -> None:
        """ User credentials.yml file over .env. If not found, fall back to .env """
        print("Hello")

def test_project_catalog(kedro_context, kedro_session):
        # Invoke catalog parsing to see if there is a problem in the catalog
        print(kedro_context.catalog.list())
        kedro_session.load_context()        # -> This invokes hooks

hook_manager = _create_hook_manager()
        _register_hooks(hook_manager, settings.HOOKS)
        _register_hooks_entry_points(hook_manager, settings.DISABLE_HOOKS_FOR_PLUGINS)