Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
@@ -0,0 +1,83 @@
from copy import deepcopy

from lfx.components.models_and_agents.policies.guard_sync_utils import (
GENERATED_GUARD_INFO_PREFIX,
sync_generated_guard_code_inputs,
)


def _generated_field(name: str, value: str) -> dict:
return {
"type": "code",
"dynamic": True,
"info": f"{GENERATED_GUARD_INFO_PREFIX}{name}",
"value": value,
}


def test_sync_preserves_generated_fields_when_step2_directory_is_missing(tmp_path):
build_config = {
"result.json": _generated_field("result.json", "persisted result"),
"test_project/guard.py": _generated_field("test_project/guard.py", "persisted guard"),
"project": {"type": "str", "value": "test_project"},
}
original_config = deepcopy(build_config)

result = sync_generated_guard_code_inputs(
build_config=build_config,
work_dir=tmp_path / "missing_project",
step2_subdir="Step_2",
project_name="test_project",
)

assert result is build_config
assert result == original_config


def test_sync_preserves_generated_fields_when_step2_path_is_not_a_directory(tmp_path):
work_dir = tmp_path / "test_project"
work_dir.mkdir()
(work_dir / "Step_2").write_text("not a directory", encoding="utf-8")
build_config = {
"result.json": _generated_field("result.json", "persisted result"),
"test_project/guard.py": _generated_field("test_project/guard.py", "persisted guard"),
}
original_config = deepcopy(build_config)

result = sync_generated_guard_code_inputs(
build_config=build_config,
work_dir=work_dir,
step2_subdir="Step_2",
project_name="test_project",
)

assert result == original_config


def test_sync_reconciles_generated_fields_when_step2_directory_exists(tmp_path):
work_dir = tmp_path / "test_project"
step2_dir = work_dir / "Step_2"
project_dir = step2_dir / "test_project"
project_dir.mkdir(parents=True)
(step2_dir / "result.json").write_text("fresh result", encoding="utf-8")
(project_dir / "guard.py").write_text("fresh guard", encoding="utf-8")
(step2_dir / "ignored.py").write_text("ignored", encoding="utf-8")

build_config = {
"result.json": _generated_field("result.json", "stale result"),
"test_project/stale.py": _generated_field("test_project/stale.py", "stale guard"),
"project": {"type": "str", "value": "test_project"},
}

result = sync_generated_guard_code_inputs(
build_config=build_config,
work_dir=work_dir,
step2_subdir="Step_2",
project_name="test_project",
)

assert result["result.json"]["value"] == "fresh result"
assert result["test_project/guard.py"]["value"] == "fresh guard"
assert "test_project/stale.py" not in result
assert "ignored.py" not in result
assert result["project"] == {"type": "str", "value": "test_project"}
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
from unittest.mock import MagicMock, patch

import pytest
from lfx.components.models_and_agents.policies.guard_sync_utils import GENERATED_GUARD_INFO_PREFIX
from lfx.components.models_and_agents.policies.module_utils import ensure_toolguard_module_path_compat
from lfx.components.models_and_agents.policies_component import (
MODE_GENERATE,
Expand All @@ -11,6 +12,8 @@
PoliciesComponent,
)

from tests.base import ComponentTestBaseWithoutClient


@pytest.fixture
def mock_tool():
Expand Down Expand Up @@ -61,6 +64,69 @@ def _make_fake_tg(**overrides):
return fake


def _generated_guard_field(name: str, value: str) -> dict:
return {
"type": "code",
"dynamic": True,
"info": f"{GENERATED_GUARD_INFO_PREFIX}{name}",
"value": value,
}


class TestPoliciesComponent(ComponentTestBaseWithoutClient):
@pytest.fixture
def component_class(self):
return PoliciesComponent

@pytest.fixture
def default_kwargs(self):
return {"enabled": False, "in_tools": []}

@pytest.fixture
def file_names_mapping(self):
return []

async def test_component_update_preserves_only_generated_guard_fields(self, component_class, default_kwargs):
"""A Breaking component update must retain guard code stored in the flow."""
component = await self.component_setup(component_class, default_kwargs)
result_field = _generated_guard_field("result.json", "persisted result")
guard_field = _generated_guard_field("test_project/guard.py", "persisted guard")
current_frontend_node = {
"display_name": "Policies",
"template": {
"code": {"type": "code", "value": "old component code"},
"project": {"type": "str", "value": "test_project"},
"removed_input": {"type": "str", "value": "do not preserve"},
"dynamic_lookalike": {
"type": "code",
"dynamic": True,
"info": "User-defined dynamic code",
"value": "do not preserve",
},
"result.json": result_field,
"test_project/guard.py": guard_field,
},
}
new_frontend_node = {
"display_name": "Policies",
"template": {
"code": {"type": "code", "value": "new component code"},
"project": {"type": "str", "value": "my_project"},
"new_input": {"type": "str", "value": "new default"},
},
}

with patch.object(PoliciesComponent, "_import_toolguard", side_effect=ImportError) as mock_import:
result = await component.update_frontend_node(new_frontend_node, current_frontend_node)

mock_import.assert_not_called()
assert result["template"]["result.json"] == result_field
assert result["template"]["test_project/guard.py"] == guard_field
assert "removed_input" not in result["template"]
assert "dynamic_lookalike" not in result["template"]
assert result["template"]["new_input"]["value"] == "new default"


@pytest.mark.asyncio
async def test_guard_tools_blocked_when_custom_components_disabled(mock_component, monkeypatch):
"""ToolGuard guard execution must be refused under allow_custom_components=False.
Expand Down
Loading
Loading