mirror of
https://github.com/MervinPraison/PraisonAI.git
synced 2026-11-07 17:33:33 -10:00
fixup: security (0.0.0.0→127.0.0.1), ImportError guards, orphan chainlit cleanup
Addresses PR #1525 reviewer feedback (CodeRabbit + Gemini + @claude review): Security + error handling (cli/main.py): - Bind realtime UI to 127.0.0.1 instead of 0.0.0.0 (2 sites) - Wrap both _launch_aiui_app calls in try/except ImportError with install hint (restores the guard the original CHAINLIT_AVAILABLE check used to provide) Orphan chainlit file cleanup: - rm ui/colab.py (module-level 'import chainlit' — callers deleted) - rm -r chat/ (chat/__init__.py wraps run_chainlit; chat/app.py gone) - rm ui/_auth.py (chainlit password_auth_callback — callers deleted) - rm ui/_pairing.py (chainlit pairing banner — callers deleted) - Trim ui/_external_agents.py: drop chainlit_switches, load_external_agent_settings_from_chainlit, save_external_agent_settings_to_chainlit (keep aiui helpers) Stale test cleanup: - rm tests/unit/ui/test_ui_bind_aware_creds.py (tested deleted _auth.py) - rm tests/integration/ui/test_ui_pairing_approval.py (tested deleted _pairing.py) - rm tests/unit/test_ui_external_agents_helper.py (tested deleted helper) - rm tests/integration/test_ui_external_agents.py (tested deleted helper) - Strip obsolete test_chainlit_app_config from test_tools_and_ui.py Cosmetic: - serve.py help text: 'Chainlit web interface' → 'aiui web interface' Validation without chainlit installed: - pip uninstall chainlit - 44/44 unit tests pass (cli/test_chat_flag + test_serve_unified + test_tools_and_ui) - All CLI imports clean Remaining chainlit refs are in docstrings/comments only (historical context) + the intentionally-kept --ui chainlit deprecation alias.
This commit is contained in:
Cascade Bot committed
1 parent
648291e3e3
commit
682a9486fd
12 files changed
+19
-1785
No files matched your search
@@ -1,106 +0,0 @@
|
||||
"""
|
||||
PraisonAI Chat Module
|
||||
|
||||
This module provides the chat UI integration for PraisonAI agents.
|
||||
It uses the PraisonAI Chat (based on Chainlit) for the frontend.
|
||||
"""
|
||||
|
||||
from typing import Optional, Any
|
||||
|
||||
__all__ = ["start_chat_server", "ChatConfig"]
|
||||
|
||||
|
||||
class ChatConfig:
|
||||
"""Configuration for the PraisonAI Chat server."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
host: str = "0.0.0.0",
|
||||
port: int = 8000,
|
||||
debug: bool = False,
|
||||
auth_enabled: bool = False,
|
||||
session_id: Optional[str] = None,
|
||||
):
|
||||
self.host = host
|
||||
self.port = port
|
||||
self.debug = debug
|
||||
self.auth_enabled = auth_enabled
|
||||
self.session_id = session_id
|
||||
|
||||
|
||||
def start_chat_server(
|
||||
agent: Optional[Any] = None,
|
||||
agents: Optional[list] = None,
|
||||
config: Optional[ChatConfig] = None,
|
||||
port: int = 8000,
|
||||
host: str = "0.0.0.0",
|
||||
debug: bool = False,
|
||||
) -> None:
|
||||
"""
|
||||
Start the PraisonAI Chat server.
|
||||
|
||||
Args:
|
||||
agent: A single PraisonAI agent to use in the chat.
|
||||
agents: A list of PraisonAI agents for multi-agent chat.
|
||||
config: ChatConfig object with server settings.
|
||||
port: Port to run the server on (default: 8000).
|
||||
host: Host to bind to (default: 0.0.0.0).
|
||||
debug: Enable debug mode (default: False).
|
||||
|
||||
Example:
|
||||
>>> from praisonaiagents import Agent
|
||||
>>> from praisonai.chat import start_chat_server
|
||||
>>>
|
||||
>>> agent = Agent(name="Assistant", instructions="You are helpful.")
|
||||
>>> start_chat_server(agent=agent, port=8000)
|
||||
"""
|
||||
# Lazy import to avoid loading chainlit unless needed
|
||||
try:
|
||||
from chainlit.cli import run_chainlit
|
||||
except ImportError:
|
||||
raise ImportError(
|
||||
"PraisonAI Chat requires the 'praisonai-chat' package. "
|
||||
"Install it with: pip install praisonai-chat"
|
||||
)
|
||||
|
||||
if config is None:
|
||||
config = ChatConfig(host=host, port=port, debug=debug)
|
||||
|
||||
# Store agents in a way accessible to the chainlit app
|
||||
import os
|
||||
os.environ["PRAISONAI_CHAT_MODE"] = "true"
|
||||
|
||||
if agent is not None:
|
||||
# Single agent mode
|
||||
_register_agent(agent)
|
||||
elif agents is not None:
|
||||
# Multi-agent mode
|
||||
for a in agents:
|
||||
_register_agent(a)
|
||||
|
||||
# Start the chainlit server
|
||||
from pathlib import Path
|
||||
|
||||
# Use the built-in app file
|
||||
app_file = Path(__file__).parent / "app.py"
|
||||
|
||||
# Set environment variables for chainlit
|
||||
os.environ["CHAINLIT_HOST"] = config.host
|
||||
os.environ["CHAINLIT_PORT"] = str(config.port)
|
||||
|
||||
run_chainlit(str(app_file))
|
||||
|
||||
|
||||
# Global registry for agents
|
||||
_REGISTERED_AGENTS: dict = {}
|
||||
|
||||
|
||||
def _register_agent(agent: Any) -> None:
|
||||
"""Register an agent for use in the chat UI."""
|
||||
agent_name = getattr(agent, "name", None) or getattr(agent, "role", "Agent")
|
||||
_REGISTERED_AGENTS[agent_name] = agent
|
||||
|
||||
|
||||
def get_registered_agents() -> dict:
|
||||
"""Get all registered agents."""
|
||||
return _REGISTERED_AGENTS
|
||||
@@ -10,7 +10,7 @@ Server Types:
|
||||
- mcp: MCP server (Model Context Protocol)
|
||||
- acp: ACP server (Agent Client Protocol for IDEs)
|
||||
- lsp: Language Server Protocol
|
||||
- ui: Web UI (Chainlit)
|
||||
- ui: Web UI (aiui)
|
||||
- rag: RAG query server
|
||||
- registry: Package registry server
|
||||
- docs: Documentation preview server
|
||||
@@ -116,7 +116,7 @@ Start any PraisonAI server with: praisonai serve <type>
|
||||
[green]mcp[/green] MCP server for Claude/Cursor (port 8080)
|
||||
[green]acp[/green] Agent Client Protocol for IDEs (STDIO)
|
||||
[green]lsp[/green] Language Server Protocol (STDIO)
|
||||
[green]ui[/green] Chainlit web interface (port 8082)
|
||||
[green]ui[/green] aiui web interface (port 8082)
|
||||
[green]rag[/green] RAG query server (port 9000)
|
||||
[green]registry[/green] Package registry server (port 7777)
|
||||
[green]docs[/green] Documentation preview (port 3000)
|
||||
@@ -351,7 +351,7 @@ def serve_ui(
|
||||
port: int = typer.Option(8082, "--port", "-p", help="Port to bind to"),
|
||||
ui_type: str = typer.Option("agents", "--type", "-t", help="UI type: agents, chat, code, realtime"),
|
||||
):
|
||||
"""Start Chainlit web UI server.
|
||||
"""Start aiui web UI server.
|
||||
|
||||
Examples:
|
||||
praisonai serve ui
|
||||
|
||||
@@ -557,11 +557,15 @@ class PraisonAI:
|
||||
return
|
||||
|
||||
# chat and code commands are now terminal-native (handled by Typer commands)
|
||||
# They no longer open Chainlit browser UI
|
||||
|
||||
if getattr(args, 'realtime', False):
|
||||
from praisonai.cli.commands.ui import _launch_aiui_app
|
||||
_launch_aiui_app("ui_realtime", "ui_realtime", 8085, "0.0.0.0", None, False, "Realtime Voice")
|
||||
try:
|
||||
from praisonai.cli.commands.ui import _launch_aiui_app
|
||||
_launch_aiui_app("ui_realtime", "ui_realtime", 8085, "127.0.0.1", None, False, "Realtime Voice")
|
||||
except ImportError:
|
||||
print("\033[91mERROR: Realtime UI is not installed.\033[0m")
|
||||
print('Install with: pip install "praisonai[ui]"')
|
||||
sys.exit(1)
|
||||
return
|
||||
|
||||
if getattr(args, 'call', False):
|
||||
@@ -1218,8 +1222,13 @@ class PraisonAI:
|
||||
sys.exit(0)
|
||||
|
||||
elif args.command == 'realtime':
|
||||
from praisonai.cli.commands.ui import _launch_aiui_app
|
||||
_launch_aiui_app("ui_realtime", "ui_realtime", 8085, "0.0.0.0", None, False, "Realtime Voice")
|
||||
try:
|
||||
from praisonai.cli.commands.ui import _launch_aiui_app
|
||||
_launch_aiui_app("ui_realtime", "ui_realtime", 8085, "127.0.0.1", None, False, "Realtime Voice")
|
||||
except ImportError:
|
||||
print("\033[91mERROR: Realtime UI is not installed.\033[0m")
|
||||
print('Install with: pip install "praisonai[ui]"')
|
||||
sys.exit(1)
|
||||
sys.exit(0)
|
||||
|
||||
elif args.command == 'train':
|
||||
|
||||
@@ -1,80 +0,0 @@
|
||||
"""
|
||||
Chainlit UI authentication helper.
|
||||
|
||||
Provides bind-aware auth for Chainlit apps - consolidates duplicated
|
||||
password auth callbacks across UI modules.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import logging
|
||||
from typing import Optional
|
||||
|
||||
from praisonaiagents.gateway.protocols import AuthMode, is_loopback, resolve_auth_mode
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class UIStartupError(Exception):
|
||||
"""Raised when UI cannot start due to security configuration."""
|
||||
pass
|
||||
|
||||
|
||||
def register_password_auth(app, *, bind_host: str) -> None:
|
||||
"""Register password authentication for Chainlit app with bind-aware security.
|
||||
|
||||
Args:
|
||||
app: Chainlit app instance (unused but kept for consistency)
|
||||
bind_host: Host/IP that the UI server is bound to
|
||||
|
||||
Raises:
|
||||
UIStartupError: If using default credentials on external interface
|
||||
"""
|
||||
import chainlit as cl # lazy — chainlit is in [ui] extra only
|
||||
# Get credentials from environment
|
||||
expected_username = os.getenv("CHAINLIT_USERNAME", "admin")
|
||||
expected_password = os.getenv("CHAINLIT_PASSWORD", "admin")
|
||||
|
||||
# Check if using default credentials
|
||||
using_defaults = (expected_username == "admin" and expected_password == "admin")
|
||||
|
||||
# Resolve auth mode based on bind interface
|
||||
auth_mode = resolve_auth_mode(bind_host)
|
||||
|
||||
# Allow default credentials only on loopback
|
||||
if using_defaults and auth_mode != "local":
|
||||
# Check for escape hatch
|
||||
allow_defaults = os.getenv("PRAISONAI_ALLOW_DEFAULT_CREDS", "").lower() in ("1", "true", "yes")
|
||||
if not allow_defaults:
|
||||
raise UIStartupError(
|
||||
f"Cannot bind to {bind_host} with default admin/admin credentials.\n"
|
||||
f"Fix: export CHAINLIT_USERNAME=myuser CHAINLIT_PASSWORD=mypass\n"
|
||||
f"Lab: export PRAISONAI_ALLOW_DEFAULT_CREDS=1 (demo only)"
|
||||
)
|
||||
else:
|
||||
logger.warning(
|
||||
f"⚠️ Using default admin/admin credentials on external interface {bind_host}. "
|
||||
f"This is UNSAFE for production. Set CHAINLIT_USERNAME and CHAINLIT_PASSWORD."
|
||||
)
|
||||
elif using_defaults and auth_mode == "local":
|
||||
logger.warning(
|
||||
f"⚠️ Using default admin/admin credentials on loopback {bind_host}. "
|
||||
f"Set CHAINLIT_USERNAME and CHAINLIT_PASSWORD environment variables for production."
|
||||
)
|
||||
|
||||
# Register the password auth callback
|
||||
@cl.password_auth_callback
|
||||
def auth_callback(username: str, password: str) -> Optional["cl.User"]:
|
||||
"""Password authentication callback."""
|
||||
logger.debug("Auth attempt received")
|
||||
|
||||
if (username, password) == (expected_username, expected_password):
|
||||
logger.info("Login successful")
|
||||
return cl.User(identifier=username, metadata={"role": "admin", "provider": "credentials"})
|
||||
else:
|
||||
logger.warning("Login failed")
|
||||
return None
|
||||
|
||||
# Log the registration
|
||||
auth_status = "permissive" if auth_mode == "local" else "strict"
|
||||
logger.info(f"Registered password auth for {bind_host} (mode: {auth_status})")
|
||||
@@ -2,12 +2,12 @@
|
||||
|
||||
Single source of truth for:
|
||||
- Listing installed external agents (lazy, cached)
|
||||
- Rendering Chainlit Switch widgets / aiui settings entries
|
||||
- Rendering aiui settings entries
|
||||
- Building the tools list from enabled agents
|
||||
"""
|
||||
|
||||
from functools import lru_cache
|
||||
from typing import Any, Callable, Dict, List, Optional
|
||||
from typing import Any, Dict, List
|
||||
|
||||
# Map of UI toggle id → (integration class path, pretty label)
|
||||
EXTERNAL_AGENTS: Dict[str, Dict[str, str]] = {
|
||||
@@ -54,16 +54,6 @@ def external_agent_tools(settings: Dict[str, Any], workspace: str = ".") -> list
|
||||
return tools
|
||||
|
||||
|
||||
def chainlit_switches(current_settings: Dict[str, bool]):
|
||||
"""Return Chainlit Switch widgets for installed external agents only."""
|
||||
from chainlit.input_widget import Switch
|
||||
return [
|
||||
Switch(id=toggle_id, label=EXTERNAL_AGENTS[toggle_id]["label"],
|
||||
initial=current_settings.get(toggle_id, False))
|
||||
for toggle_id in installed_external_agents()
|
||||
]
|
||||
|
||||
|
||||
def aiui_settings_entries() -> Dict[str, Any]:
|
||||
"""Return aiui settings entries for installed external agents."""
|
||||
settings = {}
|
||||
@@ -83,59 +73,3 @@ def _parse_setting_bool(value: Any) -> bool:
|
||||
if value is None:
|
||||
return False
|
||||
return str(value).strip().lower() in {"true", "1", "yes", "on"}
|
||||
|
||||
|
||||
def load_external_agent_settings_from_chainlit(
|
||||
load_setting_fn: Optional[Callable[[str], Any]] = None,
|
||||
) -> Dict[str, bool]:
|
||||
"""Load external agent settings from Chainlit session and persistent storage.
|
||||
|
||||
Args:
|
||||
load_setting_fn: Optional callback used to load persisted settings by key.
|
||||
If omitted, falls back to importing ``praisonai.ui.chat.load_setting``
|
||||
for backward compatibility.
|
||||
"""
|
||||
import chainlit as cl
|
||||
settings = {toggle_id: False for toggle_id in EXTERNAL_AGENTS}
|
||||
loader = load_setting_fn
|
||||
|
||||
# Try to load from persistent storage (if load_setting is available)
|
||||
if loader is None:
|
||||
try:
|
||||
# Backward-compatible fallback for callers that don't pass a loader
|
||||
from praisonai.ui.chat import load_setting as loader # type: ignore
|
||||
except ImportError:
|
||||
loader = None
|
||||
|
||||
if loader is not None:
|
||||
legacy_claude = _parse_setting_bool(loader("claude_code_enabled"))
|
||||
|
||||
# Load all current toggles from persistent storage first
|
||||
for toggle_id in EXTERNAL_AGENTS:
|
||||
persistent_value = loader(toggle_id)
|
||||
if persistent_value: # non-empty string means explicitly stored
|
||||
settings[toggle_id] = _parse_setting_bool(persistent_value)
|
||||
|
||||
# Apply legacy migration only where no explicit value was stored
|
||||
if legacy_claude and not loader("claude_enabled"):
|
||||
settings["claude_enabled"] = True
|
||||
|
||||
# Load from session (may override persistent settings)
|
||||
# Check for legacy key in session
|
||||
if _parse_setting_bool(cl.user_session.get("claude_code_enabled", False)):
|
||||
settings["claude_enabled"] = True
|
||||
|
||||
# Load all current toggles from session
|
||||
for toggle_id in EXTERNAL_AGENTS:
|
||||
session_value = cl.user_session.get(toggle_id)
|
||||
if session_value is not None:
|
||||
settings[toggle_id] = _parse_setting_bool(session_value)
|
||||
|
||||
return settings
|
||||
|
||||
|
||||
def save_external_agent_settings_to_chainlit(settings: Dict[str, bool]):
|
||||
"""Save external agent settings to Chainlit session."""
|
||||
import chainlit as cl
|
||||
for toggle_id, enabled in settings.items():
|
||||
cl.user_session.set(toggle_id, enabled)
|
||||
@@ -1,264 +0,0 @@
|
||||
"""
|
||||
Chainlit UI components for pairing approval banner.
|
||||
|
||||
Provides admin banner functionality for approving pending pairing requests.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# Gateway client configuration
|
||||
GATEWAY_HOST = os.environ.get("GATEWAY_HOST", "127.0.0.1")
|
||||
GATEWAY_PORT = int(os.environ.get("GATEWAY_PORT", "8765"))
|
||||
GATEWAY_TOKEN = os.environ.get("GATEWAY_AUTH_TOKEN", "")
|
||||
|
||||
|
||||
async def get_pending_pairings() -> List[Dict]:
|
||||
"""Fetch pending pairing requests from gateway API."""
|
||||
if not GATEWAY_TOKEN:
|
||||
logger.warning("No GATEWAY_AUTH_TOKEN set, cannot fetch pending pairings")
|
||||
return []
|
||||
|
||||
try:
|
||||
import aiohttp
|
||||
|
||||
url = f"http://{GATEWAY_HOST}:{GATEWAY_PORT}/api/pairing/pending"
|
||||
headers = {"Authorization": f"Bearer {GATEWAY_TOKEN}"}
|
||||
|
||||
timeout = aiohttp.ClientTimeout(total=5)
|
||||
async with aiohttp.ClientSession(timeout=timeout) as session:
|
||||
async with session.get(url, headers=headers) as resp:
|
||||
if resp.status == 200:
|
||||
data = await resp.json()
|
||||
return data.get("pending", [])
|
||||
else:
|
||||
logger.warning(f"Failed to fetch pending pairings: {resp.status}")
|
||||
return []
|
||||
except Exception as e:
|
||||
logger.error(f"Error fetching pending pairings: {e}")
|
||||
return []
|
||||
|
||||
|
||||
async def approve_pairing(channel: str, code: str) -> bool:
|
||||
"""Approve a pairing request via gateway API."""
|
||||
if not GATEWAY_TOKEN:
|
||||
logger.warning("No GATEWAY_AUTH_TOKEN set, cannot approve pairing")
|
||||
return False
|
||||
|
||||
try:
|
||||
import aiohttp
|
||||
|
||||
url = f"http://{GATEWAY_HOST}:{GATEWAY_PORT}/api/pairing/approve"
|
||||
headers = {
|
||||
"Authorization": f"Bearer {GATEWAY_TOKEN}",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
data = {"channel": channel, "code": code}
|
||||
|
||||
timeout = aiohttp.ClientTimeout(total=5)
|
||||
async with aiohttp.ClientSession(timeout=timeout) as session:
|
||||
async with session.post(url, headers=headers, json=data) as resp:
|
||||
success = resp.status == 200
|
||||
if not success:
|
||||
logger.warning(f"Failed to approve pairing: {resp.status}")
|
||||
return success
|
||||
except Exception as e:
|
||||
logger.error(f"Error approving pairing: {e}")
|
||||
return False
|
||||
|
||||
|
||||
async def deny_pairing(channel: str, code: str) -> bool:
|
||||
"""Deny a pairing request (logs denial - code will expire naturally)."""
|
||||
# Note: Currently there's no dedicated deny endpoint in the gateway
|
||||
# The pending request will expire naturally based on TTL
|
||||
# This is a UI-only action for admin feedback
|
||||
logger.info(f"Admin denied pairing request: {channel} code {code}")
|
||||
return True # Always succeed for UI purposes
|
||||
|
||||
|
||||
async def refresh_pending_banner():
|
||||
"""Display or update the pending pairing banner for admin users."""
|
||||
import chainlit as cl
|
||||
|
||||
# Check if user is admin
|
||||
user = cl.user_session.get("user")
|
||||
if not user or user.metadata.get("role") != "admin":
|
||||
return
|
||||
|
||||
pending = await get_pending_pairings()
|
||||
|
||||
if not pending:
|
||||
# No pending requests - don't show banner
|
||||
return
|
||||
|
||||
# Create approval actions for each pending request
|
||||
actions = []
|
||||
for p in pending:
|
||||
channel = p.get("channel", "unknown")
|
||||
code = p.get("code", "")
|
||||
user_name = p.get("user_name", f"User {code}")
|
||||
age = p.get("age_seconds", 0)
|
||||
|
||||
# Format age nicely
|
||||
if age < 60:
|
||||
age_str = f"{age}s ago"
|
||||
elif age < 3600:
|
||||
age_str = f"{age//60}m ago"
|
||||
else:
|
||||
age_str = f"{age//3600}h ago"
|
||||
|
||||
actions.append(
|
||||
cl.Action(
|
||||
name="approve_pairing",
|
||||
value=f"{channel}:{code}",
|
||||
label=f"✅ Approve {user_name} ({channel}) - {age_str}",
|
||||
description=f"Approve pairing request from {user_name} on {channel}"
|
||||
)
|
||||
)
|
||||
actions.append(
|
||||
cl.Action(
|
||||
name="deny_pairing",
|
||||
value=f"{channel}:{code}",
|
||||
label=f"❌ Deny {user_name} ({channel})",
|
||||
description=f"Deny pairing request from {user_name} on {channel}",
|
||||
)
|
||||
)
|
||||
|
||||
# Display or update banner message with actions
|
||||
banner_content = f"🔔 **{len(pending)} pending pairing request(s)**\n\nClick to approve:"
|
||||
|
||||
# Check if we already have a banner message stored in the session
|
||||
banner_msg_id = cl.user_session.get("pending_banner_id")
|
||||
|
||||
if banner_msg_id:
|
||||
# Try to update existing message
|
||||
try:
|
||||
# Get the existing message and update it
|
||||
# Note: Chainlit's update functionality depends on the specific version
|
||||
# For now, send a new message and store its ID
|
||||
msg = await cl.Message(
|
||||
content=banner_content,
|
||||
actions=actions,
|
||||
author="System"
|
||||
).send()
|
||||
cl.user_session.set("pending_banner_id", msg.id)
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to update banner message: {e}")
|
||||
# Fall back to sending new message
|
||||
msg = await cl.Message(
|
||||
content=banner_content,
|
||||
actions=actions,
|
||||
author="System"
|
||||
).send()
|
||||
cl.user_session.set("pending_banner_id", msg.id)
|
||||
else:
|
||||
# Send new banner message and store its ID
|
||||
msg = await cl.Message(
|
||||
content=banner_content,
|
||||
actions=actions,
|
||||
author="System"
|
||||
).send()
|
||||
cl.user_session.set("pending_banner_id", msg.id)
|
||||
|
||||
|
||||
async def on_approve_pairing(action):
|
||||
"""Handle approval action from banner."""
|
||||
import chainlit as cl
|
||||
|
||||
try:
|
||||
# Parse channel:code from action value
|
||||
channel, code = action.value.split(":", 1)
|
||||
|
||||
# Show loading message
|
||||
await cl.Message(
|
||||
content=f"⏳ Approving pairing for {channel} code {code}...",
|
||||
author="System"
|
||||
).send()
|
||||
|
||||
# Approve the pairing
|
||||
success = await approve_pairing(channel, code)
|
||||
|
||||
if success:
|
||||
await cl.Message(
|
||||
content=f"✅ Successfully approved pairing for {channel} code {code}",
|
||||
author="System"
|
||||
).send()
|
||||
else:
|
||||
await cl.Message(
|
||||
content=f"❌ Failed to approve pairing for {channel} code {code}",
|
||||
author="System"
|
||||
).send()
|
||||
|
||||
# Refresh the banner to update count
|
||||
await refresh_pending_banner()
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error in approval handler: {e}")
|
||||
await cl.Message(
|
||||
content=f"❌ Error processing approval: {str(e)}",
|
||||
author="System"
|
||||
).send()
|
||||
|
||||
|
||||
async def on_deny_pairing(action):
|
||||
"""Handle denial action from banner."""
|
||||
import chainlit as cl
|
||||
|
||||
try:
|
||||
# Parse channel:code from action value
|
||||
channel, code = action.value.split(":", 1)
|
||||
|
||||
# Show loading message
|
||||
await cl.Message(
|
||||
content=f"⏳ Denying pairing for {channel} code {code}...",
|
||||
author="System"
|
||||
).send()
|
||||
|
||||
# Deny the pairing
|
||||
success = await deny_pairing(channel, code)
|
||||
|
||||
if success:
|
||||
await cl.Message(
|
||||
content=f"✅ Successfully denied pairing for {channel} code {code}",
|
||||
author="System"
|
||||
).send()
|
||||
else:
|
||||
await cl.Message(
|
||||
content=f"❌ Failed to deny pairing for {channel} code {code}",
|
||||
author="System"
|
||||
).send()
|
||||
|
||||
# Refresh the banner to update count
|
||||
await refresh_pending_banner()
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error in denial handler: {e}")
|
||||
await cl.Message(
|
||||
content=f"❌ Error processing denial: {str(e)}",
|
||||
author="System"
|
||||
).send()
|
||||
|
||||
|
||||
def setup_pairing_callbacks():
|
||||
"""Setup pairing action callbacks (call this at module level after chainlit import)."""
|
||||
import chainlit as cl
|
||||
|
||||
@cl.action_callback("approve_pairing")
|
||||
async def _approve_callback(action):
|
||||
await on_approve_pairing(action)
|
||||
|
||||
@cl.action_callback("deny_pairing")
|
||||
async def _deny_callback(action):
|
||||
await on_deny_pairing(action)
|
||||
|
||||
|
||||
async def setup_pairing_banner():
|
||||
"""Setup pairing banner on chat start (call this from @cl.on_chat_start)."""
|
||||
await refresh_pending_banner()
|
||||
@@ -1,474 +0,0 @@
|
||||
from praisonaiagents import Agent, Task, AgentTeam
|
||||
import os
|
||||
import importlib
|
||||
import inspect
|
||||
import yaml
|
||||
import logging
|
||||
from .callbacks import trigger_callback
|
||||
import asyncio
|
||||
import chainlit as cl
|
||||
from queue import Queue
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
agent_file = "agents.yaml"
|
||||
|
||||
with open(agent_file, 'r') as f:
|
||||
config = yaml.safe_load(f)
|
||||
|
||||
topic = "get from the message content from the chainlit user message"
|
||||
|
||||
# Create a message queue
|
||||
message_queue = Queue()
|
||||
|
||||
async def process_message_queue():
|
||||
"""Process messages in the queue and send them to Chainlit"""
|
||||
while True:
|
||||
try:
|
||||
if not message_queue.empty():
|
||||
msg_data = message_queue.get()
|
||||
await cl.Message(**msg_data).send()
|
||||
await asyncio.sleep(0.1) # Small delay to prevent busy waiting
|
||||
except Exception as e:
|
||||
logger.error(f"Error processing message queue: {e}")
|
||||
|
||||
def load_tools_from_tools_py():
|
||||
"""
|
||||
Imports and returns all contents from tools.py file.
|
||||
Also adds the tools to the global namespace.
|
||||
|
||||
Returns:
|
||||
list: A list of callable functions with proper formatting
|
||||
"""
|
||||
tools_list = []
|
||||
try:
|
||||
# Try to import tools.py from current directory
|
||||
spec = importlib.util.spec_from_file_location("tools", "tools.py")
|
||||
logger.info(f"Spec: {spec}")
|
||||
if spec is None:
|
||||
logger.info("tools.py not found in current directory")
|
||||
return tools_list
|
||||
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
|
||||
# Get all module attributes except private ones and classes
|
||||
for name, obj in inspect.getmembers(module):
|
||||
if (not name.startswith('_') and
|
||||
callable(obj) and
|
||||
not inspect.isclass(obj)):
|
||||
# Add the function to global namespace
|
||||
globals()[name] = obj
|
||||
# Format the tool as an OpenAI function
|
||||
tool = {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": name,
|
||||
"description": obj.__doc__ or f"Function to {name.replace('_', ' ')}",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"query": {
|
||||
"type": "string",
|
||||
"description": "The search query to look up information about"
|
||||
}
|
||||
},
|
||||
"required": ["query"]
|
||||
}
|
||||
}
|
||||
}
|
||||
# Add formatted tool to tools list
|
||||
tools_list.append(tool)
|
||||
logger.info(f"Loaded and globalized tool function: {name}")
|
||||
|
||||
logger.info(f"Loaded {len(tools_list)} tool functions from tools.py")
|
||||
logger.info(f"Tools list: {tools_list}")
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"Error loading tools from tools.py: {e}")
|
||||
|
||||
return tools_list
|
||||
|
||||
async def step_callback(step_details):
|
||||
"""Callback for agent steps"""
|
||||
logger.info(f"[CALLBACK DEBUG] Step callback triggered with details: {step_details}")
|
||||
try:
|
||||
# Queue message for agent response
|
||||
if step_details.get("response"):
|
||||
message_queue.put({
|
||||
"content": f"Agent Response: {step_details.get('response')}",
|
||||
"author": step_details.get("agent_name", "Agent")
|
||||
})
|
||||
logger.info("[CALLBACK DEBUG] Queued agent response message")
|
||||
|
||||
# Queue message for tool usage
|
||||
if step_details.get("tool_name"):
|
||||
message_queue.put({
|
||||
"content": f"🛠️ Using tool: {step_details.get('tool_name')}",
|
||||
"author": "System"
|
||||
})
|
||||
logger.info("[CALLBACK DEBUG] Queued tool usage message")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"[CALLBACK DEBUG] Error in step callback: {str(e)}", exc_info=True)
|
||||
|
||||
async def task_callback(task_output):
|
||||
"""Callback for task completion"""
|
||||
logger.info(f"[CALLBACK DEBUG] Task callback triggered with output: {task_output}")
|
||||
try:
|
||||
# Create message content
|
||||
if hasattr(task_output, 'raw'):
|
||||
content = task_output.raw
|
||||
elif hasattr(task_output, 'content'):
|
||||
content = task_output.content
|
||||
else:
|
||||
content = str(task_output)
|
||||
|
||||
# Queue the message
|
||||
message_queue.put({
|
||||
"content": f"Task Output: {content}",
|
||||
"author": "Task"
|
||||
})
|
||||
logger.info("[CALLBACK DEBUG] Queued task completion message")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"[CALLBACK DEBUG] Error in task callback: {str(e)}", exc_info=True)
|
||||
|
||||
async def step_callback_wrapper(step_details):
|
||||
logger.info(f"[CALLBACK DEBUG] Step callback wrapper triggered with details: {step_details}")
|
||||
try:
|
||||
# Check if we have a Chainlit context
|
||||
if not cl.context.context_var.get():
|
||||
logger.warning("[CALLBACK DEBUG] No Chainlit context available in wrapper")
|
||||
return
|
||||
logger.info("[CALLBACK DEBUG] Chainlit context found in wrapper")
|
||||
|
||||
# Create a message for the agent's response
|
||||
if step_details.get("response"):
|
||||
logger.info(f"[CALLBACK DEBUG] Sending agent response from wrapper: {step_details.get('response')}")
|
||||
try:
|
||||
await cl.Message(
|
||||
content=f"{role_name}: {step_details.get('response')}",
|
||||
author=role_name,
|
||||
).send()
|
||||
logger.info("[CALLBACK DEBUG] Successfully sent agent response message from wrapper")
|
||||
except Exception as e:
|
||||
logger.error(f"[CALLBACK DEBUG] Error sending agent response message from wrapper: {str(e)}")
|
||||
|
||||
# Create a message for any tool usage
|
||||
if step_details.get("tool_name"):
|
||||
logger.info(f"[CALLBACK DEBUG] Sending tool usage from wrapper: {step_details.get('tool_name')}")
|
||||
try:
|
||||
await cl.Message(
|
||||
content=f"🛠️ {role_name} is using tool: {step_details.get('tool_name')}",
|
||||
author="System",
|
||||
).send()
|
||||
logger.info("[CALLBACK DEBUG] Successfully sent tool usage message from wrapper")
|
||||
except Exception as e:
|
||||
logger.error(f"[CALLBACK DEBUG] Error sending tool usage message from wrapper: {str(e)}")
|
||||
|
||||
# Create a message for any thoughts or reasoning
|
||||
if step_details.get("thought"):
|
||||
logger.info(f"[CALLBACK DEBUG] Sending thought from wrapper: {step_details.get('thought')}")
|
||||
try:
|
||||
await cl.Message(
|
||||
content=f"💭 {role_name}'s thought: {step_details.get('thought')}",
|
||||
author=role_name,
|
||||
).send()
|
||||
logger.info("[CALLBACK DEBUG] Successfully sent thought message from wrapper")
|
||||
except Exception as e:
|
||||
logger.error(f"[CALLBACK DEBUG] Error sending thought message from wrapper: {str(e)}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"[CALLBACK DEBUG] Error in step callback wrapper: {str(e)}", exc_info=True)
|
||||
try:
|
||||
await cl.Message(
|
||||
content=f"Error in step callback: {str(e)}",
|
||||
author="System",
|
||||
).send()
|
||||
except Exception as send_error:
|
||||
logger.error(f"[CALLBACK DEBUG] Error sending error message: {str(send_error)}")
|
||||
|
||||
async def task_callback_wrapper(task_output):
|
||||
logger.info(f"[CALLBACK DEBUG] Task callback wrapper triggered with output type: {type(task_output)}")
|
||||
try:
|
||||
# Check if we have a Chainlit context
|
||||
if not cl.context.context_var.get():
|
||||
logger.warning("[CALLBACK DEBUG] No Chainlit context available in task wrapper")
|
||||
return
|
||||
logger.info("[CALLBACK DEBUG] Chainlit context found in task wrapper")
|
||||
|
||||
# Create a message for task completion
|
||||
if hasattr(task_output, 'raw'):
|
||||
content = task_output.raw
|
||||
logger.info("[CALLBACK DEBUG] Using raw output")
|
||||
elif hasattr(task_output, 'content'):
|
||||
content = task_output.content
|
||||
logger.info("[CALLBACK DEBUG] Using content output")
|
||||
else:
|
||||
content = str(task_output)
|
||||
logger.info("[CALLBACK DEBUG] Using string representation of output")
|
||||
|
||||
logger.info(f"[CALLBACK DEBUG] Sending task completion message from wrapper: {content[:100]}...")
|
||||
try:
|
||||
await cl.Message(
|
||||
content=f"✅ {role_name} completed task:\n{content}",
|
||||
author=role_name,
|
||||
).send()
|
||||
logger.info("[CALLBACK DEBUG] Successfully sent task completion message from wrapper")
|
||||
except Exception as e:
|
||||
logger.error(f"[CALLBACK DEBUG] Error sending task completion message from wrapper: {str(e)}")
|
||||
|
||||
# If there are any additional task details
|
||||
if hasattr(task_output, 'details'):
|
||||
logger.info("[CALLBACK DEBUG] Task has additional details")
|
||||
try:
|
||||
await cl.Message(
|
||||
content=f"📝 Additional details:\n{task_output.details}",
|
||||
author=role_name,
|
||||
).send()
|
||||
logger.info("[CALLBACK DEBUG] Successfully sent additional details message")
|
||||
except Exception as e:
|
||||
logger.error(f"[CALLBACK DEBUG] Error sending additional details message: {str(e)}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"[CALLBACK DEBUG] Error in task callback wrapper: {str(e)}", exc_info=True)
|
||||
try:
|
||||
await cl.Message(
|
||||
content=f"Error in task callback: {str(e)}",
|
||||
author="System",
|
||||
).send()
|
||||
except Exception as send_error:
|
||||
logger.error(f"[CALLBACK DEBUG] Error sending error message: {str(send_error)}")
|
||||
|
||||
def sync_task_callback_wrapper(task_output):
|
||||
logger.info("[CALLBACK DEBUG] Sync task callback wrapper triggered")
|
||||
try:
|
||||
# Create a new event loop for this thread if there isn't one
|
||||
try:
|
||||
loop = asyncio.get_event_loop()
|
||||
logger.info("[CALLBACK DEBUG] Got existing event loop")
|
||||
except RuntimeError:
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
logger.info("[CALLBACK DEBUG] Created new event loop")
|
||||
|
||||
if loop.is_running():
|
||||
# If loop is running, schedule the callback
|
||||
logger.info("[CALLBACK DEBUG] Loop is running, scheduling callback")
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
task_callback_wrapper(task_output),
|
||||
loop
|
||||
)
|
||||
else:
|
||||
# If loop is not running, run it directly
|
||||
logger.info("[CALLBACK DEBUG] Loop is not running, running callback directly")
|
||||
loop.run_until_complete(task_callback_wrapper(task_output))
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"[CALLBACK DEBUG] Error in sync task callback: {str(e)}", exc_info=True)
|
||||
|
||||
def sync_step_callback_wrapper(step_details):
|
||||
logger.info("[CALLBACK DEBUG] Sync step callback wrapper triggered")
|
||||
try:
|
||||
# Create a new event loop for this thread if there isn't one
|
||||
try:
|
||||
loop = asyncio.get_event_loop()
|
||||
logger.info("[CALLBACK DEBUG] Got existing event loop")
|
||||
except RuntimeError:
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
logger.info("[CALLBACK DEBUG] Created new event loop")
|
||||
|
||||
if loop.is_running():
|
||||
# If loop is running, schedule the callback
|
||||
logger.info("[CALLBACK DEBUG] Loop is running, scheduling callback")
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
step_callback_wrapper(step_details),
|
||||
loop
|
||||
)
|
||||
else:
|
||||
# If loop is not running, run it directly
|
||||
logger.info("[CALLBACK DEBUG] Loop is not running, running callback directly")
|
||||
loop.run_until_complete(step_callback_wrapper(step_details))
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"[CALLBACK DEBUG] Error in sync step callback: {str(e)}", exc_info=True)
|
||||
|
||||
async def ui_run_praisonai(config, topic, tools_dict):
|
||||
"""Run PraisonAI with the given configuration and topic."""
|
||||
logger = logging.getLogger(__name__)
|
||||
logger.setLevel(logging.DEBUG)
|
||||
agents = {}
|
||||
tasks = []
|
||||
tasks_dict = {}
|
||||
|
||||
try:
|
||||
# Start message queue processor
|
||||
queue_processor = asyncio.create_task(process_message_queue())
|
||||
|
||||
# Create agents for each role
|
||||
for role, details in config['roles'].items():
|
||||
# Format the role name and other details
|
||||
role_name = details.get('name', role).format(topic=topic)
|
||||
role_filled = details.get('role', role).format(topic=topic)
|
||||
goal_filled = details['goal'].format(topic=topic)
|
||||
backstory_filled = details['backstory'].format(topic=topic)
|
||||
|
||||
# Test message to verify Chainlit is working
|
||||
await cl.Message(
|
||||
content=f"[DEBUG] Creating agent: {role_name}",
|
||||
author="System"
|
||||
).send()
|
||||
|
||||
# Create a sync wrapper for the step callback
|
||||
def step_callback_sync(step_details):
|
||||
try:
|
||||
# Create a new event loop for this thread
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
|
||||
# Add agent name to step details
|
||||
step_details["agent_name"] = role_name
|
||||
|
||||
# Run the callback
|
||||
loop.run_until_complete(step_callback(step_details))
|
||||
loop.close()
|
||||
except Exception as e:
|
||||
logger.error(f"[CALLBACK DEBUG] Error in step callback: {str(e)}", exc_info=True)
|
||||
|
||||
agent = Agent(
|
||||
name=role_name,
|
||||
role=role_filled,
|
||||
goal=goal_filled,
|
||||
backstory=backstory_filled,
|
||||
llm=details.get('llm', 'gpt-5-nano'),
|
||||
verbose=True,
|
||||
allow_delegation=details.get('allow_delegation', False),
|
||||
max_iter=details.get('max_iter', 15),
|
||||
max_rpm=details.get('max_rpm'),
|
||||
max_execution_time=details.get('max_execution_time'),
|
||||
cache=details.get('cache', True),
|
||||
step_callback=step_callback_sync
|
||||
)
|
||||
agents[role] = agent
|
||||
|
||||
# Create tasks for each role
|
||||
for role, details in config['roles'].items():
|
||||
agent = agents[role]
|
||||
tools_list = []
|
||||
|
||||
# Get tools for this role
|
||||
for tool_name in details.get('tools', []):
|
||||
if tool_name in tools_dict:
|
||||
tool_func = tools_dict[tool_name]
|
||||
tools_list.append(tool_func)
|
||||
|
||||
# Create tasks for the agent
|
||||
for task_name, task_details in details.get('tasks', {}).items():
|
||||
description_filled = task_details['description'].format(topic=topic)
|
||||
expected_output_filled = task_details['expected_output'].format(topic=topic)
|
||||
|
||||
# Test message to verify task creation
|
||||
await cl.Message(
|
||||
content=f"[DEBUG] Created task: {task_name} for agent {role_name}",
|
||||
author="System"
|
||||
).send()
|
||||
|
||||
# Create a sync wrapper for the task callback
|
||||
def task_callback_sync(task_output):
|
||||
try:
|
||||
# Create a new event loop for this thread
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
|
||||
# Run the callback
|
||||
loop.run_until_complete(task_callback(task_output))
|
||||
loop.close()
|
||||
except Exception as e:
|
||||
logger.error(f"[CALLBACK DEBUG] Error in task callback: {str(e)}", exc_info=True)
|
||||
|
||||
task = Task(
|
||||
description=description_filled,
|
||||
expected_output=expected_output_filled,
|
||||
agent=agent,
|
||||
tools=tools_list,
|
||||
async_execution=True,
|
||||
context=[],
|
||||
config=task_details.get('config', {}),
|
||||
output_json=task_details.get('output_json'),
|
||||
output_pydantic=task_details.get('output_pydantic'),
|
||||
output_file=task_details.get('output_file', ""),
|
||||
callback=task_callback_sync,
|
||||
create_directory=task_details.get('create_directory', False)
|
||||
)
|
||||
|
||||
tasks.append(task)
|
||||
tasks_dict[task_name] = task
|
||||
|
||||
# Set up task contexts
|
||||
for role, details in config['roles'].items():
|
||||
for task_name, task_details in details.get('tasks', {}).items():
|
||||
task = tasks_dict[task_name]
|
||||
context_tasks = [tasks_dict[ctx] for ctx in task_details.get('context', [])
|
||||
if ctx in tasks_dict]
|
||||
task.context = context_tasks
|
||||
|
||||
# Send the start message
|
||||
await cl.Message(
|
||||
content="Starting PraisonAI agents execution...",
|
||||
author="System"
|
||||
).send()
|
||||
|
||||
# Create and run the PraisonAI agents
|
||||
if config.get('process') == 'hierarchical':
|
||||
crew = AgentTeam(
|
||||
agents=list(agents.values()),
|
||||
tasks=tasks,
|
||||
verbose=True,
|
||||
process="hierarchical",
|
||||
manager_llm=config.get('manager_llm', 'gpt-5-nano')
|
||||
)
|
||||
else:
|
||||
crew = AgentTeam(
|
||||
agents=list(agents.values()),
|
||||
tasks=tasks,
|
||||
verbose=2
|
||||
)
|
||||
|
||||
# Store the crew in the user session
|
||||
cl.user_session.set("crew", crew)
|
||||
|
||||
# Run the agents in a separate thread
|
||||
loop = asyncio.get_event_loop()
|
||||
response = await loop.run_in_executor(None, crew.start)
|
||||
|
||||
logger.debug(f"[CALLBACK DEBUG] Result: {response}")
|
||||
|
||||
# Convert response to string if it's not already
|
||||
if hasattr(response, 'raw'):
|
||||
result = response.raw
|
||||
elif hasattr(response, 'content'):
|
||||
result = response.content
|
||||
else:
|
||||
result = str(response)
|
||||
|
||||
# Send the completion message
|
||||
await cl.Message(
|
||||
content="PraisonAI agents execution completed.",
|
||||
author="System"
|
||||
).send()
|
||||
|
||||
# After getting the response, wait a bit for remaining messages
|
||||
await asyncio.sleep(1) # Give time for final messages to be processed
|
||||
queue_processor.cancel() # Stop the queue processor
|
||||
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"Error in ui_run_praisonai: {str(e)}"
|
||||
logger.error(error_msg, exc_info=True)
|
||||
await cl.Message(
|
||||
content=error_msg,
|
||||
author="System"
|
||||
).send()
|
||||
raise
|
||||
@@ -1,196 +0,0 @@
|
||||
"""
|
||||
Integration tests for external agents UI functionality.
|
||||
|
||||
Tests the shared helper and UI integration for external agents (Claude/Gemini/Codex/Cursor).
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import shutil
|
||||
from unittest.mock import patch, MagicMock
|
||||
import sys
|
||||
import os
|
||||
|
||||
# Mark most tests as unit tests to bypass provider gating
|
||||
# Only TestRealAgenticIntegration is truly integration-level
|
||||
pytestmark = pytest.mark.unit
|
||||
|
||||
# Add src path for imports
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "../../"))
|
||||
|
||||
from praisonai.ui._external_agents import (
|
||||
EXTERNAL_AGENTS,
|
||||
installed_external_agents,
|
||||
external_agent_tools,
|
||||
chainlit_switches,
|
||||
aiui_settings_entries,
|
||||
)
|
||||
|
||||
|
||||
class TestExternalAgentsHelper:
|
||||
"""Test the shared external agents helper."""
|
||||
|
||||
def test_external_agents_config(self):
|
||||
"""Test that EXTERNAL_AGENTS is properly configured."""
|
||||
assert isinstance(EXTERNAL_AGENTS, dict)
|
||||
assert "claude_enabled" in EXTERNAL_AGENTS
|
||||
assert "gemini_enabled" in EXTERNAL_AGENTS
|
||||
assert "codex_enabled" in EXTERNAL_AGENTS
|
||||
assert "cursor_enabled" in EXTERNAL_AGENTS
|
||||
|
||||
# Check structure of each agent config
|
||||
for toggle_id, config in EXTERNAL_AGENTS.items():
|
||||
assert "module" in config
|
||||
assert "cls" in config
|
||||
assert "label" in config
|
||||
assert "cli" in config
|
||||
|
||||
@patch('shutil.which')
|
||||
def test_installed_external_agents_caching(self, mock_which):
|
||||
"""Test that installed_external_agents properly caches results."""
|
||||
# Clear cache
|
||||
from praisonai.ui._external_agents import installed_external_agents
|
||||
installed_external_agents.cache_clear()
|
||||
|
||||
# Mock some CLIs as available
|
||||
def mock_which_side_effect(cli):
|
||||
return "/usr/bin/claude" if cli == "claude" else None
|
||||
|
||||
mock_which.side_effect = mock_which_side_effect
|
||||
|
||||
# First call should check all CLIs
|
||||
result1 = installed_external_agents()
|
||||
assert result1 == ["claude_enabled"]
|
||||
assert mock_which.call_count == 4 # Called for each CLI
|
||||
|
||||
# Second call should use cache
|
||||
mock_which.reset_mock()
|
||||
result2 = installed_external_agents()
|
||||
assert result2 == ["claude_enabled"]
|
||||
assert mock_which.call_count == 0 # Not called due to cache
|
||||
|
||||
@patch('importlib.import_module')
|
||||
def test_external_agent_tools(self, mock_import):
|
||||
"""Test external_agent_tools builds tools correctly."""
|
||||
# Mock integration classes
|
||||
mock_integration = MagicMock()
|
||||
mock_integration.is_available = True
|
||||
mock_tool_func = MagicMock()
|
||||
mock_integration.as_tool.return_value = mock_tool_func
|
||||
|
||||
mock_module = MagicMock()
|
||||
mock_module.ClaudeCodeIntegration.return_value = mock_integration
|
||||
mock_import.return_value = mock_module
|
||||
|
||||
settings = {"claude_enabled": True, "gemini_enabled": False}
|
||||
tools = external_agent_tools(settings)
|
||||
|
||||
assert len(tools) == 1
|
||||
assert tools[0] == mock_tool_func
|
||||
mock_import.assert_called_once_with("praisonai.integrations.claude_code")
|
||||
mock_integration.as_tool.assert_called_once()
|
||||
|
||||
def test_external_agent_tools_unavailable_integration(self):
|
||||
"""Test external_agent_tools handles missing integrations gracefully."""
|
||||
settings = {"nonexistent_enabled": True}
|
||||
tools = external_agent_tools(settings)
|
||||
assert tools == []
|
||||
|
||||
@patch('chainlit.input_widget.Switch')
|
||||
@patch('praisonai.ui._external_agents.installed_external_agents')
|
||||
def test_chainlit_switches(self, mock_installed, mock_switch):
|
||||
"""Test chainlit_switches generates correct switches."""
|
||||
mock_installed.return_value = ["claude_enabled", "gemini_enabled"]
|
||||
mock_switch_instance = MagicMock()
|
||||
mock_switch.return_value = mock_switch_instance
|
||||
|
||||
current_settings = {"claude_enabled": True, "gemini_enabled": False}
|
||||
switches = chainlit_switches(current_settings)
|
||||
|
||||
assert len(switches) == 2
|
||||
assert all(s == mock_switch_instance for s in switches)
|
||||
assert mock_switch.call_count == 2
|
||||
|
||||
@patch('praisonai.ui._external_agents.installed_external_agents')
|
||||
def test_aiui_settings_entries(self, mock_installed):
|
||||
"""Test aiui_settings_entries generates correct settings."""
|
||||
mock_installed.return_value = ["claude_enabled", "gemini_enabled"]
|
||||
|
||||
settings = aiui_settings_entries()
|
||||
|
||||
assert isinstance(settings, dict)
|
||||
assert "claude_enabled" in settings
|
||||
assert "gemini_enabled" in settings
|
||||
assert settings["claude_enabled"]["type"] == "checkbox"
|
||||
assert settings["claude_enabled"]["default"] is False
|
||||
|
||||
|
||||
class TestBackwardCompatibility:
|
||||
"""Test backward compatibility with legacy settings."""
|
||||
|
||||
def test_claude_code_enabled_legacy_support(self):
|
||||
"""Test that claude_code_enabled is mapped to claude_enabled."""
|
||||
# This would require mocking chainlit session which is complex
|
||||
# In practice, this is tested by the load_external_agent_settings_from_chainlit function
|
||||
# which handles the legacy mapping
|
||||
pass
|
||||
|
||||
|
||||
class TestAvailabilityGating:
|
||||
"""Test that unavailable CLIs are properly hidden."""
|
||||
|
||||
@patch('shutil.which')
|
||||
def test_only_available_agents_shown(self, mock_which):
|
||||
"""Test that only available external agents are shown in UI."""
|
||||
# Clear cache
|
||||
from praisonai.ui._external_agents import installed_external_agents
|
||||
installed_external_agents.cache_clear()
|
||||
|
||||
# Mock only Claude as available
|
||||
mock_which.side_effect = lambda cli: "/usr/bin/claude" if cli == "claude" else None
|
||||
|
||||
available = installed_external_agents()
|
||||
assert available == ["claude_enabled"]
|
||||
|
||||
# Verify all CLIs were checked
|
||||
expected_calls = [cli_config["cli"] for cli_config in EXTERNAL_AGENTS.values()]
|
||||
actual_calls = [call[0][0] for call in mock_which.call_args_list]
|
||||
assert set(actual_calls) == set(expected_calls)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.provider_google # Only this class needs provider gating
|
||||
class TestRealAgenticIntegration:
|
||||
"""Real agentic tests - requires actual integrations to be available."""
|
||||
|
||||
@pytest.mark.skipif(not shutil.which("echo"), reason="echo command not available")
|
||||
def test_external_agent_tools_real_execution(self):
|
||||
"""Test that external agent tools can be created and are callable."""
|
||||
# This is a simplified test using echo command to simulate external CLI
|
||||
# In a real test environment with Claude Code installed, this would test actual execution
|
||||
|
||||
settings = {"claude_enabled": False} # Disable to avoid actual execution
|
||||
tools = external_agent_tools(settings)
|
||||
|
||||
# Should return empty list when all disabled
|
||||
assert tools == []
|
||||
|
||||
# Test with mock available integration would go here if Claude Code was installed
|
||||
|
||||
def test_settings_to_tools_mapping_consistency(self):
|
||||
"""Test that settings keys map consistently to integration modules."""
|
||||
for toggle_id, config in EXTERNAL_AGENTS.items():
|
||||
# Verify the toggle_id pattern is consistent
|
||||
assert toggle_id.endswith("_enabled")
|
||||
|
||||
# Verify module name is valid identifier
|
||||
module_name = config["module"]
|
||||
assert module_name.replace("_", "").isalnum()
|
||||
|
||||
# Verify class name follows PascalCase Integration pattern
|
||||
class_name = config["cls"]
|
||||
assert class_name.endswith("Integration")
|
||||
assert class_name[0].isupper()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
@@ -1,364 +0,0 @@
|
||||
"""
|
||||
Integration tests for UI pairing approval.
|
||||
|
||||
Tests the end-to-end flow: pending request → banner shown → approve action → PairingStore.is_approved()==True
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import asyncio
|
||||
import tempfile
|
||||
import os
|
||||
from unittest.mock import Mock, AsyncMock, patch
|
||||
|
||||
# Skip if chainlit is not available (optional [ui] extra)
|
||||
pytest.importorskip("chainlit", reason="chainlit is an optional [ui] extra")
|
||||
|
||||
from praisonai.gateway.pairing import PairingStore
|
||||
from praisonai.ui._pairing import (
|
||||
get_pending_pairings,
|
||||
approve_pairing,
|
||||
refresh_pending_banner,
|
||||
setup_pairing_banner,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_pairing_store():
|
||||
"""Create a temporary PairingStore for testing."""
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
store = PairingStore(store_dir=temp_dir)
|
||||
yield store
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_gateway_server(temp_pairing_store):
|
||||
"""Mock gateway server with actual PairingStore."""
|
||||
server = Mock()
|
||||
server.pairing_store = temp_pairing_store
|
||||
return server
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_chainlit_user_admin():
|
||||
"""Mock Chainlit user session with admin role."""
|
||||
with patch("chainlit.user_session") as mock_session:
|
||||
mock_user = Mock()
|
||||
mock_user.metadata = {"role": "admin"}
|
||||
mock_session.get.return_value = mock_user
|
||||
yield mock_session
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_chainlit_user_regular():
|
||||
"""Mock Chainlit user session with regular role."""
|
||||
with patch("chainlit.user_session") as mock_session:
|
||||
mock_user = Mock()
|
||||
mock_user.metadata = {"role": "user"}
|
||||
mock_session.get.return_value = mock_user
|
||||
yield mock_session
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_chainlit_message():
|
||||
"""Mock Chainlit Message class."""
|
||||
with patch("chainlit.Message") as mock_message_class:
|
||||
mock_message = Mock()
|
||||
mock_message.send = AsyncMock()
|
||||
mock_message_class.return_value = mock_message
|
||||
yield mock_message_class
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_chainlit_action():
|
||||
"""Mock Chainlit Action class."""
|
||||
with patch("chainlit.Action") as mock_action_class:
|
||||
yield mock_action_class
|
||||
|
||||
|
||||
class TestGetPendingPairings:
|
||||
"""Test the get_pending_pairings function."""
|
||||
|
||||
@patch("praisonai.ui._pairing.GATEWAY_TOKEN", "test-token")
|
||||
@patch("aiohttp.ClientSession")
|
||||
async def test_get_pending_success(self, mock_session):
|
||||
"""Test successful retrieval of pending pairings."""
|
||||
# Mock successful HTTP response
|
||||
mock_response = AsyncMock()
|
||||
mock_response.status = 200
|
||||
mock_response.json = AsyncMock(return_value={
|
||||
"pending": [
|
||||
{
|
||||
"channel": "telegram",
|
||||
"code": "ABCD1234",
|
||||
"user_id": "ABCD1234",
|
||||
"user_name": "User ABCD1234",
|
||||
"age_seconds": 30
|
||||
}
|
||||
]
|
||||
})
|
||||
|
||||
mock_session_instance = AsyncMock()
|
||||
mock_session_instance.__aenter__ = AsyncMock(return_value=mock_session_instance)
|
||||
mock_session_instance.__aexit__ = AsyncMock(return_value=None)
|
||||
mock_session_instance.get = AsyncMock()
|
||||
mock_session_instance.get.return_value.__aenter__ = AsyncMock(return_value=mock_response)
|
||||
mock_session_instance.get.return_value.__aexit__ = AsyncMock(return_value=None)
|
||||
mock_session.return_value = mock_session_instance
|
||||
|
||||
result = await get_pending_pairings()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0]["channel"] == "telegram"
|
||||
assert result[0]["code"] == "ABCD1234"
|
||||
|
||||
@patch("praisonai.ui._pairing.GATEWAY_TOKEN", "")
|
||||
async def test_get_pending_no_token(self, caplog):
|
||||
"""Test get_pending_pairings with no token."""
|
||||
result = await get_pending_pairings()
|
||||
|
||||
assert result == []
|
||||
assert "No GATEWAY_AUTH_TOKEN set" in caplog.text
|
||||
|
||||
@patch("praisonai.ui._pairing.GATEWAY_TOKEN", "test-token")
|
||||
@patch("aiohttp.ClientSession")
|
||||
async def test_get_pending_http_error(self, mock_session):
|
||||
"""Test get_pending_pairings with HTTP error."""
|
||||
# Mock HTTP error response
|
||||
mock_response = AsyncMock()
|
||||
mock_response.status = 401
|
||||
|
||||
mock_session_instance = AsyncMock()
|
||||
mock_session_instance.__aenter__ = AsyncMock(return_value=mock_session_instance)
|
||||
mock_session_instance.__aexit__ = AsyncMock(return_value=None)
|
||||
mock_session_instance.get = AsyncMock()
|
||||
mock_session_instance.get.return_value.__aenter__ = AsyncMock(return_value=mock_response)
|
||||
mock_session_instance.get.return_value.__aexit__ = AsyncMock(return_value=None)
|
||||
mock_session.return_value = mock_session_instance
|
||||
|
||||
result = await get_pending_pairings()
|
||||
|
||||
assert result == []
|
||||
|
||||
|
||||
class TestApprovePairing:
|
||||
"""Test the approve_pairing function."""
|
||||
|
||||
@patch("praisonai.ui._pairing.GATEWAY_TOKEN", "test-token")
|
||||
@patch("aiohttp.ClientSession")
|
||||
async def test_approve_success(self, mock_session):
|
||||
"""Test successful pairing approval."""
|
||||
# Mock successful HTTP response
|
||||
mock_response = AsyncMock()
|
||||
mock_response.status = 200
|
||||
|
||||
mock_session_instance = AsyncMock()
|
||||
mock_session_instance.__aenter__ = AsyncMock(return_value=mock_session_instance)
|
||||
mock_session_instance.__aexit__ = AsyncMock(return_value=None)
|
||||
mock_session_instance.post = AsyncMock()
|
||||
mock_session_instance.post.return_value.__aenter__ = AsyncMock(return_value=mock_response)
|
||||
mock_session_instance.post.return_value.__aexit__ = AsyncMock(return_value=None)
|
||||
mock_session.return_value = mock_session_instance
|
||||
|
||||
result = await approve_pairing("telegram", "ABCD1234")
|
||||
|
||||
assert result is True
|
||||
|
||||
@patch("praisonai.ui._pairing.GATEWAY_TOKEN", "")
|
||||
async def test_approve_no_token(self):
|
||||
"""Test approve_pairing with no token."""
|
||||
result = await approve_pairing("telegram", "ABCD1234")
|
||||
|
||||
assert result is False
|
||||
|
||||
@patch("praisonai.ui._pairing.GATEWAY_TOKEN", "test-token")
|
||||
@patch("aiohttp.ClientSession")
|
||||
async def test_approve_http_error(self, mock_session):
|
||||
"""Test approve_pairing with HTTP error."""
|
||||
# Mock HTTP error response
|
||||
mock_response = AsyncMock()
|
||||
mock_response.status = 404
|
||||
|
||||
mock_session_instance = AsyncMock()
|
||||
mock_session_instance.__aenter__ = AsyncMock(return_value=mock_session_instance)
|
||||
mock_session_instance.__aexit__ = AsyncMock(return_value=None)
|
||||
mock_session_instance.post = AsyncMock()
|
||||
mock_session_instance.post.return_value.__aenter__ = AsyncMock(return_value=mock_response)
|
||||
mock_session_instance.post.return_value.__aexit__ = AsyncMock(return_value=None)
|
||||
mock_session.return_value = mock_session_instance
|
||||
|
||||
result = await approve_pairing("telegram", "ABCD1234")
|
||||
|
||||
assert result is False
|
||||
|
||||
|
||||
class TestRefreshPendingBanner:
|
||||
"""Test the refresh_pending_banner function."""
|
||||
|
||||
async def test_banner_not_shown_for_non_admin(
|
||||
self,
|
||||
mock_chainlit_user_regular,
|
||||
mock_chainlit_message
|
||||
):
|
||||
"""Test that banner is not shown for non-admin users."""
|
||||
await refresh_pending_banner()
|
||||
|
||||
# Message should not be sent for non-admin
|
||||
mock_chainlit_message.assert_not_called()
|
||||
|
||||
async def test_banner_not_shown_with_no_pending(
|
||||
self,
|
||||
mock_chainlit_user_admin,
|
||||
mock_chainlit_message
|
||||
):
|
||||
"""Test that banner is not shown when no pending requests."""
|
||||
with patch("praisonai.ui._pairing.get_pending_pairings", return_value=[]):
|
||||
await refresh_pending_banner()
|
||||
|
||||
# Message should not be sent when no pending
|
||||
mock_chainlit_message.assert_not_called()
|
||||
|
||||
async def test_banner_shown_for_admin_with_pending(
|
||||
self,
|
||||
mock_chainlit_user_admin,
|
||||
mock_chainlit_message,
|
||||
mock_chainlit_action
|
||||
):
|
||||
"""Test that banner is shown for admin with pending requests."""
|
||||
mock_pending = [
|
||||
{
|
||||
"channel": "telegram",
|
||||
"code": "ABCD1234",
|
||||
"user_name": "User ABCD1234",
|
||||
"age_seconds": 30
|
||||
}
|
||||
]
|
||||
|
||||
with patch("praisonai.ui._pairing.get_pending_pairings", return_value=mock_pending):
|
||||
await refresh_pending_banner()
|
||||
|
||||
# Message should be created and sent
|
||||
mock_chainlit_message.assert_called_once()
|
||||
mock_message_call = mock_chainlit_message.call_args
|
||||
|
||||
# Check message content
|
||||
assert "🔔" in mock_message_call[1]["content"]
|
||||
assert "1 pending pairing request(s)" in mock_message_call[1]["content"]
|
||||
|
||||
# Verify action contract
|
||||
actions = mock_message_call[1]["actions"]
|
||||
assert len(actions) == 2 # Should have both approve and deny actions
|
||||
|
||||
# Check approve action
|
||||
approve_action = actions[0]
|
||||
assert approve_action.name == "approve_pairing"
|
||||
assert approve_action.value == "telegram:ABCD1234"
|
||||
assert "✅ Approve" in approve_action.label
|
||||
|
||||
# Check deny action
|
||||
deny_action = actions[1]
|
||||
assert deny_action.name == "deny_pairing"
|
||||
assert deny_action.value == "telegram:ABCD1234"
|
||||
assert "❌ Deny" in deny_action.label
|
||||
|
||||
async def test_banner_multiple_pending(
|
||||
self,
|
||||
mock_chainlit_user_admin,
|
||||
mock_chainlit_message,
|
||||
mock_chainlit_action
|
||||
):
|
||||
"""Test banner with multiple pending requests."""
|
||||
mock_pending = [
|
||||
{
|
||||
"channel": "telegram",
|
||||
"code": "ABCD1234",
|
||||
"user_name": "User ABCD1234",
|
||||
"age_seconds": 30
|
||||
},
|
||||
{
|
||||
"channel": "slack",
|
||||
"code": "EFGH5678",
|
||||
"user_name": "User EFGH5678",
|
||||
"age_seconds": 120
|
||||
}
|
||||
]
|
||||
|
||||
with patch("praisonai.ui._pairing.get_pending_pairings", return_value=mock_pending):
|
||||
await refresh_pending_banner()
|
||||
|
||||
# Message should be created with 2 actions
|
||||
mock_chainlit_message.assert_called_once()
|
||||
mock_message_call = mock_chainlit_message.call_args
|
||||
|
||||
assert "2 pending pairing request(s)" in mock_message_call[1]["content"]
|
||||
|
||||
# Should have 4 actions total (2 pending × 2 actions each)
|
||||
actions = mock_message_call[1]["actions"]
|
||||
assert len(actions) == 4
|
||||
|
||||
# Verify first pairing actions
|
||||
assert actions[0].name == "approve_pairing"
|
||||
assert actions[0].value == "telegram:ABCD1234"
|
||||
assert actions[1].name == "deny_pairing"
|
||||
assert actions[1].value == "telegram:ABCD1234"
|
||||
|
||||
# Verify second pairing actions
|
||||
assert actions[2].name == "approve_pairing"
|
||||
assert actions[2].value == "slack:EFGH5678"
|
||||
assert actions[3].name == "deny_pairing"
|
||||
assert actions[3].value == "slack:EFGH5678"
|
||||
|
||||
|
||||
class TestIntegrationFlow:
|
||||
"""Test the complete integration flow."""
|
||||
|
||||
async def test_end_to_end_pairing_flow(self, temp_pairing_store):
|
||||
"""Test complete flow: generate code → pending request → approve → is_paired."""
|
||||
# 1. Generate a pairing code
|
||||
code = temp_pairing_store.generate_code(channel_type="telegram")
|
||||
assert len(code) == 8
|
||||
|
||||
# 2. Verify it shows up in pending list
|
||||
pending = temp_pairing_store.list_pending()
|
||||
assert len(pending) == 1
|
||||
assert pending[0]["code"] == code
|
||||
assert pending[0]["channel"] == "telegram"
|
||||
|
||||
# 3. Approve the pairing
|
||||
success = temp_pairing_store.approve("telegram", code, user_id="12345")
|
||||
assert success is True
|
||||
|
||||
# 4. Verify it's now paired
|
||||
assert temp_pairing_store.is_paired("12345", "telegram") is True
|
||||
|
||||
# 5. Verify no longer in pending
|
||||
pending_after = temp_pairing_store.list_pending()
|
||||
assert len(pending_after) == 0
|
||||
|
||||
async def test_invalid_code_approval_flow(self, temp_pairing_store):
|
||||
"""Test approval flow with invalid code."""
|
||||
# Try to approve non-existent code
|
||||
success = temp_pairing_store.approve("telegram", "INVALID", user_id="12345")
|
||||
assert success is False
|
||||
|
||||
# Verify not paired
|
||||
assert temp_pairing_store.is_paired("12345", "telegram") is False
|
||||
|
||||
async def test_expired_code_flow(self, temp_pairing_store):
|
||||
"""Test that expired codes are cleaned up."""
|
||||
# Generate a code with very short TTL
|
||||
store_short_ttl = PairingStore(
|
||||
store_dir=temp_pairing_store._dir,
|
||||
code_ttl=0.1 # 100ms TTL
|
||||
)
|
||||
|
||||
code = store_short_ttl.generate_code(channel_type="telegram")
|
||||
|
||||
# Wait for expiration
|
||||
await asyncio.sleep(0.2)
|
||||
|
||||
# Try to approve expired code
|
||||
success = store_short_ttl.approve("telegram", code, user_id="12345")
|
||||
assert success is False
|
||||
|
||||
# Verify not paired
|
||||
assert store_short_ttl.is_paired("12345", "telegram") is False
|
||||
@@ -193,21 +193,6 @@ class TestUIIntegration:
|
||||
assert streamlit_config["page_icon"] == "🤖"
|
||||
assert streamlit_config["layout"] == "wide"
|
||||
|
||||
def test_chainlit_app_config(self):
|
||||
"""Test Chainlit app configuration."""
|
||||
chainlit_config = {
|
||||
"name": "PraisonAI Agent",
|
||||
"description": "Interact with PraisonAI agents",
|
||||
"author": "PraisonAI Team",
|
||||
"tags": ["ai", "agents", "chat"],
|
||||
"public": False,
|
||||
"authentication": True
|
||||
}
|
||||
|
||||
assert chainlit_config["name"] == "PraisonAI Agent"
|
||||
assert chainlit_config["authentication"] is True
|
||||
assert "ai" in chainlit_config["tags"]
|
||||
|
||||
def test_ui_agent_wrapper(self, sample_agent_config):
|
||||
"""Test UI agent wrapper functionality."""
|
||||
class UIAgentWrapper:
|
||||
|
||||
@@ -1,42 +0,0 @@
|
||||
import sys
|
||||
import types
|
||||
|
||||
from praisonai.ui import _external_agents as ext
|
||||
|
||||
|
||||
class _FakeUserSession:
|
||||
def __init__(self, values):
|
||||
self._values = values
|
||||
|
||||
def get(self, key, default=None):
|
||||
return self._values.get(key, default)
|
||||
|
||||
|
||||
def test_load_external_agent_settings_uses_explicit_loader(monkeypatch):
|
||||
fake_chainlit = types.SimpleNamespace(user_session=_FakeUserSession({}))
|
||||
monkeypatch.setitem(sys.modules, "chainlit", fake_chainlit)
|
||||
|
||||
values = {
|
||||
"claude_code_enabled": "true",
|
||||
"claude_enabled": "false",
|
||||
"gemini_enabled": "true",
|
||||
"codex_enabled": "1",
|
||||
}
|
||||
settings = ext.load_external_agent_settings_from_chainlit(lambda key: values.get(key))
|
||||
|
||||
assert set(settings.keys()) == set(ext.EXTERNAL_AGENTS.keys())
|
||||
assert settings["claude_enabled"] is False
|
||||
assert settings["gemini_enabled"] is True
|
||||
assert settings["codex_enabled"] is True
|
||||
|
||||
|
||||
def test_load_external_agent_settings_session_overrides_persistent(monkeypatch):
|
||||
fake_chainlit = types.SimpleNamespace(
|
||||
user_session=_FakeUserSession({"gemini_enabled": True, "claude_code_enabled": False})
|
||||
)
|
||||
monkeypatch.setitem(sys.modules, "chainlit", fake_chainlit)
|
||||
|
||||
values = {"gemini_enabled": "false"}
|
||||
settings = ext.load_external_agent_settings_from_chainlit(lambda key: values.get(key))
|
||||
|
||||
assert settings["gemini_enabled"] is True
|
||||
@@ -1,168 +0,0 @@
|
||||
"""
|
||||
Unit tests for UI bind-aware credential enforcement.
|
||||
|
||||
Tests that UI refuses default admin/admin credentials on external interfaces.
|
||||
"""
|
||||
|
||||
import os
|
||||
import pytest
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
pytest.importorskip("chainlit", reason="chainlit is an optional [ui] extra")
|
||||
|
||||
from praisonai.ui._auth import register_password_auth, UIStartupError
|
||||
|
||||
|
||||
class TestUIAuthValidation:
|
||||
"""Test UI authentication validation logic."""
|
||||
|
||||
@patch.dict(os.environ, {
|
||||
"CHAINLIT_USERNAME": "admin",
|
||||
"CHAINLIT_PASSWORD": "admin",
|
||||
"PRAISONAI_ALLOW_DEFAULT_CREDS": ""
|
||||
})
|
||||
def test_default_creds_on_loopback_warns_only(self):
|
||||
"""Test default credentials on loopback show warning but allow start."""
|
||||
mock_app = MagicMock()
|
||||
|
||||
# Should not raise exception on loopback
|
||||
register_password_auth(mock_app, bind_host="127.0.0.1")
|
||||
register_password_auth(mock_app, bind_host="localhost")
|
||||
register_password_auth(mock_app, bind_host="::1")
|
||||
|
||||
@patch.dict(os.environ, {
|
||||
"CHAINLIT_USERNAME": "admin",
|
||||
"CHAINLIT_PASSWORD": "admin",
|
||||
"PRAISONAI_ALLOW_DEFAULT_CREDS": ""
|
||||
})
|
||||
def test_default_creds_on_external_blocks_start(self):
|
||||
"""Test default credentials on external interface block start."""
|
||||
mock_app = MagicMock()
|
||||
|
||||
with pytest.raises(UIStartupError) as exc_info:
|
||||
register_password_auth(mock_app, bind_host="0.0.0.0")
|
||||
|
||||
assert "Cannot bind to 0.0.0.0 with default admin/admin credentials" in str(exc_info.value)
|
||||
assert "CHAINLIT_USERNAME" in str(exc_info.value)
|
||||
assert "CHAINLIT_PASSWORD" in str(exc_info.value)
|
||||
|
||||
# Test other external interfaces
|
||||
external_hosts = ["192.168.1.1", "10.0.0.1", "8.8.8.8"]
|
||||
for host in external_hosts:
|
||||
with pytest.raises(UIStartupError):
|
||||
register_password_auth(mock_app, bind_host=host)
|
||||
|
||||
@patch.dict(os.environ, {
|
||||
"CHAINLIT_USERNAME": "admin",
|
||||
"CHAINLIT_PASSWORD": "admin",
|
||||
"PRAISONAI_ALLOW_DEFAULT_CREDS": "1"
|
||||
})
|
||||
def test_escape_hatch_allows_default_creds(self):
|
||||
"""Test escape hatch allows default credentials on external interface."""
|
||||
mock_app = MagicMock()
|
||||
|
||||
# Should not raise with escape hatch
|
||||
register_password_auth(mock_app, bind_host="0.0.0.0")
|
||||
register_password_auth(mock_app, bind_host="192.168.1.1")
|
||||
|
||||
@patch.dict(os.environ, {
|
||||
"CHAINLIT_USERNAME": "myuser",
|
||||
"CHAINLIT_PASSWORD": "mypass"
|
||||
})
|
||||
def test_custom_creds_always_allowed(self):
|
||||
"""Test custom credentials are always allowed."""
|
||||
mock_app = MagicMock()
|
||||
|
||||
# Should not raise on any interface with custom creds
|
||||
register_password_auth(mock_app, bind_host="127.0.0.1")
|
||||
register_password_auth(mock_app, bind_host="0.0.0.0")
|
||||
register_password_auth(mock_app, bind_host="192.168.1.1")
|
||||
|
||||
@patch.dict(os.environ, {
|
||||
"CHAINLIT_USERNAME": "admin",
|
||||
"CHAINLIT_PASSWORD": "different"
|
||||
})
|
||||
def test_non_default_admin_allowed(self):
|
||||
"""Test non-default admin credentials are allowed."""
|
||||
mock_app = MagicMock()
|
||||
|
||||
# Should not raise - only username is default
|
||||
register_password_auth(mock_app, bind_host="0.0.0.0")
|
||||
|
||||
@patch.dict(os.environ, {
|
||||
"CHAINLIT_USERNAME": "different",
|
||||
"CHAINLIT_PASSWORD": "admin"
|
||||
})
|
||||
def test_non_default_password_allowed(self):
|
||||
"""Test non-default password with admin username is allowed."""
|
||||
mock_app = MagicMock()
|
||||
|
||||
# Should not raise - only password is default
|
||||
register_password_auth(mock_app, bind_host="0.0.0.0")
|
||||
|
||||
|
||||
class TestChainlitAuthCallback:
|
||||
"""Test the actual Chainlit auth callback behavior."""
|
||||
|
||||
def test_auth_callback_registration(self):
|
||||
"""Test that auth callback gets registered with Chainlit."""
|
||||
mock_app = MagicMock()
|
||||
|
||||
with patch('chainlit.password_auth_callback') as mock_decorator:
|
||||
mock_decorator.return_value = lambda f: f # Mock decorator
|
||||
|
||||
register_password_auth(mock_app, bind_host="127.0.0.1")
|
||||
|
||||
# Verify decorator was called
|
||||
mock_decorator.assert_called_once()
|
||||
|
||||
|
||||
class TestUISecurityScenarios:
|
||||
"""Test real-world UI security scenarios."""
|
||||
|
||||
@patch.dict(os.environ, {
|
||||
"CHAINLIT_USERNAME": "admin",
|
||||
"CHAINLIT_PASSWORD": "admin"
|
||||
})
|
||||
def test_local_development_scenario(self):
|
||||
"""Test local development scenario (default creds, loopback)."""
|
||||
mock_app = MagicMock()
|
||||
|
||||
# Should warn but allow (typical development scenario)
|
||||
register_password_auth(mock_app, bind_host="127.0.0.1")
|
||||
|
||||
@patch.dict(os.environ, {
|
||||
"CHAINLIT_USERNAME": "admin",
|
||||
"CHAINLIT_PASSWORD": "admin",
|
||||
"PRAISONAI_ALLOW_DEFAULT_CREDS": ""
|
||||
})
|
||||
def test_accidental_production_deploy_blocked(self):
|
||||
"""Test accidental production deploy with default creds is blocked."""
|
||||
mock_app = MagicMock()
|
||||
|
||||
# Should block (prevents shipping with admin/admin to LAN/internet)
|
||||
with pytest.raises(UIStartupError):
|
||||
register_password_auth(mock_app, bind_host="0.0.0.0")
|
||||
|
||||
@patch.dict(os.environ, {
|
||||
"CHAINLIT_USERNAME": "prod-user",
|
||||
"CHAINLIT_PASSWORD": "secure-pass-123"
|
||||
})
|
||||
def test_secure_production_deploy_allowed(self):
|
||||
"""Test secure production deploy with custom creds is allowed."""
|
||||
mock_app = MagicMock()
|
||||
|
||||
# Should allow (secure production deployment)
|
||||
register_password_auth(mock_app, bind_host="0.0.0.0")
|
||||
|
||||
@patch.dict(os.environ, {
|
||||
"CHAINLIT_USERNAME": "admin",
|
||||
"CHAINLIT_PASSWORD": "admin",
|
||||
"PRAISONAI_ALLOW_DEFAULT_CREDS": "1"
|
||||
})
|
||||
def test_lab_demo_escape_hatch(self):
|
||||
"""Test lab/demo escape hatch scenario."""
|
||||
mock_app = MagicMock()
|
||||
|
||||
# Should allow with warning (demo/lab scenario)
|
||||
register_password_auth(mock_app, bind_host="0.0.0.0")
|
||||
Reference in new issue
Block a user