fix: complete n8n integration minor gaps and enhancements (fixes #1402)

- Fix Typer test assertion for modern API (handles both registered_commands list and legacy commands dict)
- Add loop: → splitInBatches mapping in YAMLToN8nConverter with proper node creation
- Add n8n module documentation in main __init__.py (direct import recommended)
- Create comprehensive N8N_INTEGRATION_TEST_GUIDE.md for GitHub Actions setup
- All 18 tests now pass with enhanced CLI command detection logic

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-authored-by: Claude <noreply@anthropic.com>

Author: praisonai-triage-agent[bot]

N8N_INTEGRATION_TEST_GUIDE.md

@@ -0,0 +1,210 @@
+# n8n Integration Test Guide
+
+This document describes how to set up live n8n integration tests for the PraisonAI n8n integration feature.
+
+## GitHub Actions Configuration
+
+To add live n8n integration testing to your CI/CD pipeline, create a workflow file `.github/workflows/n8n-integration.yml`:
+
+```yaml
+name: n8n Integration Tests
+
+on:
+  push:
+    branches: [ main, develop ]
+  pull_request:
+    branches: [ main, develop ]
+    paths:
+      - 'src/praisonai/praisonai/n8n/**'
+      - 'src/praisonai/tests/test_n8n_integration.py'
+
+jobs:
+  n8n-integration:
+    runs-on: ubuntu-latest
+    
+    services:
+      n8n:
+        image: docker.n8n.io/n8nio/n8n
+        ports:
+          - 5678:5678
+        env:
+          N8N_BASIC_AUTH_ACTIVE: true
+          N8N_BASIC_AUTH_USER: admin
+          N8N_BASIC_AUTH_PASSWORD: password
+          DB_TYPE: sqlite
+          N8N_ENCRYPTION_KEY: test-encryption-key
+        options: >-
+          --health-cmd "wget --no-verbose --tries=1 --spider http://localhost:5678 || exit 1"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+    
+    steps:
+      - uses: actions/checkout@v4
+      
+      - name: Set up Python 3.11
+        uses: actions/setup-python@v4
+        with:
+          python-version: 3.11
+          
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          cd src/praisonai && pip install -e ".[n8n]"
+          pip install pytest
+          
+      - name: Wait for n8n to be ready
+        run: |
+          timeout 120 bash -c 'until curl -f http://localhost:5678; do sleep 2; done'
+          
+      - name: Run n8n integration tests
+        run: |
+          cd src/praisonai
+          pytest tests/test_n8n_integration.py -v -m integration
+        env:
+          N8N_URL: http://localhost:5678
+          N8N_USER: admin
+          N8N_PASSWORD: password
+
+  test-workflow-creation:
+    runs-on: ubuntu-latest
+    needs: n8n-integration
+    
+    services:
+      n8n:
+        image: docker.n8n.io/n8nio/n8n
+        ports:
+          - 5678:5678
+        env:
+          N8N_BASIC_AUTH_ACTIVE: true
+          N8N_BASIC_AUTH_USER: admin
+          N8N_BASIC_AUTH_PASSWORD: password
+          DB_TYPE: sqlite
+          N8N_ENCRYPTION_KEY: test-encryption-key
+    
+    steps:
+      - uses: actions/checkout@v4
+      
+      - name: Set up Python 3.11
+        uses: actions/setup-python@v4
+        with:
+          python-version: 3.11
+          
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          cd src/praisonai && pip install -e ".[n8n]"
+          
+      - name: Test workflow creation and execution
+        run: |
+          cd src/praisonai
+          python -c "
+          from praisonai.n8n import YAMLToN8nConverter, N8nClient
+          import yaml
+          import os
+          
+          # Test YAML workflow
+          test_workflow = {
+              'name': 'Test Workflow',
+              'agents': {
+                  'researcher': {'instructions': 'Research topics'},
+                  'writer': {'instructions': 'Write content'}
+              },
+              'steps': [
+                  'researcher',
+                  {'agent': 'writer', 'action': 'write_summary'}
+              ]
+          }
+          
+          # Convert to n8n format
+          converter = YAMLToN8nConverter()
+          n8n_workflow = converter.convert(test_workflow)
+          
+          # Test connection to n8n
+          client = N8nClient(
+              base_url=os.getenv('N8N_URL', 'http://localhost:5678'),
+              auth=('admin', 'password')
+          )
+          
+          # Create workflow
+          created = client.create_workflow('test-workflow', n8n_workflow)
+          print(f'Created workflow: {created}')
+          
+          # List workflows
+          workflows = client.list_workflows()
+          print(f'Available workflows: {len(workflows)}')
+          
+          print('✅ Live n8n integration test passed!')
+          "
+        env:
+          N8N_URL: http://localhost:5678
+```
+
+## Test Markers
+
+To properly organize the integration tests, add these pytest markers to your test file:
+
+```python
+import pytest
+
+@pytest.mark.integration
+def test_live_n8n_workflow_creation():
+    \"\"\"Test actual workflow creation with live n8n instance.\"\"\"
+    # Test implementation here
+    pass
+
+@pytest.mark.integration
+def test_live_n8n_execution():
+    \"\"\"Test workflow execution with live n8n instance.\"\"\"
+    # Test implementation here
+    pass
+```
+
+## Running Tests Locally
+
+To run the integration tests locally:
+
+1. Start n8n with Docker:
+```bash
+docker run -d --name n8n -p 5678:5678 \
+  -e N8N_BASIC_AUTH_ACTIVE=true \
+  -e N8N_BASIC_AUTH_USER=admin \
+  -e N8N_BASIC_AUTH_PASSWORD=password \
+  docker.n8n.io/n8nio/n8n
+```
+
+2. Run the tests:
+```bash
+cd src/praisonai
+pytest tests/test_n8n_integration.py -v -m integration
+```
+
+## Environment Variables
+
+The integration tests use these environment variables:
+
+- `N8N_URL`: n8n instance URL (default: http://localhost:5678)
+- `N8N_USER`: Basic auth username (default: admin)
+- `N8N_PASSWORD`: Basic auth password (default: password)
+- `N8N_API_KEY`: Alternative to basic auth (optional)
+
+## Test Coverage
+
+The integration tests should cover:
+
+- [ ] Workflow creation via API
+- [ ] Workflow execution
+- [ ] YAML to n8n JSON conversion with real data
+- [ ] n8n JSON to YAML reverse conversion
+- [ ] Loop pattern with splitInBatches nodes
+- [ ] Route pattern with switch nodes
+- [ ] Parallel execution patterns
+- [ ] Error handling and recovery
+
+## Notes
+
+- Tests marked with `@pytest.mark.integration` require a live n8n instance
+- Unit tests (without the marker) can run without n8n
+- The n8n Docker service includes health checks to ensure readiness
+- Basic authentication is used for simplicity in tests
+- SQLite is used as the database for test isolation

src/praisonai/praisonai/__init__.py

@@ -34,6 +34,7 @@
     'AnthropicManagedAgent',
     'LocalManagedAgent',
     'LocalManagedConfig',
+    'n8n',  # n8n workflow integration
 ]
 
 
@@ -100,6 +101,8 @@ def __getattr__(name):
     elif name in ('DB', 'PraisonAIDB', 'PraisonDB'):
         from .db.adapter import DB
         return DB
+    # Note: n8n is available via direct import: from praisonai.n8n import YAMLToN8nConverter
+    # Lazy loading from main package causes recursion, so use direct import for now
 
     # Try praisonaiagents exports
     try:

src/praisonai/praisonai/n8n/converter.py

@@ -245,6 +245,43 @@ def _steps_to_connections(
                         # For simplicity, use last parallel agent as previous
                         if parallel_targets:
                             previous_node = parallel_targets[-1]["node"]
+                            
+                elif "loop" in step:
+                    # Loop step - create splitInBatches node for loop iteration
+                    loop_node = self._create_split_in_batches_node(step["loop"])
+                    all_nodes.append(loop_node)
+                    
+                    # Connect previous to split node
+                    connections[previous_node] = {
+                        "main": [[{"node": loop_node.name, "type": "main", "index": 0}]]
+                    }
+                    
+                    # Connect split node to loop body agents
+                    loop_targets = []
+                    loop_config = step["loop"]
+                    if "steps" in loop_config:
+                        for loop_step in loop_config["steps"]:
+                            if isinstance(loop_step, str) and loop_step in agent_nodes:
+                                loop_targets.append({
+                                    "node": agent_nodes[loop_step],
+                                    "type": "main",
+                                    "index": 0
+                                })
+                            elif isinstance(loop_step, dict) and "agent" in loop_step:
+                                agent_id = loop_step["agent"]
+                                if agent_id in agent_nodes:
+                                    loop_targets.append({
+                                        "node": agent_nodes[agent_id],
+                                        "type": "main", 
+                                        "index": 0
+                                    })
+                    
+                    if loop_targets:
+                        connections[loop_node.name] = {"main": [loop_targets]}
+                        # Use the last loop target as previous node
+                        previous_node = loop_targets[-1]["node"]
+                    else:
+                        previous_node = loop_node.name
 
         return connections
 
@@ -275,6 +312,27 @@ def _create_switch_node(self, route_config: Dict[str, Any]) -> N8nNode:
             }
         )
 
+    def _create_split_in_batches_node(self, loop_config: Dict[str, Any]) -> N8nNode:
+        """Convert loop configuration to n8n SplitInBatches node."""
+        self.node_counter += 1
+        
+        # Extract loop configuration
+        batch_size = loop_config.get("batch_size", 1)
+        input_field = loop_config.get("items", "items")
+        
+        return N8nNode(
+            name=f"Loop {self.node_counter}",
+            type="n8n-nodes-base.splitInBatches",
+            position=[
+                self.position_x_start + (self.node_counter * self.position_x_increment),
+                self.position_y
+            ],
+            parameters={
+                "batchSize": batch_size,
+                "options": {}
+            }
+        )
+    
     def _calculate_position(self, node_index: int) -> List[int]:
         """Calculate position for node layout."""
         x = self.position_x_start + (node_index * self.position_x_increment)

src/praisonai/tests/test_n8n_integration.py

@@ -333,10 +333,22 @@ def test_cli_command_structure(self):
             pytest.skip("CLI commands not available")
 
         # Check that the main app is a typer app
-        assert hasattr(app, 'commands')
+        assert hasattr(app, 'registered_commands') or hasattr(app, 'commands')
 
         # Check that expected commands exist
-        command_names = [cmd.name for cmd in app.commands.values()]
+        if hasattr(app, 'registered_commands'):
+            # Modern Typer uses a list of CommandInfo objects
+            if isinstance(app.registered_commands, list):
+                command_names = []
+                for cmd in app.registered_commands:
+                    # Use name if available, otherwise fallback to callback name
+                    name = cmd.name if cmd.name else (cmd.callback.__name__ if cmd.callback else 'unknown')
+                    command_names.append(name)
+            else:
+                command_names = [name for name in app.registered_commands.keys()]
+        else:
+            command_names = [cmd.name for cmd in app.commands.values()]
+        # Some commands use explicit names, others use function names
         expected_commands = ['export', 'import', 'preview', 'push', 'pull', 'test', 'list']
 
         for cmd in expected_commands: