Meta Ads MCP

"""Comprehensive regression tests for duplication module."""

import os
import json
import pytest
import httpx
from unittest.mock import patch, AsyncMock, MagicMock
import importlib


class TestDuplicationFeatureToggle:
    """Test feature toggle functionality to prevent regression."""
    
    def test_feature_disabled_by_default(self):
        """Ensure duplication is disabled by default."""
        with patch.dict(os.environ, {}, clear=True):
            # Force reload to pick up environment changes
            import importlib
            from meta_ads_mcp.core import duplication
            importlib.reload(duplication)
            
            assert not duplication.ENABLE_DUPLICATION
            # Note: Functions may persist in module namespace due to previous test runs
            # The important thing is that ENABLE_DUPLICATION flag is False
    
    def test_feature_enabled_with_env_var(self):
        """Ensure duplication is enabled when environment variable is set."""
        with patch.dict(os.environ, {"META_ADS_ENABLE_DUPLICATION": "1"}):
            # Force reload to pick up environment changes
            import importlib
            from meta_ads_mcp.core import duplication
            importlib.reload(duplication)
            
            assert duplication.ENABLE_DUPLICATION
            assert hasattr(duplication, 'duplicate_campaign')
            assert hasattr(duplication, 'duplicate_adset')
            assert hasattr(duplication, 'duplicate_ad')
            assert hasattr(duplication, 'duplicate_creative')
    
    def test_feature_enabled_with_various_truthy_values(self):
        """Test that various truthy values enable the feature."""
        truthy_values = ["1", "true", "TRUE", "yes", "YES", "on", "ON", "enabled"]
        
        for value in truthy_values:
            with patch.dict(os.environ, {"META_ADS_ENABLE_DUPLICATION": value}):
                import importlib
                from meta_ads_mcp.core import duplication
                importlib.reload(duplication)
                
                assert duplication.ENABLE_DUPLICATION, f"Value '{value}' should enable the feature"
    
    def test_feature_disabled_with_empty_string(self):
        """Test that empty string disables the feature."""
        with patch.dict(os.environ, {"META_ADS_ENABLE_DUPLICATION": ""}):
            import importlib
            from meta_ads_mcp.core import duplication
            importlib.reload(duplication)
            
            assert not duplication.ENABLE_DUPLICATION


class TestDuplicationDecorators:
    """Test that decorators are applied correctly to prevent regression."""
    
    @pytest.fixture(autouse=True)
    def enable_feature(self):
        """Enable the duplication feature for these tests."""
        with patch.dict(os.environ, {"META_ADS_ENABLE_DUPLICATION": "1"}):
            import importlib
            from meta_ads_mcp.core import duplication
            importlib.reload(duplication)
            yield duplication
    
    def test_functions_have_meta_api_tool_decorator(self, enable_feature):
        """Ensure all duplication functions have @meta_api_tool decorator."""
        duplication = enable_feature
        
        functions = ['duplicate_campaign', 'duplicate_adset', 'duplicate_ad', 'duplicate_creative']
        
        for func_name in functions:
            func = getattr(duplication, func_name)
            
            # Check that function has been wrapped by meta_api_tool
            # The meta_api_tool decorator should add access token handling
            assert callable(func), f"{func_name} should be callable"
            
            # Check function signature includes access_token parameter
            import inspect
            sig = inspect.signature(func)
            assert 'access_token' in sig.parameters, f"{func_name} should have access_token parameter"
            assert sig.parameters['access_token'].default is None, f"{func_name} access_token should default to None"
    
    @pytest.mark.asyncio
    async def test_functions_are_mcp_tools(self, enable_feature):
        """Ensure all duplication functions are registered as MCP tools."""
        # This test ensures the @mcp_server.tool() decorator is working
        from meta_ads_mcp.core.server import mcp_server
        
        # Get all registered tool names (list_tools is async)
        tools = await mcp_server.list_tools()
        tool_names = [tool.name for tool in tools]
        
        expected_tools = ['duplicate_campaign', 'duplicate_adset', 'duplicate_ad', 'duplicate_creative']
        
        for tool_name in expected_tools:
            assert tool_name in tool_names, f"{tool_name} should be registered as an MCP tool"


class TestDuplicationAPIContract:
    """Test API contract to prevent regression in external API calls."""
    
    @pytest.fixture(autouse=True)
    def enable_feature(self):
        """Enable the duplication feature for these tests."""
        with patch.dict(os.environ, {"META_ADS_ENABLE_DUPLICATION": "1"}):
            import importlib
            from meta_ads_mcp.core import duplication
            importlib.reload(duplication)
            yield duplication
    
    @pytest.mark.asyncio
    async def test_api_endpoint_construction(self, enable_feature):
        """Test that API endpoints are constructed correctly."""
        duplication = enable_feature
        
        test_cases = [
            ("campaign", "123456789", "https://mcp.pipeboard.co/api/meta/duplicate/campaign/123456789"),
            ("adset", "987654321", "https://mcp.pipeboard.co/api/meta/duplicate/adset/987654321"),
            ("ad", "555666777", "https://mcp.pipeboard.co/api/meta/duplicate/ad/555666777"),
            ("creative", "111222333", "https://mcp.pipeboard.co/api/meta/duplicate/creative/111222333"),
        ]
        
        for resource_type, resource_id, expected_url in test_cases:
            # Mock dual-header authentication
            with patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration") as mock_auth:
                mock_auth.get_pipeboard_token.return_value = "pipeboard_token"
                mock_auth.get_auth_token.return_value = "facebook_token"
                
                with patch("meta_ads_mcp.core.duplication.httpx.AsyncClient") as mock_client:
                    mock_response = MagicMock()
                    mock_response.status_code = 200
                    mock_response.json.return_value = {"success": True}
                    mock_client.return_value.__aenter__.return_value.post.return_value = mock_response
                    
                    await duplication._forward_duplication_request(
                        resource_type, resource_id, "test_token", {}
                    )
                    
                    # Verify the correct URL was called
                    call_args = mock_client.return_value.__aenter__.return_value.post.call_args
                    actual_url = call_args[0][0]
                    assert actual_url == expected_url, f"Expected {expected_url}, got {actual_url}"

    @pytest.mark.asyncio
    async def test_request_headers_format(self, enable_feature):
        """Test that request headers are formatted correctly."""
        duplication = enable_feature
        
        # Mock dual-header authentication
        with patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration") as mock_auth:
            mock_auth.get_pipeboard_token.return_value = "pipeboard_token_12345"
            mock_auth.get_auth_token.return_value = "facebook_token_67890"
            
            with patch("meta_ads_mcp.core.duplication.httpx.AsyncClient") as mock_client:
                mock_response = MagicMock()
                mock_response.status_code = 200
                mock_response.json.return_value = {"success": True}
                mock_client.return_value.__aenter__.return_value.post.return_value = mock_response
                
                await duplication._forward_duplication_request(
                    "campaign", "123456789", "test_token_12345", {"name_suffix": " - Test"}
                )
                
                # Verify dual headers are sent correctly
                call_args = mock_client.return_value.__aenter__.return_value.post.call_args
                headers = call_args[1]["headers"]
                
                # Check the dual-header authentication pattern
                assert headers["Authorization"] == "Bearer facebook_token_67890"  # Facebook token for Meta API
                assert headers["X-Pipeboard-Token"] == "pipeboard_token_12345"   # Pipeboard token for auth
                assert headers["Content-Type"] == "application/json"
                assert headers["User-Agent"] == "meta-ads-mcp/1.0"

    @pytest.mark.asyncio
    async def test_request_timeout_configuration(self, enable_feature):
        """Test that request timeout is configured correctly."""
        duplication = enable_feature
        
        # Mock dual-header authentication
        with patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration") as mock_auth:
            mock_auth.get_pipeboard_token.return_value = "pipeboard_token"
            mock_auth.get_auth_token.return_value = "facebook_token"
            
            with patch("meta_ads_mcp.core.duplication.httpx.AsyncClient") as mock_client:
                mock_response = MagicMock()
                mock_response.status_code = 200
                mock_response.json.return_value = {"success": True}
                mock_client.return_value.__aenter__.return_value.post.return_value = mock_response
                
                await duplication._forward_duplication_request(
                    "campaign", "123456789", "test_token", {}
                )
                
                # Verify timeout is set to 120 seconds
                mock_client.assert_called_once_with(timeout=120.0)


class TestDuplicationErrorHandling:
    """Test error handling to prevent regression in error scenarios."""
    
    @pytest.fixture(autouse=True)
    def enable_feature(self):
        """Enable the duplication feature for these tests."""
        with patch.dict(os.environ, {"META_ADS_ENABLE_DUPLICATION": "1"}):
            import importlib
            from meta_ads_mcp.core import duplication
            importlib.reload(duplication)
            yield duplication
    
    @pytest.mark.asyncio
    async def test_missing_access_token_error(self, enable_feature):
        """Test error handling when authentication tokens are missing."""
        duplication = enable_feature
        
        from meta_ads_mcp.core.duplication import DuplicationError

        # Test missing Pipeboard token (primary authentication failure)
        with patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration") as mock_auth:
            mock_auth.get_pipeboard_token.return_value = None  # No Pipeboard token
            mock_auth.get_auth_token.return_value = "facebook_token"  # Has Facebook token

            with pytest.raises(DuplicationError) as exc_info:
                await duplication._forward_duplication_request("campaign", "123", None, {})
            result_json = json.loads(str(exc_info.value))
            assert result_json["error"] == "authentication_required"
            assert "Pipeboard API token not found" in result_json["message"]

        # Test missing Facebook token (secondary authentication failure)
        with patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration") as mock_auth:
            mock_auth.get_pipeboard_token.return_value = "pipeboard_token"  # Has Pipeboard token
            mock_auth.get_auth_token.return_value = None  # No Facebook token

            with patch("meta_ads_mcp.core.auth.get_current_access_token") as mock_get_token:
                mock_get_token.return_value = None  # No fallback token

                with pytest.raises(DuplicationError) as exc_info:
                    await duplication._forward_duplication_request("campaign", "123", None, {})
                result_json = json.loads(str(exc_info.value))
                assert result_json["error"] == "authentication_required"
                assert "Meta Ads access token not found" in result_json["message"]
    
    @pytest.mark.asyncio
    async def test_http_status_code_handling(self, enable_feature):
        """Test handling of various HTTP status codes."""
        duplication = enable_feature
        
        status_code_tests = [
            (200, "success_response", "json"),
            (400, "validation_failed", "error"),
            (401, "authentication_error", "error"),
            (402, "subscription_required", "error"),
            (403, "facebook_connection_required", "error"),
            (404, "resource_not_found", "error"),
            (429, "rate_limit_exceeded", "error"),
            (502, "meta_api_error", "error"),
            (500, "duplication_failed", "error"),
        ]
        
        for status_code, expected_error_type, response_type in status_code_tests:
            with patch("meta_ads_mcp.core.duplication.httpx.AsyncClient") as mock_client, \
                 patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration.get_pipeboard_token", return_value="test_pipeboard_token"), \
                 patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration.get_auth_token", return_value="test_facebook_token"):
                # Use MagicMock instead of AsyncMock for more predictable behavior
                mock_response = MagicMock()
                mock_response.status_code = status_code
                
                if status_code == 200:
                    mock_response.json.return_value = {"success": True, "id": "new_123"}
                elif status_code == 400:
                    mock_response.json.return_value = {"errors": ["Invalid parameter"], "warnings": []}
                elif status_code == 401:
                    mock_response.json.side_effect = Exception("No JSON")
                    mock_response.text = "Unauthorized"
                elif status_code == 402:
                    mock_response.json.return_value = {
                        "message": "This feature is not available in your current plan",
                        "upgrade_url": "https://pipeboard.co/upgrade",
                        "suggestion": "Please upgrade your account to access this feature"
                    }
                elif status_code == 403:
                    mock_response.json.return_value = {
                        "message": "You need to connect your Facebook account first",
                        "details": {
                            "login_flow_url": "/connections",
                            "auth_flow_url": "/api/meta/auth"
                        }
                    }
                elif status_code == 404:
                    mock_response.json.side_effect = Exception("No JSON")
                    mock_response.text = "Not Found"
                elif status_code == 429:
                    mock_response.headers.get.return_value = "60"
                    mock_response.json.side_effect = Exception("No JSON")
                    mock_response.text = "Rate limited"
                elif status_code == 502:
                    mock_response.json.return_value = {"message": "Facebook API error"}
                else:
                    mock_response.json.side_effect = Exception("No JSON")
                    mock_response.text = f"Error {status_code}"
                
                mock_client.return_value.__aenter__.return_value.post.return_value = mock_response

                if status_code == 200:
                    result = await duplication._forward_duplication_request(
                        "campaign", "123", "token", {}
                    )
                    result_json = json.loads(result)
                    assert "success" in result_json or "id" in result_json
                elif status_code == 429:
                    # 429 raises RateLimitError so FastMCP sets isError: true
                    from meta_ads_mcp.core.duplication import RateLimitError
                    with pytest.raises(RateLimitError) as exc_info:
                        await duplication._forward_duplication_request(
                            "campaign", "123", "token", {}
                        )
                    exc_json = json.loads(str(exc_info.value))
                    assert exc_json["error"] == expected_error_type
                else:
                    # All non-200/non-429 errors raise DuplicationError so FastMCP
                    # sets isError: true and the usage credit is refunded
                    from meta_ads_mcp.core.duplication import DuplicationError
                    with pytest.raises(DuplicationError) as exc_info:
                        await duplication._forward_duplication_request(
                            "campaign", "123", "token", {}
                        )
                    exc_json = json.loads(str(exc_info.value))
                    assert exc_json["error"] == expected_error_type
    
    @pytest.mark.asyncio
    async def test_network_error_handling(self, enable_feature):
        """Test handling of network errors."""
        duplication = enable_feature
        
        network_errors = [
            (httpx.TimeoutException("Timeout"), "request_timeout"),
            (httpx.RequestError("Connection failed"), "network_error"),
            (Exception("Unexpected error"), "unexpected_error"),
        ]
        
        from meta_ads_mcp.core.duplication import DuplicationError
        for exception, expected_error in network_errors:
            with patch("meta_ads_mcp.core.duplication.httpx.AsyncClient") as mock_client, \
                 patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration.get_pipeboard_token", return_value="test_pipeboard_token"), \
                 patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration.get_auth_token", return_value="test_facebook_token"):
                mock_client.return_value.__aenter__.return_value.post.side_effect = exception

                # Network/unexpected errors raise DuplicationError so FastMCP sets isError: true
                with pytest.raises(DuplicationError) as exc_info:
                    await duplication._forward_duplication_request(
                        "campaign", "123", "token", {}
                    )
                exc_json = json.loads(str(exc_info.value))
                assert exc_json["error"] == expected_error


class TestDuplicationParameterHandling:
    """Test parameter handling to prevent regression in data processing."""
    
    @pytest.fixture(autouse=True)
    def enable_feature(self):
        """Enable the duplication feature for these tests."""
        with patch.dict(os.environ, {"META_ADS_ENABLE_DUPLICATION": "1"}):
            import importlib
            from meta_ads_mcp.core import duplication
            importlib.reload(duplication)
            yield duplication
    
    @pytest.mark.asyncio
    async def test_none_values_filtered_from_options(self, enable_feature):
        """Test that None values are filtered from options."""
        duplication = enable_feature
        
        # Mock dual-header authentication  
        with patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration") as mock_auth:
            mock_auth.get_pipeboard_token.return_value = "pipeboard_token"
            mock_auth.get_auth_token.return_value = "facebook_token"
            
            with patch("meta_ads_mcp.core.duplication.httpx.AsyncClient") as mock_client:
                mock_response = MagicMock()
                mock_response.status_code = 200
                mock_response.json.return_value = {"success": True}
                mock_client.return_value.__aenter__.return_value.post.return_value = mock_response
                
                # Test with options containing None values
                options_with_none = {
                    "name_suffix": " - Test",
                    "new_daily_budget": None,
                    "new_status": "PAUSED",
                    "new_headline": None,
                }
                
                await duplication._forward_duplication_request(
                    "campaign", "123", "token", options_with_none
                )
                
                # Verify None values were filtered out
                call_args = mock_client.return_value.__aenter__.return_value.post.call_args
                json_payload = call_args[1]["json"]
                
                assert "name_suffix" in json_payload
                assert "new_status" in json_payload
                assert "new_daily_budget" not in json_payload
                assert "new_headline" not in json_payload
    
    @pytest.mark.asyncio
    async def test_campaign_duplication_parameter_forwarding(self, enable_feature):
        """Test that campaign duplication forwards all parameters correctly."""
        duplication = enable_feature
        
        with patch("meta_ads_mcp.core.duplication._forward_duplication_request") as mock_forward:
            mock_forward.return_value = '{"success": true}'
            
            # Test with all parameters
            result = await duplication.duplicate_campaign(
                campaign_id="123456789",
                access_token="test_token",
                name_suffix=" - New Copy",
                include_ad_sets=False,
                include_ads=True,
                include_creatives=False,
                copy_schedule=True,
                new_daily_budget=100.50,
                new_status="ACTIVE"
            )
            
            # Verify parameters were forwarded correctly
            mock_forward.assert_called_once_with(
                "campaign",
                "123456789",
                "test_token",
                {
                    "name_suffix": " - New Copy",
                    "include_ad_sets": False,
                    "include_ads": True,
                    "include_creatives": False,
                    "copy_schedule": True,
                    "new_daily_budget": 100.50,
                    "new_status": "ACTIVE",
                    "pb_token": None
                }
            )
    
    @pytest.mark.asyncio
    async def test_adset_duplication_parameter_forwarding(self, enable_feature):
        """Test that ad set duplication forwards all parameters correctly including new_targeting."""
        duplication = enable_feature
        
        with patch("meta_ads_mcp.core.duplication._forward_duplication_request") as mock_forward:
            mock_forward.return_value = '{"success": true}'
            
            # Test with all parameters including new_targeting
            result = await duplication.duplicate_adset(
                adset_id="987654321",
                access_token="test_token",
                target_campaign_id="campaign_123",
                name_suffix=" - Targeted Copy",
                include_ads=False,
                include_creatives=True,
                new_daily_budget=200.00,
                new_targeting={
                    "age_min": 25,
                    "age_max": 45,
                    "geo_locations": {
                        "countries": ["US", "CA"]
                    }
                },
                new_status="ACTIVE"
            )
            
            # Verify parameters were forwarded correctly
            mock_forward.assert_called_once_with(
                "adset",
                "987654321",
                "test_token",
                {
                    "target_campaign_id": "campaign_123",
                    "name_suffix": " - Targeted Copy",
                    "include_ads": False,
                    "include_creatives": True,
                    "new_daily_budget": 200.00,
                    "new_targeting": {
                        "age_min": 25,
                        "age_max": 45,
                        "geo_locations": {
                            "countries": ["US", "CA"]
                        }
                    },
                    "new_status": "ACTIVE",
                    "pb_token": None
                }
            )
    
    def test_estimated_components_calculation(self, enable_feature):
        """Test that estimated components are calculated correctly."""
        duplication = enable_feature
        
        test_cases = [
            # Campaign with all components
            ("campaign", {"include_ad_sets": True, "include_ads": True, "include_creatives": True}, 
             {"campaigns": 1, "ad_sets": "3-5 (estimated)", "ads": "5-15 (estimated)", "creatives": "5-15 (estimated)"}),
            
            # Campaign with no sub-components
            ("campaign", {"include_ad_sets": False, "include_ads": False, "include_creatives": False},
             {"campaigns": 1}),
            
            # Ad set with ads
            ("adset", {"include_ads": True, "include_creatives": True},
             {"ad_sets": 1, "ads": "2-5 (estimated)", "creatives": "2-5 (estimated)"}),
            
            # Ad set without ads
            ("adset", {"include_ads": False, "include_creatives": False},
             {"ad_sets": 1}),
            
            # Single ad with creative
            ("ad", {"duplicate_creative": True},
             {"ads": 1, "creatives": 1}),
            
            # Single ad without creative
            ("ad", {"duplicate_creative": False},
             {"ads": 1}),
            
            # Single creative
            ("creative", {},
             {"creatives": 1}),
        ]
        
        for resource_type, options, expected in test_cases:
            result = duplication._get_estimated_components(resource_type, options)
            assert result == expected, f"Failed for {resource_type} with {options}"


class TestDuplicationIntegration:
    """Integration tests to prevent regression in end-to-end functionality."""
    
    @pytest.fixture(autouse=True)
    def enable_feature(self):
        """Enable the duplication feature for these tests."""
        with patch.dict(os.environ, {"META_ADS_ENABLE_DUPLICATION": "1"}):
            import importlib
            from meta_ads_mcp.core import duplication
            importlib.reload(duplication)
            yield duplication
    
    @pytest.mark.asyncio
    async def test_end_to_end_successful_duplication(self, enable_feature):
        """Test complete successful duplication flow."""
        duplication = enable_feature
        
        # Mock the auth system completely to bypass the @meta_api_tool decorator checks
        with patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration") as mock_auth_integration:
            # Mock dual authentication tokens
            mock_auth_integration.get_pipeboard_token.return_value = "pipeboard_token"
            mock_auth_integration.get_auth_token.return_value = "facebook_token"
            
            with patch("meta_ads_mcp.core.duplication.httpx.AsyncClient") as mock_client:
                        # Mock successful response
                        mock_response = MagicMock()
                        mock_response.status_code = 200
                        mock_response.json.return_value = {
                            "success": True,
                            "original_campaign_id": "123456789",
                            "new_campaign_id": "987654321",
                            "duplicated_components": {
                                "campaign": {"id": "987654321", "name": "Test Campaign - Copy"},
                                "ad_sets": [{"id": "111", "name": "Ad Set 1 - Copy"}],
                                "ads": [{"id": "222", "name": "Ad 1 - Copy"}],
                                "creatives": [{"id": "333", "name": "Creative 1 - Copy"}]
                            },
                            "warnings": [],
                            "subscription": {
                                "status": "active"
                            }
                        }
                        mock_client.return_value.__aenter__.return_value.post.return_value = mock_response
                        
                        # Call the function with explicit token
                        result = await duplication.duplicate_campaign(
                            campaign_id="123456789",
                            access_token="facebook_token",  # Use the mocked token
                            name_suffix=" - Test Copy"
                        )
                        
                        # Verify result - handle @meta_api_tool wrapper
                        result_json = json.loads(result)
                        if "data" in result_json:
                            actual_result = json.loads(result_json["data"])
                        else:
                            actual_result = result_json
                            
                        assert actual_result["success"] is True
                        assert actual_result["new_campaign_id"] == "987654321"
                        assert "duplicated_components" in actual_result
    
    @pytest.mark.asyncio
    async def test_facebook_connection_error_flow(self, enable_feature):
        """Test Facebook connection required error flow."""
        duplication = enable_feature
        
        # Mock the auth system completely to bypass the @meta_api_tool decorator checks  
        with patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration") as mock_auth_integration:
            # Mock dual authentication tokens
            mock_auth_integration.get_pipeboard_token.return_value = "pipeboard_token"
            mock_auth_integration.get_auth_token.return_value = "facebook_token"
            
            with patch("meta_ads_mcp.core.duplication.httpx.AsyncClient") as mock_client:
                        # Mock 403 response (Facebook connection required)
                        mock_response = MagicMock()
                        mock_response.status_code = 403
                        mock_response.json.return_value = {
                            "message": "You need to connect your Facebook account first",
                            "details": {
                                "login_flow_url": "/connections",
                                "auth_flow_url": "/api/meta/auth"
                            }
                        }
                        mock_client.return_value.__aenter__.return_value.post.return_value = mock_response
                        
                        from meta_ads_mcp.core.duplication import DuplicationError
                        with pytest.raises(DuplicationError) as exc_info:
                            await duplication.duplicate_campaign(
                                campaign_id="123456789",
                                access_token="facebook_token"
                            )
                        actual_result = json.loads(str(exc_info.value))
                        assert actual_result["error"] == "facebook_connection_required"
                        assert actual_result["message"] == "You need to connect your Facebook account first"
                        assert "details" in actual_result
                        assert actual_result["details"]["login_flow_url"] == "/connections"
    
    @pytest.mark.asyncio
    async def test_subscription_required_error_flow(self, enable_feature):
        """Test subscription required error flow."""
        duplication = enable_feature
        
        # Mock the auth system completely to bypass the @meta_api_tool decorator checks
        with patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration") as mock_auth_integration:
            # Mock dual authentication tokens
            mock_auth_integration.get_pipeboard_token.return_value = "pipeboard_token"
            mock_auth_integration.get_auth_token.return_value = "facebook_token"
            
            with patch("meta_ads_mcp.core.duplication.httpx.AsyncClient") as mock_client:
                        # Mock 402 response (subscription required)
                        mock_response = MagicMock()
                        mock_response.status_code = 402
                        mock_response.json.return_value = {
                            "message": "This feature is not available in your current plan",
                            "upgrade_url": "https://pipeboard.co/upgrade",
                            "suggestion": "Please upgrade your account to access this feature"
                        }
                        mock_client.return_value.__aenter__.return_value.post.return_value = mock_response
                        
                        from meta_ads_mcp.core.duplication import DuplicationError
                        with pytest.raises(DuplicationError) as exc_info:
                            await duplication.duplicate_campaign(
                                campaign_id="123456789",
                                access_token="facebook_token"
                            )
                        actual_result = json.loads(str(exc_info.value))
                        assert actual_result["error"] == "subscription_required"
                        assert actual_result["message"] == "This feature is not available in your current plan"
                        assert actual_result["upgrade_url"] == "https://pipeboard.co/upgrade"
                        assert actual_result["suggestion"] == "Please upgrade your account to access this feature"


class TestDuplicationTokenHandling:
    """Test access token handling to prevent auth regression."""
    
    @pytest.fixture(autouse=True)
    def enable_feature(self):
        """Enable the duplication feature for these tests."""
        with patch.dict(os.environ, {"META_ADS_ENABLE_DUPLICATION": "1"}):
            import importlib
            from meta_ads_mcp.core import duplication
            importlib.reload(duplication)
            yield duplication
    
    @pytest.mark.asyncio
    async def test_meta_api_tool_decorator_token_handling(self, enable_feature):
        """Test that @meta_api_tool decorator properly handles explicit tokens."""
        duplication = enable_feature
        
        # Test with explicit token - this should bypass auth system entirely
        with patch("meta_ads_mcp.core.duplication._forward_duplication_request") as mock_forward:
            mock_forward.return_value = '{"success": true}'
            
            # Call with explicit access_token
            await duplication.duplicate_campaign(
                campaign_id="123456789",
                access_token="explicit_token_12345"
            )
            
            # Verify the explicit token was passed through
            mock_forward.assert_called_once()
            call_args = mock_forward.call_args[0]
            assert call_args[2] == "explicit_token_12345"  # access_token is 3rd argument
    
    @pytest.mark.asyncio
    async def test_explicit_token_overrides_injection(self, enable_feature):
        """Test that explicit token overrides auto-injection."""
        duplication = enable_feature
        
        with patch("meta_ads_mcp.core.auth.get_current_access_token") as mock_get_token:
            mock_get_token.return_value = "injected_token"
            
            with patch("meta_ads_mcp.core.duplication._forward_duplication_request") as mock_forward:
                mock_forward.return_value = '{"success": true}'
                
                # Call with explicit access_token
                await duplication.duplicate_campaign(
                    campaign_id="123456789",
                    access_token="explicit_token_12345"
                )
                
                # Verify the explicit token was used, not the injected one
                mock_forward.assert_called_once()
                call_args = mock_forward.call_args[0]
                assert call_args[2] == "explicit_token_12345"  # access_token is 3rd argument


class TestDuplicationRegressionEdgeCases:
    """Test edge cases that could cause regressions."""
    
    @pytest.fixture(autouse=True)
    def enable_feature(self):
        """Enable the duplication feature for these tests."""
        with patch.dict(os.environ, {"META_ADS_ENABLE_DUPLICATION": "1"}):
            import importlib
            from meta_ads_mcp.core import duplication
            importlib.reload(duplication)
            yield duplication
    
    @pytest.mark.asyncio
    async def test_empty_string_parameters(self, enable_feature):
        """Test handling of empty string parameters."""
        duplication = enable_feature
        
        with patch("meta_ads_mcp.core.duplication._forward_duplication_request") as mock_forward:
            mock_forward.return_value = '{"success": true}'
            
            # Test with empty strings
            await duplication.duplicate_campaign(
                campaign_id="123456789",
                access_token="token",
                name_suffix="",  # Empty string
                new_status=""    # Empty string
            )
            
            # Verify empty strings are preserved (not filtered like None)
            call_args = mock_forward.call_args[0]
            options = call_args[3]
            assert options["name_suffix"] == ""
            assert options["new_status"] == ""
    
    @pytest.mark.asyncio
    async def test_unicode_parameters(self, enable_feature):
        """Test handling of unicode parameters."""
        duplication = enable_feature
        
        # Mock dual-header authentication
        with patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration") as mock_auth:
            mock_auth.get_pipeboard_token.return_value = "pipeboard_token"
            mock_auth.get_auth_token.return_value = "facebook_token"
            
            with patch("meta_ads_mcp.core.duplication.httpx.AsyncClient") as mock_client:
                mock_response = MagicMock()
                mock_response.status_code = 200
                mock_response.json.return_value = {"success": True}
                mock_client.return_value.__aenter__.return_value.post.return_value = mock_response
                
                # Test with unicode characters
                unicode_suffix = " - 复制版本 🚀"
                await duplication._forward_duplication_request(
                    "campaign", "123", "token", {"name_suffix": unicode_suffix}
                )
                
                # Verify unicode is preserved in the request
                call_args = mock_client.return_value.__aenter__.return_value.post.call_args
                json_payload = call_args[1]["json"]
                assert json_payload["name_suffix"] == unicode_suffix

    @pytest.mark.asyncio
    async def test_large_parameter_values(self, enable_feature):
        """Test handling of large parameter values."""
        duplication = enable_feature
        
        # Mock dual-header authentication
        with patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration") as mock_auth:
            mock_auth.get_pipeboard_token.return_value = "pipeboard_token"
            mock_auth.get_auth_token.return_value = "facebook_token"
            
            with patch("meta_ads_mcp.core.duplication.httpx.AsyncClient") as mock_client:
                mock_response = MagicMock()
                mock_response.status_code = 200
                mock_response.json.return_value = {"success": True}
                mock_client.return_value.__aenter__.return_value.post.return_value = mock_response
                
                # Test with very large budget value
                large_budget = 999999999.99
                await duplication._forward_duplication_request(
                    "campaign", "123", "token", {"new_daily_budget": large_budget}
                )
                
                # Verify large values are preserved
                call_args = mock_client.return_value.__aenter__.return_value.post.call_args
                json_payload = call_args[1]["json"]
                assert json_payload["new_daily_budget"] == large_budget
    
    def test_module_reload_safety(self):
        """Test that module can be safely reloaded without side effects."""
        # This tests for common issues like global state pollution
        
        # Enable feature
        with patch.dict(os.environ, {"META_ADS_ENABLE_DUPLICATION": "1"}):
            import importlib
            from meta_ads_mcp.core import duplication
            importlib.reload(duplication)
            
            assert duplication.ENABLE_DUPLICATION
            assert hasattr(duplication, 'duplicate_campaign')
        
        # Disable feature and reload
        with patch.dict(os.environ, {"META_ADS_ENABLE_DUPLICATION": ""}):
            importlib.reload(duplication)
            
            assert not duplication.ENABLE_DUPLICATION
            # Note: Functions may still exist in the module namespace due to Python's 
            # module loading behavior, but they won't be registered as MCP tools
            # This is expected behavior and not a problem for the feature toggle
        
        # Re-enable feature and reload
        with patch.dict(os.environ, {"META_ADS_ENABLE_DUPLICATION": "1"}):
            importlib.reload(duplication)
            
            assert duplication.ENABLE_DUPLICATION
            assert hasattr(duplication, 'duplicate_campaign')