diff --git a/ccflow/base.py b/ccflow/base.py
index 2a35454..2309577 100644
--- a/ccflow/base.py
+++ b/ccflow/base.py
@@ -241,6 +241,34 @@ def get_widget(
# Can't use self.model_dump_json or self.model_dump because they don't expose the fallback argument
return JSON(self.__pydantic_serializer__.to_python(self, **kwargs), **(widget_kwargs or {}))
+ def __panel__(self):
+ """Return a Panel viewable for this model.
+
+ Requires ccflow UI dependencies (panel, panel_material_ui).
+ """
+ try:
+ from ccflow.ui.model import ModelViewer
+ except ImportError:
+ raise ImportError(
+ "panel and other optional dependencies must be installed to use ModelViewer. Pip install ccflow[full] to install all optional dependencies."
+ ) from None
+
+ return ModelViewer(model=self)
+
+ def get_panel(self):
+ """Get a Panel pane for this model.
+
+ Requires panel to be installed.
+ """
+ try:
+ import panel as pn
+ except ImportError:
+ raise ImportError(
+ "panel and other optional dependencies must be installed to use get_panel(). Pip install ccflow[full] to install all optional dependencies."
+ ) from None
+
+ return pn.panel(self)
+
@model_validator(mode="wrap")
def _base_model_validator(cls, v, handler, info):
if isinstance(v, str):
@@ -400,6 +428,15 @@ def models(self) -> MappingProxyType:
"""Return an immutable pointer to the models dictionary."""
return MappingProxyType(self._models)
+ def __panel__(self):
+ """Return a Panel viewable for this registry.
+
+ Requires ccflow UI dependencies (panel, panel_material_ui).
+ """
+ from ccflow.ui.registry import ModelRegistryViewer
+
+ return ModelRegistryViewer(self)
+
@classmethod
def root(cls) -> Self:
"""Return a static instance of the root registry."""
diff --git a/ccflow/tests/test_base.py b/ccflow/tests/test_base.py
index e53e0af..26623fb 100644
--- a/ccflow/tests/test_base.py
+++ b/ccflow/tests/test_base.py
@@ -173,6 +173,32 @@ def test_widget(self):
),
)
+ def test_panel(self):
+ from ccflow import ModelRegistry
+ from ccflow.ui.model import ModelViewer
+ from ccflow.ui.registry import ModelRegistryViewer
+
+ m = ModelA(x="foo")
+ panel_obj = m.__panel__()
+ self.assertIsInstance(panel_obj, ModelViewer)
+
+ registry = ModelRegistry(name="test", models={"a": m})
+ registry_panel_obj = registry.__panel__()
+ self.assertIsInstance(registry_panel_obj, ModelRegistryViewer)
+
+ def test_get_panel(self):
+ import panel as pn
+
+ from ccflow import ModelRegistry
+
+ m = ModelA(x="foo")
+ panel_pane = m.get_panel()
+ self.assertIsInstance(panel_pane, pn.viewable.Viewable)
+
+ registry = ModelRegistry(name="test", models={"a": m})
+ registry_pane = registry.get_panel()
+ self.assertIsInstance(registry_pane, pn.viewable.Viewable)
+
class TestLocalRegistration(TestCase):
def test_local_class_registered_for_base_model(self):
diff --git a/ccflow/tests/ui/__init__.py b/ccflow/tests/ui/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/ccflow/tests/ui/test_cli.py b/ccflow/tests/ui/test_cli.py
new file mode 100644
index 0000000..d209aea
--- /dev/null
+++ b/ccflow/tests/ui/test_cli.py
@@ -0,0 +1,66 @@
+"""Unit tests for ccflow.ui.cli module."""
+
+from ccflow.ui.cli import _get_ui_args_parser
+
+
+class TestGetUIArgsParser:
+ """Tests for _get_ui_args_parser function.
+
+ Note: Default values for hydra config args and panel server args are tested
+ in ccflow/tests/utils/test_hydra.py. These tests focus on viewer-specific
+ arguments and verifying the parser composition works correctly.
+ """
+
+ def test_parser_composition(self):
+ """Test parser includes args from both helper functions."""
+ parser = _get_ui_args_parser()
+ args = parser.parse_args([])
+
+ # From add_hydra_config_args
+ assert hasattr(args, "overrides")
+ assert hasattr(args, "config_path")
+ assert hasattr(args, "config_name")
+
+ # From add_panel_server_args
+ assert hasattr(args, "address")
+ assert hasattr(args, "port")
+ assert hasattr(args, "show")
+
+ # Viewer-specific
+ assert hasattr(args, "browser_width")
+ assert hasattr(args, "browser_height")
+ assert hasattr(args, "viewer_width")
+
+ def test_viewer_layout_defaults(self):
+ """Test default values for viewer-specific arguments."""
+ parser = _get_ui_args_parser()
+ args = parser.parse_args([])
+
+ assert args.browser_width == 400
+ assert args.browser_height == 700
+ assert args.viewer_width is None
+
+ def test_viewer_layout_custom_values(self):
+ """Test setting custom values for viewer layout arguments."""
+ parser = _get_ui_args_parser()
+ args = parser.parse_args(
+ [
+ "--browser-width",
+ "500",
+ "--browser-height",
+ "800",
+ "--viewer-width",
+ "600",
+ ]
+ )
+
+ assert args.browser_width == 500
+ assert args.browser_height == 800
+ assert args.viewer_width == 600
+
+ def test_overrides_positional(self):
+ """Test overrides are captured as positional arguments."""
+ parser = _get_ui_args_parser()
+ args = parser.parse_args(["key1=value1", "key2=value2"])
+
+ assert args.overrides == ["key1=value1", "key2=value2"]
diff --git a/ccflow/tests/ui/test_model.py b/ccflow/tests/ui/test_model.py
new file mode 100644
index 0000000..043dbd6
--- /dev/null
+++ b/ccflow/tests/ui/test_model.py
@@ -0,0 +1,555 @@
+"""Unit tests for ccflow.ui.model module."""
+
+import panel as pn
+from pydantic import Field
+
+from ccflow import BaseModel, CallableModel, ContextBase, Flow, GenericResult, MetaData, ModelRegistry
+from ccflow.ui.model import ModelConfigViewer, ModelTypeViewer, ModelViewer
+
+from .utils import find_components_by_type
+
+
+class SimpleModel(BaseModel):
+ """A simple test model with documentation."""
+
+ name: str
+ value: int = 0
+
+
+class NoDocModel(BaseModel):
+ field: str
+
+
+class DescribedModel(BaseModel):
+ """Model with field descriptions."""
+
+ name: str = Field(description="The name field")
+ count: int = Field(description="The count field")
+
+
+class ContainerModel(BaseModel):
+ """Model that contains another model."""
+
+ inner: SimpleModel
+
+
+class SampleContext(ContextBase):
+ """Sample context for callable models."""
+
+ input_value: str = ""
+
+
+class SampleResult(GenericResult):
+ """Sample result for callable models."""
+
+ output_value: str = ""
+
+
+class SimpleCallableModel(CallableModel):
+ """A simple callable model for testing."""
+
+ multiplier: int = 1
+
+ @Flow.call
+ def __call__(self, context: SampleContext) -> SampleResult:
+ return SampleResult(output_value=context.input_value * self.multiplier)
+
+
+class TestModelTypeViewer:
+ """Tests for ModelTypeViewer class."""
+
+ def test_init_returns_viewable(self):
+ """Test ModelTypeViewer returns a Panel viewable."""
+ viewer = ModelTypeViewer()
+ panel = viewer.__panel__()
+ assert isinstance(panel, pn.viewable.Viewable)
+
+ def test_panel_contains_html_pane(self):
+ """Test that the panel contains an HTML pane for displaying content."""
+ viewer = ModelTypeViewer()
+ panel = viewer.__panel__()
+ html_panes = find_components_by_type(panel, pn.pane.HTML)
+ assert len(html_panes) > 0
+
+ def test_on_type_change_with_none(self):
+ """Test that setting model_type to None clears the display."""
+ viewer = ModelTypeViewer()
+ viewer.model_type = SimpleModel
+ viewer.model_type = None
+
+ panel = viewer.__panel__()
+ html_panes = find_components_by_type(panel, pn.pane.HTML)
+ # All HTML panes should be empty
+ for pane in html_panes:
+ assert pane.object == ""
+
+ def test_on_type_change_with_model(self):
+ """Test that setting model_type displays type information."""
+ viewer = ModelTypeViewer()
+ viewer.model_type = SimpleModel
+
+ panel = viewer.__panel__()
+ html_panes = find_components_by_type(panel, pn.pane.HTML)
+ html_content = "".join(pane.object for pane in html_panes)
+
+ # Check that type name is displayed
+ assert "SimpleModel" in html_content
+ # Check that documentation is displayed
+ assert "A simple test model with documentation" in html_content
+ # Check that fields are displayed
+ assert "name" in html_content
+ assert "value" in html_content
+ assert "str" in html_content
+ assert "int" in html_content
+
+ def test_on_type_change_without_docstring(self):
+ """Test model type without a docstring."""
+ viewer = ModelTypeViewer()
+ viewer.model_type = NoDocModel
+
+ panel = viewer.__panel__()
+ html_panes = find_components_by_type(panel, pn.pane.HTML)
+ html_content = "".join(pane.object for pane in html_panes)
+
+ assert "NoDocModel" in html_content
+ assert "field" in html_content
+ # Should not have documentation section
+ assert "Class Documentation:" not in html_content
+
+ def test_on_type_change_with_field_descriptions(self):
+ """Test that field descriptions are displayed."""
+ viewer = ModelTypeViewer()
+ viewer.model_type = DescribedModel
+
+ panel = viewer.__panel__()
+ html_panes = find_components_by_type(panel, pn.pane.HTML)
+ html_content = "".join(pane.object for pane in html_panes)
+
+ assert "The name field" in html_content
+ assert "The count field" in html_content
+
+
+class TestModelConfigViewer:
+ """Tests for ModelConfigViewer class."""
+
+ def test_init_returns_viewable(self):
+ """Test ModelConfigViewer returns a Panel viewable."""
+ viewer = ModelConfigViewer()
+ panel = viewer.__panel__()
+ assert isinstance(panel, pn.viewable.Viewable)
+
+ def test_panel_contains_html_pane(self):
+ """Test that the panel contains an HTML pane for metadata display."""
+ viewer = ModelConfigViewer()
+ panel = viewer.__panel__()
+ html_panes = find_components_by_type(panel, pn.pane.HTML)
+ assert len(html_panes) > 0
+
+ def test_on_model_change_with_none(self):
+ """Test that setting model to None clears the metadata display."""
+ viewer = ModelConfigViewer()
+ model = SimpleModel(name="test")
+ viewer.model = model
+ viewer.model = None
+
+ panel = viewer.__panel__()
+ html_panes = find_components_by_type(panel, pn.pane.HTML)
+ for pane in html_panes:
+ assert pane.object == ""
+
+ def test_on_model_change_with_model(self):
+ """Test that setting model displays metadata."""
+ viewer = ModelConfigViewer()
+ model = SimpleModel(name="test", value=42)
+ viewer.model = model
+
+ panel = viewer.__panel__()
+ html_panes = find_components_by_type(panel, pn.pane.HTML)
+ # Should not crash, content depends on model metadata
+ assert len(html_panes) > 0
+
+ def test_on_model_change_with_description(self):
+ """Test model with meta description."""
+ viewer = ModelConfigViewer()
+ model = SimpleCallableModel(
+ multiplier=2,
+ meta=MetaData(description="This is a test model description"),
+ )
+ viewer.model = model
+
+ panel = viewer.__panel__()
+ html_panes = find_components_by_type(panel, pn.pane.HTML)
+ html_content = "".join(pane.object for pane in html_panes)
+
+ assert "This is a test model description" in html_content
+
+ def test_render_dependencies_empty(self):
+ """Test _render_dependencies with no dependencies."""
+ viewer = ModelConfigViewer()
+ model = SimpleModel(name="test")
+ result = viewer._render_dependencies(model)
+ assert result == ""
+
+ def test_render_dependencies_with_deps(self):
+ """Test _render_dependencies with dependencies."""
+ root = ModelRegistry.root()
+ registry = ModelRegistry(name="test_dep_registry")
+ root.add("test_dep_registry", registry)
+
+ try:
+ model = SimpleModel(name="test")
+ registry.add("my_model", model)
+
+ container = ContainerModel(inner=model)
+ registry.add("container", container)
+
+ viewer = ModelConfigViewer()
+ result = viewer._render_dependencies(container)
+
+ assert "Registry Dependencies" in result
+ assert "my_model" in result
+ finally:
+ root.remove("test_dep_registry")
+
+
+class TestModelViewer:
+ """Tests for ModelViewer class."""
+
+ def test_init_returns_viewable(self):
+ """Test ModelViewer returns a Panel viewable."""
+ viewer = ModelViewer()
+ panel = viewer.__panel__()
+ assert isinstance(panel, pn.viewable.Viewable)
+
+ def test_panel_contains_json_editor(self):
+ """Test that the panel contains a JSON editor widget."""
+ viewer = ModelViewer()
+ panel = viewer.__panel__()
+ json_editors = find_components_by_type(panel, pn.widgets.JSONEditor)
+ assert len(json_editors) > 0
+
+ def test_json_editor_initially_empty(self):
+ """Test that JSON editor is initially empty."""
+ viewer = ModelViewer()
+ panel = viewer.__panel__()
+ json_editors = find_components_by_type(panel, pn.widgets.JSONEditor)
+ assert json_editors[0].value == {}
+
+ def test_on_model_change_with_none(self):
+ """Test that setting model to None clears the JSON editor."""
+ viewer = ModelViewer()
+ viewer.model = SimpleModel(name="test")
+ viewer.model = None
+
+ panel = viewer.__panel__()
+ json_editors = find_components_by_type(panel, pn.widgets.JSONEditor)
+ assert json_editors[0].value == {}
+
+ def test_on_model_change_with_base_model(self):
+ """Test that setting a BaseModel populates the JSON editor."""
+ viewer = ModelViewer()
+ model = SimpleModel(name="test", value=42)
+ viewer.model = model
+
+ panel = viewer.__panel__()
+ json_editors = find_components_by_type(panel, pn.widgets.JSONEditor)
+ json_value = json_editors[0].value
+
+ assert "name" in json_value
+ assert json_value["name"] == "test"
+ assert json_value["value"] == 42
+
+ def test_on_model_change_with_callable_model(self):
+ """Test that setting a CallableModel sets up type viewers correctly."""
+ viewer = ModelViewer()
+ model = SimpleCallableModel(multiplier=2)
+ viewer.model = model
+
+ # Verify the internal type viewers are properly configured
+ assert viewer._config_viewer.model == model
+ assert viewer._type_viewer.model_type is type(model)
+ assert viewer._context_type_viewer.model_type == model.context_type
+ assert viewer._result_type_viewer.model_type == model.result_type
+
+ def test_on_model_change_updates_viewers(self):
+ """Test that changing model updates the viewers."""
+ viewer = ModelViewer()
+
+ # Set a callable model first
+ callable_model = SimpleCallableModel(multiplier=2)
+ viewer.model = callable_model
+
+ # Then set a base model
+ base_model = SimpleModel(name="test")
+ viewer.model = base_model
+
+ # Viewers should be updated for the new model
+ assert viewer._config_viewer.model == base_model
+ assert viewer._type_viewer.model_type is type(base_model)
+
+ def test_json_serialization(self):
+ """Test that JSON editor correctly serializes model data."""
+ viewer = ModelViewer()
+ model = SimpleModel(name="test_name", value=123)
+ viewer.model = model
+
+ panel = viewer.__panel__()
+ json_editors = find_components_by_type(panel, pn.widgets.JSONEditor)
+ json_value = json_editors[0].value
+
+ assert json_value["name"] == "test_name"
+ assert json_value["value"] == 123
+
+
+class TestModelSwitching:
+ """Tests for switching between different models to ensure proper state reset."""
+
+ def test_switch_between_base_models_same_type(self):
+ """Test switching between two BaseModel instances of the same type."""
+ viewer = ModelViewer()
+
+ model1 = SimpleModel(name="first", value=1)
+ model2 = SimpleModel(name="second", value=2)
+
+ # Set first model
+ viewer.model = model1
+ assert viewer._config_viewer.model == model1
+ assert viewer._type_viewer.model_type == SimpleModel
+
+ panel = viewer.__panel__()
+ json_editors = find_components_by_type(panel, pn.widgets.JSONEditor)
+ assert json_editors[0].value["name"] == "first"
+ assert json_editors[0].value["value"] == 1
+
+ # Switch to second model
+ viewer.model = model2
+ assert viewer._config_viewer.model == model2
+ assert viewer._type_viewer.model_type == SimpleModel
+
+ # Verify JSON editor updated (not still showing old model)
+ assert json_editors[0].value["name"] == "second"
+ assert json_editors[0].value["value"] == 2
+
+ def test_switch_between_base_models_different_types(self):
+ """Test switching between two BaseModel instances of different types."""
+ viewer = ModelViewer()
+
+ model1 = SimpleModel(name="simple", value=42)
+ model2 = DescribedModel(name="described", count=100)
+
+ # Set first model
+ viewer.model = model1
+ assert viewer._type_viewer.model_type == SimpleModel
+
+ panel = viewer.__panel__()
+ json_editors = find_components_by_type(panel, pn.widgets.JSONEditor)
+ assert json_editors[0].value["name"] == "simple"
+ assert "value" in json_editors[0].value
+
+ # Switch to different type
+ viewer.model = model2
+ assert viewer._config_viewer.model == model2
+ assert viewer._type_viewer.model_type == DescribedModel
+
+ # JSON editor should show new model's fields, not old ones
+ assert json_editors[0].value["name"] == "described"
+ assert json_editors[0].value["count"] == 100
+ assert "value" not in json_editors[0].value
+
+ def test_switch_from_callable_to_base_model(self):
+ """Test switching from CallableModel to BaseModel removes context/result tabs."""
+ viewer = ModelViewer()
+
+ callable_model = SimpleCallableModel(multiplier=5)
+ base_model = SimpleModel(name="base", value=10)
+
+ # Set callable model first
+ viewer.model = callable_model
+ assert viewer._context_type_viewer.model_type == callable_model.context_type
+ assert viewer._result_type_viewer.model_type == callable_model.result_type
+
+ # Should have 4 tabs: Summary, Model Type, Context Type, Result Type
+ assert len(viewer._tabs) == 4
+
+ # Switch to base model
+ viewer.model = base_model
+ assert viewer._config_viewer.model == base_model
+ assert viewer._type_viewer.model_type == SimpleModel
+
+ # Should have only 2 tabs now: Summary, Model Type
+ assert len(viewer._tabs) == 2
+
+ # JSON editor should show base model data
+ panel = viewer.__panel__()
+ json_editors = find_components_by_type(panel, pn.widgets.JSONEditor)
+ assert json_editors[0].value["name"] == "base"
+ assert json_editors[0].value["value"] == 10
+ assert "multiplier" not in json_editors[0].value
+
+ def test_switch_from_base_to_callable_model(self):
+ """Test switching from BaseModel to CallableModel adds context/result tabs."""
+ viewer = ModelViewer()
+
+ base_model = SimpleModel(name="base", value=10)
+ callable_model = SimpleCallableModel(multiplier=5)
+
+ # Set base model first
+ viewer.model = base_model
+ assert len(viewer._tabs) == 2
+
+ # Switch to callable model
+ viewer.model = callable_model
+ assert viewer._config_viewer.model == callable_model
+ assert viewer._type_viewer.model_type == SimpleCallableModel
+ assert viewer._context_type_viewer.model_type == callable_model.context_type
+ assert viewer._result_type_viewer.model_type == callable_model.result_type
+
+ # Should have 4 tabs now
+ assert len(viewer._tabs) == 4
+
+ # JSON editor should show callable model data
+ panel = viewer.__panel__()
+ json_editors = find_components_by_type(panel, pn.widgets.JSONEditor)
+ assert "multiplier" in json_editors[0].value
+ assert json_editors[0].value["multiplier"] == 5
+
+ def test_switch_to_none_clears_state(self):
+ """Test switching from a model to None clears all state."""
+ viewer = ModelViewer()
+
+ model = SimpleCallableModel(multiplier=3)
+
+ # Set model
+ viewer.model = model
+ assert len(viewer._tabs) == 4
+ assert viewer._json_container.visible is True
+
+ # Clear model
+ viewer.model = None
+ assert len(viewer._tabs) == 0
+ assert viewer._json_container.visible is False
+
+ panel = viewer.__panel__()
+ json_editors = find_components_by_type(panel, pn.widgets.JSONEditor)
+ assert json_editors[0].value == {}
+
+ def test_tabs_reset_to_first_on_model_switch(self):
+ """Test that tab selection resets to first tab when switching models."""
+ viewer = ModelViewer()
+
+ model1 = SimpleCallableModel(multiplier=1)
+ model2 = SimpleCallableModel(multiplier=2)
+
+ # Set first model and change active tab
+ viewer.model = model1
+ viewer._tabs.active = 2 # Select "Context Type" tab
+
+ # Switch to second model
+ viewer.model = model2
+
+ # Tab should reset to first (Summary)
+ assert viewer._tabs.active == 0
+
+
+class TestModelTypeViewerSwitching:
+ """Tests for ModelTypeViewer switching between types."""
+
+ def test_switch_model_types(self):
+ """Test switching between different model types updates display."""
+ viewer = ModelTypeViewer()
+
+ # Set first type
+ viewer.model_type = SimpleModel
+ panel = viewer.__panel__()
+ html_panes = find_components_by_type(panel, pn.pane.HTML)
+ html_content = "".join(pane.object for pane in html_panes)
+
+ assert "SimpleModel" in html_content
+ assert "name" in html_content
+ assert "value" in html_content
+
+ # Switch to different type
+ viewer.model_type = DescribedModel
+ html_content = "".join(pane.object for pane in html_panes)
+
+ # Should show new type, not old
+ assert "DescribedModel" in html_content
+ assert "count" in html_content
+ assert "SimpleModel" not in html_content
+ # "name" is in both, but "value" should not be present
+ assert "value" not in html_content
+
+ def test_switch_to_none_clears_display(self):
+ """Test switching to None clears the display."""
+ viewer = ModelTypeViewer()
+
+ viewer.model_type = SimpleModel
+ panel = viewer.__panel__()
+ html_panes = find_components_by_type(panel, pn.pane.HTML)
+
+ # Verify content is present
+ html_content = "".join(pane.object for pane in html_panes)
+ assert "SimpleModel" in html_content
+
+ # Clear
+ viewer.model_type = None
+ html_content = "".join(pane.object for pane in html_panes)
+ assert html_content == ""
+
+
+class TestModelConfigViewerSwitching:
+ """Tests for ModelConfigViewer switching between models."""
+
+ def test_switch_models_with_different_metadata(self):
+ """Test switching between models with different metadata."""
+ viewer = ModelConfigViewer()
+
+ model1 = SimpleCallableModel(
+ multiplier=1,
+ meta=MetaData(description="First model description"),
+ )
+ model2 = SimpleCallableModel(
+ multiplier=2,
+ meta=MetaData(description="Second model description"),
+ )
+
+ # Set first model
+ viewer.model = model1
+ panel = viewer.__panel__()
+ html_panes = find_components_by_type(panel, pn.pane.HTML)
+ html_content = "".join(pane.object for pane in html_panes)
+
+ assert "First model description" in html_content
+
+ # Switch to second model
+ viewer.model = model2
+ html_content = "".join(pane.object for pane in html_panes)
+
+ # Should show new description, not old
+ assert "Second model description" in html_content
+ assert "First model description" not in html_content
+
+ def test_switch_from_model_with_description_to_without(self):
+ """Test switching from model with description to one without."""
+ viewer = ModelConfigViewer()
+
+ model_with_desc = SimpleCallableModel(
+ multiplier=1,
+ meta=MetaData(description="Has a description"),
+ )
+ model_without_desc = SimpleModel(name="no desc")
+
+ # Set model with description
+ viewer.model = model_with_desc
+ panel = viewer.__panel__()
+ html_panes = find_components_by_type(panel, pn.pane.HTML)
+ html_content = "".join(pane.object for pane in html_panes)
+
+ assert "Has a description" in html_content
+
+ # Switch to model without description
+ viewer.model = model_without_desc
+ html_content = "".join(pane.object for pane in html_panes)
+
+ # Old description should not be present
+ assert "Has a description" not in html_content
diff --git a/ccflow/tests/ui/test_registry.py b/ccflow/tests/ui/test_registry.py
new file mode 100644
index 0000000..c1266f9
--- /dev/null
+++ b/ccflow/tests/ui/test_registry.py
@@ -0,0 +1,602 @@
+"""Unit tests for ccflow.ui.registry module."""
+
+from unittest import mock
+
+import panel as pn
+
+from ccflow import BaseModel, ModelRegistry
+from ccflow.ui.registry import ModelRegistryViewer, RegistryBrowser
+
+from .utils import find_components_by_type
+
+
+class SimpleModel(BaseModel):
+ """A simple test model."""
+
+ name: str
+ value: int = 0
+
+
+class AnotherModel(BaseModel):
+ """Another test model."""
+
+ data: str = ""
+
+
+class TestRegistryBrowser:
+ """Tests for RegistryBrowser class."""
+
+ def test_init_returns_viewable(self):
+ """Test RegistryBrowser returns a Panel viewable."""
+ registry = ModelRegistry(name="test")
+ browser = RegistryBrowser(registry)
+ panel = browser.__panel__()
+ assert isinstance(panel, pn.viewable.Viewable)
+
+ def test_panel_contains_autocomplete_search(self):
+ """Test that the panel contains an autocomplete search widget."""
+ registry = ModelRegistry(name="test")
+ browser = RegistryBrowser(registry)
+ panel = browser.__panel__()
+ autocomplete = find_components_by_type(panel, pn.widgets.AutocompleteInput)
+ assert len(autocomplete) > 0
+
+ def test_init_with_empty_registry(self):
+ """Test RegistryBrowser initialization with empty registry."""
+ registry = ModelRegistry(name="test")
+ browser = RegistryBrowser(registry)
+
+ assert browser.selected_model is None
+ # Search options should be empty
+ panel = browser.__panel__()
+ autocomplete = find_components_by_type(panel, pn.widgets.AutocompleteInput)
+ assert autocomplete[0].options == []
+
+ def test_init_with_models(self):
+ """Test RegistryBrowser initialization with models in registry."""
+ registry = ModelRegistry(name="test")
+ model1 = SimpleModel(name="test1", value=1)
+ model2 = AnotherModel(data="test2")
+ registry.add("model1", model1)
+ registry.add("model2", model2)
+
+ browser = RegistryBrowser(registry)
+
+ # Search options should contain the model names
+ panel = browser.__panel__()
+ autocomplete = find_components_by_type(panel, pn.widgets.AutocompleteInput)
+ assert "model1" in autocomplete[0].options
+ assert "model2" in autocomplete[0].options
+
+ def test_build_tree_simple(self):
+ """Test _build_tree with flat registry."""
+ registry = ModelRegistry(name="test")
+ model = SimpleModel(name="test", value=1)
+ registry.add("my_model", model)
+
+ browser = RegistryBrowser(registry)
+ tree_items = browser._tree_items
+
+ assert len(tree_items) == 1
+ assert tree_items[0]["label"] == "my_model"
+ assert tree_items[0]["model"] == model
+ assert tree_items[0]["_index_path"] == (0,)
+
+ def test_build_tree_nested(self):
+ """Test _build_tree with nested registries."""
+ root = ModelRegistry(name="root")
+ sub = ModelRegistry(name="sub")
+ model = SimpleModel(name="test", value=1)
+
+ sub.add("nested_model", model)
+ root.add("subregistry", sub)
+
+ browser = RegistryBrowser(root)
+ tree_items = browser._tree_items
+
+ assert len(tree_items) == 1
+ assert tree_items[0]["label"] == "subregistry"
+ assert "items" in tree_items[0]
+ assert len(tree_items[0]["items"]) == 1
+ assert tree_items[0]["items"][0]["label"] == "nested_model"
+ assert tree_items[0]["items"][0]["model"] == model
+
+ def test_build_node_index(self):
+ """Test _build_node_index creates correct path mappings."""
+ root = ModelRegistry(name="root")
+ sub = ModelRegistry(name="sub")
+ model1 = SimpleModel(name="test1", value=1)
+ model2 = SimpleModel(name="test2", value=2)
+
+ root.add("top_model", model1)
+ sub.add("nested_model", model2)
+ root.add("subregistry", sub)
+
+ browser = RegistryBrowser(root)
+
+ assert "top_model" in browser._node_index
+ assert "subregistry/nested_model" in browser._node_index
+ assert browser._node_index["top_model"]["model"] == model1
+ assert browser._node_index["subregistry/nested_model"]["model"] == model2
+
+ def test_expanded_from_index_path(self):
+ """Test _expanded_from_index_path generates correct expanded paths."""
+ # Single level - no expansion needed
+ result = RegistryBrowser._expanded_from_index_path((0,))
+ assert result == []
+
+ # Two levels
+ result = RegistryBrowser._expanded_from_index_path((0, 1))
+ assert result == [(0,)]
+
+ # Three levels
+ result = RegistryBrowser._expanded_from_index_path((0, 1, 2))
+ assert result == [(0,), (0, 1)]
+
+ # Four levels
+ result = RegistryBrowser._expanded_from_index_path((1, 2, 3, 4))
+ assert result == [(1,), (1, 2), (1, 2, 3)]
+
+ def test_on_search_select_empty(self):
+ """Test _on_search_select with empty value."""
+ registry = ModelRegistry(name="test")
+ model = SimpleModel(name="test", value=1)
+ registry.add("my_model", model)
+ browser = RegistryBrowser(registry)
+
+ event = mock.Mock()
+ event.new = ""
+
+ browser._on_search_select(event)
+ assert browser.selected_model is None
+
+ def test_on_search_select_invalid_path(self):
+ """Test _on_search_select with non-existent path."""
+ registry = ModelRegistry(name="test")
+ model = SimpleModel(name="test", value=1)
+ registry.add("my_model", model)
+ browser = RegistryBrowser(registry)
+
+ event = mock.Mock()
+ event.new = "nonexistent"
+
+ browser._on_search_select(event)
+ assert browser.selected_model is None
+
+ def test_on_search_select_valid_path(self):
+ """Test _on_search_select with valid path."""
+ registry = ModelRegistry(name="test")
+ model = SimpleModel(name="test", value=1)
+ registry.add("my_model", model)
+ browser = RegistryBrowser(registry)
+
+ event = mock.Mock()
+ event.new = "my_model"
+
+ browser._on_search_select(event)
+
+ # Tree value should be set
+ assert len(browser._tree.value) == 1
+ assert browser._tree.value[0]["label"] == "my_model"
+
+ def test_on_tree_select_empty(self):
+ """Test _on_tree_select with empty selection."""
+ registry = ModelRegistry(name="test")
+ model = SimpleModel(name="test", value=1)
+ registry.add("my_model", model)
+ browser = RegistryBrowser(registry)
+
+ event = mock.Mock()
+ event.new = []
+
+ browser._on_tree_select(event)
+ assert browser.selected_model is None
+
+ def test_on_tree_select_model(self):
+ """Test _on_tree_select with model selection."""
+ registry = ModelRegistry(name="test")
+ model = SimpleModel(name="test", value=1)
+ registry.add("my_model", model)
+ browser = RegistryBrowser(registry)
+
+ event = mock.Mock()
+ event.new = [{"label": "my_model", "model": model}]
+
+ browser._on_tree_select(event)
+ assert browser.selected_model == model
+
+ def test_on_tree_select_registry(self):
+ """Test _on_tree_select with registry selection (no model key)."""
+ registry = ModelRegistry(name="test")
+ sub = ModelRegistry(name="sub")
+ registry.add("subregistry", sub)
+ browser = RegistryBrowser(registry)
+
+ event = mock.Mock()
+ event.new = [{"label": "subregistry", "items": []}]
+
+ browser._on_tree_select(event)
+ assert browser.selected_model is None
+
+ def test_search_options_sorted(self):
+ """Test that search widget options are sorted."""
+ root = ModelRegistry(name="root")
+ sub = ModelRegistry(name="sub")
+ model1 = SimpleModel(name="test1", value=1)
+ model2 = SimpleModel(name="test2", value=2)
+
+ root.add("zebra", model1)
+ sub.add("alpha", model2)
+ root.add("subregistry", sub)
+
+ browser = RegistryBrowser(root)
+ panel = browser.__panel__()
+ autocomplete = find_components_by_type(panel, pn.widgets.AutocompleteInput)
+
+ assert autocomplete[0].options == sorted(autocomplete[0].options)
+
+
+class TestModelRegistryViewer:
+ """Tests for ModelRegistryViewer class."""
+
+ def test_init_returns_viewable(self):
+ """Test ModelRegistryViewer returns a Panel viewable."""
+ registry = ModelRegistry(name="test")
+ viewer = ModelRegistryViewer(registry)
+ panel = viewer.__panel__()
+ assert isinstance(panel, pn.viewable.Viewable)
+
+ def test_panel_is_row_layout(self):
+ """Test that the panel is a Row layout (browser + viewer side by side)."""
+ registry = ModelRegistry(name="test")
+ viewer = ModelRegistryViewer(registry)
+ panel = viewer.__panel__()
+ assert isinstance(panel, pn.Row)
+
+ def test_init_with_custom_dimensions(self):
+ """Test initialization with custom width/height."""
+ registry = ModelRegistry(name="test")
+ viewer = ModelRegistryViewer(
+ registry,
+ browser_width=500,
+ browser_height=800,
+ viewer_width=600,
+ )
+
+ assert viewer.browser_width == 500
+ assert viewer.browser_height == 800
+ assert viewer.viewer_width == 600
+
+ def test_browser_viewer_wiring(self):
+ """Test that browser selection updates viewer."""
+ registry = ModelRegistry(name="test")
+ model = SimpleModel(name="test", value=42)
+ registry.add("my_model", model)
+
+ viewer = ModelRegistryViewer(registry)
+
+ # Simulate browser selection
+ viewer._browser.selected_model = model
+
+ # Viewer should be updated
+ assert viewer._viewer.model == model
+
+ def test_default_browser_dimensions(self):
+ """Test default browser dimensions."""
+ registry = ModelRegistry(name="test")
+ viewer = ModelRegistryViewer(registry)
+
+ assert viewer.browser_width == 400
+ assert viewer.browser_height == 700
+ assert viewer.viewer_width is None
+
+ def test_make_browser_column(self):
+ """Test _make_browser_column creates proper column."""
+ registry = ModelRegistry(name="test")
+ viewer = ModelRegistryViewer(registry)
+
+ column = viewer._make_browser_column()
+ assert isinstance(column, pn.Column)
+ assert column.width == viewer.browser_width
+ assert column.height == viewer.browser_height
+ assert column.scroll is True
+
+ def test_make_viewer_column_with_width(self):
+ """Test _make_viewer_column with specified width."""
+ registry = ModelRegistry(name="test")
+ viewer = ModelRegistryViewer(registry, viewer_width=600)
+
+ column = viewer._make_viewer_column()
+ assert isinstance(column, pn.Column)
+ assert column.width == 600
+
+ def test_make_viewer_column_without_width(self):
+ """Test _make_viewer_column without specified width (stretch)."""
+ registry = ModelRegistry(name="test")
+ viewer = ModelRegistryViewer(registry)
+
+ column = viewer._make_viewer_column()
+ assert isinstance(column, pn.Column)
+ assert column.sizing_mode == "stretch_width"
+
+ def test_model_param_default_none(self):
+ """Test that model param starts as None."""
+ registry = ModelRegistry(name="test")
+ viewer = ModelRegistryViewer(registry)
+ assert viewer.model is None
+
+ def test_model_param_updated_on_selection(self):
+ """Test that model param is updated when a model is selected."""
+ registry = ModelRegistry(name="test")
+ model = SimpleModel(name="test", value=42)
+ registry.add("my_model", model)
+
+ viewer = ModelRegistryViewer(registry)
+ viewer._browser.selected_model = model
+
+ assert viewer.model == model
+
+ def test_model_param_cleared_on_deselection(self):
+ """Test that model param is cleared when selection is cleared."""
+ registry = ModelRegistry(name="test")
+ model = SimpleModel(name="test", value=42)
+ registry.add("my_model", model)
+
+ viewer = ModelRegistryViewer(registry)
+ viewer._browser.selected_model = model
+ assert viewer.model == model
+
+ viewer._browser.selected_model = None
+ assert viewer.model is None
+
+
+class TestIntegration:
+ """Integration tests for UI components."""
+
+ def test_full_workflow(self):
+ """Test complete workflow from registry to viewer."""
+ # Create a registry with nested structure
+ root = ModelRegistry(name="root")
+ models_reg = ModelRegistry(name="models")
+ configs_reg = ModelRegistry(name="configs")
+
+ model1 = SimpleModel(name="model1", value=1)
+ model2 = SimpleModel(name="model2", value=2)
+ config1 = AnotherModel(data="config1")
+
+ models_reg.add("first", model1)
+ models_reg.add("second", model2)
+ configs_reg.add("main", config1)
+
+ root.add("models", models_reg)
+ root.add("configs", configs_reg)
+
+ # Create viewer
+ viewer = ModelRegistryViewer(root)
+
+ # Verify browser has all paths indexed
+ browser = viewer._browser
+ assert "models/first" in browser._node_index
+ assert "models/second" in browser._node_index
+ assert "configs/main" in browser._node_index
+
+ # Simulate selecting a model
+ browser.selected_model = model1
+
+ # Verify viewer is updated
+ assert viewer._viewer.model == model1
+
+ def test_nested_registry_expansion(self):
+ """Test that nested paths generate correct expansion."""
+ root = ModelRegistry(name="root")
+ level1 = ModelRegistry(name="level1")
+ level2 = ModelRegistry(name="level2")
+ model = SimpleModel(name="deep", value=99)
+
+ level2.add("deep_model", model)
+ level1.add("level2", level2)
+ root.add("level1", level1)
+
+ browser = RegistryBrowser(root)
+
+ # Find the deep model node
+ node = browser._node_index["level1/level2/deep_model"]
+
+ # Check expansion path
+ expanded = browser._expanded_from_index_path(node["_index_path"])
+ assert len(expanded) == 2 # level1 and level1/level2
+
+ def test_switch_model_selection(self):
+ """Test switching between different models updates viewer correctly."""
+ root = ModelRegistry(name="root")
+ model1 = SimpleModel(name="first", value=1)
+ model2 = SimpleModel(name="second", value=2)
+
+ root.add("model1", model1)
+ root.add("model2", model2)
+
+ viewer = ModelRegistryViewer(root)
+
+ # Select first model
+ viewer._browser.selected_model = model1
+ assert viewer._viewer.model == model1
+ assert viewer._viewer._config_viewer.model == model1
+
+ # Switch to second model
+ viewer._browser.selected_model = model2
+ assert viewer._viewer.model == model2
+ assert viewer._viewer._config_viewer.model == model2
+
+ # Verify JSON editor shows second model's data
+ json_editors = find_components_by_type(viewer._viewer.__panel__(), pn.widgets.JSONEditor)
+ assert json_editors[0].value["name"] == "second"
+ assert json_editors[0].value["value"] == 2
+
+ def test_switch_between_different_model_types(self):
+ """Test switching between models of different types in the viewer."""
+ root = ModelRegistry(name="root")
+ simple_model = SimpleModel(name="simple", value=42)
+ another_model = AnotherModel(data="test data")
+
+ root.add("simple", simple_model)
+ root.add("another", another_model)
+
+ viewer = ModelRegistryViewer(root)
+
+ # Select SimpleModel
+ viewer._browser.selected_model = simple_model
+ assert viewer._viewer._type_viewer.model_type == SimpleModel
+
+ json_editors = find_components_by_type(viewer._viewer.__panel__(), pn.widgets.JSONEditor)
+ assert "name" in json_editors[0].value
+ assert "value" in json_editors[0].value
+
+ # Switch to AnotherModel
+ viewer._browser.selected_model = another_model
+ assert viewer._viewer._type_viewer.model_type == AnotherModel
+
+ # JSON should now show AnotherModel's fields
+ assert "data" in json_editors[0].value
+ assert "value" not in json_editors[0].value
+ assert json_editors[0].value["data"] == "test data"
+
+ def test_deselect_model_clears_viewer(self):
+ """Test that deselecting a model (selecting registry) clears the viewer."""
+ root = ModelRegistry(name="root")
+ sub = ModelRegistry(name="sub")
+ model = SimpleModel(name="test", value=1)
+
+ sub.add("model", model)
+ root.add("subregistry", sub)
+
+ viewer = ModelRegistryViewer(root)
+
+ # Select model first
+ viewer._browser.selected_model = model
+ assert viewer._viewer.model == model
+ assert viewer._viewer._json_container.visible is True
+
+ # Deselect (simulate selecting registry node which has no model)
+ viewer._browser.selected_model = None
+ assert viewer._viewer.model is None
+ assert viewer._viewer._json_container.visible is False
+
+ def test_tree_selection_updates_viewer(self):
+ """Test that tree selection properly updates the viewer through the wiring."""
+ root = ModelRegistry(name="root")
+ model1 = SimpleModel(name="first", value=1)
+ model2 = SimpleModel(name="second", value=2)
+
+ root.add("model1", model1)
+ root.add("model2", model2)
+
+ viewer = ModelRegistryViewer(root)
+ browser = viewer._browser
+
+ # Simulate tree selection of first model
+ event = mock.Mock()
+ event.new = [{"label": "model1", "model": model1}]
+ browser._on_tree_select(event)
+
+ assert browser.selected_model == model1
+ assert viewer._viewer.model == model1
+
+ # Simulate tree selection of second model
+ event.new = [{"label": "model2", "model": model2}]
+ browser._on_tree_select(event)
+
+ assert browser.selected_model == model2
+ assert viewer._viewer.model == model2
+
+ def test_search_then_switch_models(self):
+ """Test using search to select models and then switching."""
+ root = ModelRegistry(name="root")
+ model1 = SimpleModel(name="alpha", value=1)
+ model2 = SimpleModel(name="beta", value=2)
+
+ root.add("alpha_model", model1)
+ root.add("beta_model", model2)
+
+ viewer = ModelRegistryViewer(root)
+ browser = viewer._browser
+
+ # Search and select first model
+ event = mock.Mock()
+ event.new = "alpha_model"
+ browser._on_search_select(event)
+
+ # Tree should be updated
+ assert len(browser._tree.value) == 1
+ assert browser._tree.value[0]["label"] == "alpha_model"
+
+ # Simulate the tree select callback that would happen
+ tree_event = mock.Mock()
+ tree_event.new = browser._tree.value
+ browser._on_tree_select(tree_event)
+
+ assert viewer._viewer.model == model1
+
+ # Now search and select second model
+ event.new = "beta_model"
+ browser._on_search_select(event)
+
+ tree_event.new = browser._tree.value
+ browser._on_tree_select(tree_event)
+
+ assert viewer._viewer.model == model2
+
+ # Verify viewer shows second model
+ json_editors = find_components_by_type(viewer._viewer.__panel__(), pn.widgets.JSONEditor)
+ assert json_editors[0].value["name"] == "beta"
+
+ def test_rapid_model_switching(self):
+ """Test rapidly switching between multiple models."""
+ root = ModelRegistry(name="root")
+ models = [SimpleModel(name=f"model_{i}", value=i) for i in range(5)]
+
+ for i, model in enumerate(models):
+ root.add(f"model_{i}", model)
+
+ viewer = ModelRegistryViewer(root)
+
+ # Rapidly switch through all models
+ for i, model in enumerate(models):
+ viewer._browser.selected_model = model
+ assert viewer._viewer.model == model
+ assert viewer._viewer._config_viewer.model == model
+
+ json_editors = find_components_by_type(viewer._viewer.__panel__(), pn.widgets.JSONEditor)
+ assert json_editors[0].value["name"] == f"model_{i}"
+ assert json_editors[0].value["value"] == i
+
+ def test_switch_models_in_nested_registries(self):
+ """Test switching between models in different nested registries."""
+ root = ModelRegistry(name="root")
+ reg_a = ModelRegistry(name="reg_a")
+ reg_b = ModelRegistry(name="reg_b")
+
+ model_a = SimpleModel(name="in_a", value=100)
+ model_b = AnotherModel(data="in_b")
+
+ reg_a.add("model", model_a)
+ reg_b.add("model", model_b)
+ root.add("registry_a", reg_a)
+ root.add("registry_b", reg_b)
+
+ viewer = ModelRegistryViewer(root)
+
+ # Verify both paths are indexed
+ assert "registry_a/model" in viewer._browser._node_index
+ assert "registry_b/model" in viewer._browser._node_index
+
+ # Select model from registry_a
+ viewer._browser.selected_model = model_a
+ assert viewer._viewer._type_viewer.model_type == SimpleModel
+
+ # Switch to model from registry_b (different type)
+ viewer._browser.selected_model = model_b
+ assert viewer._viewer._type_viewer.model_type == AnotherModel
+
+ json_editors = find_components_by_type(viewer._viewer.__panel__(), pn.widgets.JSONEditor)
+ assert "data" in json_editors[0].value
+ assert "value" not in json_editors[0].value
diff --git a/ccflow/tests/ui/utils.py b/ccflow/tests/ui/utils.py
new file mode 100644
index 0000000..f1a4391
--- /dev/null
+++ b/ccflow/tests/ui/utils.py
@@ -0,0 +1,29 @@
+"""Utility functions for UI tests."""
+
+import panel as pn
+
+
+def find_components_by_type(layout, component_type):
+ """Recursively find all components of a given type in a Panel layout.
+
+ Args:
+ layout: A Panel layout or component to search
+ component_type: The type of component to find
+
+ Returns:
+ A list of all components matching the given type
+ """
+ found = []
+ if isinstance(layout, component_type):
+ found.append(layout)
+ if hasattr(layout, "objects"):
+ for obj in layout.objects:
+ found.extend(find_components_by_type(obj, component_type))
+ if hasattr(layout, "__iter__") and not isinstance(layout, str):
+ try:
+ for item in layout:
+ if hasattr(item, "objects") or isinstance(item, pn.viewable.Viewable):
+ found.extend(find_components_by_type(item, component_type))
+ except TypeError:
+ pass
+ return found
diff --git a/ccflow/tests/utils/test_hydra.py b/ccflow/tests/utils/test_hydra.py
index 06e2d9d..7c9886e 100644
--- a/ccflow/tests/utils/test_hydra.py
+++ b/ccflow/tests/utils/test_hydra.py
@@ -1,10 +1,196 @@
+import argparse
import sys
from pathlib import Path
+from unittest.mock import MagicMock
import pytest
from hydra import compose, initialize
-from ccflow.utils.hydra import get_args_parser_default, load_config
+from ccflow.utils.hydra import (
+ add_hydra_config_args,
+ add_panel_server_args,
+ get_args_parser_default,
+ load_config,
+ resolve_config_paths,
+)
+
+
+class TestAddHydraConfigArgs:
+ """Tests for add_hydra_config_args helper function."""
+
+ def test_adds_all_arguments(self):
+ """Test that all expected arguments are added."""
+ parser = argparse.ArgumentParser()
+ add_hydra_config_args(parser)
+ args = parser.parse_args([])
+
+ assert hasattr(args, "overrides")
+ assert hasattr(args, "config_path")
+ assert hasattr(args, "config_name")
+ assert hasattr(args, "config_dir")
+ assert hasattr(args, "config_dir_config_name")
+ assert hasattr(args, "basepath")
+
+ def test_default_values(self):
+ """Test default values for all arguments."""
+ parser = argparse.ArgumentParser()
+ add_hydra_config_args(parser)
+ args = parser.parse_args([])
+
+ assert args.overrides == []
+ assert args.config_path is None
+ assert args.config_name is None
+ assert args.config_dir is None
+ assert args.config_dir_config_name is None
+ assert args.basepath is None
+
+ def test_short_flags(self):
+ """Test short flag aliases work correctly."""
+ parser = argparse.ArgumentParser()
+ add_hydra_config_args(parser)
+ args = parser.parse_args(["-cp", "/path", "-cn", "name", "-cd", "/dir", "-cdcn", "dirname"])
+
+ assert args.config_path == "/path"
+ assert args.config_name == "name"
+ assert args.config_dir == "/dir"
+ assert args.config_dir_config_name == "dirname"
+
+ def test_overrides_positional(self):
+ """Test overrides are captured as positional arguments."""
+ parser = argparse.ArgumentParser()
+ add_hydra_config_args(parser)
+ args = parser.parse_args(["key1=value1", "key2=value2", "+group=option"])
+
+ assert args.overrides == ["key1=value1", "key2=value2", "+group=option"]
+
+
+class TestAddPanelServerArgs:
+ """Tests for add_panel_server_args helper function."""
+
+ def test_adds_all_arguments(self):
+ """Test that all expected arguments are added."""
+ parser = argparse.ArgumentParser()
+ add_panel_server_args(parser)
+ args = parser.parse_args([])
+
+ assert hasattr(args, "address")
+ assert hasattr(args, "port")
+ assert hasattr(args, "allow_websocket_origin")
+ assert hasattr(args, "basic_auth")
+ assert hasattr(args, "cookie_secret")
+ assert hasattr(args, "show")
+
+ def test_default_values(self):
+ """Test default values for all arguments."""
+ parser = argparse.ArgumentParser()
+ add_panel_server_args(parser)
+ args = parser.parse_args([])
+
+ assert args.address == "127.0.0.1"
+ assert args.port == 8080
+ assert args.allow_websocket_origin == ["*"]
+ assert args.basic_auth is None
+ assert args.cookie_secret == "secret"
+ assert args.show is False
+
+ def test_sets_epilog(self):
+ """Test that epilog is set on the parser."""
+ parser = argparse.ArgumentParser()
+ add_panel_server_args(parser)
+
+ assert parser.epilog is not None
+ assert "server" in parser.epilog.lower()
+
+ def test_custom_values(self):
+ """Test setting custom values for arguments."""
+ parser = argparse.ArgumentParser()
+ add_panel_server_args(parser)
+ args = parser.parse_args(
+ [
+ "--address",
+ "0.0.0.0",
+ "--port",
+ "9000",
+ "--basic-auth",
+ "user:pass",
+ "--cookie-secret",
+ "mysecret",
+ "--show",
+ ]
+ )
+
+ assert args.address == "0.0.0.0"
+ assert args.port == 9000
+ assert args.basic_auth == "user:pass"
+ assert args.cookie_secret == "mysecret"
+ assert args.show is True
+
+ def test_websocket_origin_multiple(self):
+ """Test multiple websocket origins."""
+ parser = argparse.ArgumentParser()
+ add_panel_server_args(parser)
+ args = parser.parse_args(["--allow-websocket-origin", "localhost:8080", "example.com"])
+
+ assert args.allow_websocket_origin == ["localhost:8080", "example.com"]
+
+
+class TestResolveConfigPaths:
+ """Tests for resolve_config_paths helper function."""
+
+ def test_uses_args_config_path(self):
+ """Test that args.config_path takes precedence."""
+ args = argparse.Namespace(config_path="/from/args", config_name="name")
+ root_dir, root_name = resolve_config_paths(args, config_path="/default", config_name="default")
+
+ assert root_dir == "/from/args"
+ assert root_name == "name"
+
+ def test_uses_args_config_name(self):
+ """Test that args.config_name takes precedence."""
+ args = argparse.Namespace(config_path="/path", config_name="from_args")
+ root_dir, root_name = resolve_config_paths(args, config_path="", config_name="default")
+
+ assert root_name == "from_args"
+
+ def test_falls_back_to_config_name_default(self):
+ """Test fallback to config_name parameter when args.config_name is None."""
+ args = argparse.Namespace(config_path="/path", config_name=None)
+ root_dir, root_name = resolve_config_paths(args, config_path="", config_name="default_name")
+
+ assert root_name == "default_name"
+
+ def test_raises_without_config_path(self):
+ """Test ValueError when no config_path available."""
+ args = argparse.Namespace(config_path=None, config_name="name")
+
+ with pytest.raises(ValueError, match="Must provide --config-path"):
+ resolve_config_paths(args, config_path="", config_name="name", hydra_main=None)
+
+ def test_raises_without_config_name(self):
+ """Test ValueError when no config_name available."""
+ args = argparse.Namespace(config_path="/path", config_name=None)
+
+ with pytest.raises(ValueError, match="Must provide --config-name"):
+ resolve_config_paths(args, config_path="", config_name="", hydra_main=None)
+
+ def test_uses_hydra_main_for_config_path(self):
+ """Test that hydra_main is used to resolve config_path."""
+ # Create a mock hydra_main with __wrapped__ attribute
+ mock_func = MagicMock()
+ mock_func.__wrapped__ = lambda: None
+ # Mock inspect.getfile to return a known path
+ import ccflow.utils.hydra as hydra_module
+
+ original_getfile = hydra_module.inspect.getfile
+
+ try:
+ hydra_module.inspect.getfile = lambda x: "/some/module/path.py"
+ args = argparse.Namespace(config_path=None, config_name="name")
+ root_dir, root_name = resolve_config_paths(args, config_path="config", config_name="name", hydra_main=mock_func)
+
+ assert root_dir == "/some/module/config"
+ finally:
+ hydra_module.inspect.getfile = original_getfile
@pytest.fixture
diff --git a/ccflow/ui/__init__.py b/ccflow/ui/__init__.py
new file mode 100644
index 0000000..417aeab
--- /dev/null
+++ b/ccflow/ui/__init__.py
@@ -0,0 +1,3 @@
+from .cli import *
+from .model import *
+from .registry import *
diff --git a/ccflow/ui/cli.py b/ccflow/ui/cli.py
new file mode 100644
index 0000000..a6b7d3f
--- /dev/null
+++ b/ccflow/ui/cli.py
@@ -0,0 +1,106 @@
+"""CLI for serving ModelRegistryViewer as a Panel application."""
+
+import argparse
+from typing import Callable, Optional
+
+import panel as pn
+
+from ccflow import ModelRegistry
+from ccflow.utils.hydra import add_hydra_config_args, add_panel_server_args, load_config, resolve_config_paths
+
+from .registry import ModelRegistryViewer
+
+__all__ = ("registry_viewer_cli",)
+
+
+def _get_ui_args_parser() -> argparse.ArgumentParser:
+ """Create argument parser with UI server configuration options."""
+ parser = argparse.ArgumentParser(
+ add_help=True,
+ description="Serve ModelRegistryViewer as a Panel application",
+ )
+
+ # Standard hydra config loading arguments
+ add_hydra_config_args(parser)
+
+ # Standard Panel server arguments
+ add_panel_server_args(parser)
+
+ # Viewer-specific arguments
+ parser.add_argument(
+ "--browser-width",
+ type=int,
+ default=400,
+ help="Width of the registry browser panel (default: 400)",
+ )
+ parser.add_argument(
+ "--browser-height",
+ type=int,
+ default=700,
+ help="Height of the registry browser panel (default: 700)",
+ )
+ parser.add_argument(
+ "--viewer-width",
+ type=int,
+ default=None,
+ help="Fixed width for model viewer panel (default: stretch)",
+ )
+
+ return parser
+
+
+def registry_viewer_cli(
+ config_path: str = "",
+ config_name: str = "",
+ hydra_main: Optional[Callable] = None,
+):
+ """CLI entry point for serving ModelRegistryViewer.
+
+ Parameters
+ ----------
+ config_path
+ The config_path specified in hydra.main()
+ config_name
+ The config_name specified in hydra.main()
+ hydra_main
+ The function decorated with hydra.main(). Used to resolve config_path
+ relative to the decorated function's file location.
+ """
+ parser = _get_ui_args_parser()
+ args = parser.parse_args()
+
+ # Resolve config paths using shared helper
+ root_config_dir, root_config_name = resolve_config_paths(args, config_path, config_name, hydra_main)
+
+ # Load config using hydra utilities
+ result = load_config(
+ root_config_dir=root_config_dir,
+ root_config_name=root_config_name,
+ config_dir=args.config_dir,
+ config_name=args.config_dir_config_name,
+ overrides=args.overrides,
+ basepath=args.basepath,
+ )
+
+ # Load registry from config
+ registry = ModelRegistry.root()
+ registry.load_config(cfg=result.cfg, overwrite=True)
+
+ # Create app factory for per-session instances
+ def create_app():
+ viewer = ModelRegistryViewer(
+ registry,
+ browser_width=args.browser_width,
+ browser_height=args.browser_height,
+ viewer_width=args.viewer_width,
+ )
+ return viewer.__panel__()
+
+ # Serve the panel app (callable = fresh instance per session)
+ pn.serve(
+ create_app,
+ address=args.address,
+ port=args.port,
+ allow_websocket_origin=args.allow_websocket_origin,
+ show=args.show,
+ )
diff --git a/ccflow/ui/model.py b/ccflow/ui/model.py
new file mode 100644
index 0000000..a631e3b
--- /dev/null
+++ b/ccflow/ui/model.py
@@ -0,0 +1,258 @@
+import html
+
+import panel as pn
+import panel_material_ui # noqa: F401 Must be imported like this to register the extension
+import panel_material_ui as pmui
+import param
+from pydantic._internal._repr import display_as_type
+
+import ccflow
+
+pn.extension()
+pn.extension("jsoneditor")
+
+
+__all__ = ("ModelTypeViewer", "ModelViewer", "ModelConfigViewer")
+
+
+class ModelTypeViewer(param.Parameterized):
+ """
+ Displays type name, class docstring, and fields for a Pydantic model type.
+ """
+
+ model_type = param.Parameter(default=None)
+
+ def __init__(self, **params):
+ super().__init__(**params)
+
+ self._pane = pn.pane.HTML("", width=1200)
+ self._layout = pn.Column(
+ self._pane,
+ )
+
+ self.param.watch(self._on_type_change, "model_type")
+
+ def __panel__(self):
+ return self._layout
+
+ def _on_type_change(self, event):
+ model_cls = event.new
+ if model_cls is None:
+ self._pane.object = ""
+ return
+
+ type_name = display_as_type(model_cls)
+
+ # Class documentation
+ docs = (model_cls.__doc__ or "").strip()
+ docs_html = ""
+ if docs:
+ escaped = html.escape(docs).replace("\n", "
")
+ docs_html = f"""
+
{escaped}
+ {html.escape(name)} ({html.escape(field_type)}){': ' + html.escape(desc) if desc else ''}{html.escape(type_name)}
+