Orchestrator/shared/AI-Orchestrator/tests/test_secret_updater.py

"""Tests for the SecretUpdater module."""

from __future__ import annotations

import base64

import httpx
import pytest
import respx
from nacl.public import PrivateKey

from scripts.pat_manager.errors import SecretUpdateError
from scripts.pat_manager.secret_updater import SecretUpdater


@pytest.fixture
def keypair():
    """Generate a NaCl keypair for testing encryption."""
    private_key = PrivateKey.generate()
    public_key = private_key.public_key
    public_key_b64 = base64.b64encode(bytes(public_key)).decode("utf-8")
    return private_key, public_key, public_key_b64


@pytest.fixture
def updater():
    """Create a SecretUpdater instance for testing."""
    return SecretUpdater(
        github_token="ghp_test_token_123",
        repo_owner="test-owner",
        repo_name="test-repo",
    )


@pytest.mark.asyncio
@respx.mock
async def test_update_secret_success(updater, keypair):
    """Test successful secret update flow."""
    _, _, public_key_b64 = keypair

    # Mock GET public key endpoint
    respx.get(
        "https://api.github.com/repos/test-owner/test-repo/actions/secrets/public-key"
    ).mock(
        return_value=httpx.Response(
            200,
            json={"key": public_key_b64, "key_id": "key-id-123"},
        )
    )

    # Mock PUT secret endpoint
    respx.put(
        "https://api.github.com/repos/test-owner/test-repo/actions/secrets/MY_SECRET"
    ).mock(return_value=httpx.Response(204))

    result = await updater.update_secret("MY_SECRET", "super-secret-value")
    assert result is True


@pytest.mark.asyncio
@respx.mock
async def test_update_secret_created_201(updater, keypair):
    """Test secret creation returns True on 201."""
    _, _, public_key_b64 = keypair

    respx.get(
        "https://api.github.com/repos/test-owner/test-repo/actions/secrets/public-key"
    ).mock(
        return_value=httpx.Response(
            200,
            json={"key": public_key_b64, "key_id": "key-id-456"},
        )
    )

    respx.put(
        "https://api.github.com/repos/test-owner/test-repo/actions/secrets/NEW_SECRET"
    ).mock(return_value=httpx.Response(201))

    result = await updater.update_secret("NEW_SECRET", "new-value")
    assert result is True


@pytest.mark.asyncio
@respx.mock
async def test_update_secret_public_key_fetch_fails_http_error(updater):
    """Test SecretUpdateError raised when public key fetch returns non-200."""
    respx.get(
        "https://api.github.com/repos/test-owner/test-repo/actions/secrets/public-key"
    ).mock(return_value=httpx.Response(403))

    with pytest.raises(SecretUpdateError) as exc_info:
        await updater.update_secret("MY_SECRET", "value")

    assert "public key" in str(exc_info.value).lower()
    assert "403" in str(exc_info.value)
    # Ensure token value is NOT in the error message
    assert "value" not in str(exc_info.value) or "secret value" not in str(
        exc_info.value
    )


@pytest.mark.asyncio
@respx.mock
async def test_update_secret_public_key_network_error(updater):
    """Test SecretUpdateError raised on network failure during key fetch."""
    respx.get(
        "https://api.github.com/repos/test-owner/test-repo/actions/secrets/public-key"
    ).mock(side_effect=httpx.ConnectError("Connection refused"))

    with pytest.raises(SecretUpdateError) as exc_info:
        await updater.update_secret("MY_SECRET", "value")

    assert "MY_SECRET" in str(exc_info.value)
    assert "ConnectError" in str(exc_info.value)


@pytest.mark.asyncio
@respx.mock
async def test_update_secret_put_fails_http_error(updater, keypair):
    """Test SecretUpdateError raised when PUT returns unexpected status."""
    _, _, public_key_b64 = keypair

    respx.get(
        "https://api.github.com/repos/test-owner/test-repo/actions/secrets/public-key"
    ).mock(
        return_value=httpx.Response(
            200,
            json={"key": public_key_b64, "key_id": "key-id-789"},
        )
    )

    respx.put(
        "https://api.github.com/repos/test-owner/test-repo/actions/secrets/MY_SECRET"
    ).mock(return_value=httpx.Response(422))

    with pytest.raises(SecretUpdateError) as exc_info:
        await updater.update_secret("MY_SECRET", "secret-token-value")

    assert "422" in str(exc_info.value)
    # Ensure the actual secret value is NOT exposed in the error
    assert "secret-token-value" not in str(exc_info.value)


@pytest.mark.asyncio
@respx.mock
async def test_update_secret_put_network_error(updater, keypair):
    """Test SecretUpdateError raised on network failure during PUT."""
    _, _, public_key_b64 = keypair

    respx.get(
        "https://api.github.com/repos/test-owner/test-repo/actions/secrets/public-key"
    ).mock(
        return_value=httpx.Response(
            200,
            json={"key": public_key_b64, "key_id": "key-id-000"},
        )
    )

    respx.put(
        "https://api.github.com/repos/test-owner/test-repo/actions/secrets/MY_SECRET"
    ).mock(side_effect=httpx.TimeoutException("Request timed out"))

    with pytest.raises(SecretUpdateError) as exc_info:
        await updater.update_secret("MY_SECRET", "value")

    assert "MY_SECRET" in str(exc_info.value)
    assert "TimeoutException" in str(exc_info.value)


@pytest.mark.asyncio
@respx.mock
async def test_update_secret_does_not_expose_token_value(updater, keypair):
    """Verify that error messages never contain the secret value."""
    _, _, public_key_b64 = keypair
    sensitive_value = "ghp_SUPER_SECRET_TOKEN_12345"

    respx.get(
        "https://api.github.com/repos/test-owner/test-repo/actions/secrets/public-key"
    ).mock(
        return_value=httpx.Response(
            200,
            json={"key": public_key_b64, "key_id": "key-id-sec"},
        )
    )

    respx.put(
        "https://api.github.com/repos/test-owner/test-repo/actions/secrets/TOKEN_SECRET"
    ).mock(return_value=httpx.Response(500))

    with pytest.raises(SecretUpdateError) as exc_info:
        await updater.update_secret("TOKEN_SECRET", sensitive_value)

    error_str = str(exc_info.value)
    assert sensitive_value not in error_str
    assert "ghp_SUPER_SECRET" not in error_str


@pytest.mark.asyncio
async def test_encrypt_secret_produces_valid_output(updater, keypair):
    """Test that _encrypt_secret produces base64-encoded encrypted data."""
    _, _, public_key_b64 = keypair

    encrypted = updater._encrypt_secret(public_key_b64, "test-secret")

    # Should be valid base64
    decoded = base64.b64decode(encrypted)
    # NaCl sealed box output is always 48 bytes longer than the plaintext
    # (32 bytes ephemeral public key + 16 bytes MAC)
    assert len(decoded) == len("test-secret".encode("utf-8")) + 48


@pytest.mark.asyncio
@respx.mock
async def test_update_secret_sends_correct_headers(updater, keypair):
    """Verify correct authorization headers are sent."""
    _, _, public_key_b64 = keypair

    key_route = respx.get(
        "https://api.github.com/repos/test-owner/test-repo/actions/secrets/public-key"
    ).mock(
        return_value=httpx.Response(
            200,
            json={"key": public_key_b64, "key_id": "key-id-hdr"},
        )
    )

    put_route = respx.put(
        "https://api.github.com/repos/test-owner/test-repo/actions/secrets/HDR_SECRET"
    ).mock(return_value=httpx.Response(204))

    await updater.update_secret("HDR_SECRET", "value")

    # Verify authorization header was sent
    assert key_route.called
    key_request = key_route.calls[0].request
    assert key_request.headers["Authorization"] == "Bearer ghp_test_token_123"

    assert put_route.called
    put_request = put_route.calls[0].request
    assert put_request.headers["Authorization"] == "Bearer ghp_test_token_123"