Compare commits

...
2 Commits
Author SHA1 Message Date
Mervin Praison 585b68d02c Release v3.6.2 2026-01-10 09:08:53 +00:00
Mervin Praison 66481cc51c Release v3.6.1 2026-01-09 17:16:27 +00:00
30 changed files with 2739 additions and 38 deletions

No files matched your search

+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=3.6.0" \
"praisonai>=3.6.2" \
"praisonai[chat]" \
"embedchain[github,youtube]"
+1 -1
View File
@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=3.6.0" \
"praisonai>=3.6.2" \
"praisonai[ui]" \
"praisonai[chat]" \
"praisonai[realtime]" \
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=3.6.0" \
"praisonai>=3.6.2" \
"praisonai[ui]" \
"praisonai[crewai]"
+16
View File
@@ -23,6 +23,7 @@ examples/
| Category | Examples | Docs |
|----------|----------|------|
| **Consolidated Params** | [consolidated_params/](consolidated_params/) | [📖](https://docs.praison.ai/api/consolidated-params) |
| **Agents** | [python/agents/](python/agents/) | [📖](https://docs.praison.ai/concepts/agents) |
| **Workflows** | [python/workflows/](python/workflows/) | [📖](https://docs.praison.ai/features/workflows) |
| **Model Router** | [python/agents/router-agent-cost-optimization.py](python/agents/router-agent-cost-optimization.py) | [📖](https://docs.praison.ai/features/model-router) |
@@ -35,6 +36,21 @@ examples/
| **Serve** | [serve/](serve/) | [📖](https://docs.praison.ai/cli/serve) |
| **Endpoints** | [serve/](serve/) | [📖](https://docs.praison.ai/cli/endpoints) |
## Consolidated Params Examples
Agent-centric API with unified parameter resolution. Precedence: Instance > Config > Array > Dict > String > Bool > Default
| Example | Description |
|---------|-------------|
| [basic_agent.py](consolidated_params/basic_agent.py) | Minimal agent with memory |
| [basic_agents.py](consolidated_params/basic_agents.py) | Multi-agent with memory+planning |
| [basic_workflow.py](consolidated_params/basic_workflow.py) | Workflow with consolidated params |
| [basic_memory.py](consolidated_params/basic_memory.py) | Memory presets (file, redis, postgres) |
| [basic_guardrails.py](consolidated_params/basic_guardrails.py) | Guardrails with callable or config |
| [basic_workflow_agentlike.py](consolidated_params/basic_workflow_agentlike.py) | Workflow with agent-like params |
| [basic_step_override.py](consolidated_params/basic_step_override.py) | Step-level override of workflow defaults |
| [advanced_workflow_full_features.py](consolidated_params/advanced_workflow_full_features.py) | All consolidated params |
## Serve Examples
| Example | Description | CLI Command |
@@ -0,0 +1,55 @@
"""Advanced Workflow with all consolidated params configured."""
from praisonaiagents import Agent
from praisonaiagents.workflows import Workflow, WorkflowStep
from praisonaiagents.workflows.workflow_configs import (
WorkflowOutputConfig, WorkflowPlanningConfig, WorkflowMemoryConfig,
)
# Full-featured workflow with all consolidated params
workflow = Workflow(
name="FullFeaturedWorkflow",
description="Demonstrates all agent-like consolidated params",
steps=[
WorkflowStep(
name="researcher",
action="Research: {{input}}",
agent=Agent(instructions="You are a researcher."),
knowledge=["research_docs/"], # Step-specific knowledge
web="duckduckgo", # Use DuckDuckGo for this step
),
WorkflowStep(
name="analyzer",
action="Analyze the research",
agent=Agent(instructions="You are an analyst."),
reflection="thorough", # Thorough reflection for analysis
guardrails="strict", # Strict validation
),
WorkflowStep(
name="writer",
action="Write the final report",
agent=Agent(instructions="You are a writer."),
autonomy=True, # Enable autonomy for writing
),
],
# Workflow-level consolidated params
output=WorkflowOutputConfig(verbose=True, stream=True),
planning=WorkflowPlanningConfig(enabled=True, reasoning=True),
memory=WorkflowMemoryConfig(backend="file"),
# Agent-like params
autonomy=True,
knowledge=["shared_docs/"],
guardrails=True,
web=True,
reflection=True,
)
if __name__ == "__main__":
print(f"Workflow: {workflow.name}")
print(f"Output verbose: {workflow._verbose}")
print(f"Planning enabled: {workflow._planning_enabled}")
print(f"Memory config: {workflow._memory_config}")
print(f"Autonomy: {workflow.autonomy}")
print(f"Knowledge: {workflow.knowledge}")
print(f"Guardrails: {workflow.guardrails}")
print(f"Web: {workflow.web}")
print(f"Reflection: {workflow.reflection}")
@@ -0,0 +1,12 @@
"""Basic Agent example with consolidated params."""
from praisonaiagents import Agent
# Minimal agent with memory
agent = Agent(
instructions="You are a helpful assistant.",
memory=True,
)
if __name__ == "__main__":
response = agent.chat("Hello!")
print(response)
@@ -0,0 +1,17 @@
"""Basic Agents (multi-agent) example with consolidated params."""
from praisonaiagents import Agent, Agents
# Create agents
writer = Agent(instructions="You write content.")
editor = Agent(instructions="You edit content.")
# Multi-agent with consolidated params
agents = Agents(
agents=[writer, editor],
memory=True,
planning=True,
)
if __name__ == "__main__":
result = agents.start("Write a haiku about AI")
print(result)
@@ -0,0 +1,33 @@
"""Basic example showing step-level override of workflow defaults."""
from praisonaiagents import Agent
from praisonaiagents.workflows import Workflow, WorkflowStep
# Workflow with defaults
workflow = Workflow(
name="StepOverrideWorkflow",
steps=[
WorkflowStep(
name="step1",
action="Process with workflow defaults",
agent=Agent(instructions="You are a processor."),
# Uses workflow-level web=True (inherited)
),
WorkflowStep(
name="step2",
action="Process with step override",
agent=Agent(instructions="You are a processor."),
# Override workflow default
web=False, # Disable web for this step
reflection="thorough", # Use thorough reflection preset
),
],
# Workflow defaults
web=True,
reflection=True,
)
if __name__ == "__main__":
print(f"Workflow web default: {workflow.web}")
print(f"Step1 web: {workflow.steps[0].web}") # None (uses workflow default)
print(f"Step2 web: {workflow.steps[1].web}") # False (overridden)
print(f"Step2 reflection: {workflow.steps[1].reflection}") # thorough
@@ -0,0 +1,32 @@
"""Basic Workflow with Agent-like consolidated params."""
from praisonaiagents import Agent
from praisonaiagents.workflows import Workflow, WorkflowStep
# Workflow with agent-like params (knowledge, web, guardrails, reflection)
workflow = Workflow(
name="AgentLikeWorkflow",
steps=[
WorkflowStep(
name="researcher",
action="Research the topic: {{input}}",
agent=Agent(instructions="You are a researcher."),
),
WorkflowStep(
name="writer",
action="Write about the research findings",
agent=Agent(instructions="You are a writer."),
),
],
# Agent-like consolidated params
knowledge=["docs/"], # Enable RAG
web=True, # Enable web search
guardrails=True, # Enable guardrails
reflection=True, # Enable self-reflection
)
if __name__ == "__main__":
print(f"Workflow: {workflow.name}")
print(f"Knowledge: {workflow.knowledge}")
print(f"Web: {workflow.web}")
print(f"Guardrails: {workflow.guardrails}")
print(f"Reflection: {workflow.reflection}")
@@ -0,0 +1,23 @@
"""Advanced memory example - with config and overrides."""
from praisonaiagents import Agent, MemoryConfig
# With full config
agent = Agent(
instructions="You are a helpful assistant with memory.",
memory=MemoryConfig(
backend="redis",
user_id="user123",
session_id="session456",
auto_memory=True,
),
)
# With array [preset, overrides]
agent_override = Agent(
instructions="You are a helpful assistant.",
memory=["redis", {"user_id": "custom_user"}],
)
if __name__ == "__main__":
response = agent.chat("Remember that my favorite color is blue.")
print(response)
@@ -0,0 +1,24 @@
"""Basic memory example - minimal usage."""
from praisonaiagents import Agent
# Simple enable (uses default file-based memory)
agent = Agent(
instructions="You are a helpful assistant with memory.",
memory=True,
)
# With preset
agent_redis = Agent(
instructions="You are a helpful assistant.",
memory="redis", # Uses redis preset
)
# With URL
agent_postgres = Agent(
instructions="You are a helpful assistant.",
memory="postgresql://localhost/mydb", # URL auto-detected
)
if __name__ == "__main__":
response = agent.chat("Remember that my favorite color is blue.")
print(response)
@@ -0,0 +1,13 @@
"""Basic output and execution example."""
from praisonaiagents import Agent
# Simple preset usage
agent = Agent(
instructions="You are a helpful assistant.",
output="verbose", # Preset: verbose output
execution="thorough", # Preset: thorough execution
)
if __name__ == "__main__":
response = agent.chat("What is 2+2?")
print(response)
@@ -0,0 +1,22 @@
"""Basic workflow example with consolidated params."""
from praisonaiagents import Agent
from praisonaiagents.workflows import Workflow, WorkflowStep
# Create agents
writer = Agent(instructions="You are a content writer.")
editor = Agent(instructions="You are an editor.")
# Create workflow with consolidated params
workflow = Workflow(
name="Content Pipeline",
steps=[
WorkflowStep(name="write", agent=writer, action="Write about {{topic}}"),
WorkflowStep(name="edit", agent=editor, action="Edit the content", context=["write"]),
],
output="verbose",
planning=True,
)
if __name__ == "__main__":
result = workflow.run(variables={"topic": "AI agents"})
print(result)
@@ -142,6 +142,12 @@ class Agents:
output: Optional[Any] = None, # Union[str, MultiAgentOutputConfig]
execution: Optional[Any] = None, # Union[str, MultiAgentExecutionConfig]
hooks: Optional[Any] = None, # MultiAgentHooksConfig
# Additional consolidated params for feature parity with Agent
autonomy: Optional[Any] = None, # Union[bool, AutonomyConfig] - agent autonomy
knowledge: Optional[Any] = None, # Union[bool, List[str], KnowledgeConfig] - RAG
guardrails: Optional[Any] = None, # Union[bool, Callable, GuardrailConfig] - validation
web: Optional[Any] = None, # Union[bool, WebConfig] - web search/fetch
reflection: Optional[Any] = None, # Union[bool, ReflectionConfig] - self-reflection
):
"""
Initialize PraisonAIAgents with consolidated feature parameters.
@@ -159,7 +165,18 @@ class Agents:
output: Output configuration (str | MultiAgentOutputConfig)
execution: Execution configuration (str | MultiAgentExecutionConfig)
hooks: Hooks configuration (MultiAgentHooksConfig)
autonomy: Autonomy configuration (bool | AutonomyConfig)
knowledge: Knowledge/RAG configuration (bool | List[str] | KnowledgeConfig)
guardrails: Guardrails configuration (bool | Callable | GuardrailConfig)
web: Web search/fetch configuration (bool | WebConfig)
reflection: Self-reflection configuration (bool | ReflectionConfig)
"""
# Store new params for propagation to agents
self._autonomy = autonomy
self._knowledge = knowledge
self._guardrails = guardrails
self._web = web
self._reflection = reflection
# ─────────────────────────────────────────────────────────────────────
# Extract values from consolidated params using UNIFIED CANONICAL resolver
# Precedence: Instance > Config > Dict > Array > String > Bool > Default
@@ -54,7 +54,7 @@ def resolve(
) -> Any:
"""
Resolve a consolidated parameter following precedence rules:
Instance > Config > Array > String > Bool > Default
Instance > Config > Array > Dict > String > Bool > Default
Args:
value: The parameter value
@@ -92,7 +92,20 @@ def resolve(
return value
# =========================================================================
# 4. Dict -> convert to config (STRICT validation)
# 4. Array handling (BEFORE Dict per AC-1 precedence)
# =========================================================================
if isinstance(value, (list, tuple)):
return _resolve_array(
value=value,
param_name=param_name,
config_class=config_class,
presets=presets,
url_schemes=url_schemes,
array_mode=array_mode,
)
# =========================================================================
# 5. Dict -> convert to config (STRICT validation)
# =========================================================================
if isinstance(value, dict):
if config_class:
@@ -120,19 +133,6 @@ def resolve(
f"Use a supported type: bool, str, or instance."
)
# =========================================================================
# 5. Array handling (before string to respect precedence)
# =========================================================================
if isinstance(value, (list, tuple)):
return _resolve_array(
value=value,
param_name=param_name,
config_class=config_class,
presets=presets,
url_schemes=url_schemes,
array_mode=array_mode,
)
# =========================================================================
# 6. String handling
# =========================================================================
@@ -221,7 +221,7 @@ def _resolve_array(
return config_class(sources=sources, **config_override)
return sources
# Single or list mode
# Single or list mode - SINGLE item only, multiple items raise error
if array_mode == ArrayMode.SINGLE_OR_LIST:
if len(value) == 1:
single_value = value[0]
@@ -233,10 +233,17 @@ def _resolve_array(
# Check if single item is a preset
if isinstance(single_value, str) and presets and single_value in presets:
return _apply_preset(single_value, presets, config_class)
# Multiple items - treat as sources
if config_class:
return config_class(sources=list(value))
return list(value)
# Single non-preset string - try as URL or return as-is
if isinstance(single_value, str):
if config_class:
return config_class()
return single_value
# Multiple items - ERROR (not allowed for SINGLE_OR_LIST mode)
raise TypeError(
f"Multiple values not allowed for {param_name}. "
f"Use a single value: {param_name}='preset' or {param_name}='url://...' "
f"Got {len(value)} items: {value}"
)
# Preset override mode - [preset, {overrides}] or [preset]
if array_mode == ArrayMode.PRESET_OVERRIDE:
@@ -231,20 +231,29 @@ class WorkflowStep:
loop_over: Optional[str] = None # Variable name to iterate over (e.g., "items")
loop_var: str = "item" # Variable name for current item in loop
# Guardrail (like process="workflow")
guardrail: Optional[Callable[['StepResult'], Tuple[bool, Any]]] = None # Returns (is_valid, feedback)
# Guardrails (standardized plural naming)
# Accepts: callable, str (LLM prompt), GuardrailConfig, or list [preset, overrides]
guardrails: Optional[Any] = None # Union[Callable, str, GuardrailConfig] - validation
# Backward compatibility alias (deprecated - use guardrails instead)
guardrail: Optional[Callable[['StepResult'], Tuple[bool, Any]]] = None # DEPRECATED: use guardrails
# Image handling (migrated from Task)
images: Optional[List[str]] = None # Image paths/URLs for vision tasks
# ============================================================
# CONSOLIDATED FEATURE PARAMS (workflow-centric API)
# Precedence: Instance > Config > String > Bool > Default
# Precedence: Instance > Config > Array > Dict > String > Bool > Default
# ============================================================
context: Optional[Any] = None # Union[bool, List[str], WorkflowStepContextConfig] - context handling
output: Optional[Any] = None # Union[str, WorkflowStepOutputConfig] - output handling
execution: Optional[Any] = None # Union[str, WorkflowStepExecutionConfig] - execution control
routing: Optional[Any] = None # Union[List[str], WorkflowStepRoutingConfig] - branching
# NEW: Agent-like consolidated params for feature parity
autonomy: Optional[Any] = None # Union[bool, AutonomyConfig] - agent autonomy
knowledge: Optional[Any] = None # Union[bool, List[str], KnowledgeConfig] - RAG
web: Optional[Any] = None # Union[bool, WebConfig] - web search/fetch
reflection: Optional[Any] = None # Union[bool, ReflectionConfig] - self-reflection
# Status tracking
status: str = "pending" # pending, running, completed, failed, skipped
@@ -264,6 +273,11 @@ class WorkflowStep:
_on_error: str = field(default="stop", repr=False)
_next_steps: Optional[List[str]] = field(default=None, repr=False)
_branch_condition: Optional[Dict[str, List[str]]] = field(default=None, repr=False)
# NEW: Resolved configs for agent-like params
_autonomy_config: Optional[Any] = field(default=None, repr=False)
_knowledge_config: Optional[Any] = field(default=None, repr=False)
_web_config: Optional[Any] = field(default=None, repr=False)
_reflection_config: Optional[Any] = field(default=None, repr=False)
def __post_init__(self):
"""Resolve consolidated params to internal values."""
@@ -272,6 +286,18 @@ class WorkflowStep:
resolve_step_execution_config, resolve_step_routing_config,
)
# Handle guardrails/guardrail compatibility
# guardrails (plural) takes precedence; guardrail (singular) is deprecated
if self.guardrails is None and self.guardrail is not None:
# Use deprecated guardrail value, map to guardrails
import warnings
warnings.warn(
"WorkflowStep 'guardrail' parameter is deprecated. Use 'guardrails' instead.",
DeprecationWarning,
stacklevel=2,
)
object.__setattr__(self, 'guardrails', self.guardrail)
# Resolve context param
ctx_cfg = resolve_step_context_config(self.context)
if ctx_cfg:
@@ -300,6 +326,50 @@ class WorkflowStep:
if route_cfg:
self._next_steps = route_cfg.next_steps
self._branch_condition = route_cfg.branches
# Resolve NEW agent-like params using canonical resolver
from ..config.param_resolver import resolve, ArrayMode
from ..config.presets import (
WEB_PRESETS, REFLECTION_PRESETS, AUTONOMY_PRESETS, KNOWLEDGE_PRESETS,
)
try:
from ..config.feature_configs import (
AutonomyConfig, KnowledgeConfig, WebConfig, ReflectionConfig,
)
except ImportError:
AutonomyConfig = KnowledgeConfig = WebConfig = ReflectionConfig = None
# Resolve autonomy
if AutonomyConfig and self.autonomy is not None:
self._autonomy_config = resolve(
value=self.autonomy, param_name="autonomy",
config_class=AutonomyConfig, presets=AUTONOMY_PRESETS,
default=None,
)
# Resolve knowledge
if KnowledgeConfig and self.knowledge is not None:
self._knowledge_config = resolve(
value=self.knowledge, param_name="knowledge",
config_class=KnowledgeConfig, presets=KNOWLEDGE_PRESETS,
array_mode=ArrayMode.SOURCES, default=None,
)
# Resolve web
if WebConfig and self.web is not None:
self._web_config = resolve(
value=self.web, param_name="web",
config_class=WebConfig, presets=WEB_PRESETS,
default=None,
)
# Resolve reflection
if ReflectionConfig and self.reflection is not None:
self._reflection_config = resolve(
value=self.reflection, param_name="reflection",
config_class=ReflectionConfig, presets=REFLECTION_PRESETS,
default=None,
)
# Property accessors for backward compatibility
@property
@@ -423,13 +493,20 @@ class Workflow:
# ============================================================
# CONSOLIDATED FEATURE PARAMS (workflow-centric API)
# Precedence: Instance > Config > String > Bool > Default
# Precedence: Instance > Config > Array > Dict > String > Bool > Default
# Workflow is "Agents in an organised way" - supports same params as Agents
# ============================================================
output: Optional[Any] = None # Union[str, WorkflowOutputConfig] - verbose/stream
planning: Optional[Any] = False # Union[bool, str, WorkflowPlanningConfig] - planning mode
memory: Optional[Any] = None # Union[bool, WorkflowMemoryConfig, instance] - memory
hooks: Optional[Any] = None # WorkflowHooksConfig - lifecycle callbacks
context: Optional[Any] = False # Union[bool, ManagerConfig, ContextManager] - context management
# NEW: Agent-like consolidated params for feature parity
autonomy: Optional[Any] = None # Union[bool, AutonomyConfig] - agent autonomy
knowledge: Optional[Any] = None # Union[bool, List[str], KnowledgeConfig] - RAG
guardrails: Optional[Any] = None # Union[bool, Callable, GuardrailConfig] - validation
web: Optional[Any] = None # Union[bool, WebConfig] - web search/fetch
reflection: Optional[Any] = None # Union[bool, ReflectionConfig] - self-reflection
# Status tracking
status: str = "not_started" # not_started, running, completed, failed
@@ -445,6 +522,12 @@ class Workflow:
_hooks_config: Optional[Any] = field(default=None, repr=False)
_context_manager: Optional[Any] = field(default=None, repr=False)
_context_manager_initialized: bool = field(default=False, repr=False)
# NEW: Resolved configs for agent-like params (for propagation to steps/agents)
_autonomy_config: Optional[Any] = field(default=None, repr=False)
_knowledge_config: Optional[Any] = field(default=None, repr=False)
_guardrails_config: Optional[Any] = field(default=None, repr=False)
_web_config: Optional[Any] = field(default=None, repr=False)
_reflection_config: Optional[Any] = field(default=None, repr=False)
def __post_init__(self):
"""Resolve consolidated params to internal values."""
@@ -474,6 +557,62 @@ class Workflow:
# Resolve hooks param
self._hooks_config = resolve_hooks_config(self.hooks)
# Resolve NEW agent-like params using canonical resolver
# These are stored for propagation to steps/agents
from ..config.param_resolver import resolve, ArrayMode
from ..config.presets import (
WEB_PRESETS, REFLECTION_PRESETS, GUARDRAIL_PRESETS,
AUTONOMY_PRESETS, KNOWLEDGE_PRESETS,
)
try:
from ..config.feature_configs import (
AutonomyConfig, KnowledgeConfig, GuardrailConfig,
WebConfig, ReflectionConfig,
)
except ImportError:
AutonomyConfig = KnowledgeConfig = GuardrailConfig = None
WebConfig = ReflectionConfig = None
# Resolve autonomy
if AutonomyConfig and self.autonomy is not None:
self._autonomy_config = resolve(
value=self.autonomy, param_name="autonomy",
config_class=AutonomyConfig, presets=AUTONOMY_PRESETS,
default=None,
)
# Resolve knowledge
if KnowledgeConfig and self.knowledge is not None:
self._knowledge_config = resolve(
value=self.knowledge, param_name="knowledge",
config_class=KnowledgeConfig, presets=KNOWLEDGE_PRESETS,
array_mode=ArrayMode.SOURCES, default=None,
)
# Resolve guardrails
if GuardrailConfig and self.guardrails is not None:
self._guardrails_config = resolve(
value=self.guardrails, param_name="guardrails",
config_class=GuardrailConfig, presets=GUARDRAIL_PRESETS,
default=None,
)
# Resolve web
if WebConfig and self.web is not None:
self._web_config = resolve(
value=self.web, param_name="web",
config_class=WebConfig, presets=WEB_PRESETS,
default=None,
)
# Resolve reflection
if ReflectionConfig and self.reflection is not None:
self._reflection_config = resolve(
value=self.reflection, param_name="reflection",
config_class=ReflectionConfig, presets=REFLECTION_PRESETS,
default=None,
)
def to_dict(self) -> Dict[str, Any]:
return {
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "praisonaiagents"
version = "0.10.8"
version = "0.10.10"
description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
readme = "README.md"
requires-python = ">=3.10"
@@ -0,0 +1,720 @@
"""
TDD Tests for Resolver Standardization.
These tests verify:
1. Precedence order: None → Instance → Config → Array → Dict → String → Bool → Default
2. Memory array semantics: single=ok, multiple=error
3. Caching array support with PRESET_OVERRIDE
4. Workflow/WorkflowStep using canonical resolve
5. CLI resolver integration
6. Naming standardization (guardrails plural)
7. All precedence levels per consolidated param
Run with: pytest tests/unit/config/test_resolver_standardization.py -v
"""
import pytest
from dataclasses import dataclass
from typing import Any, Dict, List, Optional
# =============================================================================
# Test Fixtures - Config Classes
# =============================================================================
@dataclass
class MockConfig:
"""Mock config class for testing."""
enabled: bool = True
value: str = "default"
count: int = 0
@dataclass
class MockMemoryConfig:
"""Mock memory config for testing."""
backend: str = "file"
user_id: Optional[str] = None
config: Optional[Dict[str, Any]] = None
sources: Optional[List[str]] = None
@dataclass
class MockCachingConfig:
"""Mock caching config for testing."""
enabled: bool = True
ttl: int = 3600
prompt_caching: bool = False
# =============================================================================
# TODO 1.1: Tests for Precedence Order (Array BEFORE Dict)
# =============================================================================
class TestPrecedenceOrder:
"""Test that precedence order is: None → Instance → Config → Array → Dict → String → Bool → Default"""
def test_array_takes_precedence_over_dict_interpretation(self):
"""Array input should be processed as array, not as dict."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
# When given an array, it should use array handling, not dict handling
result = resolve(
value=["preset1"],
param_name="test",
config_class=MockConfig,
presets={"preset1": {"enabled": True, "value": "from_preset"}},
array_mode=ArrayMode.PRESET_OVERRIDE,
)
# Should resolve via array path (preset), not dict path
assert isinstance(result, MockConfig)
assert result.value == "from_preset"
def test_precedence_instance_over_config(self):
"""Instance check should take precedence over config class check."""
from praisonaiagents.config.param_resolver import resolve
class CustomInstance:
def __init__(self):
self.is_custom = True
custom = CustomInstance()
result = resolve(
value=custom,
param_name="test",
config_class=MockConfig,
instance_check=lambda v: hasattr(v, 'is_custom'),
)
# Should return instance as-is, not try to convert
assert result is custom
assert result.is_custom is True
def test_precedence_config_over_array(self):
"""Config class instance should take precedence over array handling."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
config = MockConfig(enabled=False, value="config_value")
result = resolve(
value=config,
param_name="test",
config_class=MockConfig,
array_mode=ArrayMode.PRESET_OVERRIDE,
)
# Should return config as-is
assert result is config
assert result.value == "config_value"
def test_precedence_array_over_dict(self):
"""Array should be checked BEFORE dict in precedence."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
# This test verifies the fix: Array must be checked before Dict
# A list input should go through array handling, not dict handling
result = resolve(
value=["verbose", {"count": 5}],
param_name="test",
config_class=MockConfig,
presets={"verbose": {"enabled": True, "value": "verbose_mode"}},
array_mode=ArrayMode.PRESET_OVERRIDE,
)
assert isinstance(result, MockConfig)
assert result.value == "verbose_mode"
assert result.count == 5 # Override applied
def test_precedence_dict_over_string(self):
"""Dict should take precedence over string handling."""
from praisonaiagents.config.param_resolver import resolve
result = resolve(
value={"enabled": False, "value": "dict_value"},
param_name="test",
config_class=MockConfig,
presets={"dict_value": {"enabled": True}}, # Should NOT use this
)
assert isinstance(result, MockConfig)
assert result.value == "dict_value"
assert result.enabled is False
def test_precedence_string_over_bool(self):
"""String preset should take precedence over bool handling."""
from praisonaiagents.config.param_resolver import resolve
result = resolve(
value="verbose",
param_name="test",
config_class=MockConfig,
presets={"verbose": {"enabled": True, "value": "verbose_mode"}},
)
assert isinstance(result, MockConfig)
assert result.value == "verbose_mode"
def test_precedence_bool_over_default(self):
"""Bool True should create default config, not return default value."""
from praisonaiagents.config.param_resolver import resolve
result = resolve(
value=True,
param_name="test",
config_class=MockConfig,
default={"should": "not_use"},
)
assert isinstance(result, MockConfig)
# Should be default MockConfig(), not the default dict
def test_precedence_none_returns_default(self):
"""None should return the default value."""
from praisonaiagents.config.param_resolver import resolve
default_config = MockConfig(value="default_value")
result = resolve(
value=None,
param_name="test",
config_class=MockConfig,
default=default_config,
)
assert result is default_config
# =============================================================================
# TODO 1.2: Tests for Memory Array Semantics
# =============================================================================
class TestMemoryArraySemantics:
"""Test memory array semantics: single=ok, multiple=error."""
def test_memory_single_item_array_resolves_as_scalar(self):
"""memory=["redis"] should resolve as scalar preset."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
result = resolve(
value=["redis"],
param_name="memory",
config_class=MockMemoryConfig,
presets={"redis": {"backend": "redis"}},
array_mode=ArrayMode.SINGLE_OR_LIST,
)
assert isinstance(result, MockMemoryConfig)
assert result.backend == "redis"
def test_memory_multiple_items_raises_error(self):
"""memory=["redis", "file"] should raise TypeError."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
with pytest.raises(TypeError) as exc_info:
resolve(
value=["redis", "file"],
param_name="memory",
config_class=MockMemoryConfig,
presets={"redis": {"backend": "redis"}, "file": {"backend": "file"}},
array_mode=ArrayMode.SINGLE_OR_LIST,
)
# Error message should be helpful
assert "memory" in str(exc_info.value).lower()
assert "multiple" in str(exc_info.value).lower() or "single" in str(exc_info.value).lower()
def test_memory_single_url_in_array_resolves(self):
"""memory=["postgresql://..."] should resolve as URL."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
result = resolve(
value=["postgresql://localhost/db"],
param_name="memory",
config_class=MockMemoryConfig,
url_schemes={"postgresql": "postgres"},
array_mode=ArrayMode.SINGLE_OR_LIST,
)
assert isinstance(result, MockMemoryConfig)
assert result.backend == "postgres"
# =============================================================================
# TODO 1.3: Tests for Caching Array Support
# =============================================================================
class TestCachingArraySupport:
"""Test caching array support with PRESET_OVERRIDE."""
def test_caching_array_preset_override(self):
"""caching=["enabled", {"ttl": 7200}] should work."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
result = resolve(
value=["enabled", {"ttl": 7200}],
param_name="caching",
config_class=MockCachingConfig,
presets={"enabled": {"enabled": True, "ttl": 3600}},
array_mode=ArrayMode.PRESET_OVERRIDE,
)
assert isinstance(result, MockCachingConfig)
assert result.enabled is True
assert result.ttl == 7200 # Override applied
def test_caching_string_preset(self):
"""caching="enabled" should work."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
result = resolve(
value="enabled",
param_name="caching",
config_class=MockCachingConfig,
presets={"enabled": {"enabled": True}},
array_mode=ArrayMode.PRESET_OVERRIDE,
)
assert isinstance(result, MockCachingConfig)
assert result.enabled is True
# =============================================================================
# TODO 1.4: Tests for Workflow/WorkflowStep Using Canonical Resolve
# =============================================================================
class TestWorkflowCanonicalResolve:
"""Test that Workflow and WorkflowStep use canonical resolve directly."""
def test_workflow_output_uses_canonical_resolve(self):
"""Workflow output param should use canonical resolve."""
from praisonaiagents.workflows.workflows import Workflow
workflow = Workflow(
name="test",
steps=[],
output="verbose",
)
assert workflow._verbose is True
def test_workflow_output_dict_validation(self):
"""Workflow output dict should have strict validation."""
from praisonaiagents.workflows.workflows import Workflow
with pytest.raises(TypeError) as exc_info:
Workflow(
name="test",
steps=[],
output={"invalid_key": True},
)
assert "invalid_key" in str(exc_info.value).lower() or "unknown" in str(exc_info.value).lower()
def test_workflowstep_context_uses_canonical_resolve(self):
"""WorkflowStep context param should use canonical resolve."""
from praisonaiagents.workflows.workflows import WorkflowStep
step = WorkflowStep(
name="test",
context=["step1", "step2"],
)
assert step._context_from == ["step1", "step2"]
def test_workflowstep_execution_preset(self):
"""WorkflowStep execution preset should work."""
from praisonaiagents.workflows.workflows import WorkflowStep
step = WorkflowStep(
name="test",
execution="thorough",
)
# Should have thorough preset values
assert step._max_retries >= 3 # Thorough has more retries
# =============================================================================
# TODO 1.5: Tests for CLI Resolver Integration
# =============================================================================
class TestCLIResolverIntegration:
"""Test CLI flags use canonical resolver."""
def test_cli_memory_flag_true(self):
"""CLI --memory flag (true) should resolve correctly."""
# This tests the integration point - CLI should pass value to canonical resolver
from praisonaiagents.config.param_resolver import resolve, ArrayMode
from praisonaiagents.config.presets import MEMORY_PRESETS, MEMORY_URL_SCHEMES
# Simulate CLI passing True
result = resolve(
value=True,
param_name="memory",
config_class=MockMemoryConfig,
presets=MEMORY_PRESETS,
url_schemes=MEMORY_URL_SCHEMES,
array_mode=ArrayMode.SINGLE_OR_LIST,
)
assert isinstance(result, MockMemoryConfig)
def test_cli_memory_flag_preset_string(self):
"""CLI --memory=redis should resolve as preset."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
from praisonaiagents.config.presets import MEMORY_PRESETS, MEMORY_URL_SCHEMES
result = resolve(
value="redis",
param_name="memory",
config_class=MockMemoryConfig,
presets=MEMORY_PRESETS,
url_schemes=MEMORY_URL_SCHEMES,
array_mode=ArrayMode.SINGLE_OR_LIST,
)
assert isinstance(result, MockMemoryConfig)
assert result.backend == "redis"
def test_cli_memory_flag_url(self):
"""CLI --memory=postgresql://... should resolve as URL."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
from praisonaiagents.config.presets import MEMORY_PRESETS, MEMORY_URL_SCHEMES
result = resolve(
value="postgresql://localhost/db",
param_name="memory",
config_class=MockMemoryConfig,
presets=MEMORY_PRESETS,
url_schemes=MEMORY_URL_SCHEMES,
array_mode=ArrayMode.SINGLE_OR_LIST,
)
assert isinstance(result, MockMemoryConfig)
assert result.backend == "postgres"
# =============================================================================
# TODO 1.6: Tests for Naming Standardization (guardrails plural)
# =============================================================================
class TestNamingStandardization:
"""Test guardrail → guardrails naming standardization."""
def test_workflowstep_accepts_guardrails_plural(self):
"""WorkflowStep should accept guardrails= (plural)."""
from praisonaiagents.workflows.workflows import WorkflowStep
def my_guardrail(result):
return (True, result)
# Should accept guardrails= (plural)
step = WorkflowStep(
name="test",
guardrails=my_guardrail,
)
# Should store in guardrails attribute
assert step.guardrails is my_guardrail or step.guardrail is my_guardrail
def test_workflowstep_guardrail_singular_compat(self):
"""WorkflowStep should accept guardrail= (singular) for compatibility."""
from praisonaiagents.workflows.workflows import WorkflowStep
def my_guardrail(result):
return (True, result)
# Should still accept guardrail= (singular) for backward compat
step = WorkflowStep(
name="test",
guardrail=my_guardrail,
)
# Should work (backward compat)
assert step.guardrail is my_guardrail
# =============================================================================
# TODO 1.7: Tests for All Precedence Levels Per Consolidated Param
# =============================================================================
class TestAllPrecedenceLevels:
"""Test each consolidated param has all 7 precedence levels working."""
def test_output_all_precedence_levels(self):
"""Test output param at all precedence levels."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
from praisonaiagents.config.feature_configs import OutputConfig
from praisonaiagents.config.presets import OUTPUT_PRESETS
# 1. None → Default
result = resolve(value=None, param_name="output", config_class=OutputConfig,
presets=OUTPUT_PRESETS, default=OutputConfig())
assert isinstance(result, OutputConfig)
# 2. Instance (custom object with expected attributes)
class CustomOutput:
verbose = True
stream = True
custom = CustomOutput()
result = resolve(value=custom, param_name="output", config_class=OutputConfig,
presets=OUTPUT_PRESETS, instance_check=lambda v: hasattr(v, 'verbose'))
assert result is custom
# 3. Config class instance
config = OutputConfig(verbose=True, stream=False)
result = resolve(value=config, param_name="output", config_class=OutputConfig,
presets=OUTPUT_PRESETS)
assert result is config
# 4. Array [preset, overrides]
result = resolve(value=["verbose", {"stream": False}], param_name="output",
config_class=OutputConfig, presets=OUTPUT_PRESETS,
array_mode=ArrayMode.PRESET_OVERRIDE)
assert isinstance(result, OutputConfig)
# 5. Dict
result = resolve(value={"verbose": True, "stream": False}, param_name="output",
config_class=OutputConfig, presets=OUTPUT_PRESETS)
assert isinstance(result, OutputConfig)
assert result.verbose is True
# 6. String preset
result = resolve(value="verbose", param_name="output", config_class=OutputConfig,
presets=OUTPUT_PRESETS)
assert isinstance(result, OutputConfig)
# 7. Bool True
result = resolve(value=True, param_name="output", config_class=OutputConfig,
presets=OUTPUT_PRESETS)
assert isinstance(result, OutputConfig)
# 8. Bool False → None
result = resolve(value=False, param_name="output", config_class=OutputConfig,
presets=OUTPUT_PRESETS)
assert result is None
def test_memory_all_precedence_levels(self):
"""Test memory param at all precedence levels."""
from praisonaiagents.config.param_resolver import resolve, ArrayMode
from praisonaiagents.config.feature_configs import MemoryConfig
from praisonaiagents.config.presets import MEMORY_PRESETS, MEMORY_URL_SCHEMES
# 1. None → Default
result = resolve(value=None, param_name="memory", config_class=MemoryConfig,
presets=MEMORY_PRESETS, url_schemes=MEMORY_URL_SCHEMES,
array_mode=ArrayMode.SINGLE_OR_LIST, default=None)
assert result is None
# 2. Config class instance
config = MemoryConfig(backend="redis")
result = resolve(value=config, param_name="memory", config_class=MemoryConfig,
presets=MEMORY_PRESETS, url_schemes=MEMORY_URL_SCHEMES,
array_mode=ArrayMode.SINGLE_OR_LIST)
assert result is config
# 3. Dict
result = resolve(value={"backend": "sqlite"}, param_name="memory",
config_class=MemoryConfig, presets=MEMORY_PRESETS,
url_schemes=MEMORY_URL_SCHEMES, array_mode=ArrayMode.SINGLE_OR_LIST)
assert isinstance(result, MemoryConfig)
assert result.backend == "sqlite"
# 4. String preset
result = resolve(value="redis", param_name="memory", config_class=MemoryConfig,
presets=MEMORY_PRESETS, url_schemes=MEMORY_URL_SCHEMES,
array_mode=ArrayMode.SINGLE_OR_LIST)
assert isinstance(result, MemoryConfig)
# 5. String URL
result = resolve(value="postgresql://localhost/db", param_name="memory",
config_class=MemoryConfig, presets=MEMORY_PRESETS,
url_schemes=MEMORY_URL_SCHEMES, array_mode=ArrayMode.SINGLE_OR_LIST)
assert isinstance(result, MemoryConfig)
# 6. Bool True
result = resolve(value=True, param_name="memory", config_class=MemoryConfig,
presets=MEMORY_PRESETS, url_schemes=MEMORY_URL_SCHEMES,
array_mode=ArrayMode.SINGLE_OR_LIST)
assert isinstance(result, MemoryConfig)
# 7. Bool False → None
result = resolve(value=False, param_name="memory", config_class=MemoryConfig,
presets=MEMORY_PRESETS, url_schemes=MEMORY_URL_SCHEMES,
array_mode=ArrayMode.SINGLE_OR_LIST)
assert result is None
# =============================================================================
# Tests for Error Messages and Typo Suggestions
# =============================================================================
class TestErrorMessages:
"""Test error messages are helpful with typo suggestions."""
def test_invalid_preset_suggests_similar(self):
"""Invalid preset should suggest similar valid presets."""
from praisonaiagents.config.param_resolver import resolve
with pytest.raises(ValueError) as exc_info:
resolve(
value="verbos", # Typo
param_name="output",
config_class=MockConfig,
presets={"verbose": {}, "minimal": {}, "debug": {}},
)
error_msg = str(exc_info.value).lower()
assert "verbose" in error_msg # Should suggest "verbose"
def test_unknown_dict_keys_lists_valid_keys(self):
"""Unknown dict keys should list valid keys."""
from praisonaiagents.config.param_resolver import resolve
with pytest.raises(TypeError) as exc_info:
resolve(
value={"invalid_key": True},
param_name="test",
config_class=MockConfig,
)
error_msg = str(exc_info.value).lower()
assert "invalid_key" in error_msg or "unknown" in error_msg
assert "enabled" in error_msg or "valid" in error_msg
# =============================================================================
# Tests for Strict Dict Validation
# =============================================================================
class TestStrictDictValidation:
"""Test strict dict validation."""
def test_dict_without_config_class_raises_error(self):
"""Dict input without config_class should raise TypeError."""
from praisonaiagents.config.param_resolver import resolve
with pytest.raises(TypeError) as exc_info:
resolve(
value={"key": "value"},
param_name="test",
config_class=None, # No config class
)
assert "dict" in str(exc_info.value).lower()
def test_dict_with_unknown_keys_raises_error(self):
"""Dict with unknown keys should raise TypeError."""
from praisonaiagents.config.param_resolver import resolve
with pytest.raises(TypeError) as exc_info:
resolve(
value={"enabled": True, "unknown_field": "value"},
param_name="test",
config_class=MockConfig,
)
error_msg = str(exc_info.value)
assert "unknown_field" in error_msg.lower() or "unknown" in error_msg.lower()
# =============================================================================
# PHASE 1 TDD: Tests for Workflow/WorkflowStep New Params (ALL GREEN target)
# =============================================================================
class TestWorkflowNewParams:
"""Test Workflow supports all consolidated params like Agents."""
def test_workflow_has_autonomy_param(self):
"""Workflow should have autonomy param."""
from praisonaiagents.workflows.workflows import Workflow
# Should accept autonomy param without error
workflow = Workflow(steps=[], autonomy=True)
assert hasattr(workflow, 'autonomy') or hasattr(workflow, '_autonomy_config')
def test_workflow_has_knowledge_param(self):
"""Workflow should have knowledge param."""
from praisonaiagents.workflows.workflows import Workflow
workflow = Workflow(steps=[], knowledge=["docs/"])
assert hasattr(workflow, 'knowledge') or hasattr(workflow, '_knowledge_config')
def test_workflow_has_guardrails_param(self):
"""Workflow should have guardrails param."""
from praisonaiagents.workflows.workflows import Workflow
workflow = Workflow(steps=[], guardrails=True)
assert hasattr(workflow, 'guardrails') or hasattr(workflow, '_guardrails_config')
def test_workflow_has_web_param(self):
"""Workflow should have web param."""
from praisonaiagents.workflows.workflows import Workflow
workflow = Workflow(steps=[], web=True)
assert hasattr(workflow, 'web') or hasattr(workflow, '_web_config')
def test_workflow_has_reflection_param(self):
"""Workflow should have reflection param."""
from praisonaiagents.workflows.workflows import Workflow
workflow = Workflow(steps=[], reflection=True)
assert hasattr(workflow, 'reflection') or hasattr(workflow, '_reflection_config')
class TestWorkflowStepNewParams:
"""Test WorkflowStep supports all consolidated params."""
def test_workflowstep_has_autonomy_param(self):
"""WorkflowStep should have autonomy param."""
from praisonaiagents.workflows.workflows import WorkflowStep
step = WorkflowStep(name="test", autonomy=True)
assert hasattr(step, 'autonomy') or hasattr(step, '_autonomy_config')
def test_workflowstep_has_knowledge_param(self):
"""WorkflowStep should have knowledge param."""
from praisonaiagents.workflows.workflows import WorkflowStep
step = WorkflowStep(name="test", knowledge=["docs/"])
assert hasattr(step, 'knowledge') or hasattr(step, '_knowledge_config')
def test_workflowstep_has_web_param(self):
"""WorkflowStep should have web param."""
from praisonaiagents.workflows.workflows import WorkflowStep
step = WorkflowStep(name="test", web=True)
assert hasattr(step, 'web') or hasattr(step, '_web_config')
def test_workflowstep_has_reflection_param(self):
"""WorkflowStep should have reflection param."""
from praisonaiagents.workflows.workflows import WorkflowStep
step = WorkflowStep(name="test", reflection=True)
assert hasattr(step, 'reflection') or hasattr(step, '_reflection_config')
class TestWorkflowPropagation:
"""Test workflow defaults propagate to steps/agents."""
def test_workflow_defaults_available_for_steps(self):
"""Workflow-level configs should be accessible for step propagation."""
from praisonaiagents.workflows.workflows import Workflow
workflow = Workflow(
steps=[],
knowledge=["docs/"],
web=True,
guardrails=True,
)
# Workflow should store resolved configs for propagation
assert workflow.knowledge is not None or hasattr(workflow, '_knowledge_config')
assert workflow.web is not None or hasattr(workflow, '_web_config')
assert workflow.guardrails is not None or hasattr(workflow, '_guardrails_config')
if __name__ == "__main__":
pytest.main([__file__, "-v"])
+1 -1
View File
@@ -2422,7 +2422,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "0.10.8"
version = "0.10.10"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },
+2 -2
View File
@@ -3,8 +3,8 @@ class Praisonai < Formula
desc "AI tools for various AI applications"
homepage "https://github.com/MervinPraison/PraisonAI"
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.6.0.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.6.0.tar.gz | shasum -a 256`.split.first
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.6.2.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v3.6.2.tar.gz | shasum -a 256`.split.first
license "MIT"
depends_on "[email protected]"
@@ -68,6 +68,31 @@ def chat_main(
no_acp: bool = typer.Option(False, "--no-acp", help="Disable ACP tools"),
no_lsp: bool = typer.Option(False, "--no-lsp", help="Disable LSP tools"),
autonomy: bool = typer.Option(True, "--autonomy/--no-autonomy", help="Enable agent autonomy for complex tasks"),
# NEW: Agent-like consolidated params for ALL GREEN consistency
knowledge: Optional[str] = typer.Option(
None, "--knowledge", "-k",
help="Enable knowledge/RAG. Use --knowledge for default, --knowledge=docs/ for sources",
is_flag=False,
flag_value="true",
),
guardrails: Optional[str] = typer.Option(
None, "--guardrails",
help="Enable guardrails. Use --guardrails for default, --guardrails=strict for preset",
is_flag=False,
flag_value="true",
),
web: Optional[str] = typer.Option(
None, "--web",
help="Enable web search. Use --web for default, --web=duckduckgo for preset",
is_flag=False,
flag_value="true",
),
reflection: Optional[str] = typer.Option(
None, "--reflection",
help="Enable self-reflection. Use --reflection for default, --reflection=thorough for preset",
is_flag=False,
flag_value="true",
),
profile: bool = typer.Option(False, "--profile", help="Enable CLI profiling (timing breakdown)"),
profile_deep: bool = typer.Option(False, "--profile-deep", help="Enable deep profiling (cProfile stats, higher overhead)"),
):
@@ -724,3 +724,92 @@ def docs_serve(
pass
finally:
sys.argv = original_argv
@app.command("report")
def docs_report(
report_dir: Optional[Path] = typer.Argument(
None,
help="Report directory path (default: auto-detect latest)",
),
base_dir: Optional[Path] = typer.Option(
None,
"--base-dir", "-b",
help="Base reports directory (default: ~/Downloads/reports/docs)",
),
limit: int = typer.Option(
50,
"--limit", "-n",
help="Limit rows in tables (0 = no limit)",
),
errors: bool = typer.Option(
True,
"--errors/--no-errors",
help="Show error grouping section",
),
wide: bool = typer.Option(
False,
"--wide/--no-wide",
help="Wide output: full error messages and item lists",
),
show_paths: bool = typer.Option(
True,
"--show-paths/--no-show-paths",
help="Show affected item paths in error groups",
),
match: Optional[List[str]] = typer.Option(
None,
"--match", "-m",
help="Filter by status (repeatable: --match failed --match timeout)",
),
group: Optional[List[str]] = typer.Option(
None,
"--group", "-g",
help="Filter by group name (repeatable)",
),
contains: Optional[str] = typer.Option(
None,
"--contains", "-c",
help="Filter items whose error contains text (case-insensitive)",
),
open_dir: bool = typer.Option(
False,
"--open",
help="Show report artifacts listing",
),
output_format: str = typer.Option(
"table",
"--format", "-f",
help="Output format: table or json",
),
):
"""
View execution report with failures and error grouping.
Examples:
praisonai docs report # View latest report
praisonai docs report ./my-report # View specific report
praisonai docs report --wide --limit 0 # Full details
praisonai docs report --match failed # Only failures
praisonai docs report --format json # JSON output
"""
# Lazy import to avoid loading at CLI startup
from praisonai.suite_runner.report_viewer import view_report
output, exit_code = view_report(
report_dir=report_dir,
suite="docs",
base_dir=base_dir,
limit=limit,
show_errors=errors,
wide=wide,
show_paths=show_paths,
match_statuses=match,
match_groups=group,
contains=contains,
show_artifacts=open_dir,
output_format=output_format,
)
typer.echo(output)
raise typer.Exit(exit_code)
@@ -691,5 +691,94 @@ def examples_run_all(
raise typer.Exit(0)
@app.command("report")
def examples_report(
report_dir: Optional[Path] = typer.Argument(
None,
help="Report directory path (default: auto-detect latest)",
),
base_dir: Optional[Path] = typer.Option(
None,
"--base-dir", "-b",
help="Base reports directory (default: ~/Downloads/reports/examples)",
),
limit: int = typer.Option(
50,
"--limit", "-n",
help="Limit rows in tables (0 = no limit)",
),
errors: bool = typer.Option(
True,
"--errors/--no-errors",
help="Show error grouping section",
),
wide: bool = typer.Option(
False,
"--wide/--no-wide",
help="Wide output: full error messages and item lists",
),
show_paths: bool = typer.Option(
True,
"--show-paths/--no-show-paths",
help="Show affected item paths in error groups",
),
match: Optional[List[str]] = typer.Option(
None,
"--match", "-m",
help="Filter by status (repeatable: --match failed --match timeout)",
),
group: Optional[List[str]] = typer.Option(
None,
"--group", "-g",
help="Filter by group name (repeatable)",
),
contains: Optional[str] = typer.Option(
None,
"--contains", "-c",
help="Filter items whose error contains text (case-insensitive)",
),
open_dir: bool = typer.Option(
False,
"--open",
help="Show report artifacts listing",
),
output_format: str = typer.Option(
"table",
"--format", "-f",
help="Output format: table or json",
),
):
"""
View execution report with failures and error grouping.
Examples:
praisonai examples report # View latest report
praisonai examples report ./my-report # View specific report
praisonai examples report --wide --limit 0 # Full details
praisonai examples report --match failed # Only failures
praisonai examples report --format json # JSON output
"""
# Lazy import to avoid loading at CLI startup
from praisonai.suite_runner.report_viewer import view_report
output, exit_code = view_report(
report_dir=report_dir,
suite="examples",
base_dir=base_dir,
limit=limit,
show_errors=errors,
wide=wide,
show_paths=show_paths,
match_statuses=match,
match_groups=group,
contains=contains,
show_artifacts=open_dir,
output_format=output_format,
)
typer.echo(output)
raise typer.Exit(exit_code)
if __name__ == "__main__":
app()
+1 -1
View File
@@ -57,7 +57,7 @@ class CloudDeployer:
file.write("FROM python:3.11-slim\n")
file.write("WORKDIR /app\n")
file.write("COPY . .\n")
file.write("RUN pip install flask praisonai==3.6.0 gunicorn markdown\n")
file.write("RUN pip install flask praisonai==3.6.2 gunicorn markdown\n")
file.write("EXPOSE 8080\n")
file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
@@ -0,0 +1,793 @@
"""
Report Viewer for suite execution reports.
Provides CLI-friendly viewing of JSON/CSV reports with:
- Latest report auto-detection
- Failure tables with filtering
- Error grouping and deduplication
- ASCII table rendering (no external deps)
Used by both `praisonai examples report` and `praisonai docs report`.
"""
from __future__ import annotations
import json
import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, List, Optional, Tuple
# =============================================================================
# Data Models
# =============================================================================
@dataclass
class ReportItem:
"""Normalized report item from JSON/CSV."""
item_id: str
suite: str
group: str
source_path: str
block_index: int = 0
status: str = "not_run"
duration_seconds: float = 0.0
exit_code: int = 0
error_type: Optional[str] = None
error_message: Optional[str] = None
error_signature: str = "" # Normalized for grouping
skip_reason: Optional[str] = None
stdout_path: Optional[str] = None
stderr_path: Optional[str] = None
start_time: str = ""
end_time: str = ""
runnable_decision: str = ""
@property
def display_name(self) -> str:
"""Human-readable display name."""
path = Path(self.source_path)
if self.suite == "docs" and self.block_index > 0:
return f"{path.name}:{self.block_index}"
return path.name
@dataclass
class ReportMetadata:
"""Report metadata from JSON."""
suite: str = ""
source_path: str = ""
timestamp: str = ""
platform: str = ""
python_version: str = ""
python_executable: str = ""
praisonai_version: str = ""
git_commit: Optional[str] = None
groups_run: List[str] = field(default_factory=list)
totals: Dict[str, int] = field(default_factory=dict)
@dataclass
class ErrorGroup:
"""Group of items with similar errors."""
signature: str
error_type: str
sample_message: str
count: int = 0
items: List[ReportItem] = field(default_factory=list)
@dataclass
class LoadedReport:
"""Complete loaded and normalized report."""
metadata: ReportMetadata
items: List[ReportItem]
report_dir: Path
@property
def failures(self) -> List[ReportItem]:
"""Get failed/timeout/error items."""
return [i for i in self.items if i.status in ("failed", "timeout", "error")]
# =============================================================================
# Error Signature Normalization
# =============================================================================
# Patterns to strip from error messages for grouping
_STRIP_PATTERNS = [
# File paths
(r'File "[^"]+[/\\][^"]+\.py"', 'File "...py"'),
(r'/[^\s:]+\.py', '...py'),
(r'[A-Z]:\\[^\s:]+\.py', '...py'),
# Line numbers
(r', line \d+', ', line N'),
(r'line \d+', 'line N'),
# Memory addresses
(r'0x[0-9a-fA-F]+', '0x...'),
# Request IDs, UUIDs
(r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}', 'UUID'),
(r'req_[a-zA-Z0-9]+', 'req_...'),
# Timestamps
(r'\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}[^\s]*', 'TIMESTAMP'),
# Numbers in brackets
(r'\[\d+\]', '[N]'),
# Port numbers
(r':\d{4,5}(?=/|$|\s)', ':PORT'),
# Whitespace normalization
(r'\s+', ' '),
]
def normalize_error_signature(error_type: Optional[str], error_message: Optional[str]) -> str:
"""
Create normalized error signature for grouping.
Strips variable parts (paths, line numbers, addresses, timestamps)
to group similar errors together.
"""
if not error_message and not error_type:
return "unknown_error"
# Start with error type
sig_parts = []
if error_type:
sig_parts.append(error_type)
if error_message:
msg = error_message.strip()
# Apply all normalization patterns
for pattern, replacement in _STRIP_PATTERNS:
msg = re.sub(pattern, replacement, msg)
# Take first 100 chars of normalized message
msg = msg[:100].strip()
if msg:
sig_parts.append(msg)
return ": ".join(sig_parts) if sig_parts else "unknown_error"
# =============================================================================
# Report Discovery
# =============================================================================
def find_latest_report_dir(suite: str, base_dir: Optional[Path] = None) -> Optional[Path]:
"""
Find the latest report directory for a suite.
Args:
suite: "examples" or "docs"
base_dir: Base reports directory (default: ~/Downloads/reports/<suite>)
Returns:
Path to latest report directory, or None if not found.
"""
if base_dir is None:
base_dir = Path.home() / "Downloads" / "reports" / suite
if not base_dir.exists():
return None
# Check if base_dir itself contains report.json (user passed specific run dir)
if (base_dir / "report.json").exists():
return base_dir
# Find subdirectories with report.json
candidates = []
for subdir in base_dir.iterdir():
if subdir.is_dir():
# Check for report.json in subdir or in group subdirs
if (subdir / "report.json").exists():
candidates.append(subdir)
else:
# Check if this is a run dir with group subdirs
for group_dir in subdir.iterdir():
if group_dir.is_dir() and (group_dir / "report.json").exists():
candidates.append(subdir)
break
if not candidates:
return None
# Sort by mtime (newest first)
candidates.sort(key=lambda p: p.stat().st_mtime, reverse=True)
return candidates[0]
def find_report_files(report_dir: Path) -> List[Path]:
"""
Find all report.json files in a report directory.
Handles both single-group and multi-group report structures.
"""
report_files = []
# Check for report.json directly in report_dir
direct = report_dir / "report.json"
if direct.exists():
report_files.append(direct)
# Check for group subdirectories
for subdir in report_dir.iterdir():
if subdir.is_dir():
group_report = subdir / "report.json"
if group_report.exists():
report_files.append(group_report)
return report_files
# =============================================================================
# Report Loading
# =============================================================================
def load_report(report_dir: Path) -> LoadedReport:
"""
Load and normalize a report from a directory.
Aggregates all group reports if multiple exist.
Args:
report_dir: Path to report run directory.
Returns:
LoadedReport with normalized items.
Raises:
FileNotFoundError: If no report.json found.
ValueError: If report cannot be parsed.
"""
report_files = find_report_files(report_dir)
if not report_files:
raise FileNotFoundError(f"No report.json found in {report_dir}")
all_items: List[ReportItem] = []
metadata = ReportMetadata()
groups_seen = set()
for report_file in report_files:
try:
data = json.loads(report_file.read_text(encoding='utf-8'))
except json.JSONDecodeError as e:
raise ValueError(f"Invalid JSON in {report_file}: {e}")
# Parse metadata from first report
if not metadata.suite and "metadata" in data:
meta = data["metadata"]
metadata = ReportMetadata(
suite=meta.get("suite", ""),
source_path=meta.get("source_path", ""),
timestamp=meta.get("timestamp", ""),
platform=meta.get("platform", ""),
python_version=meta.get("python_version", ""),
python_executable=meta.get("python_executable", ""),
praisonai_version=meta.get("praisonai_version", ""),
git_commit=meta.get("git_commit"),
groups_run=meta.get("groups_run", []),
totals=meta.get("totals", {}),
)
# Parse results
for r in data.get("results", []):
error_sig = normalize_error_signature(
r.get("error_type"),
r.get("error_message")
)
item = ReportItem(
item_id=r.get("item_id", ""),
suite=r.get("suite", ""),
group=r.get("group", ""),
source_path=r.get("source_path", ""),
block_index=r.get("block_index", 0),
status=r.get("status", "not_run"),
duration_seconds=r.get("duration_seconds", 0.0),
exit_code=r.get("exit_code", 0),
error_type=r.get("error_type"),
error_message=r.get("error_message"),
error_signature=error_sig,
skip_reason=r.get("skip_reason"),
stdout_path=r.get("stdout_path"),
stderr_path=r.get("stderr_path"),
start_time=r.get("start_time", ""),
end_time=r.get("end_time", ""),
runnable_decision=r.get("runnable_decision", ""),
)
all_items.append(item)
groups_seen.add(item.group)
# Update metadata with aggregated groups
if groups_seen:
metadata.groups_run = sorted(groups_seen)
# Recalculate totals from all items
totals = {"passed": 0, "failed": 0, "skipped": 0, "timeout": 0, "not_run": 0, "xfail": 0, "total": 0}
for item in all_items:
totals["total"] += 1
if item.status in totals:
totals[item.status] += 1
metadata.totals = totals
return LoadedReport(
metadata=metadata,
items=all_items,
report_dir=report_dir,
)
# =============================================================================
# Error Grouping
# =============================================================================
def group_errors(items: List[ReportItem]) -> List[ErrorGroup]:
"""
Group items by error signature.
Returns groups sorted by count (descending), then signature.
"""
groups: Dict[str, ErrorGroup] = {}
for item in items:
if item.status not in ("failed", "timeout", "error"):
continue
sig = item.error_signature or "unknown_error"
if sig not in groups:
groups[sig] = ErrorGroup(
signature=sig,
error_type=item.error_type or "",
sample_message=item.error_message or "",
)
groups[sig].count += 1
groups[sig].items.append(item)
# Sort by count desc, then signature asc
sorted_groups = sorted(
groups.values(),
key=lambda g: (-g.count, g.signature)
)
return sorted_groups
# =============================================================================
# ASCII Table Rendering
# =============================================================================
def _truncate(text: str, max_len: int, ellipsis: str = "...") -> str:
"""Truncate text with ellipsis if too long."""
if not text:
return ""
text = text.replace("\n", " ").replace("\r", "")
if len(text) <= max_len:
return text
return text[:max_len - len(ellipsis)] + ellipsis
def _render_table(
headers: List[str],
rows: List[List[str]],
col_widths: Optional[List[int]] = None,
max_width: int = 120,
) -> str:
"""
Render a simple ASCII table.
Args:
headers: Column headers.
rows: List of row data (list of strings).
col_widths: Optional fixed column widths.
max_width: Maximum total table width.
Returns:
Formatted table string.
"""
if not headers:
return ""
num_cols = len(headers)
# Calculate column widths
if col_widths is None:
col_widths = [len(h) for h in headers]
for row in rows:
for i, cell in enumerate(row[:num_cols]):
col_widths[i] = max(col_widths[i], len(str(cell)))
# Ensure we don't exceed max_width
total_width = sum(col_widths) + (num_cols - 1) * 3 + 4 # separators + borders
if total_width > max_width and col_widths:
# Shrink largest columns proportionally
excess = total_width - max_width
while excess > 0 and max(col_widths) > 10:
max_idx = col_widths.index(max(col_widths))
col_widths[max_idx] -= 1
excess -= 1
lines = []
# Header separator
sep = "+" + "+".join("-" * (w + 2) for w in col_widths) + "+"
lines.append(sep)
# Header row
header_cells = [
f" {_truncate(h, w).ljust(w)} " for h, w in zip(headers, col_widths)
]
lines.append("|" + "|".join(header_cells) + "|")
lines.append(sep)
# Data rows
for row in rows:
cells = []
for i, w in enumerate(col_widths):
cell = str(row[i]) if i < len(row) else ""
cells.append(f" {_truncate(cell, w).ljust(w)} ")
lines.append("|" + "|".join(cells) + "|")
lines.append(sep)
return "\n".join(lines)
# =============================================================================
# Report Rendering
# =============================================================================
STATUS_ICONS = {
"passed": "✅",
"failed": "❌",
"skipped": "⏭️",
"timeout": "⏱️",
"not_run": "📝",
"xfail": "⚠️",
"error": "💥",
}
def render_overview(report: LoadedReport, wide: bool = False) -> str:
"""Render overview section with totals and metadata."""
meta = report.metadata
totals = meta.totals
lines = [
"",
"=" * 60,
f" REPORT: {meta.suite.upper()} EXECUTION",
"=" * 60,
"",
f" Timestamp: {meta.timestamp}",
f" Platform: {meta.platform}",
f" Python: {meta.python_version}",
f" PraisonAI: {meta.praisonai_version}",
]
if meta.git_commit:
lines.append(f" Git Commit: {meta.git_commit}")
if meta.groups_run:
groups_str = ", ".join(meta.groups_run[:10])
if len(meta.groups_run) > 10:
groups_str += f" (+{len(meta.groups_run) - 10} more)"
lines.append(f" Groups: {groups_str}")
lines.append(f" Report Dir: {report.report_dir}")
lines.append("")
# Summary table
lines.append(" SUMMARY")
lines.append(" " + "-" * 40)
total = totals.get("total", 0)
for status in ["passed", "failed", "timeout", "skipped", "not_run", "xfail"]:
count = totals.get(status, 0)
icon = STATUS_ICONS.get(status, "?")
pct = f"({count * 100 / total:.1f}%)" if total > 0 else ""
lines.append(f" {icon} {status.capitalize():12} {count:5} {pct}")
lines.append(" " + "-" * 40)
lines.append(f" {'Total':15} {total:5}")
lines.append("")
return "\n".join(lines)
def render_failures_table(
report: LoadedReport,
limit: int = 50,
match_statuses: Optional[List[str]] = None,
match_groups: Optional[List[str]] = None,
contains: Optional[str] = None,
wide: bool = False,
) -> str:
"""
Render failures table.
Args:
report: Loaded report.
limit: Max rows to show (0 = no limit).
match_statuses: Filter by status(es).
match_groups: Filter by group(s).
contains: Filter by error message substring.
wide: Show full error messages.
Returns:
Formatted table string.
"""
# Default to failure statuses
if match_statuses is None:
match_statuses = ["failed", "timeout", "error"]
# Filter items
items = []
for item in report.items:
if item.status not in match_statuses:
continue
if match_groups and item.group not in match_groups:
continue
if contains:
msg = (item.error_message or "").lower()
if contains.lower() not in msg:
continue
items.append(item)
if not items:
return "\n No matching failures found.\n"
# Apply limit
total_count = len(items)
if limit > 0 and len(items) > limit:
items = items[:limit]
lines = [
"",
f" FAILURES ({total_count} total, showing {len(items)})",
" " + "-" * 50,
"",
]
# Build table
headers = ["Group", "Item", "Status", "Duration", "Error"]
rows = []
error_width = 60 if wide else 30
for item in items:
icon = STATUS_ICONS.get(item.status, "?")
duration = f"{item.duration_seconds:.2f}s" if item.duration_seconds else "-"
error = item.error_message or item.skip_reason or "-"
rows.append([
item.group,
item.display_name,
f"{icon} {item.status}",
duration,
_truncate(error, error_width) if not wide else error[:200],
])
col_widths = [15, 30, 12, 10, error_width]
lines.append(_render_table(headers, rows, col_widths if not wide else None))
if limit > 0 and total_count > limit:
lines.append(f"\n ... and {total_count - limit} more (use --limit 0 to show all)")
lines.append("")
return "\n".join(lines)
def render_error_groups(
report: LoadedReport,
limit: int = 20,
wide: bool = False,
show_paths: bool = True,
) -> str:
"""
Render error grouping section.
Args:
report: Loaded report.
limit: Max groups to show (0 = no limit).
wide: Show full error messages and all paths.
show_paths: Show affected item paths.
Returns:
Formatted error groups string.
"""
groups = group_errors(report.items)
if not groups:
return "\n No errors to group.\n"
total_groups = len(groups)
if limit > 0 and len(groups) > limit:
groups = groups[:limit]
lines = [
"",
f" ERROR GROUPS ({total_groups} unique errors)",
" " + "=" * 50,
"",
]
for i, grp in enumerate(groups, 1):
lines.append(f" [{i}] {grp.error_type or 'Error'} ({grp.count} occurrences)")
lines.append(" " + "-" * 40)
# Show signature/sample message
if wide:
msg = grp.sample_message or grp.signature
lines.append(f" {msg[:500]}")
else:
lines.append(f" {_truncate(grp.signature, 80)}")
# Show affected items
if show_paths:
lines.append("")
lines.append(" Affected items:")
items_to_show = grp.items if wide else grp.items[:5]
for item in items_to_show:
lines.append(f" - {item.group}/{item.display_name}")
if not wide and len(grp.items) > 5:
lines.append(f" ... and {len(grp.items) - 5} more")
lines.append("")
if limit > 0 and total_groups > limit:
lines.append(f" ... and {total_groups - limit} more error groups (use --limit 0 to show all)")
lines.append("")
return "\n".join(lines)
def render_artifacts(report: LoadedReport) -> str:
"""Render list of report artifacts."""
lines = [
"",
" REPORT ARTIFACTS",
" " + "-" * 40,
f" Directory: {report.report_dir}",
"",
]
# List files
for f in sorted(report.report_dir.rglob("*")):
if f.is_file():
rel = f.relative_to(report.report_dir)
size = f.stat().st_size
size_str = f"{size:,} bytes" if size < 1024 else f"{size / 1024:.1f} KB"
lines.append(f" {rel}: {size_str}")
lines.append("")
return "\n".join(lines)
# =============================================================================
# Main View Function
# =============================================================================
def view_report(
report_dir: Optional[Path] = None,
suite: str = "examples",
base_dir: Optional[Path] = None,
limit: int = 50,
show_errors: bool = True,
wide: bool = False,
show_paths: bool = True,
match_statuses: Optional[List[str]] = None,
match_groups: Optional[List[str]] = None,
contains: Optional[str] = None,
show_artifacts: bool = False,
output_format: str = "table",
) -> Tuple[str, int]:
"""
Main entry point for viewing a report.
Args:
report_dir: Explicit report directory (or None for auto-detect).
suite: Suite name for auto-detection.
base_dir: Base reports directory for auto-detection.
limit: Max rows in tables.
show_errors: Show error grouping section.
wide: Wide output mode.
show_paths: Show paths in error groups.
match_statuses: Filter by status(es).
match_groups: Filter by group(s).
contains: Filter by error message substring.
show_artifacts: Show artifact listing.
output_format: "table" or "json".
Returns:
Tuple of (output_string, exit_code).
Exit code is 0 if no failures, 1 if failures exist.
"""
# Find report directory
if report_dir is None:
report_dir = find_latest_report_dir(suite, base_dir)
if report_dir is None:
return f"Error: No reports found for suite '{suite}'", 1
if not report_dir.exists():
return f"Error: Report directory not found: {report_dir}", 1
# Load report
try:
report = load_report(report_dir)
except FileNotFoundError as e:
return f"Error: {e}", 1
except ValueError as e:
return f"Error: {e}", 1
# JSON output
if output_format == "json":
output = {
"metadata": {
"suite": report.metadata.suite,
"timestamp": report.metadata.timestamp,
"totals": report.metadata.totals,
"groups_run": report.metadata.groups_run,
},
"failures": [
{
"item_id": i.item_id,
"group": i.group,
"display_name": i.display_name,
"status": i.status,
"error_type": i.error_type,
"error_message": i.error_message,
}
for i in report.failures
],
"error_groups": [
{
"signature": g.signature,
"error_type": g.error_type,
"count": g.count,
"items": [i.display_name for i in g.items],
}
for g in group_errors(report.items)
],
}
return json.dumps(output, indent=2), 0 if not report.failures else 1
# Table output
parts = []
# Overview
parts.append(render_overview(report, wide=wide))
# Failures table
parts.append(render_failures_table(
report,
limit=limit,
match_statuses=match_statuses,
match_groups=match_groups,
contains=contains,
wide=wide,
))
# Error groups
if show_errors:
parts.append(render_error_groups(
report,
limit=limit,
wide=wide,
show_paths=show_paths,
))
# Artifacts
if show_artifacts:
parts.append(render_artifacts(report))
output = "\n".join(parts)
exit_code = 0 if not report.failures else 1
return output, exit_code
+1 -1
View File
@@ -1 +1 @@
__version__ = "3.6.0"
__version__ = "3.6.2"
+1 -1
View File
@@ -12,7 +12,7 @@ dependencies = [
"rich>=13.7",
"markdown>=3.5",
"pyparsing>=3.0.0",
"praisonaiagents>=0.10.8",
"praisonaiagents>=0.10.10",
"python-dotenv>=0.19.0",
"instructor>=1.3.3",
"PyYAML>=6.0",
+18
View File
@@ -130,6 +130,19 @@ def bump_version(new_version: str, agents_version: Optional[str] = None):
print("\n✨ Version bump complete!")
def validate_dependencies() -> bool:
"""Validate that uv lock will succeed before making any changes."""
praisonai_dir = get_praisonai_dir()
print("\n🔍 Validating dependencies (dry-run)...")
result = run(["uv", "lock", "--dry-run"], cwd=praisonai_dir, check=False)
if result.returncode != 0:
print("\n❌ Dependency validation failed. Fix the issues above before releasing.")
return False
print(" ✅ Dependencies validated successfully")
return True
def release(version: str):
"""Run the release process."""
root = get_project_root()
@@ -215,6 +228,11 @@ def main():
# Run bump version
bump_version(args.version, args.agents)
# Validate dependencies after version bump (before git operations)
if not validate_dependencies():
print("\n💡 Tip: Revert changes with 'git checkout .' if needed")
sys.exit(1)
# Run release
release(args.version)
@@ -0,0 +1,557 @@
"""
Unit tests for report_viewer module.
Tests cover:
- Latest report directory detection
- JSON parsing and loading
- Error signature normalization
- Error grouping
- Table rendering
- CLI integration smoke test
"""
import json
import pytest
from praisonai.suite_runner.report_viewer import (
normalize_error_signature,
find_latest_report_dir,
find_report_files,
load_report,
group_errors,
render_overview,
render_failures_table,
render_error_groups,
view_report,
_truncate,
_render_table,
)
# =============================================================================
# Fixtures
# =============================================================================
@pytest.fixture
def sample_report_json():
"""Sample report JSON structure."""
return {
"metadata": {
"suite": "examples",
"source_path": "/path/to/examples",
"timestamp": "2026-01-09T14:26:07.808316+00:00",
"platform": "Darwin-25.3.0-arm64",
"python_version": "3.12.11",
"python_executable": "/usr/bin/python",
"praisonai_version": "3.5.8",
"git_commit": "abc1234",
"cli_args": [],
"groups_run": ["acp", "tools"],
"totals": {
"passed": 5,
"failed": 2,
"skipped": 1,
"timeout": 1,
"not_run": 0,
"xfail": 0,
"total": 9,
},
},
"results": [
{
"item_id": "acp__test1",
"suite": "examples",
"group": "acp",
"source_path": "/path/to/examples/acp/test1.py",
"block_index": 0,
"language": "python",
"status": "passed",
"duration_seconds": 1.5,
"exit_code": 0,
},
{
"item_id": "acp__test2",
"suite": "examples",
"group": "acp",
"source_path": "/path/to/examples/acp/test2.py",
"block_index": 0,
"language": "python",
"status": "failed",
"duration_seconds": 2.0,
"exit_code": 1,
"error_type": "ImportError",
"error_message": "No module named 'missing_module'",
},
{
"item_id": "tools__test3",
"suite": "examples",
"group": "tools",
"source_path": "/path/to/examples/tools/test3.py",
"block_index": 0,
"language": "python",
"status": "failed",
"duration_seconds": 0.5,
"exit_code": 1,
"error_type": "ImportError",
"error_message": "No module named 'missing_module'",
},
{
"item_id": "tools__test4",
"suite": "examples",
"group": "tools",
"source_path": "/path/to/examples/tools/test4.py",
"block_index": 0,
"language": "python",
"status": "timeout",
"duration_seconds": 60.0,
"exit_code": -1,
"error_type": "TimeoutError",
"error_message": "Execution timed out after 60 seconds",
},
],
}
@pytest.fixture
def report_dir(tmp_path, sample_report_json):
"""Create a temporary report directory with sample data."""
report_path = tmp_path / "report.json"
report_path.write_text(json.dumps(sample_report_json), encoding='utf-8')
return tmp_path
@pytest.fixture
def multi_group_report_dir(tmp_path, sample_report_json):
"""Create a report directory with multiple group subdirs."""
# Create group subdirectories
acp_dir = tmp_path / "acp"
acp_dir.mkdir()
tools_dir = tmp_path / "tools"
tools_dir.mkdir()
# Write separate reports for each group
acp_report = {
"metadata": sample_report_json["metadata"].copy(),
"results": [r for r in sample_report_json["results"] if r["group"] == "acp"],
}
acp_report["metadata"]["groups_run"] = ["acp"]
(acp_dir / "report.json").write_text(json.dumps(acp_report), encoding='utf-8')
tools_report = {
"metadata": sample_report_json["metadata"].copy(),
"results": [r for r in sample_report_json["results"] if r["group"] == "tools"],
}
tools_report["metadata"]["groups_run"] = ["tools"]
(tools_dir / "report.json").write_text(json.dumps(tools_report), encoding='utf-8')
return tmp_path
# =============================================================================
# Error Signature Normalization Tests
# =============================================================================
class TestNormalizeErrorSignature:
"""Tests for error signature normalization."""
def test_basic_error(self):
"""Test basic error type and message."""
sig = normalize_error_signature("ImportError", "No module named 'foo'")
assert "ImportError" in sig
assert "No module named" in sig
def test_strips_file_paths(self):
"""Test that file paths are normalized."""
msg = 'File "/Users/praison/project/test.py", line 42, in foo'
sig = normalize_error_signature("SyntaxError", msg)
assert "/Users/praison" not in sig
assert "...py" in sig or "line N" in sig
def test_strips_line_numbers(self):
"""Test that line numbers are normalized."""
msg = "Error at line 123 in function"
sig = normalize_error_signature("Error", msg)
assert "123" not in sig
assert "line N" in sig
def test_strips_memory_addresses(self):
"""Test that memory addresses are normalized."""
msg = "Object at 0x7f8b1c2d3e4f crashed"
sig = normalize_error_signature("RuntimeError", msg)
assert "0x7f8b1c2d3e4f" not in sig
assert "0x..." in sig
def test_strips_uuids(self):
"""Test that UUIDs are normalized."""
msg = "Request 550e8400-e29b-41d4-a716-446655440000 failed"
sig = normalize_error_signature("Error", msg)
assert "550e8400" not in sig
assert "UUID" in sig
def test_strips_timestamps(self):
"""Test that timestamps are normalized."""
msg = "Error at 2026-01-09T14:30:00Z"
sig = normalize_error_signature("Error", msg)
assert "2026-01-09" not in sig
assert "TIMESTAMP" in sig
def test_empty_message(self):
"""Test with empty message."""
sig = normalize_error_signature("Error", "")
assert sig == "Error"
def test_none_values(self):
"""Test with None values."""
sig = normalize_error_signature(None, None)
assert sig == "unknown_error"
def test_truncates_long_messages(self):
"""Test that long messages are truncated."""
long_msg = "x" * 200
sig = normalize_error_signature("Error", long_msg)
assert len(sig) <= 110 # Error: + 100 chars
# =============================================================================
# Report Discovery Tests
# =============================================================================
class TestFindLatestReportDir:
"""Tests for latest report directory detection."""
def test_finds_direct_report(self, report_dir):
"""Test finding report.json directly in directory."""
result = find_latest_report_dir("examples", report_dir)
assert result == report_dir
def test_finds_latest_by_mtime(self, tmp_path):
"""Test finding latest directory by modification time."""
# Create older directory
old_dir = tmp_path / "20260101_120000"
old_dir.mkdir()
(old_dir / "report.json").write_text('{"metadata": {}, "results": []}')
# Create newer directory
new_dir = tmp_path / "20260109_140000"
new_dir.mkdir()
(new_dir / "report.json").write_text('{"metadata": {}, "results": []}')
result = find_latest_report_dir("examples", tmp_path)
assert result == new_dir
def test_returns_none_for_empty_dir(self, tmp_path):
"""Test returns None when no reports found."""
result = find_latest_report_dir("examples", tmp_path)
assert result is None
def test_returns_none_for_nonexistent_dir(self, tmp_path):
"""Test returns None for nonexistent directory."""
result = find_latest_report_dir("examples", tmp_path / "nonexistent")
assert result is None
class TestFindReportFiles:
"""Tests for finding report files in a directory."""
def test_finds_direct_report(self, report_dir):
"""Test finding report.json directly."""
files = find_report_files(report_dir)
assert len(files) == 1
assert files[0].name == "report.json"
def test_finds_group_reports(self, multi_group_report_dir):
"""Test finding reports in group subdirectories."""
files = find_report_files(multi_group_report_dir)
assert len(files) == 2
names = {f.parent.name for f in files}
assert names == {"acp", "tools"}
# =============================================================================
# Report Loading Tests
# =============================================================================
class TestLoadReport:
"""Tests for loading and parsing reports."""
def test_loads_single_report(self, report_dir):
"""Test loading a single report.json."""
report = load_report(report_dir)
assert report.metadata.suite == "examples"
assert len(report.items) == 4
def test_loads_multi_group_report(self, multi_group_report_dir):
"""Test loading and aggregating multiple group reports."""
report = load_report(multi_group_report_dir)
assert len(report.items) == 4
assert set(report.metadata.groups_run) == {"acp", "tools"}
def test_calculates_totals(self, report_dir):
"""Test that totals are calculated correctly."""
report = load_report(report_dir)
totals = report.metadata.totals
assert totals["passed"] == 1
assert totals["failed"] == 2
assert totals["timeout"] == 1
assert totals["total"] == 4
def test_raises_for_missing_report(self, tmp_path):
"""Test raises FileNotFoundError for missing report."""
with pytest.raises(FileNotFoundError):
load_report(tmp_path)
def test_raises_for_invalid_json(self, tmp_path):
"""Test raises ValueError for invalid JSON."""
(tmp_path / "report.json").write_text("not valid json")
with pytest.raises(ValueError):
load_report(tmp_path)
def test_normalizes_error_signatures(self, report_dir):
"""Test that error signatures are normalized during load."""
report = load_report(report_dir)
failed_items = [i for i in report.items if i.status == "failed"]
assert all(i.error_signature for i in failed_items)
# =============================================================================
# Error Grouping Tests
# =============================================================================
class TestGroupErrors:
"""Tests for error grouping functionality."""
def test_groups_similar_errors(self, report_dir):
"""Test that similar errors are grouped together."""
report = load_report(report_dir)
groups = group_errors(report.items)
# Should have 2 groups: ImportError and TimeoutError
assert len(groups) == 2
def test_counts_occurrences(self, report_dir):
"""Test that occurrence counts are correct."""
report = load_report(report_dir)
groups = group_errors(report.items)
# Find ImportError group (should have 2 occurrences)
import_group = next((g for g in groups if "ImportError" in g.error_type), None)
assert import_group is not None
assert import_group.count == 2
def test_sorts_by_count_desc(self, report_dir):
"""Test that groups are sorted by count descending."""
report = load_report(report_dir)
groups = group_errors(report.items)
counts = [g.count for g in groups]
assert counts == sorted(counts, reverse=True)
def test_ignores_passed_items(self, report_dir):
"""Test that passed items are not grouped."""
report = load_report(report_dir)
groups = group_errors(report.items)
all_items = [i for g in groups for i in g.items]
assert all(i.status in ("failed", "timeout", "error") for i in all_items)
# =============================================================================
# Rendering Tests
# =============================================================================
class TestTruncate:
"""Tests for text truncation."""
def test_short_text_unchanged(self):
"""Test that short text is not truncated."""
assert _truncate("hello", 10) == "hello"
def test_long_text_truncated(self):
"""Test that long text is truncated with ellipsis."""
result = _truncate("hello world", 8)
assert len(result) == 8
assert result.endswith("...")
def test_empty_text(self):
"""Test empty text handling."""
assert _truncate("", 10) == ""
assert _truncate(None, 10) == ""
def test_newlines_removed(self):
"""Test that newlines are removed."""
assert "\n" not in _truncate("hello\nworld", 20)
class TestRenderTable:
"""Tests for ASCII table rendering."""
def test_renders_headers(self):
"""Test that headers are rendered."""
table = _render_table(["A", "B"], [["1", "2"]])
assert "A" in table
assert "B" in table
def test_renders_data(self):
"""Test that data rows are rendered."""
table = _render_table(["Col"], [["value1"], ["value2"]])
assert "value1" in table
assert "value2" in table
def test_empty_table(self):
"""Test empty table handling."""
table = _render_table([], [])
assert table == ""
class TestRenderOverview:
"""Tests for overview rendering."""
def test_includes_metadata(self, report_dir):
"""Test that overview includes metadata."""
report = load_report(report_dir)
output = render_overview(report)
assert "EXAMPLES" in output.upper()
assert "Python" in output or "python" in output
def test_includes_totals(self, report_dir):
"""Test that overview includes totals."""
report = load_report(report_dir)
output = render_overview(report)
assert "Passed" in output or "passed" in output
assert "Failed" in output or "failed" in output
class TestRenderFailuresTable:
"""Tests for failures table rendering."""
def test_shows_failures(self, report_dir):
"""Test that failures are shown."""
report = load_report(report_dir)
output = render_failures_table(report)
assert "test2.py" in output or "test3.py" in output
def test_respects_limit(self, report_dir):
"""Test that limit is respected."""
report = load_report(report_dir)
output = render_failures_table(report, limit=1)
# Should mention there are more
assert "more" in output.lower() or "1" in output
def test_filters_by_status(self, report_dir):
"""Test filtering by status."""
report = load_report(report_dir)
output = render_failures_table(report, match_statuses=["timeout"])
assert "timeout" in output.lower()
class TestRenderErrorGroups:
"""Tests for error groups rendering."""
def test_shows_groups(self, report_dir):
"""Test that error groups are shown."""
report = load_report(report_dir)
output = render_error_groups(report)
assert "ERROR GROUPS" in output
assert "ImportError" in output or "TimeoutError" in output
def test_shows_counts(self, report_dir):
"""Test that occurrence counts are shown."""
report = load_report(report_dir)
output = render_error_groups(report)
assert "occurrence" in output.lower()
# =============================================================================
# Integration Tests
# =============================================================================
class TestViewReport:
"""Integration tests for view_report function."""
def test_table_output(self, report_dir):
"""Test table output format."""
output, exit_code = view_report(
report_dir=report_dir,
suite="examples",
output_format="table",
)
assert "EXAMPLES" in output.upper()
assert exit_code == 1 # Has failures
def test_json_output(self, report_dir):
"""Test JSON output format."""
output, exit_code = view_report(
report_dir=report_dir,
suite="examples",
output_format="json",
)
data = json.loads(output)
assert "metadata" in data
assert "failures" in data
assert "error_groups" in data
def test_exit_code_zero_for_no_failures(self, tmp_path):
"""Test exit code is 0 when no failures."""
report = {
"metadata": {"suite": "examples", "totals": {"passed": 1, "total": 1}},
"results": [
{
"item_id": "test1",
"suite": "examples",
"group": "test",
"source_path": "/test.py",
"status": "passed",
}
],
}
(tmp_path / "report.json").write_text(json.dumps(report))
output, exit_code = view_report(report_dir=tmp_path, suite="examples")
assert exit_code == 0
def test_error_for_missing_report(self, tmp_path):
"""Test error message for missing report."""
output, exit_code = view_report(
report_dir=tmp_path / "nonexistent",
suite="examples",
)
assert "Error" in output
assert exit_code == 1
def test_auto_detect_latest(self, tmp_path):
"""Test auto-detection of latest report."""
# Create a report directory
run_dir = tmp_path / "20260109_140000"
run_dir.mkdir()
report = {
"metadata": {"suite": "examples", "totals": {"passed": 1, "total": 1}},
"results": [
{
"item_id": "test1",
"suite": "examples",
"group": "test",
"source_path": "/test.py",
"status": "passed",
}
],
}
(run_dir / "report.json").write_text(json.dumps(report))
output, exit_code = view_report(
report_dir=None,
suite="examples",
base_dir=tmp_path,
)
assert "EXAMPLES" in output.upper()
assert exit_code == 0
+4 -4
View File
@@ -4276,7 +4276,7 @@ requires-dist = [
{ name = "praisonai-tools", marker = "extra == 'autogen'", specifier = ">=0.1.0" },
{ name = "praisonai-tools", marker = "extra == 'autogen-v4'", specifier = ">=0.1.0" },
{ name = "praisonai-tools", marker = "extra == 'crewai'", specifier = ">=0.1.0" },
{ name = "praisonaiagents", specifier = ">=0.10.8" },
{ name = "praisonaiagents", specifier = ">=0.10.10" },
{ name = "pyautogen", marker = "extra == 'autogen'", specifier = "==0.2.29" },
{ name = "pydantic", marker = "extra == 'chat'", specifier = "<=2.10.1" },
{ name = "pydantic", marker = "extra == 'code'", specifier = "<=2.10.1" },
@@ -4324,7 +4324,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "0.10.8"
version = "0.10.10"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "aiohttp" },
@@ -4333,9 +4333,9 @@ dependencies = [
{ name = "pydantic" },
{ name = "rich" },
]
sdist = { url = "https://files.pythonhosted.org/packages/d9/29/3d3d35fc6b0d9f8284c2099015dbeb08dcaab0c90c224c42f817f14c3d76/praisonaiagents-0.10.8.tar.gz", hash = "sha256:5f134ade419025ef12c3770945e91b40b11b6ac0c78b11f1354767d87d8158b7", size = 961134, upload-time = "2026-01-09T15:41:12.101Z" }
sdist = { url = "https://files.pythonhosted.org/packages/56/4f/d01c588827a74ddc58a8bf30cb7eb97671d210b7f5ef46f9eb2a7249f919/praisonaiagents-0.10.10.tar.gz", hash = "sha256:b6cfea93fef40b898575d0c1574134fdc9ec49754d2287e9ade90ba4381ca3d7", size = 962453, upload-time = "2026-01-10T08:57:35.82Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/1e/db/be17baa03ed7fb02aaef008364179e4846e4d47c19cd6a268bf81fbeb799/praisonaiagents-0.10.8-py3-none-any.whl", hash = "sha256:eaeef1f568966cf004cda274ba030b406e9308a118f651badbec0946a2563442", size = 940406, upload-time = "2026-01-09T15:41:10.094Z" },
{ url = "https://files.pythonhosted.org/packages/07/fe/edd179c792d7f6143d3278c01daf37015fe4a2d98c5dc326fb309b263336/praisonaiagents-0.10.10-py3-none-any.whl", hash = "sha256:fa9bc6db1fb12ae25ea491db8e30eb98ac16c1e9798952cf617594c09999b826", size = 941797, upload-time = "2026-01-10T08:57:34.07Z" },
]
[[package]]