diff --git a/docs/lifecycle.md b/docs/lifecycle.md new file mode 100644 index 00000000..2b60743a --- /dev/null +++ b/docs/lifecycle.md @@ -0,0 +1,200 @@ +# Kernel Lifecycle State Management + +The `jupyter_client.lifecycle` module provides a state machine implementation for tracking kernel lifecycle states across different kernel management operations. This feature is designed to help developers build more robust kernel management systems with better observability and error handling. + +## Quick Start + +To add lifecycle state tracking to a kernel manager, simply inherit from `KernelManagerStateMixin`: + +```python +from jupyter_client.lifecycle import KernelManagerStateMixin +from jupyter_client.manager import KernelManager + + +class StatefulKernelManager(KernelManagerStateMixin, KernelManager): + pass + + +# The mixin automatically tracks state during kernel operations +manager = StatefulKernelManager() +print(manager.lifecycle_state) # "unknown" + +await manager.start_kernel() +print(manager.lifecycle_state) # "started" +print(manager.is_started) # True +``` + +## Core Components + +### LifecycleState Enum + +The `LifecycleState` enum defines the possible states a kernel can be in: + +- **UNKNOWN**: Initial state or state after errors +- **STARTING**: Kernel is in the process of starting +- **STARTED**: Kernel has been started successfully +- **RESTARTING**: Kernel is in the process of restarting +- **RESTARTED**: Kernel has been restarted successfully +- **TERMINATING**: Kernel is in the process of shutting down +- **DEAD**: Kernel has been shut down + +The enum inherits from `str` for convenient usage: + +```python +from jupyter_client.lifecycle import LifecycleState + +# Direct string comparison works +assert LifecycleState.UNKNOWN == "unknown" +assert "started" == LifecycleState.STARTED + +# No need for .value attribute +state = LifecycleState.STARTED +print(state) # "started" +``` + +### KernelManagerStateMixin + +The `KernelManagerStateMixin` class provides automatic state tracking for kernel managers. It uses the `__init_subclass__` hook to automatically wrap kernel management methods with state transition decorators. + +Key features: + +- Automatic method wrapping for `start_kernel`, `restart_kernel`, and `shutdown_kernel` +- Support for both synchronous and asynchronous methods +- Automatic error handling (failed operations reset state to UNKNOWN) +- Configurable `lifecycle_state` trait +- Convenient state checking properties + +```python +class MyKernelManager(KernelManagerStateMixin, SomeBaseManager): + def start_kernel(self): + # Your start logic here + pass + + +manager = MyKernelManager() + +# State checking properties +print(manager.is_unknown) # True +print(manager.is_started) # False + +manager.start_kernel() +print(manager.is_started) # True +``` + +### state_transition Decorator + +For custom state management scenarios, you can use the `state_transition` decorator directly: + +```python +from jupyter_client.lifecycle import state_transition, LifecycleState + + +class CustomManager: + lifecycle_state = LifecycleState.UNKNOWN + + @state_transition(LifecycleState.STARTING, LifecycleState.STARTED) + def custom_start(self): + # Custom start logic + return "started" + + @state_transition(LifecycleState.TERMINATING, LifecycleState.DEAD) + async def custom_shutdown(self): + # Custom async shutdown logic + return "shutdown complete" +``` + +## State Transitions + +The state machine handles these automatic transitions: + +1. **start_kernel**: `unknown` → `starting` → `started` (or `unknown` on failure) +1. **restart_kernel**: `*` → `restarting` → `restarted` (or `unknown` on failure) +1. **shutdown_kernel**: `*` → `terminating` → `dead` (or `unknown` on failure) + +```mermaid +graph LR + UNKNOWN --> STARTING + STARTING --> STARTED + STARTING --> UNKNOWN + STARTED --> RESTARTING + RESTARTING --> RESTARTED + RESTARTING --> UNKNOWN + STARTED --> TERMINATING + RESTARTED --> TERMINATING + TERMINATING --> DEAD + TERMINATING --> UNKNOWN +``` + +## Advanced Usage + +### Custom Kernel Managers + +You can use the mixin with any kernel manager that follows standard method naming conventions: + +```python +class CustomKernelManager(KernelManagerStateMixin): + def __init__(self): + super().__init__() + self.kernel_id = "custom-kernel" + self.log = logging.getLogger(__name__) + + def start_kernel(self, **kwargs): + # Custom start logic + return {"kernel_id": self.kernel_id} + + async def restart_kernel(self, **kwargs): + # Custom async restart logic + await asyncio.sleep(1) # Simulate restart time + return {"kernel_id": self.kernel_id} + + def shutdown_kernel(self, immediate=False, **kwargs): + # Custom shutdown logic + return True +``` + +### Manual State Management + +While automatic state management is the primary use case, you can also manually control states: + +```python +manager = MyKernelManager() + +# Manual state setting +manager.set_lifecycle_state(LifecycleState.STARTED) +assert manager.is_started + +# Direct assignment +manager.lifecycle_state = LifecycleState.DEAD +assert manager.is_dead +``` + +### Error Handling and Recovery + +Failed operations automatically reset the state to `UNKNOWN`: + +```python +class FailingKernelManager(KernelManagerStateMixin): + def start_kernel(self): + raise RuntimeError("Start failed") + + +manager = FailingKernelManager() +try: + manager.start_kernel() +except RuntimeError: + pass + +print(manager.lifecycle_state) # "unknown" + +# You can implement retry logic based on state +if manager.is_unknown: + # Retry or handle the error + pass +``` + +## Best Practices + +1. **Use state checking properties**: Prefer `manager.is_started` over `manager.lifecycle_state == "started"` +1. **Handle UNKNOWN states**: Always have logic to handle when state is UNKNOWN after failures +1. **Test state transitions**: Include state assertions in your kernel manager tests +1. **Don't override mixin methods**: Avoid overriding the state checking properties or internal methods diff --git a/jupyter_client/__init__.py b/jupyter_client/__init__.py index 3b30a6f4..6d91baf2 100644 --- a/jupyter_client/__init__.py +++ b/jupyter_client/__init__.py @@ -5,6 +5,7 @@ from .client import KernelClient from .connect import * # noqa from .launcher import * # noqa +from .lifecycle import KernelManagerStateMixin, LifecycleState, state_transition from .manager import AsyncKernelManager, KernelManager, run_kernel from .multikernelmanager import AsyncMultiKernelManager, MultiKernelManager from .provisioning import KernelProvisionerBase, LocalProvisioner diff --git a/jupyter_client/lifecycle.py b/jupyter_client/lifecycle.py new file mode 100644 index 00000000..febe2825 --- /dev/null +++ b/jupyter_client/lifecycle.py @@ -0,0 +1,354 @@ +"""Kernel lifecycle state management for Jupyter kernel managers. + +This module provides a state machine mixin for tracking kernel lifecycle states +across different kernel management operations. It's designed to work with any +kernel manager that follows the KernelManagerABC interface. + +Basic Usage +----------- + +To add lifecycle state tracking to a kernel manager:: + + from jupyter_client.lifecycle import KernelManagerStateMixin + from jupyter_client.manager import KernelManager + + class StatefulKernelManager(KernelManagerStateMixin, KernelManager): + pass + + # The mixin automatically tracks state during kernel operations + manager = StatefulKernelManager() + print(manager.lifecycle_state) # "unknown" + + await manager.start_kernel() + print(manager.lifecycle_state) # "started" + print(manager.is_started) # True + +State Transitions +----------------- + +The state machine handles these automatic transitions: + +- **start_kernel**: unknown → starting → started (or unknown on failure) +- **restart_kernel**: * → restarting → restarted (or unknown on failure) +- **shutdown_kernel**: * → terminating → dead (or unknown on failure) + +States can also be checked using convenient properties:: + + manager.is_unknown # True if state is "unknown" + manager.is_starting # True if state is "starting" + manager.is_started # True if state is "started" + manager.is_restarting # True if state is "restarting" + manager.is_restarted # True if state is "restarted" + manager.is_terminating # True if state is "terminating" + manager.is_dead # True if state is "dead" + +Advanced Usage +-------------- + +The state machine can be used with custom kernel managers and supports +both synchronous and asynchronous operations:: + + class CustomKernelManager(KernelManagerStateMixin): + def start_kernel(self): + # Custom start logic + return "started" + + async def restart_kernel(self): + # Custom async restart logic + return "restarted" + +Manual state management is also supported:: + + manager.set_lifecycle_state(LifecycleState.STARTED) + manager.lifecycle_state = LifecycleState.DEAD + +Error Handling +-------------- + +When kernel operations fail, the state is automatically reset to "unknown":: + + class FailingKernelManager(KernelManagerStateMixin): + def start_kernel(self): + raise RuntimeError("Start failed") + + manager = FailingKernelManager() + try: + manager.start_kernel() + except RuntimeError: + pass + + print(manager.lifecycle_state) # "unknown" +""" + +import asyncio +import enum +from functools import wraps +from typing import Callable + +from traitlets import HasTraits, Unicode, observe + + +class LifecycleState(str, enum.Enum): + """Enumeration of kernel lifecycle states. + + This enum inherits from str to allow direct string comparisons without + needing to access the .value attribute:: + + assert LifecycleState.UNKNOWN == "unknown" + assert "started" == LifecycleState.STARTED + + States: + UNKNOWN: Initial state or state after errors + STARTING: Kernel is in the process of starting + STARTED: Kernel has been started successfully + RESTARTING: Kernel is in the process of restarting + RESTARTED: Kernel has been restarted successfully + TERMINATING: Kernel is in the process of shutting down + DEAD: Kernel has been shut down + """ + + UNKNOWN = "unknown" + STARTING = "starting" + STARTED = "started" + RESTARTING = "restarting" + RESTARTED = "restarted" + TERMINATING = "terminating" + DEAD = "dead" + + +def state_transition(start_state: LifecycleState, end_state: LifecycleState): + """Decorator to handle state transitions for kernel manager methods. + + This decorator automatically manages state transitions around method calls, + setting the start state before the method executes and the end state after + successful completion. If the method raises an exception, the state is set + to UNKNOWN. + + Parameters + ---------- + start_state : LifecycleState + The state to set before calling the method + end_state : LifecycleState + The state to set after successful method completion + + Returns + ------- + Callable + The decorated method with automatic state management + + Examples + -------- + >>> class Manager: + ... lifecycle_state = LifecycleState.UNKNOWN + ... + ... @state_transition(LifecycleState.STARTING, LifecycleState.STARTED) + ... def start_kernel(self): + ... return "kernel started" + ... + >>> manager = Manager() + >>> result = manager.start_kernel() + >>> assert manager.lifecycle_state == LifecycleState.STARTED + """ + + def decorator(method: Callable) -> Callable: + @wraps(method) + async def async_wrapper(self, *args, **kwargs): + # Set the starting state + self.lifecycle_state = start_state + try: + # Call the original method + result = await method(self, *args, **kwargs) + # Set the end state on success + self.lifecycle_state = end_state + return result + except Exception: + # Set to unknown state on failure + self.lifecycle_state = LifecycleState.UNKNOWN + raise + + @wraps(method) + def sync_wrapper(self, *args, **kwargs): + # Set the starting state + self.lifecycle_state = start_state + try: + # Call the original method + result = method(self, *args, **kwargs) + # Set the end state on success + self.lifecycle_state = end_state + return result + except Exception: + # Set to unknown state on failure + self.lifecycle_state = LifecycleState.UNKNOWN + raise + + # Return the appropriate wrapper based on whether the method is async + if asyncio.iscoroutinefunction(method): + return async_wrapper + else: + return sync_wrapper + + return decorator + + +class KernelManagerStateMixin(HasTraits): + """Mixin class that adds lifecycle state tracking to kernel managers. + + This mixin automatically tracks kernel lifecycle states during standard + kernel management operations. It works with any kernel manager that follows + the KernelManagerABC interface and uses method name conventions for kernel + operations. + + The mixin uses the `__init_subclass__` hook to automatically wrap kernel + management methods (start_kernel, restart_kernel, shutdown_kernel) with + state transition decorators. + + Attributes + ---------- + lifecycle_state : Unicode + The current lifecycle state of the kernel. Configurable trait that + defaults to LifecycleState.UNKNOWN. + + Examples + -------- + Basic usage with inheritance:: + + class MyKernelManager(KernelManagerStateMixin, SomeBaseManager): + def start_kernel(self): + # Your start logic here + pass + + manager = MyKernelManager() + print(manager.is_unknown) # True + + manager.start_kernel() + print(manager.is_started) # True + + Custom state management:: + + manager = MyKernelManager() + manager.set_lifecycle_state(LifecycleState.STARTED) + assert manager.lifecycle_state == "started" + + State checking:: + + if manager.is_started: + manager.restart_kernel() + elif manager.is_dead: + manager.start_kernel() + + Notes + ----- + - State transitions are logged via the manager's logger if available + - Failed operations automatically reset state to UNKNOWN + - The mixin supports both sync and async kernel operations + - Method wrapping only occurs if the methods exist on the class + """ + + lifecycle_state = Unicode( + default_value=LifecycleState.UNKNOWN, help="The current lifecycle state of the kernel" + ).tag(config=True) + + @observe("lifecycle_state") + def _lifecycle_state_changed(self, change): + """Log lifecycle state changes for debugging. + + Parameters + ---------- + change : dict + The change notification dict from traitlets containing + 'old' and 'new' values + """ + old_state = change["old"] + new_state = change["new"] + kernel_id = getattr(self, "kernel_id", "unknown") + if hasattr(self, "log"): + self.log.debug(f"Kernel {kernel_id} state changed: {old_state} -> {new_state}") + + def __init_subclass__(cls, **kwargs): + """Automatically wrap kernel management methods when the class is subclassed. + + This method is called when a class inherits from KernelManagerStateMixin + and automatically applies state transition decorators to standard kernel + management methods if they exist. + + Parameters + ---------- + **kwargs + Additional keyword arguments passed to parent __init_subclass__ + """ + super().__init_subclass__(**kwargs) + + # Wrap start_kernel method if it exists + if hasattr(cls, "start_kernel"): + original_start = cls.start_kernel + cls.start_kernel = state_transition(LifecycleState.STARTING, LifecycleState.STARTED)( + original_start + ) + + # Wrap restart_kernel method if it exists + if hasattr(cls, "restart_kernel"): + original_restart = cls.restart_kernel + cls.restart_kernel = state_transition( + LifecycleState.RESTARTING, LifecycleState.RESTARTED + )(original_restart) + + # Wrap shutdown_kernel method if it exists + if hasattr(cls, "shutdown_kernel"): + original_shutdown = cls.shutdown_kernel + cls.shutdown_kernel = state_transition(LifecycleState.TERMINATING, LifecycleState.DEAD)( + original_shutdown + ) + + # State checking properties + + @property + def is_starting(self) -> bool: + """True if kernel is in starting state.""" + return self.lifecycle_state == LifecycleState.STARTING + + @property + def is_started(self) -> bool: + """True if kernel is in started state.""" + return self.lifecycle_state == LifecycleState.STARTED + + @property + def is_restarting(self) -> bool: + """True if kernel is in restarting state.""" + return self.lifecycle_state == LifecycleState.RESTARTING + + @property + def is_restarted(self) -> bool: + """True if kernel is in restarted state.""" + return self.lifecycle_state == LifecycleState.RESTARTED + + @property + def is_terminating(self) -> bool: + """True if kernel is in terminating state.""" + return self.lifecycle_state == LifecycleState.TERMINATING + + @property + def is_dead(self) -> bool: + """True if kernel is in dead state.""" + return self.lifecycle_state == LifecycleState.DEAD + + @property + def is_unknown(self) -> bool: + """True if kernel is in unknown state.""" + return self.lifecycle_state == LifecycleState.UNKNOWN + + # State management methods + + def set_lifecycle_state(self, state: LifecycleState) -> None: + """Manually set the lifecycle state. + + Parameters + ---------- + state : LifecycleState + The state to set + + Examples + -------- + >>> manager.set_lifecycle_state(LifecycleState.STARTED) + >>> assert manager.is_started + """ + self.lifecycle_state = state diff --git a/tests/test_lifecycle.py b/tests/test_lifecycle.py new file mode 100644 index 00000000..fb20ff43 --- /dev/null +++ b/tests/test_lifecycle.py @@ -0,0 +1,437 @@ +""" +Unit tests for jupyter_client.lifecycle module +""" +from unittest.mock import Mock + +import pytest +from traitlets import HasTraits + +from jupyter_client.lifecycle import KernelManagerStateMixin, LifecycleState, state_transition + + +class TestLifecycleState: + """Test cases for LifecycleState enum""" + + def test_enum_values(self): + """Test that enum values are correct strings""" + assert LifecycleState.UNKNOWN == "unknown" + assert LifecycleState.STARTING == "starting" + assert LifecycleState.STARTED == "started" + assert LifecycleState.RESTARTING == "restarting" + assert LifecycleState.RESTARTED == "restarted" + assert LifecycleState.TERMINATING == "terminating" + assert LifecycleState.DEAD == "dead" + + def test_enum_string_behavior(self): + """Test that enum inherits from str and behaves like strings""" + assert isinstance(LifecycleState.UNKNOWN, str) + assert LifecycleState.UNKNOWN == "unknown" + assert LifecycleState.UNKNOWN == "unknown" + # Note: str() returns the enum name, but direct comparison works + assert LifecycleState.STARTING == "starting" + + def test_enum_comparison(self): + """Test enum value comparisons""" + assert LifecycleState.UNKNOWN != LifecycleState.STARTING + assert LifecycleState.STARTED == LifecycleState.STARTED + assert LifecycleState.DEAD == "dead" + + +class TestStateTransitionDecorator: + """Test cases for state_transition decorator""" + + def test_sync_method_success(self): + """Test state transition decorator with sync method - success case""" + + class TestManager(HasTraits): + lifecycle_state = LifecycleState.UNKNOWN + + @state_transition(LifecycleState.STARTING, LifecycleState.STARTED) + def start_method(self): + return "success" + + manager = TestManager() + result = manager.start_method() + + assert result == "success" + assert manager.lifecycle_state == LifecycleState.STARTED + + def test_sync_method_failure(self): + """Test state transition decorator with sync method - failure case""" + + class TestManager(HasTraits): + lifecycle_state = LifecycleState.UNKNOWN + + @state_transition(LifecycleState.STARTING, LifecycleState.STARTED) + def failing_method(self): + raise ValueError("Test error") + + manager = TestManager() + + with pytest.raises(ValueError): + manager.failing_method() + + assert manager.lifecycle_state == LifecycleState.UNKNOWN + + @pytest.mark.asyncio + async def test_async_method_success(self): + """Test state transition decorator with async method - success case""" + + class TestManager(HasTraits): + lifecycle_state = LifecycleState.UNKNOWN + + @state_transition(LifecycleState.STARTING, LifecycleState.STARTED) + async def async_start_method(self): + return "async_success" + + manager = TestManager() + result = await manager.async_start_method() + + assert result == "async_success" + assert manager.lifecycle_state == LifecycleState.STARTED + + @pytest.mark.asyncio + async def test_async_method_failure(self): + """Test state transition decorator with async method - failure case""" + + class TestManager(HasTraits): + lifecycle_state = LifecycleState.UNKNOWN + + @state_transition(LifecycleState.STARTING, LifecycleState.STARTED) + async def async_failing_method(self): + raise RuntimeError("Async test error") + + manager = TestManager() + + with pytest.raises(RuntimeError): + await manager.async_failing_method() + + assert manager.lifecycle_state == LifecycleState.UNKNOWN + + def test_state_transition_sequence(self): + """Test that state transitions happen in correct order""" + states_seen = [] + + class TestManager(HasTraits): + _lifecycle_state = LifecycleState.UNKNOWN + + @property + def lifecycle_state(self): + return self._lifecycle_state + + @lifecycle_state.setter + def lifecycle_state(self, value): + states_seen.append(value) + self._lifecycle_state = value + + @state_transition(LifecycleState.STARTING, LifecycleState.STARTED) + def start_method(self): + states_seen.append("method_called") + return "done" + + manager = TestManager() + manager.start_method() + + # Should see: STARTING (before method), method_called, STARTED (after method) + assert states_seen == [LifecycleState.STARTING, "method_called", LifecycleState.STARTED] + + +class TestKernelManagerStateMixin: + """Test cases for KernelManagerStateMixin""" + + def test_mixin_initialization(self): + """Test that mixin initializes with correct default state""" + + class TestManager(KernelManagerStateMixin, HasTraits): + pass + + manager = TestManager() + assert manager.lifecycle_state == LifecycleState.UNKNOWN + + def test_state_properties(self): + """Test state check properties""" + + class TestManager(KernelManagerStateMixin, HasTraits): + pass + + manager = TestManager() + + # Test initial unknown state + assert manager.is_unknown + assert not manager.is_starting + assert not manager.is_started + assert not manager.is_restarting + assert not manager.is_restarted + assert not manager.is_terminating + assert not manager.is_dead + + # Test starting state + manager.lifecycle_state = LifecycleState.STARTING + assert not manager.is_unknown + assert manager.is_starting + assert not manager.is_started + + # Test started state + manager.lifecycle_state = LifecycleState.STARTED + assert not manager.is_starting + assert manager.is_started + assert not manager.is_dead + + # Test dead state + manager.lifecycle_state = LifecycleState.DEAD + assert not manager.is_started + assert manager.is_dead + + def test_set_lifecycle_state(self): + """Test manual state setting""" + + class TestManager(KernelManagerStateMixin, HasTraits): + pass + + manager = TestManager() + manager.set_lifecycle_state(LifecycleState.STARTED) + + assert manager.lifecycle_state == LifecycleState.STARTED + assert manager.is_started + + def test_lifecycle_state_observer(self): + """Test that state changes are logged""" + + class TestManager(KernelManagerStateMixin, HasTraits): + def __init__(self): + super().__init__() + self.log = Mock() + + manager = TestManager() + manager.kernel_id = "test-kernel-123" + + # Change state to trigger observer + manager.lifecycle_state = LifecycleState.STARTING + + # Check that log.debug was called + manager.log.debug.assert_called_once() + call_args = manager.log.debug.call_args[0][0] + assert "test-kernel-123" in call_args + # The actual log format shows enum representation + assert "LifecycleState.UNKNOWN -> LifecycleState.STARTING" in call_args + + def test_automatic_method_wrapping_start_kernel(self): + """Test that start_kernel is automatically wrapped with state transitions""" + + class TestManager(KernelManagerStateMixin, HasTraits): + def start_kernel(self): + return "kernel_started" + + manager = TestManager() + assert manager.lifecycle_state == LifecycleState.UNKNOWN + + result = manager.start_kernel() + + assert result == "kernel_started" + assert manager.lifecycle_state == LifecycleState.STARTED + + @pytest.mark.asyncio + async def test_automatic_method_wrapping_async_start_kernel(self): + """Test that async start_kernel is automatically wrapped""" + + class TestManager(KernelManagerStateMixin, HasTraits): + async def start_kernel(self): + return "async_kernel_started" + + manager = TestManager() + assert manager.lifecycle_state == LifecycleState.UNKNOWN + + result = await manager.start_kernel() + + assert result == "async_kernel_started" + assert manager.lifecycle_state == LifecycleState.STARTED + + def test_automatic_method_wrapping_restart_kernel(self): + """Test that restart_kernel is automatically wrapped""" + + class TestManager(KernelManagerStateMixin, HasTraits): + def restart_kernel(self): + return "kernel_restarted" + + manager = TestManager() + manager.lifecycle_state = LifecycleState.STARTED # Set initial state + + result = manager.restart_kernel() + + assert result == "kernel_restarted" + assert manager.lifecycle_state == LifecycleState.RESTARTED + + def test_automatic_method_wrapping_shutdown_kernel(self): + """Test that shutdown_kernel is automatically wrapped""" + + class TestManager(KernelManagerStateMixin, HasTraits): + def shutdown_kernel(self): + return "kernel_shutdown" + + manager = TestManager() + manager.lifecycle_state = LifecycleState.STARTED # Set initial state + + result = manager.shutdown_kernel() + + assert result == "kernel_shutdown" + assert manager.lifecycle_state == LifecycleState.DEAD + + def test_method_wrapping_failure_handling(self): + """Test that method failures set state to UNKNOWN""" + + class TestManager(KernelManagerStateMixin, HasTraits): + def start_kernel(self): + raise Exception("Start failed") + + manager = TestManager() + + with pytest.raises(Exception): + manager.start_kernel() + + assert manager.lifecycle_state == LifecycleState.UNKNOWN + + def test_mixin_with_inheritance(self): + """Test that mixin works correctly with complex inheritance""" + + class BaseManager(HasTraits): + def base_method(self): + return "base" + + class TestManager(KernelManagerStateMixin, BaseManager): + def start_kernel(self): + return "started" + + def restart_kernel(self): + return "restarted" + + def shutdown_kernel(self): + return "shutdown" + + manager = TestManager() + + # Test base functionality still works + assert manager.base_method() == "base" + + # Test state transitions work + assert manager.lifecycle_state == LifecycleState.UNKNOWN + + manager.start_kernel() + assert manager.lifecycle_state == LifecycleState.STARTED + + manager.restart_kernel() + assert manager.lifecycle_state == LifecycleState.RESTARTED + + manager.shutdown_kernel() + assert manager.lifecycle_state == LifecycleState.DEAD + + def test_mixin_without_kernel_methods(self): + """Test that mixin works even when kernel methods don't exist""" + + class TestManager(KernelManagerStateMixin, HasTraits): + def some_other_method(self): + return "other" + + # Should not raise any errors during class creation + manager = TestManager() + assert manager.lifecycle_state == LifecycleState.UNKNOWN + assert manager.some_other_method() == "other" + + def test_state_transitions_complete_lifecycle(self): + """Test a complete kernel lifecycle with state transitions""" + + class TestManager(KernelManagerStateMixin, HasTraits): + def start_kernel(self): + return "started" + + def restart_kernel(self): + return "restarted" + + def shutdown_kernel(self): + return "shutdown" + + manager = TestManager() + + # Initial state + assert manager.is_unknown + + # Start kernel + manager.start_kernel() + assert manager.is_started + assert not manager.is_unknown + + # Restart kernel + manager.restart_kernel() + assert manager.is_restarted + assert not manager.is_started + + # Shutdown kernel + manager.shutdown_kernel() + assert manager.is_dead + assert not manager.is_restarted + + +class TestJupyterClientIntegration: + """Test integration with jupyter_client patterns""" + + def test_with_mock_kernel_manager_interface(self): + """Test mixin works with typical KernelManager interface""" + + class MockKernelManager(KernelManagerStateMixin, HasTraits): + """Mock kernel manager following jupyter_client patterns""" + + def __init__(self, **kwargs): + super().__init__(**kwargs) + self.kernel_id = "test-kernel-123" + self.log = Mock() + + def start_kernel(self, **kwargs): + """Mock start_kernel method""" + return {"kernel_id": self.kernel_id} + + async def restart_kernel(self, **kwargs): + """Mock async restart_kernel method""" + return {"kernel_id": self.kernel_id} + + def shutdown_kernel(self, immediate=False, **kwargs): + """Mock shutdown_kernel method""" + return True + + manager = MockKernelManager() + + # Test initial state + assert manager.is_unknown + + # Test start + result = manager.start_kernel() + assert result["kernel_id"] == "test-kernel-123" + assert manager.is_started + + # Test restart + import asyncio + + async def test_restart(): + result = await manager.restart_kernel() + assert result["kernel_id"] == "test-kernel-123" + assert manager.is_restarted + + asyncio.run(test_restart()) + + # Test shutdown + result = manager.shutdown_kernel(immediate=True) + assert result is True + assert manager.is_dead + + def test_configurable_lifecycle_state(self): + """Test that lifecycle_state is properly configurable""" + + class ConfigurableManager(KernelManagerStateMixin, HasTraits): + pass + + # Test default configuration + manager = ConfigurableManager() + assert manager.lifecycle_state == LifecycleState.UNKNOWN + + # Test configuration override + manager_configured = ConfigurableManager(lifecycle_state="started") + assert manager_configured.lifecycle_state == "started" + assert manager_configured.is_started