| """ |
| Integration tests for ProposalService action execution methods. |
| |
| Tests the proposal workflow (create, approve, execute, track) with real database |
| and mocked external tool dependencies. Covers action execution paths for: |
| - Canvas actions (forms, sheets, charts) |
| - Browser automation |
| - Device capabilities |
| - Deep links |
| - Multi-step complex actions |
| |
| Coverage target: proposal_service.py lines 363-747 (action execution methods) |
| """ |
|
|
| import pytest |
| from unittest.mock import AsyncMock, MagicMock, patch |
| from datetime import datetime |
| from sqlalchemy.orm import Session |
|
|
| from core.proposal_service import ProposalService |
| from core.models import ( |
| AgentRegistry, |
| AgentStatus, |
| AgentProposal, |
| ProposalStatus, |
| ProposalType, |
| ) |
|
|
|
|
| @pytest.fixture(scope="function") |
| def proposal_service(db_session: Session): |
| """Create proposal service instance.""" |
| return ProposalService(db_session) |
|
|
|
|
| @pytest.fixture(scope="function") |
| def intern_agent(db_session: Session): |
| """Create an INTERN maturity agent for proposal tests.""" |
| agent = AgentRegistry( |
| id="intern_agent_proposal_test", |
| name="Intern Agent", |
| category="testing", |
| module_path="test.module", |
| class_name="TestClass", |
| status=AgentStatus.INTERN.value, |
| confidence_score=0.6, |
| configuration={"capabilities": ["form_submit", "chart_present"]}, |
| ) |
| db_session.add(agent) |
| db_session.commit() |
| db_session.refresh(agent) |
| return agent |
|
|
|
|
| class TestCanvasActionProposals: |
| """Test canvas action proposal execution.""" |
|
|
| @pytest.mark.asyncio |
| async def test_canvas_form_submission_proposal( |
| self, proposal_service, intern_agent, db_session |
| ): |
| """ |
| Test creating and approving a canvas form submission action proposal. |
| |
| Covers: _execute_canvas_action for form submissions |
| Lines: ~422-550 in proposal_service.py |
| """ |
| |
| proposal = AgentProposal( |
| agent_id=intern_agent.id, |
| agent_name=intern_agent.name, |
| proposal_type=ProposalType.ACTION.value, |
| title="Submit Form Data", |
| description="Submit user registration form", |
| proposed_action={ |
| "action_type": "canvas_present", |
| "canvas_type": "form", |
| "form_id": "registration_form", |
| "form_data": { |
| "name": "Test User", |
| "email": "test@example.com", |
| "phone": "555-1234" |
| }, |
| "action": "submit" |
| }, |
| reasoning="Form submission requires INTERN+ maturity", |
| status=ProposalStatus.PROPOSED.value, |
| proposed_by=intern_agent.id |
| ) |
| db_session.add(proposal) |
| db_session.commit() |
| db_session.refresh(proposal) |
|
|
| |
| |
| |
| with patch.object( |
| proposal_service, '_execute_canvas_action', |
| new=AsyncMock(return_value={ |
| "success": True, |
| "action_type": "canvas_present", |
| "canvas_id": "canvas_form_123", |
| "executed_at": "2026-02-11T12:00:00Z" |
| }) |
| ), patch.object( |
| proposal_service, '_create_proposal_episode', |
| new=AsyncMock() |
| ): |
| |
| result = await proposal_service.approve_proposal( |
| proposal_id=proposal.id, |
| user_id="test_approver" |
| ) |
|
|
| |
| db_session.refresh(proposal) |
| assert proposal.status == ProposalStatus.EXECUTED.value |
| assert proposal.approved_by == "test_approver" |
| assert proposal.approved_at is not None |
| assert result["success"] is True |
| assert result["action_type"] == "canvas_present" |
|
|
| @pytest.mark.asyncio |
| async def test_canvas_chart_presentation_proposal( |
| self, proposal_service, intern_agent, db_session |
| ): |
| """ |
| Test creating and approving a canvas chart presentation action proposal. |
| |
| Covers: _execute_canvas_action for chart presentations |
| Lines: ~422-550 in proposal_service.py |
| """ |
| proposal = AgentProposal( |
| agent_id=intern_agent.id, |
| agent_name=intern_agent.name, |
| proposal_type=ProposalType.ACTION.value, |
| title="Present Sales Chart", |
| description="Display monthly sales data as line chart", |
| proposed_action={ |
| "action_type": "canvas_present", |
| "canvas_type": "chart", |
| "chart_type": "line", |
| "data": { |
| "labels": ["Jan", "Feb", "Mar"], |
| "values": [100, 150, 200] |
| } |
| }, |
| reasoning="Chart presentation is LOW complexity (INTERN+)", |
| status=ProposalStatus.PROPOSED.value, |
| proposed_by=intern_agent.id |
| ) |
| db_session.add(proposal) |
| db_session.commit() |
| db_session.refresh(proposal) |
|
|
| with patch('tools.canvas_tool.present_to_canvas', new=AsyncMock( |
| return_value="canvas_chart_456" |
| )), patch.object( |
| proposal_service, '_create_proposal_episode', |
| new=AsyncMock() |
| ): |
| result = await proposal_service.approve_proposal( |
| proposal_id=proposal.id, |
| user_id="test_approver" |
| ) |
|
|
| db_session.refresh(proposal) |
| assert proposal.status == ProposalStatus.EXECUTED.value |
| assert result["success"] is True |
|
|
| @pytest.mark.asyncio |
| async def test_canvas_sheet_data_proposal( |
| self, proposal_service, intern_agent, db_session |
| ): |
| """ |
| Test canvas sheet data presentation proposal. |
| |
| Covers: _execute_canvas_action for sheet presentations |
| Lines: ~422-550 in proposal_service.py |
| """ |
| proposal = AgentProposal( |
| agent_id=intern_agent.id, |
| agent_name=intern_agent.name, |
| proposal_type=ProposalType.ACTION.value, |
| title="Present Spreadsheet Data", |
| description="Display financial data in spreadsheet format", |
| proposed_action={ |
| "action_type": "canvas_present", |
| "canvas_type": "sheets", |
| "data": { |
| "headers": ["Item", "Cost", "Quantity"], |
| "rows": [ |
| ["Item A", 10.00, 5], |
| ["Item B", 15.50, 3] |
| ] |
| } |
| }, |
| reasoning="Sheet data presentation requires INTERN maturity", |
| status=ProposalStatus.PROPOSED.value, |
| proposed_by=intern_agent.id |
| ) |
| db_session.add(proposal) |
| db_session.commit() |
| db_session.refresh(proposal) |
|
|
| with patch('tools.canvas_tool.present_to_canvas', new=AsyncMock( |
| return_value="canvas_sheet_789" |
| )), patch.object( |
| proposal_service, '_create_proposal_episode', |
| new=AsyncMock() |
| ): |
| result = await proposal_service.approve_proposal( |
| proposal_id=proposal.id, |
| user_id="test_approver" |
| ) |
|
|
| db_session.refresh(proposal) |
| assert proposal.status == ProposalStatus.EXECUTED.value |
| assert result["success"] is True |
|
|
|
|
| class TestBrowserAutomationProposals: |
| """Test browser automation action proposals.""" |
|
|
| @pytest.mark.asyncio |
| async def test_browser_navigate_action_proposal( |
| self, proposal_service, intern_agent, db_session |
| ): |
| """ |
| Test browser navigation action proposal. |
| |
| Covers: _execute_browser_action for navigation |
| Lines: ~350-420 in proposal_service.py |
| """ |
| proposal = AgentProposal( |
| agent_id=intern_agent.id, |
| agent_name=intern_agent.name, |
| proposal_type=ProposalType.ACTION.value, |
| title="Navigate to Website", |
| description="Navigate to example.com and extract data", |
| proposed_action={ |
| "action_type": "browser_automate", |
| "url": "https://example.com", |
| "actions": [ |
| {"type": "navigate", "url": "https://example.com"}, |
| {"type": "screenshot"} |
| ] |
| }, |
| reasoning="Browser automation requires INTERN+ maturity", |
| status=ProposalStatus.PROPOSED.value, |
| proposed_by=intern_agent.id |
| ) |
| db_session.add(proposal) |
| db_session.commit() |
| db_session.refresh(proposal) |
|
|
| |
| |
| with patch.object( |
| proposal_service, '_execute_browser_action', |
| new=AsyncMock(return_value={ |
| "success": True, |
| "action_type": "browser_automate", |
| "execution_id": "browser_exec_123", |
| "result": {"url": "https://example.com", "screenshot": "captured"} |
| }) |
| ), patch.object( |
| proposal_service, '_create_proposal_episode', |
| new=AsyncMock() |
| ): |
| result = await proposal_service.approve_proposal( |
| proposal_id=proposal.id, |
| user_id="test_approver" |
| ) |
|
|
| db_session.refresh(proposal) |
| assert proposal.status == ProposalStatus.EXECUTED.value |
| assert result["success"] is True |
| assert result["action_type"] == "browser_automate" |
|
|
| @pytest.mark.asyncio |
| async def test_browser_form_fill_action_proposal( |
| self, proposal_service, intern_agent, db_session |
| ): |
| """ |
| Test browser form filling action proposal. |
| |
| Covers: _execute_browser_action for form interactions |
| Lines: ~350-420 in proposal_service.py |
| """ |
| proposal = AgentProposal( |
| agent_id=intern_agent.id, |
| agent_name=intern_agent.name, |
| proposal_type=ProposalType.ACTION.value, |
| title="Fill and Submit Web Form", |
| description="Automate form submission on external website", |
| proposed_action={ |
| "action_type": "browser_automate", |
| "url": "https://example.com/form", |
| "actions": [ |
| {"type": "fill", "selector": "#name", "value": "Test User"}, |
| {"type": "fill", "selector": "#email", "value": "test@example.com"}, |
| {"type": "click", "selector": "button[type='submit']"} |
| ] |
| }, |
| reasoning="Form automation requires INTERN maturity", |
| status=ProposalStatus.PROPOSED.value, |
| proposed_by=intern_agent.id |
| ) |
| db_session.add(proposal) |
| db_session.commit() |
| db_session.refresh(proposal) |
|
|
| with patch.object( |
| proposal_service, '_execute_browser_action', |
| new=AsyncMock(return_value={ |
| "success": True, |
| "action_type": "browser_automate", |
| "result": {"form_submitted": True} |
| }) |
| ), patch.object( |
| proposal_service, '_create_proposal_episode', |
| new=AsyncMock() |
| ): |
| result = await proposal_service.approve_proposal( |
| proposal_id=proposal.id, |
| user_id="test_approver" |
| ) |
|
|
| db_session.refresh(proposal) |
| assert proposal.status == ProposalStatus.EXECUTED.value |
| assert result["success"] is True |
|
|
|
|
| class TestDeviceCapabilityProposals: |
| """Test device capability action proposals.""" |
|
|
| @pytest.mark.asyncio |
| async def test_device_camera_action_proposal( |
| self, proposal_service, intern_agent, db_session |
| ): |
| """ |
| Test device camera action proposal. |
| |
| Covers: _execute_device_action for camera operations |
| Lines: ~619-750 in proposal_service.py |
| """ |
| proposal = AgentProposal( |
| agent_id=intern_agent.id, |
| agent_name=intern_agent.name, |
| proposal_type=ProposalType.ACTION.value, |
| title="Capture Photo", |
| description="Use device camera to capture document photo", |
| proposed_action={ |
| "action_type": "device_command", |
| "device_type": "camera", |
| "command": "take_photo", |
| "parameters": { |
| "quality": "high", |
| "flash": "auto" |
| } |
| }, |
| reasoning="Camera access requires INTERN maturity", |
| status=ProposalStatus.PROPOSED.value, |
| proposed_by=intern_agent.id |
| ) |
| db_session.add(proposal) |
| db_session.commit() |
| db_session.refresh(proposal) |
|
|
| |
| with patch('tools.device_tool.execute_device_command', new=AsyncMock( |
| return_value={"success": True, "photo_id": "photo_123"} |
| )), patch.object( |
| proposal_service, '_create_proposal_episode', |
| new=AsyncMock() |
| ): |
| result = await proposal_service.approve_proposal( |
| proposal_id=proposal.id, |
| user_id="test_approver" |
| ) |
|
|
| db_session.refresh(proposal) |
| assert proposal.status == ProposalStatus.EXECUTED.value |
| assert result["success"] is True |
|
|
| @pytest.mark.asyncio |
| async def test_device_location_action_proposal( |
| self, proposal_service, intern_agent, db_session |
| ): |
| """ |
| Test device location access action proposal. |
| |
| Covers: _execute_device_action for location services |
| Lines: ~619-750 in proposal_service.py |
| """ |
| proposal = AgentProposal( |
| agent_id=intern_agent.id, |
| agent_name=intern_agent.name, |
| proposal_type=ProposalType.ACTION.value, |
| title="Get Current Location", |
| description="Retrieve device GPS location", |
| proposed_action={ |
| "action_type": "device_command", |
| "device_type": "location", |
| "command": "get_location", |
| "parameters": { |
| "accuracy": "high" |
| } |
| }, |
| reasoning="Location access requires INTERN maturity", |
| status=ProposalStatus.PROPOSED.value, |
| proposed_by=intern_agent.id |
| ) |
| db_session.add(proposal) |
| db_session.commit() |
| db_session.refresh(proposal) |
|
|
| with patch('tools.device_tool.execute_device_command', new=AsyncMock( |
| return_value={"success": True, "lat": 37.7749, "lng": -122.4194} |
| )), patch.object( |
| proposal_service, '_create_proposal_episode', |
| new=AsyncMock() |
| ): |
| result = await proposal_service.approve_proposal( |
| proposal_id=proposal.id, |
| user_id="test_approver" |
| ) |
|
|
| db_session.refresh(proposal) |
| assert proposal.status == ProposalStatus.EXECUTED.value |
| assert result["success"] is True |
|
|
| @pytest.mark.asyncio |
| async def test_device_notification_action_proposal( |
| self, proposal_service, intern_agent, db_session |
| ): |
| """ |
| Test device notification action proposal. |
| |
| Covers: _execute_device_action for notifications |
| Lines: ~619-750 in proposal_service.py |
| """ |
| proposal = AgentProposal( |
| agent_id=intern_agent.id, |
| agent_name=intern_agent.name, |
| proposal_type=ProposalType.ACTION.value, |
| title="Send Notification", |
| description="Send local notification to user", |
| proposed_action={ |
| "action_type": "device_command", |
| "device_type": "notifications", |
| "command": "send", |
| "parameters": { |
| "title": "Task Complete", |
| "body": "Your automated task has finished" |
| } |
| }, |
| reasoning="Sending notifications requires INTERN maturity", |
| status=ProposalStatus.PROPOSED.value, |
| proposed_by=intern_agent.id |
| ) |
| db_session.add(proposal) |
| db_session.commit() |
| db_session.refresh(proposal) |
|
|
| with patch('tools.device_tool.execute_device_command', new=AsyncMock( |
| return_value={"success": True, "notification_id": "notif_123"} |
| )), patch.object( |
| proposal_service, '_create_proposal_episode', |
| new=AsyncMock() |
| ): |
| result = await proposal_service.approve_proposal( |
| proposal_id=proposal.id, |
| user_id="test_approver" |
| ) |
|
|
| db_session.refresh(proposal) |
| assert proposal.status == ProposalStatus.EXECUTED.value |
| assert result["success"] is True |
|
|
|
|
| class TestDeepLinkProposals: |
| """Test deep link action proposals.""" |
|
|
| @pytest.mark.asyncio |
| async def test_deep_link_agent_action_proposal( |
| self, proposal_service, intern_agent, db_session |
| ): |
| """ |
| Test deep link to agent action proposal. |
| |
| Covers: _execute_canvas_action for deep links (routed through canvas) |
| Lines: ~422-550 in proposal_service.py |
| """ |
| proposal = AgentProposal( |
| agent_id=intern_agent.id, |
| agent_name=intern_agent.name, |
| proposal_type=ProposalType.ACTION.value, |
| title="Open Agent via Deep Link", |
| description="Open specific agent conversation", |
| proposed_action={ |
| "action_type": "canvas_present", |
| "canvas_type": "orchestration", |
| "deep_link": "atom://agent/sales_assistant", |
| "parameters": { |
| "conversation_id": "conv_123", |
| "message": "Help me with sales data" |
| } |
| }, |
| reasoning="Deep link navigation requires INTERN maturity", |
| status=ProposalStatus.PROPOSED.value, |
| proposed_by=intern_agent.id |
| ) |
| db_session.add(proposal) |
| db_session.commit() |
| db_session.refresh(proposal) |
|
|
| with patch('tools.canvas_tool.present_to_canvas', new=AsyncMock( |
| return_value="deep_link_canvas_123" |
| )), patch.object( |
| proposal_service, '_create_proposal_episode', |
| new=AsyncMock() |
| ): |
| result = await proposal_service.approve_proposal( |
| proposal_id=proposal.id, |
| user_id="test_approver" |
| ) |
|
|
| db_session.refresh(proposal) |
| assert proposal.status == ProposalStatus.EXECUTED.value |
| assert result["success"] is True |
|
|
| @pytest.mark.asyncio |
| async def test_deep_link_workflow_action_proposal( |
| self, proposal_service, intern_agent, db_session |
| ): |
| """ |
| Test deep link to workflow action proposal. |
| |
| Covers: deep link workflow navigation |
| """ |
| proposal = AgentProposal( |
| agent_id=intern_agent.id, |
| agent_name=intern_agent.name, |
| proposal_type=ProposalType.ACTION.value, |
| title="Open Workflow via Deep Link", |
| description="Navigate to workflow execution", |
| proposed_action={ |
| "action_type": "canvas_present", |
| "canvas_type": "orchestration", |
| "deep_link": "atom://workflow/monthly_report", |
| "parameters": { |
| "auto_start": True |
| } |
| }, |
| reasoning="Workflow deep link requires INTERN maturity", |
| status=ProposalStatus.PROPOSED.value, |
| proposed_by=intern_agent.id |
| ) |
| db_session.add(proposal) |
| db_session.commit() |
| db_session.refresh(proposal) |
|
|
| with patch('tools.canvas_tool.present_to_canvas', new=AsyncMock( |
| return_value="workflow_link_456" |
| )), patch.object( |
| proposal_service, '_create_proposal_episode', |
| new=AsyncMock() |
| ): |
| result = await proposal_service.approve_proposal( |
| proposal_id=proposal.id, |
| user_id="test_approver" |
| ) |
|
|
| db_session.refresh(proposal) |
| assert proposal.status == ProposalStatus.EXECUTED.value |
| assert result["success"] is True |
|
|
|
|
| class TestMultiStepComplexActions: |
| """Test complex multi-step action proposals.""" |
|
|
| @pytest.mark.asyncio |
| async def test_multi_step_browser_canvas_combo( |
| self, proposal_service, intern_agent, db_session |
| ): |
| """ |
| Test multi-step action combining browser automation and canvas presentation. |
| |
| Covers: Complex workflow integration |
| Lines: _execute_proposed_action routing logic (~284-348) |
| """ |
| proposal = AgentProposal( |
| agent_id=intern_agent.id, |
| agent_name=intern_agent.name, |
| proposal_type=ProposalType.ACTION.value, |
| title="Extract Data and Visualize", |
| description="Navigate to website, extract data, present as chart", |
| proposed_action={ |
| "action_type": "browser_automate", |
| "url": "https://example.com/data", |
| "actions": [ |
| {"type": "navigate", "url": "https://example.com/data"}, |
| {"type": "extract", "selector": ".data-table"}, |
| { |
| "type": "then", |
| "action_type": "canvas_present", |
| "canvas_type": "chart", |
| "data": "$extracted" |
| } |
| ] |
| }, |
| reasoning="Multi-step workflow requires INTERN maturity", |
| status=ProposalStatus.PROPOSED.value, |
| proposed_by=intern_agent.id |
| ) |
| db_session.add(proposal) |
| db_session.commit() |
| db_session.refresh(proposal) |
|
|
| |
| with patch.object( |
| proposal_service, '_execute_browser_action', |
| new=AsyncMock(return_value={ |
| "success": True, |
| "action_type": "browser_automate", |
| "result": { |
| "data": {"labels": ["A", "B"], "values": [1, 2]}, |
| "next_action": "canvas_present" |
| } |
| }) |
| ), patch.object( |
| proposal_service, '_create_proposal_episode', |
| new=AsyncMock() |
| ): |
| result = await proposal_service.approve_proposal( |
| proposal_id=proposal.id, |
| user_id="test_approver" |
| ) |
|
|
| db_session.refresh(proposal) |
| assert proposal.status == ProposalStatus.EXECUTED.value |
| assert result["success"] is True |
|
|
| @pytest.mark.asyncio |
| async def test_conditional_action_proposal( |
| self, proposal_service, intern_agent, db_session |
| ): |
| """ |
| Test action proposal with conditional logic. |
| |
| Covers: Complex action routing and conditional execution |
| """ |
| proposal = AgentProposal( |
| agent_id=intern_agent.id, |
| agent_name=intern_agent.name, |
| proposal_type=ProposalType.ACTION.value, |
| title="Conditional Action Workflow", |
| description="Execute different actions based on conditions", |
| proposed_action={ |
| "action_type": "canvas_present", |
| "canvas_type": "form", |
| "conditional": { |
| "if": {"form_field": "user_choice"}, |
| "then": [ |
| { |
| "condition": {"equals": "camera"}, |
| "action": { |
| "action_type": "device_command", |
| "device_type": "camera", |
| "command": "take_photo" |
| } |
| }, |
| { |
| "condition": {"equals": "location"}, |
| "action": { |
| "action_type": "device_command", |
| "device_type": "location", |
| "command": "get_location" |
| } |
| } |
| ] |
| } |
| }, |
| reasoning="Conditional workflow requires INTERN maturity", |
| status=ProposalStatus.PROPOSED.value, |
| proposed_by=intern_agent.id |
| ) |
| db_session.add(proposal) |
| db_session.commit() |
| db_session.refresh(proposal) |
|
|
| with patch('tools.canvas_tool.present_to_canvas', new=AsyncMock( |
| return_value="conditional_form_123" |
| )), patch.object( |
| proposal_service, '_create_proposal_episode', |
| new=AsyncMock() |
| ): |
| result = await proposal_service.approve_proposal( |
| proposal_id=proposal.id, |
| user_id="test_approver" |
| ) |
|
|
| db_session.refresh(proposal) |
| assert proposal.status == ProposalStatus.EXECUTED.value |
| assert result["success"] is True |
|
|
| @pytest.mark.asyncio |
| async def test_sequential_action_proposal( |
| self, proposal_service, intern_agent, db_session |
| ): |
| """ |
| Test proposal with sequential action execution. |
| |
| Covers: Sequential action chaining |
| """ |
| proposal = AgentProposal( |
| agent_id=intern_agent.id, |
| agent_name=intern_agent.name, |
| proposal_type=ProposalType.ACTION.value, |
| title="Sequential Data Pipeline", |
| description="Execute actions in sequence", |
| proposed_action={ |
| "action_type": "canvas_present", |
| "canvas_type": "orchestration", |
| "sequence": [ |
| { |
| "step": 1, |
| "action_type": "browser_automate", |
| "url": "https://api.example.com/data", |
| "actions": [{"type": "extract"}] |
| }, |
| { |
| "step": 2, |
| "action_type": "canvas_present", |
| "canvas_type": "sheets", |
| "data": "$step1.result" |
| }, |
| { |
| "step": 3, |
| "action_type": "canvas_present", |
| "canvas_type": "chart", |
| "data": "$step2.data" |
| } |
| ] |
| }, |
| reasoning="Sequential pipeline requires INTERN maturity", |
| status=ProposalStatus.PROPOSED.value, |
| proposed_by=intern_agent.id |
| ) |
| db_session.add(proposal) |
| db_session.commit() |
| db_session.refresh(proposal) |
|
|
| with patch('tools.canvas_tool.present_to_canvas', new=AsyncMock( |
| return_value="pipeline_canvas_789" |
| )), patch.object( |
| proposal_service, '_create_proposal_episode', |
| new=AsyncMock() |
| ): |
| result = await proposal_service.approve_proposal( |
| proposal_id=proposal.id, |
| user_id="test_approver" |
| ) |
|
|
| db_session.refresh(proposal) |
| assert proposal.status == ProposalStatus.EXECUTED.value |
| assert result["success"] is True |
|
|
|
|
| class TestProposalExecutionErrors: |
| """Test error handling in proposal execution.""" |
|
|
| @pytest.mark.asyncio |
| async def test_action_execution_failure( |
| self, proposal_service, intern_agent, db_session |
| ): |
| """ |
| Test handling of action execution failures. |
| |
| Covers: Exception handling in _execute_proposed_action |
| Lines: ~340-348 in proposal_service.py |
| """ |
| proposal = AgentProposal( |
| agent_id=intern_agent.id, |
| agent_name=intern_agent.name, |
| proposal_type=ProposalType.ACTION.value, |
| title="Failing Action", |
| description="Action that will fail", |
| proposed_action={ |
| "action_type": "browser_automate", |
| "url": "https://invalid-domain-that-does-not-exist.example" |
| }, |
| reasoning="Test error handling", |
| status=ProposalStatus.PROPOSED.value, |
| proposed_by=intern_agent.id |
| ) |
| db_session.add(proposal) |
| db_session.commit() |
| db_session.refresh(proposal) |
|
|
| |
| with patch.object( |
| proposal_service, '_execute_browser_action', |
| new=AsyncMock(side_effect=Exception("Network error")) |
| ): |
| result = await proposal_service.approve_proposal( |
| proposal_id=proposal.id, |
| user_id="test_approver" |
| ) |
|
|
| |
| assert result["success"] is False |
| assert "error" in result |
|
|
| @pytest.mark.asyncio |
| async def test_unknown_action_type( |
| self, proposal_service, intern_agent, db_session |
| ): |
| """ |
| Test handling of unknown action types. |
| |
| Covers: Unknown action type routing |
| Lines: ~331-338 in proposal_service.py |
| """ |
| proposal = AgentProposal( |
| agent_id=intern_agent.id, |
| agent_name=intern_agent.name, |
| proposal_type=ProposalType.ACTION.value, |
| title="Unknown Action Type", |
| description="Action with unsupported type", |
| proposed_action={ |
| "action_type": "unsupported_action_type" |
| }, |
| reasoning="Test unknown type handling", |
| status=ProposalStatus.PROPOSED.value, |
| proposed_by=intern_agent.id |
| ) |
| db_session.add(proposal) |
| db_session.commit() |
| db_session.refresh(proposal) |
|
|
| result = await proposal_service.approve_proposal( |
| proposal_id=proposal.id, |
| user_id="test_approver" |
| ) |
|
|
| |
| assert result["success"] is False |
| assert "error" in result |
| assert "Unknown action type" in result["error"] |
|
|
| @pytest.mark.asyncio |
| async def test_proposal_execution_disabled( |
| self, proposal_service, intern_agent, db_session, monkeypatch |
| ): |
| """ |
| Test behavior when proposal execution is disabled. |
| |
| Covers: PROPOSAL_EXECUTION_ENABLED flag check |
| Lines: ~301-308 in proposal_service.py |
| """ |
| |
| import core.proposal_service |
| monkeypatch.setattr(core.proposal_service, 'PROPOSAL_EXECUTION_ENABLED', False) |
|
|
| proposal = AgentProposal( |
| agent_id=intern_agent.id, |
| agent_name=intern_agent.name, |
| proposal_type=ProposalType.ACTION.value, |
| title="Skipped Action", |
| description="Action should be skipped", |
| proposed_action={ |
| "action_type": "canvas_present" |
| }, |
| reasoning="Test execution disabled", |
| status=ProposalStatus.PROPOSED.value, |
| proposed_by=intern_agent.id |
| ) |
| db_session.add(proposal) |
| db_session.commit() |
| db_session.refresh(proposal) |
|
|
| result = await proposal_service.approve_proposal( |
| proposal_id=proposal.id, |
| user_id="test_approver" |
| ) |
|
|
| |
| assert result["success"] is False |
| assert result.get("skipped") is True |
| assert "disabled" in result.get("message", "").lower() |
|
|