FabledSteward/tests/core/test_git_askpass.py

"""Unit tests for GIT_ASKPASS injection in git_pull()."""
import os
import stat
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
import pytest


def test_get_sources_includes_http_token(tmp_path):
    """get_sources() must pass http_token through from config."""
    from steward.ansible.sources import get_sources

    cfg = {
        "cache_dir": str(tmp_path),
        "sources": [
            {
                "name": "repo",
                "type": "git",
                "url": "https://git.example.com/org/repo.git",
                "http_token": "mytoken123",
            }
        ],
    }
    sources = get_sources(cfg)
    src = next(s for s in sources if s["name"] == "repo")
    assert src["http_token"] == "mytoken123"


def test_get_sources_missing_http_token_defaults_empty(tmp_path):
    """http_token defaults to empty string when not in source config."""
    from steward.ansible.sources import get_sources

    cfg = {
        "cache_dir": str(tmp_path),
        "sources": [
            {"name": "r", "type": "git", "url": "https://example.com/r.git"}
        ],
    }
    sources = get_sources(cfg)
    src = next(s for s in sources if s["name"] == "r")
    assert src["http_token"] == ""


@pytest.mark.asyncio
async def test_git_pull_creates_askpass_when_token_set(tmp_path):
    """git_pull() sets GIT_ASKPASS env when source has http_token."""
    from steward.ansible.sources import git_pull

    # Simulate repo already cloned (has .git dir) so we hit the pull path.
    repo_path = tmp_path / "repo"
    (repo_path / ".git").mkdir(parents=True)

    source = {
        "name": "repo",
        "path": str(repo_path),
        "url": "https://git.example.com/org/repo.git",
        "branch": "main",
        "http_token": "secret-token",
    }

    captured_env = {}

    async def fake_subprocess(*args, **kwargs):
        captured_env.update(kwargs.get("env", {}))
        proc = MagicMock()
        proc.returncode = 0
        proc.communicate = AsyncMock(return_value=(b"", b""))
        return proc

    with patch("asyncio.create_subprocess_exec", side_effect=fake_subprocess):
        await git_pull(source)

    assert "GIT_ASKPASS" in captured_env
    askpass_path = captured_env["GIT_ASKPASS"]
    # Script should be cleaned up after the call
    assert not os.path.exists(askpass_path), "askpass script should be deleted in finally"


@pytest.mark.asyncio
async def test_git_pull_askpass_contains_token(tmp_path):
    """The askpass script content includes the token before cleanup."""
    from steward.ansible.sources import git_pull

    repo_path = tmp_path / "repo"
    (repo_path / ".git").mkdir(parents=True)

    source = {
        "name": "repo",
        "path": str(repo_path),
        "url": "https://git.example.com/org/repo.git",
        "branch": "main",
        "http_token": "secret-token",
    }

    script_content_captured = {}

    async def fake_subprocess(*args, **kwargs):
        askpass = kwargs.get("env", {}).get("GIT_ASKPASS", "")
        if askpass and os.path.exists(askpass):
            script_content_captured["content"] = Path(askpass).read_text()
            mode = stat.S_IMODE(os.stat(askpass).st_mode)
            script_content_captured["executable"] = bool(mode & stat.S_IXUSR)
        proc = MagicMock()
        proc.returncode = 0
        proc.communicate = AsyncMock(return_value=(b"", b""))
        return proc

    with patch("asyncio.create_subprocess_exec", side_effect=fake_subprocess):
        await git_pull(source)

    assert "secret-token" in script_content_captured.get("content", "")
    assert "oauth2" in script_content_captured.get("content", "")
    assert script_content_captured.get("executable") is True


@pytest.mark.asyncio
async def test_git_pull_no_askpass_without_token(tmp_path):
    """git_pull() does NOT set GIT_ASKPASS when http_token is empty."""
    from steward.ansible.sources import git_pull

    repo_path = tmp_path / "repo"
    (repo_path / ".git").mkdir(parents=True)

    source = {
        "name": "repo",
        "path": str(repo_path),
        "url": "https://git.example.com/org/repo.git",
        "branch": "main",
        "http_token": "",
    }

    captured_env = {}

    async def fake_subprocess(*args, **kwargs):
        captured_env.update(kwargs.get("env") or {})
        proc = MagicMock()
        proc.returncode = 0
        proc.communicate = AsyncMock(return_value=(b"", b""))
        return proc

    with patch("asyncio.create_subprocess_exec", side_effect=fake_subprocess):
        await git_pull(source)

    assert "GIT_ASKPASS" not in captured_env