fix: skip non-agent directories in AgentLoader.list_agents()
Merge https://github.com/google/adk-python/pull/6668 PiperOrigin-RevId: 966348937
This commit is contained in:
committed by
Copybara-Service
parent
989721746a
commit
caac070837
@@ -455,6 +455,17 @@ class AgentLoader(BaseAgentLoader):
|
||||
self._agent_cache[agent_name] = agent_or_app
|
||||
return agent_or_app
|
||||
|
||||
@staticmethod
|
||||
def _looks_like_agent_dir(dir_path: Path) -> bool:
|
||||
"""Returns True if the directory holds a loadable agent definition."""
|
||||
try:
|
||||
return (
|
||||
is_single_agent_directory(dir_path)
|
||||
or (dir_path / "__init__.py").is_file()
|
||||
)
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
@override
|
||||
def list_agents(self) -> list[str]:
|
||||
"""Lists all agents available in the agent loader (sorted alphabetically)."""
|
||||
@@ -467,6 +478,7 @@ class AgentLoader(BaseAgentLoader):
|
||||
if os.path.isdir(os.path.join(base_path, x))
|
||||
and not x.startswith(".")
|
||||
and x != "__pycache__"
|
||||
and self._looks_like_agent_dir(base_path / x)
|
||||
]
|
||||
agent_names.sort()
|
||||
return agent_names
|
||||
|
||||
@@ -287,6 +287,48 @@ class TestAgentLoader:
|
||||
assert agent2 is not agent3
|
||||
assert agent1.agent_id != agent2.agent_id != agent3.agent_id
|
||||
|
||||
def test_list_agents_skips_directories_without_a_loadable_agent(self):
|
||||
"""Stray non-agent directories under agents_dir must not be listed."""
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
temp_path = Path(temp_dir)
|
||||
|
||||
self.create_agent_structure(
|
||||
temp_path, "real_agent", "package_with_agent_module"
|
||||
)
|
||||
# A stray directory with no agent.py or root_agent.yaml,
|
||||
# e.g. one created as a side effect of local storage keyed by an
|
||||
# unmapped app name.
|
||||
(temp_path / "stray_dir").mkdir()
|
||||
|
||||
loader = AgentLoader(str(temp_path))
|
||||
|
||||
assert loader.list_agents() == ["real_agent"]
|
||||
|
||||
def test_list_agents_handles_permission_error(self):
|
||||
"""Permission errors on subdirectories must be handled gracefully."""
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
temp_path = Path(temp_dir)
|
||||
self.create_agent_structure(
|
||||
temp_path, "real_agent", "package_with_agent_module"
|
||||
)
|
||||
|
||||
# Create a directory that we will simulate permission error on
|
||||
no_perm_dir = temp_path / "no_perm_dir"
|
||||
no_perm_dir.mkdir()
|
||||
|
||||
# We want to mock is_file for paths under no_perm_dir to raise PermissionError
|
||||
original_is_file = Path.is_file
|
||||
|
||||
def mock_is_file(self_path):
|
||||
if no_perm_dir in self_path.parents or self_path == no_perm_dir:
|
||||
raise PermissionError("[Errno 13] Permission denied")
|
||||
return original_is_file(self_path)
|
||||
|
||||
loader = AgentLoader(str(temp_path))
|
||||
|
||||
with mock.patch.object(Path, "is_file", mock_is_file):
|
||||
assert loader.list_agents() == ["real_agent"]
|
||||
|
||||
def test_error_messages_use_os_sep_consistently(self):
|
||||
"""Verify error messages use os.sep instead of hardcoded '/'."""
|
||||
del self
|
||||
|
||||
Reference in New Issue
Block a user