144 lines
4.4 KiB
Python
144 lines
4.4 KiB
Python
"""Unit tests for GIT_ASKPASS injection in git_pull()."""
|
|
import asyncio
|
|
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)
|
|
assert sources[0]["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)
|
|
assert sources[0]["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
|