"""
Unit tests for agent secrets context inclusion logic.

Tests the core logic around:
1. _get_secrets_block only including fulfilled secrets (requested=False)
2. secure_credentials_request tool execution
3. Proper filtering of requested vs fulfilled secrets
"""
import re
from unittest.mock import Mock, patch
from django.test import TestCase, tag
from django.contrib.auth import get_user_model

from api.models import (
    PersistentAgent,
    PersistentAgentSecret,
    GlobalSecret,
    BrowserUseAgent,
    Organization,
    OrganizationMembership,
)
from api.agent.core.prompt_context import _get_secrets_block
from api.agent.tools.secure_credentials_request import (
    execute_secure_credentials_request,
    get_secure_credentials_request_tool,
)
from console.forms import PersistentAgentAddSecretForm, PersistentAgentEditSecretForm, PersistentAgentSecretsRequestForm

User = get_user_model()


@tag("batch_agent_secrets_ctx")
class GetSecretsBlockTests(TestCase):
    """Test the _get_secrets_block function that includes secrets in agent context."""

    def setUp(self):
        """Set up test agent and user."""
        self.user = User.objects.create_user(
            username="test@example.com",
            email="test@example.com",
            password="password"
        )
        self.browser_agent = BrowserUseAgent.objects.create(
            user=self.user,
            name="TestBrowserAgent"
        )
        self.agent = PersistentAgent.objects.create(
            user=self.user,
            browser_use_agent=self.browser_agent,
            name="TestAgent"
        )

    def test_only_fulfilled_secrets_in_context(self):
        """Test that only secrets with requested=False appear in context."""
        # Create fulfilled secrets (should appear)
        fulfilled_secret1 = PersistentAgentSecret.objects.create(
            agent=self.agent,
            domain_pattern="https://api.example.com",
            name="API Key",
            description="Main API key",
            key="api_key",
            requested=False,
            encrypted_value=b"encrypted_value_1"
        )

        fulfilled_secret2 = PersistentAgentSecret.objects.create(
            agent=self.agent,
            domain_pattern="https://auth.example.com",
            name="Auth Token",
            key="auth_token",
            requested=False,
            encrypted_value=b"encrypted_value_2"
        )

        # Create requested secrets (should NOT appear)
        requested_secret1 = PersistentAgentSecret.objects.create(
            agent=self.agent,
            domain_pattern="https://pending.example.com",
            name="Pending Secret",
            key="pending_secret",
            requested=True,
            encrypted_value=b""  # Empty as it's just requested
        )

        requested_secret2 = PersistentAgentSecret.objects.create(
            agent=self.agent,
            domain_pattern="https://waiting.example.com",
            name="Waiting Secret",
            key="waiting_secret",
            requested=True,
            encrypted_value=b""
        )

        # Get the secrets block
        result = _get_secrets_block(self.agent)

        # Verify fulfilled secrets ARE in the result
        self.assertIn("api_key", result)
        self.assertIn("auth_token", result)
        self.assertIn("API Key", result)
        self.assertIn("Auth Token", result)
        self.assertIn("https://api.example.com", result)
        self.assertIn("https://auth.example.com", result)

        # Verify requested secrets are surfaced as pending (not as available)
        self.assertIn("Pending credential requests", result)
        self.assertIn("follow up with the user", result)
        self.assertIn("pending_secret", result)
        self.assertIn("waiting_secret", result)
        self.assertIn("https://pending.example.com", result)
        self.assertIn("https://waiting.example.com", result)

    def test_secrets_context_separates_secret_types_and_hides_values(self):
        credential_secret = PersistentAgentSecret(
            agent=self.agent,
            secret_type=PersistentAgentSecret.SecretType.CREDENTIAL,
            domain_pattern="https://api.example.com",
            name="API Credential",
            key="api_credential",
            requested=False,
        )
        credential_secret.set_value("credential-secret-value")
        credential_secret.save()

        env_var_secret = PersistentAgentSecret(
            agent=self.agent,
            secret_type=PersistentAgentSecret.SecretType.ENV_VAR,
            domain_pattern=PersistentAgentSecret.ENV_VAR_DOMAIN_SENTINEL,
            name="Sandbox Token",
            key="SANDBOX_TOKEN",
            requested=False,
        )
        env_var_secret.set_value("env-var-secret-value")
        env_var_secret.save()

        PersistentAgentSecret.objects.create(
            agent=self.agent,
            secret_type=PersistentAgentSecret.SecretType.CREDENTIAL,
            domain_pattern="https://api.example.com",
            name="Pending API Credential",
            key="pending_api_credential",
            requested=True,
            encrypted_value=b"",
        )
        PersistentAgentSecret.objects.create(
            agent=self.agent,
            secret_type=PersistentAgentSecret.SecretType.ENV_VAR,
            domain_pattern=PersistentAgentSecret.ENV_VAR_DOMAIN_SENTINEL,
            name="Pending Sandbox Token",
            key="PENDING_SANDBOX_TOKEN",
            requested=True,
            encrypted_value=b"",
        )

        result = _get_secrets_block(self.agent)

        self.assertIn("Agent-specific domain-scoped credential secrets", result)
        self.assertIn("not readable via os.environ", result)
        self.assertIn("Agent-specific sandbox environment variable secrets", result)
        self.assertIn("ARE readable via os.environ", result)
        self.assertIn("Pending domain-scoped credentials", result)
        self.assertIn("Pending sandbox environment variables", result)
        self.assertIn("Key: api_credential", result)
        self.assertIn("Env Key: SANDBOX_TOKEN", result)
        self.assertNotIn("credential-secret-value", result)
        self.assertNotIn("env-var-secret-value", result)

    def test_no_secrets_returns_empty_message(self):
        """Test that when no fulfilled secrets exist, appropriate message is returned."""
        # Create only requested secrets
        PersistentAgentSecret.objects.create(
            agent=self.agent,
            domain_pattern="https://example.com",
            name="Requested Secret",
            key="requested_key",
            requested=True,
            encrypted_value=b""
        )

        result = _get_secrets_block(self.agent)

        self.assertIn("Pending credential requests", result)
        self.assertIn("requested_key", result)

    def test_secrets_grouped_by_domain(self):
        """Test that secrets are properly grouped by domain pattern."""
        # Create multiple secrets for same domain
        PersistentAgentSecret.objects.create(
            agent=self.agent,
            domain_pattern="*.google.com",
            name="Google API Key",
            key="google_api_key",
            requested=False,
            encrypted_value=b"encrypted1"
        )

        PersistentAgentSecret.objects.create(
            agent=self.agent,
            domain_pattern="*.google.com",
            name="Google Secret",
            key="google_secret",
            requested=False,
            encrypted_value=b"encrypted2"
        )

        # Create secret for different domain
        PersistentAgentSecret.objects.create(
            agent=self.agent,
            domain_pattern="*.github.com",
            name="GitHub Token",
            key="github_token",
            requested=False,
            encrypted_value=b"encrypted3"
        )

        result = _get_secrets_block(self.agent)

        # Check structure
        self.assertIn("Domain: *.github.com", result)
        self.assertIn("Domain: *.google.com", result)

        # Check that all secrets are present
        self.assertIn("google_api_key", result)
        self.assertIn("google_secret", result)
        self.assertIn("github_token", result)

        # Verify ordering - domains should be alphabetically sorted
        lines = result.split('\n')
        # Find domain lines
        domain_lines = [i for i, line in enumerate(lines) if "Domain:" in line]
        self.assertEqual(len(domain_lines), 2)

        # GitHub should come before Google alphabetically
        self.assertIn("*.github.com", lines[domain_lines[0]])
        self.assertIn("*.google.com", lines[domain_lines[1]])

    def test_mixed_requested_and_fulfilled_secrets(self):
        """Test filtering when both requested and fulfilled secrets exist."""
        # Create a mix of secrets
        fulfilled1 = PersistentAgentSecret.objects.create(
            agent=self.agent,
            domain_pattern="https://api.com",
            name="Fulfilled API Key",
            key="fulfilled_key",
            requested=False,
            encrypted_value=b"has_value"
        )

        requested1 = PersistentAgentSecret.objects.create(
            agent=self.agent,
            domain_pattern="https://api.com",  # Same domain as fulfilled
            name="Requested API Secret",
            key="requested_secret",
            requested=True,
            encrypted_value=b""
        )

        fulfilled2 = PersistentAgentSecret.objects.create(
            agent=self.agent,
            domain_pattern="https://other.com",
            name="Other Key",
            key="other_key",
            requested=False,
            encrypted_value=b"value"
        )

        result = _get_secrets_block(self.agent)

        # Fulfilled secrets should be listed as available
        self.assertIn("fulfilled_key", result)
        self.assertIn("other_key", result)

        # Pending secrets should be surfaced under the pending section
        self.assertIn("Pending credential requests", result)
        self.assertIn("requested_secret", result)

        # Both domains with fulfilled secrets should appear
        self.assertIn("https://api.com", result)
        self.assertIn("https://other.com", result)


@tag("batch_agent_secrets_ctx")
class SecureCredentialsRequestToolTests(TestCase):
    """Test the secure_credentials_request tool execution."""

    def setUp(self):
        """Set up test agent and user."""
        self.user = User.objects.create_user(
            username="test@example.com",
            email="test@example.com",
            password="password"
        )
        self.browser_agent = BrowserUseAgent.objects.create(
            user=self.user,
            name="TestBrowserAgent"
        )
        self.agent = PersistentAgent.objects.create(
            user=self.user,
            browser_use_agent=self.browser_agent,
            name="TestAgent"
        )

    def test_tool_definition_structure(self):
        """Test that the tool definition has correct structure."""
        tool_def = get_secure_credentials_request_tool()

        self.assertEqual(tool_def["type"], "function")
        self.assertEqual(tool_def["function"]["name"], "secure_credentials_request")
        self.assertIn("custom tool script", tool_def["function"]["description"])
        self.assertIn("os.environ", tool_def["function"]["description"])
        self.assertIn("ALWAYS set secret_type='env_var'", tool_def["function"]["description"])
        self.assertIn("omit domain_pattern", tool_def["function"]["description"])
        self.assertIn("credentials", tool_def["function"]["parameters"]["properties"])
        self.assertIn("credentials", tool_def["function"]["parameters"]["required"])
        item_properties = (
            tool_def["function"]["parameters"]["properties"]["credentials"]["items"]["properties"]
        )
        self.assertIn("secret_type", item_properties)
        self.assertEqual(item_properties["secret_type"]["enum"], ["credential", "env_var"])
        self.assertIn("custom tool scripts", item_properties["secret_type"]["description"])
        self.assertIn("os.environ", item_properties["secret_type"]["description"])
        self.assertIn("MUST be env_var", item_properties["secret_type"]["description"])
        self.assertIn(
            "Required for credential; omit it for env_var.",
            item_properties["domain_pattern"]["description"],
        )

    def test_create_new_credential_requests(self):
        """Test creating new credential requests."""
        params = {
            "credentials": [
                {
                    "name": "API Key",
                    "description": "Key for API access",
                    "key": "api_key",
                    "domain_pattern": "https://api.example.com"
                },
                {
                    "name": "Secret Token",
                    "description": "Authentication token",
                    "key": "auth_token",
                    "domain_pattern": "*.auth.example.com"
                }
            ]
        }

        result = execute_secure_credentials_request(self.agent, params)

        # Check result
        self.assertEqual(result["status"], "ok")
        self.assertEqual(result["created_count"], 2)
        self.assertIn("Processed 2 credential request(s)", result["message"])
        self.assertIn("API Key", result["message"])
        self.assertIn("Secret Token", result["message"])

        # Verify secrets were created with requested=True
        created_secrets = PersistentAgentSecret.objects.filter(agent=self.agent)
        self.assertEqual(created_secrets.count(), 2)

        for secret in created_secrets:
            self.assertTrue(secret.requested)
            self.assertEqual(secret.encrypted_value, b"")  # Empty for requested

        # Verify specific secrets
        api_key_secret = created_secrets.get(key="api_key")
        self.assertEqual(api_key_secret.name, "API Key")
        self.assertEqual(api_key_secret.domain_pattern, "https://api.example.com")
        self.assertEqual(api_key_secret.description, "Key for API access")

        auth_token_secret = created_secrets.get(key="auth_token")
        self.assertEqual(auth_token_secret.name, "Secret Token")
        self.assertEqual(auth_token_secret.domain_pattern, "https://*.auth.example.com")

    def test_duplicate_request_skipped(self):
        """Requesting an already-requested credential does not duplicate and reports success."""
        # Create an existing requested credential
        existing = PersistentAgentSecret.objects.create(
            agent=self.agent,
            domain_pattern="https://api.example.com",
            name="Existing API Key",
            key="api_key",
            requested=True,
            encrypted_value=b""
        )

        # Try to request the same credential again
        params = {
            "credentials": [
                {
                    "name": "API Key",
                    "description": "New description",
                    "key": "api_key",
                    "domain_pattern": "https://api.example.com"
                }
            ]
        }

        result = execute_secure_credentials_request(self.agent, params)
        # Treated as already present; we still communicate success without creating duplicates
        self.assertEqual(result["status"], "ok")
        self.assertEqual(result["created_count"], 1)

        # Verify only one secret exists
        secrets = PersistentAgentSecret.objects.filter(
            agent=self.agent,
            key="api_key",
            domain_pattern="https://api.example.com"
        )
        self.assertEqual(secrets.count(), 1)
        self.assertTrue(secrets.first().requested)

    @patch("api.agent.tools.secure_credentials_request.sandbox_compute_enabled_for_agent", return_value=True)
    def test_create_env_var_request_without_domain(self, _mock_sandbox_enabled):
        params = {
            "credentials": [
                {
                    "name": "OpenAI API Key",
                    "description": "Token for sandbox python code",
                    "key": "openai_api_key",
                    "secret_type": "env_var",
                }
            ]
        }

        result = execute_secure_credentials_request(self.agent, params)

        self.assertEqual(result["status"], "ok")
        self.assertEqual(result["created_count"], 1)
        secret = PersistentAgentSecret.objects.get(agent=self.agent, key="OPENAI_API_KEY")
        self.assertEqual(secret.secret_type, PersistentAgentSecret.SecretType.ENV_VAR)
        self.assertEqual(secret.domain_pattern, PersistentAgentSecret.ENV_VAR_DOMAIN_SENTINEL)
        self.assertTrue(secret.requested)

    @patch("api.agent.tools.secure_credentials_request.sandbox_compute_enabled_for_agent", return_value=False)
    def test_env_var_request_requires_sandbox_eligibility(self, _mock_sandbox_enabled):
        params = {
            "credentials": [
                {
                    "name": "OpenAI API Key",
                    "description": "Token for sandbox python code",
                    "key": "OPENAI_API_KEY",
                    "secret_type": "env_var",
                }
            ]
        }

        result = execute_secure_credentials_request(self.agent, params)

        self.assertEqual(result["status"], "error")
        self.assertIn("sandbox compute is not enabled", result["message"])
        self.assertEqual(PersistentAgentSecret.objects.filter(agent=self.agent).count(), 0)

    @patch("api.agent.tools.secure_credentials_request.sandbox_compute_enabled_for_agent", return_value=True)
    def test_env_var_request_is_inferred_for_custom_tool_env_usage(self, _mock_sandbox_enabled):
        params = {
            "credentials": [
                {
                    "name": "OpenAI API Key",
                    "description": "Token for a custom tool script to read from os.environ in sandbox python code",
                    "key": "openai_api_key",
                }
            ]
        }

        result = execute_secure_credentials_request(self.agent, params)

        self.assertEqual(result["status"], "ok")
        self.assertEqual(result["created_count"], 1)
        secret = PersistentAgentSecret.objects.get(agent=self.agent, key="OPENAI_API_KEY")
        self.assertEqual(secret.secret_type, PersistentAgentSecret.SecretType.ENV_VAR)
        self.assertEqual(secret.domain_pattern, PersistentAgentSecret.ENV_VAR_DOMAIN_SENTINEL)
        self.assertTrue(secret.requested)

    @patch("api.agent.tools.secure_credentials_request.sandbox_compute_enabled_for_agent", return_value=True)
    def test_env_var_request_rejects_invalid_env_key(self, _mock_sandbox_enabled):
        params = {
            "credentials": [
                {
                    "name": "Invalid Env",
                    "description": "Invalid key format",
                    "key": "bad-key",
                    "secret_type": "env_var",
                }
            ]
        }

        result = execute_secure_credentials_request(self.agent, params)

        self.assertEqual(result["status"], "error")
        self.assertIn("Failed to create any credential requests", result["message"])
        self.assertEqual(PersistentAgentSecret.objects.filter(agent=self.agent).count(), 0)

    def test_re_request_already_fulfilled_secret(self):
        """Re-requesting a fulfilled credential now leaves it fulfilled and points user to update."""
        # Create a fulfilled secret
        fulfilled = PersistentAgentSecret.objects.create(
            agent=self.agent,
            domain_pattern="https://api.example.com",
            name="Fulfilled Key",
            key="api_key",
            requested=False,
            encrypted_value=b"encrypted_value_here"
        )

        # Try to request a credential with same key and domain
        params = {
            "credentials": [
                {
                    "name": "API Key",
                    "description": "Trying to re-request",
                    "key": "api_key",
                    "domain_pattern": "https://api.example.com"
                }
            ]
        }

        result = execute_secure_credentials_request(self.agent, params)
        # Now treated as ok (converted to requested)
        self.assertEqual(result["status"], "ok")
        self.assertEqual(result["created_count"], 1)
        self.assertIn("processed 1 credential request", result["message"].lower())
        # Re-request should include update link guidance
        self.assertIn("update the existing credential(s)", result["message"].lower())

        fulfilled.refresh_from_db()
        # We no longer flip to requested or wipe the value
        self.assertFalse(fulfilled.requested)
        self.assertEqual(fulfilled.encrypted_value, b"encrypted_value_here")

        # It should remain in the available secrets block (not pending)
        context_block = _get_secrets_block(self.agent)
        self.assertIn("api_key", context_block)
        self.assertNotIn("Pending credential requests", context_block)

        # Simulate fulfilling again and ensure it appears
        fulfilled.set_value("new-value")
        fulfilled.requested = False
        fulfilled.save()
        context_block2 = _get_secrets_block(self.agent)
        self.assertIn("api_key", context_block2)


from django.urls import reverse
from django.test import Client


@tag("batch_agent_secrets_ctx")
class AgentSecretsRequestViewTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            username="test2@example.com",
            email="test2@example.com",
            password="password"
        )
        self.org = Organization.objects.create(
            name="Secrets Org",
            slug="secrets-org",
            created_by=self.user,
        )
        OrganizationMembership.objects.create(
            org=self.org,
            user=self.user,
            role=OrganizationMembership.OrgRole.OWNER,
            status=OrganizationMembership.OrgStatus.ACTIVE,
        )
        self.browser_agent = BrowserUseAgent.objects.create(
            user=self.user,
            name="BrowserAgent2"
        )
        with patch.object(PersistentAgent, "_validate_org_seats", return_value=None):
            self.agent = PersistentAgent.objects.create(
                user=self.user,
                organization=self.org,
                browser_use_agent=self.browser_agent,
                name="Agent2"
            )
        self.client = Client()
        assert self.client.login(username="test2@example.com", password="password")
        session = self.client.session
        session["context_type"] = "organization"
        session["context_id"] = str(self.org.id)
        session["context_name"] = self.org.name
        session.save()

    def _make_global_secret(self, **kwargs):
        defaults = {
            "organization": self.org,
            "name": "Existing Global Secret",
            "secret_type": GlobalSecret.SecretType.CREDENTIAL,
            "domain_pattern": "https://example.com",
            "description": "shared secret",
        }
        defaults.update(kwargs)
        secret = GlobalSecret(**defaults)
        secret.set_value("shared-value")
        secret.save()
        return secret

    def test_partial_request_save_updates_only_provided(self):
        # Two requested secrets
        s1 = PersistentAgentSecret.objects.create(
            agent=self.agent,
            domain_pattern="https://example.com",
            name="Username",
            key="username",
            requested=True,
            encrypted_value=b""
        )
        s2 = PersistentAgentSecret.objects.create(
            agent=self.agent,
            domain_pattern="https://example.com",
            name="Password",
            key="password",
            requested=True,
            encrypted_value=b""
        )

        url = reverse('agent_secrets_request', kwargs={"pk": self.agent.id})
        # Provide only username
        resp = self.client.post(url, data={f"secret_{s1.id}": "alice"})
        self.assertIn(resp.status_code, (302, 200))

        s1.refresh_from_db()
        s2.refresh_from_db()
        self.assertFalse(s1.requested)
        self.assertTrue(s2.requested)
        self.assertFalse(GlobalSecret.objects.filter(organization=self.org, name="Username").exists())

    def test_request_page_renders_make_global_option_for_organization(self):
        secret = PersistentAgentSecret.objects.create(
            agent=self.agent,
            domain_pattern="https://example.com",
            name="Password",
            key="password",
            requested=True,
            encrypted_value=b"",
        )

        response = self.client.get(reverse('agent_secrets_request', kwargs={"pk": self.agent.id}))

        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "Make global for this organization")
        self.assertContains(response, "Shared across all agents in this organization")
        self.assertContains(response, 'name="make_global"', html=False)
        checkbox = re.search(
            r'<input[^>]*name="make_global"[^>]*>',
            response.content.decode(),
        )
        self.assertIsNotNone(checkbox)
        self.assertNotIn("checked", checkbox.group(0))
        self.assertEqual(response.content.decode().count('name="make_global"'), 1)

    def test_request_page_disables_autofill_for_secret_inputs(self):
        secret = PersistentAgentSecret.objects.create(
            agent=self.agent,
            domain_pattern="https://example.com",
            name="Password",
            key="password",
            requested=True,
            encrypted_value=b"",
        )

        response = self.client.get(reverse('agent_secrets_request', kwargs={"pk": self.agent.id}))

        self.assertEqual(response.status_code, 200)
        self.assertContains(response, '<form method="post" autocomplete="off">', html=False)
        self.assertContains(response, f'name="secret_{secret.id}"', html=False)
        self.assertContains(response, 'type="password"', html=False)
        self.assertContains(response, 'autocomplete="new-password"', html=False)

    def test_request_save_can_make_credential_global(self):
        secret = PersistentAgentSecret.objects.create(
            agent=self.agent,
            domain_pattern="https://example.com",
            name="Shared Password",
            key="shared_password",
            requested=True,
            encrypted_value=b"",
        )

        response = self.client.post(
            reverse('agent_secrets_request', kwargs={"pk": self.agent.id}),
            data={
                f"secret_{secret.id}": "top-secret",
                "make_global": "on",
            },
        )

        self.assertEqual(response.status_code, 302)
        self.assertEqual(
            response.url,
            reverse('agent_secrets_request_thanks', kwargs={"pk": self.agent.id}),
        )
        self.assertFalse(PersistentAgentSecret.objects.filter(id=secret.id).exists())
        global_secret = GlobalSecret.objects.get(organization=self.org, name="Shared Password")
        self.assertEqual(global_secret.secret_type, GlobalSecret.SecretType.CREDENTIAL)
        self.assertEqual(global_secret.domain_pattern, "https://example.com")
        self.assertEqual(global_secret.get_value(), "top-secret")

    def test_request_save_can_make_env_var_global(self):
        secret = PersistentAgentSecret.objects.create(
            agent=self.agent,
            secret_type=PersistentAgentSecret.SecretType.ENV_VAR,
            domain_pattern=PersistentAgentSecret.ENV_VAR_DOMAIN_SENTINEL,
            name="Sandbox Token",
            key="SANDBOX_TOKEN",
            requested=True,
            encrypted_value=b"",
        )

        response = self.client.post(
            reverse('agent_secrets_request', kwargs={"pk": self.agent.id}),
            data={
                f"secret_{secret.id}": "env-secret",
                "make_global": "on",
            },
        )

        self.assertEqual(response.status_code, 302)
        self.assertFalse(PersistentAgentSecret.objects.filter(id=secret.id).exists())
        global_secret = GlobalSecret.objects.get(organization=self.org, name="Sandbox Token")
        self.assertEqual(global_secret.secret_type, GlobalSecret.SecretType.ENV_VAR)
        self.assertEqual(global_secret.domain_pattern, GlobalSecret.ENV_VAR_DOMAIN_SENTINEL)
        self.assertEqual(global_secret.key, "SANDBOX_TOKEN")
        self.assertEqual(global_secret.get_value(), "env-secret")

    def test_request_save_make_global_duplicate_rerenders_without_changes(self):
        secret = PersistentAgentSecret.objects.create(
            agent=self.agent,
            domain_pattern="https://example.com",
            name="Shared Password",
            key="shared_password",
            requested=True,
            encrypted_value=b"",
        )
        self._make_global_secret(
            name="Shared Password",
            key="shared_password",
            domain_pattern="https://example.com",
        )

        response = self.client.post(
            reverse('agent_secrets_request', kwargs={"pk": self.agent.id}),
            data={
                f"secret_{secret.id}": "top-secret",
                "make_global": "on",
            },
        )

        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "already exists")
        secret.refresh_from_db()
        self.assertTrue(secret.requested)
        self.assertEqual(secret.encrypted_value, b"")
        self.assertEqual(GlobalSecret.objects.filter(organization=self.org, name="Shared Password").count(), 1)

    def test_request_save_make_global_respects_global_secret_limit(self):
        secret = PersistentAgentSecret.objects.create(
            agent=self.agent,
            domain_pattern="https://example.com",
            name="Overflow Secret",
            key="overflow_secret",
            requested=True,
            encrypted_value=b"",
        )

        with patch.object(GlobalSecret, "MAX_GLOBAL_SECRETS_PER_OWNER", 1):
            self._make_global_secret(name="Existing One", key="existing_one")
            response = self.client.post(
                reverse('agent_secrets_request', kwargs={"pk": self.agent.id}),
                data={
                    f"secret_{secret.id}": "top-secret",
                    "make_global": "on",
                },
            )

        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "Maximum 1 global secrets allowed.")
        secret.refresh_from_db()
        self.assertTrue(secret.requested)
        self.assertEqual(GlobalSecret.objects.filter(organization=self.org).count(), 1)

    def test_edit_secret_keeps_existing_key_when_name_changes(self):
        secret = PersistentAgentSecret(
            agent=self.agent,
            secret_type=PersistentAgentSecret.SecretType.CREDENTIAL,
            domain_pattern="https://api.example.com",
            name="Original Name",
            key="api_key",
            requested=False,
        )
        secret.set_value("old-value")
        secret.save()

        edit_url = reverse(
            "agent_secrets_edit",
            kwargs={"pk": self.agent.id, "secret_id": secret.id},
        )
        response = self.client.post(
            edit_url,
            data={
                "secret_type": "credential",
                "domain": "https://api.example.com",
                "name": "Updated Display Name",
                "description": "updated description",
                "value": "new-value",
            },
        )

        self.assertEqual(response.status_code, 302)
        secret.refresh_from_db()
        self.assertEqual(secret.key, "api_key")
        self.assertEqual(secret.name, "Updated Display Name")


    def test_missing_required_fields(self):
        """Test that missing required fields in credentials causes error."""
        # Missing domain_pattern
        params = {
            "credentials": [
                {
                    "name": "API Key",
                    "description": "Key for API",
                    "key": "api_key"
                    # domain_pattern missing
                }
            ]
        }

        result = execute_secure_credentials_request(self.agent, params)

        self.assertIn("error", result["status"].lower())
        self.assertIn("Missing required fields", result["message"])

        # Verify no secret was created
        self.assertEqual(PersistentAgentSecret.objects.filter(agent=self.agent).count(), 0)

    def test_empty_credentials_list(self):
        """Test that empty credentials list returns error."""
        params = {
            "credentials": []
        }

        result = execute_secure_credentials_request(self.agent, params)

        self.assertEqual(result["status"], "error")
        # The actual implementation checks if credentials list is empty before checking individual items
        # But based on the test output, it seems to give a different error message
        self.assertIn("invalid", result["message"].lower())

    def test_invalid_params_format(self):
        """Test that invalid params format is handled gracefully."""
        # credentials not a list
        params = {
            "credentials": "not a list"
        }

        result = execute_secure_credentials_request(self.agent, params)

        self.assertEqual(result["status"], "error")
        self.assertIn("Missing or invalid", result["message"])

        # Missing credentials entirely
        params = {}

        result = execute_secure_credentials_request(self.agent, params)

        self.assertEqual(result["status"], "error")
        self.assertIn("Missing or invalid", result["message"])

    @patch('api.agent.tools.secure_credentials_request.reverse')
    @patch('django.contrib.sites.models.Site.objects.get_current')
    def test_url_generation_in_response(self, mock_site, mock_reverse):
        """Test that the response includes the correct URL for user to provide secrets."""
        # Mock the site and URL generation
        mock_site.return_value = Mock(domain="example.com")
        mock_reverse.return_value = f"/console/agents/{self.agent.id}/secrets/request/"

        params = {
            "credentials": [
                {
                    "name": "Test Key",
                    "description": "Test",
                    "key": "test_key",
                    "domain_pattern": "https://test.com"
                }
            ]
        }

        result = execute_secure_credentials_request(self.agent, params)

        self.assertEqual(result["status"], "ok")
        expected_url = f"https://example.com/console/agents/{self.agent.id}/secrets/request/"
        self.assertIn(expected_url, result["message"])
        self.assertIn("securely enter the requested credentials", result["message"])

    def test_mixed_success_and_errors(self):
        """Test handling of partial success when some credentials succeed and others fail."""
        # Create an existing fulfilled secret
        PersistentAgentSecret.objects.create(
            agent=self.agent,
            domain_pattern="https://existing.com",
            name="Existing",
            key="existing_key",
            requested=False,
            encrypted_value=b"value"
        )

        params = {
            "credentials": [
                {
                    "name": "New Key",
                    "description": "This should succeed",
                    "key": "new_key",
                    "domain_pattern": "https://new.com"
                },
                {
                    "name": "Existing Key",
                    "description": "This should fail",
                    "key": "existing_key",
                    "domain_pattern": "https://existing.com"
                },
                {
                    "name": "Invalid",
                    # Missing required fields - should fail
                    "key": "invalid_key"
                }
            ]
        }

        result = execute_secure_credentials_request(self.agent, params)

        # Partial success: one new and one re-request created; one invalid failed
        self.assertEqual(result["status"], "partial")
        self.assertEqual(result["created_count"], 2)
        self.assertIn("errors", result)
        self.assertEqual(len(result["errors"]), 1)

        # Verify only the successful one was created
        created = PersistentAgentSecret.objects.filter(
            agent=self.agent,
            key="new_key"
        )
        self.assertEqual(created.count(), 1)
        self.assertTrue(created.first().requested)


@tag("batch_agent_secrets_ctx")
class AgentSecretFormsTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            username="form-tests@example.com",
            email="form-tests@example.com",
            password="password",
        )
        self.browser_agent = BrowserUseAgent.objects.create(
            user=self.user,
            name="FormBrowserAgent",
        )
        self.agent = PersistentAgent.objects.create(
            user=self.user,
            browser_use_agent=self.browser_agent,
            name="FormAgent",
        )
        self.existing_secret = PersistentAgentSecret(
            agent=self.agent,
            secret_type=PersistentAgentSecret.SecretType.CREDENTIAL,
            domain_pattern="https://api.example.com",
            name="Existing Secret",
            key="existing_secret",
            requested=False,
        )
        self.existing_secret.set_value("existing-value")
        self.existing_secret.save()

    def test_add_form_requires_domain_for_credential(self):
        form = PersistentAgentAddSecretForm(
            data={
                "secret_type": "credential",
                "domain": "",
                "name": "Credential Secret",
                "description": "Credential secret",
                "value": "credential-value",
            },
            agent=self.agent,
        )

        self.assertFalse(form.is_valid())
        self.assertIn("domain", form.errors)

    def test_add_form_widgets_disable_autofill(self):
        form = PersistentAgentAddSecretForm(agent=self.agent)

        self.assertEqual(form.fields["domain"].widget.attrs["autocomplete"], "off")
        self.assertEqual(form.fields["domain"].widget.attrs["autocorrect"], "off")
        self.assertEqual(form.fields["domain"].widget.attrs["autocapitalize"], "off")
        self.assertEqual(form.fields["domain"].widget.attrs["spellcheck"], "false")
        self.assertEqual(form.fields["value"].widget.attrs["autocomplete"], "new-password")

    @patch("api.services.sandbox_compute.sandbox_compute_enabled_for_agent", return_value=True)
    def test_add_form_env_var_allows_blank_domain_and_uses_sentinel(self, _mock_sandbox_enabled):
        form = PersistentAgentAddSecretForm(
            data={
                "secret_type": "env_var",
                "domain": "",
                "name": "Sandbox Token",
                "description": "Sandbox env token",
                "value": "env-value",
            },
            agent=self.agent,
        )

        self.assertTrue(form.is_valid(), form.errors.as_json())
        self.assertEqual(
            form.cleaned_data["domain"],
            PersistentAgentSecret.ENV_VAR_DOMAIN_SENTINEL,
        )

    @patch("api.secret_key_generator.SecretKeyGenerator.generate_key_from_name", return_value="bad-key")
    @patch("api.services.sandbox_compute.sandbox_compute_enabled_for_agent", return_value=True)
    def test_add_form_env_var_rejects_invalid_generated_env_key(
        self,
        _mock_sandbox_enabled,
        _mock_generate_key,
    ):
        form = PersistentAgentAddSecretForm(
            data={
                "secret_type": "env_var",
                "domain": "",
                "name": "Sandbox Token",
                "description": "Sandbox env token",
                "value": "env-value",
            },
            agent=self.agent,
        )

        self.assertFalse(form.is_valid())
        self.assertIn("name", form.errors)

    @patch("api.services.sandbox_compute.sandbox_compute_enabled_for_agent", return_value=True)
    def test_edit_form_env_var_allows_blank_domain(self, _mock_sandbox_enabled):
        form = PersistentAgentEditSecretForm(
            data={
                "secret_type": "env_var",
                "domain": "",
                "name": "Sandbox Token",
                "description": "Updated env token",
                "value": "new-env-value",
            },
            agent=self.agent,
            secret=self.existing_secret,
        )

        self.assertTrue(form.is_valid(), form.errors.as_json())
        self.assertEqual(
            form.cleaned_data["domain"],
            PersistentAgentSecret.ENV_VAR_DOMAIN_SENTINEL,
        )

    def test_edit_form_widgets_disable_autofill(self):
        form = PersistentAgentEditSecretForm(agent=self.agent, secret=self.existing_secret)

        self.assertEqual(form.fields["domain"].widget.attrs["autocomplete"], "off")
        self.assertEqual(form.fields["domain"].widget.attrs["autocorrect"], "off")
        self.assertEqual(form.fields["domain"].widget.attrs["autocapitalize"], "off")
        self.assertEqual(form.fields["domain"].widget.attrs["spellcheck"], "false")
        self.assertEqual(form.fields["value"].widget.attrs["autocomplete"], "new-password")

    def test_requested_secret_form_widgets_use_new_password_autocomplete(self):
        requested_secret = PersistentAgentSecret.objects.create(
            agent=self.agent,
            domain_pattern="https://example.com",
            name="Requested Secret",
            key="requested_secret",
            requested=True,
            encrypted_value=b"",
        )

        form = PersistentAgentSecretsRequestForm(requested_secrets=[requested_secret])

        self.assertEqual(
            form.fields[f"secret_{requested_secret.id}"].widget.attrs["autocomplete"],
            "new-password",
        )
        self.assertIn("make_global", form.fields)
        self.assertFalse(form.fields["make_global"].initial)


@tag("batch_agent_secrets_ctx")
class SecretContextIntegrationTests(TestCase):
    """Integration tests for the full secret request → fulfill → context flow."""

    def setUp(self):
        """Set up test agent and user."""
        self.user = User.objects.create_user(
            username="test@example.com",
            email="test@example.com",
            password="password"
        )
        self.browser_agent = BrowserUseAgent.objects.create(
            user=self.user,
            name="TestBrowserAgent"
        )
        self.agent = PersistentAgent.objects.create(
            user=self.user,
            browser_use_agent=self.browser_agent,
            name="TestAgent"
        )

    def _create_agent_credential_secret(
        self,
        *,
        key,
        value=None,
        requested=False,
        name=None,
        domain_pattern="https://api.com",
    ):
        secret = PersistentAgentSecret.objects.create(
            agent=self.agent,
            domain_pattern=domain_pattern,
            name=name or key.replace("_", " ").title(),
            key=key,
            requested=requested,
            encrypted_value=b"",
        )
        if not requested:
            secret.set_value(value)
            secret.save()
        return secret

    def _create_global_credential_secret(
        self,
        *,
        key,
        value,
        name=None,
        domain_pattern="https://api.com",
    ):
        secret = GlobalSecret(
            user=self.user,
            domain_pattern=domain_pattern,
            name=name or key.replace("_", " ").title(),
            key=key,
            secret_type=GlobalSecret.SecretType.CREDENTIAL,
        )
        secret.set_value(value)
        secret.save()
        return secret

    def _execute_http_request(self, params):
        from api.agent.tools.http_request import execute_http_request

        with patch("api.agent.tools.http_request.select_proxy_for_persistent_agent") as mock_proxy:
            with patch("api.agent.tools.http_request.requests.request") as mock_request:
                mock_proxy.return_value = Mock(proxy_url="http://proxy:8080")
                mock_request.return_value = Mock(
                    status_code=200,
                    headers={"Content-Type": "text/plain"},
                    iter_content=lambda chunk_size: [b"ok"],
                    close=lambda: None,
                )

                result = execute_http_request(self.agent, params)

                mock_request.assert_called_once()
                return result, mock_request.call_args.args, mock_request.call_args.kwargs

    def test_full_request_fulfill_context_flow(self):
        """Test the complete flow from request to fulfillment to context inclusion."""
        # Step 1: Agent requests credentials
        request_params = {
            "credentials": [
                {
                    "name": "API Key",
                    "description": "Main API key",
                    "key": "api_key",
                    "domain_pattern": "https://api.example.com"
                }
            ]
        }

        result = execute_secure_credentials_request(self.agent, request_params)
        self.assertEqual(result["status"], "ok")

        # Step 2: Verify secret is surfaced as pending but not usable yet
        context_before = _get_secrets_block(self.agent)
        self.assertIn("Pending credential requests", context_before)
        self.assertIn("api_key", context_before)

        # Step 3: Simulate user providing the credential value
        secret = PersistentAgentSecret.objects.get(
            agent=self.agent,
            key="api_key"
        )
        self.assertTrue(secret.requested)

        # Fulfill the secret (this is what the console view does)
        secret.set_value("super-secret-api-key-value")
        secret.requested = False
        secret.save()

        # Step 4: Verify secret now appears in context
        context_after = _get_secrets_block(self.agent)
        self.assertNotEqual(context_after, "No secrets configured.")
        self.assertIn("api_key", context_after)
        self.assertIn("API Key", context_after)
        self.assertIn("https://api.example.com", context_after)

        # Verify the actual value is NOT in the context (only metadata)
        self.assertNotIn("super-secret-api-key-value", context_after)

    def test_http_request_only_uses_fulfilled_secrets(self):
        """Test that http_request tool only has access to fulfilled secrets."""
        # Create one fulfilled and one requested secret
        self._create_agent_credential_secret(
            key="fulfilled_key",
            value="fulfilled_value",
            name="Fulfilled Key",
        )

        self._create_agent_credential_secret(
            key="requested_key",
            requested=True,
            name="Requested Key",
        )

        result, _, request_kwargs = self._execute_http_request(
            {
                "method": "GET",
                "url": "https://api.com/test",
                "headers": {
                    "X-Fulfilled": "<<<fulfilled_key>>>",
                    "X-Requested": "<<<requested_key>>>",
                },
            }
        )

        self.assertEqual(result["status"], "ok")
        self.assertEqual(request_kwargs["headers"]["X-Fulfilled"], "fulfilled_value")
        self.assertEqual(request_kwargs["headers"]["X-Requested"], "<<<requested_key>>>")

    def test_http_request_uses_global_credential_placeholders(self):
        self._create_global_credential_secret(
            key="hubspot_api_key",
            value="global-hubspot-token",
            name="HubSpot API Key",
        )

        result, request_args, request_kwargs = self._execute_http_request(
            {
                "method": "POST",
                "url": "https://api.com/<<<hubspot_api_key>>>",
                "headers": {
                    "Authorization": "Bearer <<<hubspot_api_key>>>",
                },
                "body": {
                    "token": "<<<hubspot_api_key>>>",
                },
            }
        )

        self.assertEqual(result["status"], "ok")
        self.assertEqual(request_args[1], "https://api.com/global-hubspot-token")
        self.assertEqual(
            request_kwargs["headers"]["Authorization"],
            "Bearer global-hubspot-token",
        )
        self.assertEqual(request_kwargs["data"], '{"token": "global-hubspot-token"}')

    def test_http_request_agent_secret_overrides_global_secret_with_same_key(self):
        self._create_global_credential_secret(
            key="hubspot_api_key",
            value="global-hubspot-token",
            name="Global HubSpot API Key",
        )
        self._create_agent_credential_secret(
            key="hubspot_api_key",
            value="agent-hubspot-token",
            name="Agent HubSpot API Key",
        )

        result, _, request_kwargs = self._execute_http_request(
            {
                "method": "GET",
                "url": "https://api.com/test",
                "headers": {
                    "X-API-Key": "<<<hubspot_api_key>>>",
                },
            }
        )

        self.assertEqual(result["status"], "ok")
        self.assertEqual(request_kwargs["headers"]["X-API-Key"], "agent-hubspot-token")

    def test_http_request_requested_agent_secret_does_not_block_global_secret(self):
        self._create_global_credential_secret(
            key="hubspot_api_key",
            value="global-hubspot-token",
            name="Global HubSpot API Key",
        )
        self._create_agent_credential_secret(
            key="hubspot_api_key",
            requested=True,
            name="Requested HubSpot API Key",
        )

        result, _, request_kwargs = self._execute_http_request(
            {
                "method": "GET",
                "url": "https://api.com/test",
                "headers": {
                    "X-API-Key": "<<<hubspot_api_key>>>",
                    "X-Missing": "<<<missing_key>>>",
                },
            }
        )

        self.assertEqual(result["status"], "ok")
        self.assertEqual(request_kwargs["headers"]["X-API-Key"], "global-hubspot-token")
        self.assertEqual(request_kwargs["headers"]["X-Missing"], "<<<missing_key>>>")
