345 lines
13 KiB
Python
345 lines
13 KiB
Python
import os
|
|
import re
|
|
import pytest
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock, ANY, AsyncMock
|
|
|
|
import json
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
# Importer alt som skal testes
|
|
from server import app, TOOLS, WEB_AGENT_ALLOWED_TOOLS
|
|
from server import (
|
|
get_git_status,
|
|
get_git_branch,
|
|
get_git_log,
|
|
get_git_diff,
|
|
read_repo_file,
|
|
list_repo_files,
|
|
preview_git_change,
|
|
commit_and_push_files,
|
|
_validate_path_inside_repo,
|
|
_is_allowed_remote,
|
|
_mask_secrets,
|
|
_run_git
|
|
)
|
|
|
|
# --- Fixtures ---
|
|
|
|
@pytest.fixture
|
|
def mock_repo_root(monkeypatch):
|
|
"""Setter REPO_ROOT til en trygg, midlertidig mappe for unit-tester."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
monkeypatch.setattr("server.REPO_ROOT", Path(tmpdir))
|
|
yield Path(tmpdir)
|
|
|
|
@pytest.fixture
|
|
def isolated_git_repo(monkeypatch):
|
|
"""Setter opp et isolert Git-miljø og patcher REPO_ROOT for integrasjonstester."""
|
|
with tempfile.TemporaryDirectory() as tmpdir_str:
|
|
tmpdir = Path(tmpdir_str)
|
|
remote_repo_path = tmpdir / "remote.git"
|
|
local_repo_path = tmpdir / "local"
|
|
|
|
subprocess.run(["git", "init", "--bare", str(remote_repo_path)], check=True)
|
|
subprocess.run(["git", "clone", str(remote_repo_path), str(local_repo_path)], check=True)
|
|
subprocess.run(["git", "config", "user.name", "Test User"], cwd=local_repo_path, check=True)
|
|
subprocess.run(["git", "config", "user.email", "test@example.com"], cwd=local_repo_path, check=True)
|
|
# Add an initial commit so the repo is not empty
|
|
(local_repo_path / ".gitkeep").touch()
|
|
subprocess.run(["git", "add", ".gitkeep"], cwd=local_repo_path, check=True)
|
|
subprocess.run(["git", "commit", "-m", "Initial commit"], cwd=local_repo_path, check=True)
|
|
branch_name = subprocess.run(["git", "rev-parse", "--abbrev-ref", "HEAD"], cwd=local_repo_path, check=True, capture_output=True, text=True).stdout.strip()
|
|
subprocess.run(["git", "push", "origin", branch_name], cwd=local_repo_path, check=True)
|
|
|
|
monkeypatch.setattr("server.REPO_ROOT", local_repo_path)
|
|
yield local_repo_path, remote_repo_path
|
|
|
|
@pytest.fixture
|
|
def mcp_secret(monkeypatch):
|
|
"""Sets the MCP_SECRET environment variable for testing."""
|
|
secret = "test-secret-for-dispatch"
|
|
monkeypatch.setenv("MCP_SECRET", secret)
|
|
yield secret
|
|
|
|
# --- Unit-tester for Hjelpefunksjoner ---
|
|
|
|
def test_validate_path_inside_repo_success(mock_repo_root):
|
|
(mock_repo_root / "testdir").mkdir()
|
|
(mock_repo_root / "testdir" / "test.txt").touch()
|
|
assert _validate_path_inside_repo("testdir/test.txt") == (mock_repo_root / "testdir" / "test.txt").resolve()
|
|
|
|
@pytest.mark.parametrize(
|
|
"invalid_path, match_error",
|
|
[
|
|
(None, "Path must be a non-empty string."),
|
|
(123, "Path must be a non-empty string."),
|
|
("", "Path must be a non-empty string."),
|
|
(" ", "Path must be a non-empty string."),
|
|
('a\x00b', "Path cannot contain control characters."),
|
|
('a\rb', "Path cannot contain control characters."),
|
|
('a\nb', "Path cannot contain control characters."),
|
|
('a\\b', "Path cannot contain backslashes."),
|
|
("/etc/passwd", "Path cannot be absolute."),
|
|
("C:boot.ini", "Path cannot be absolute."),
|
|
("//server/share", "Path cannot be absolute."),
|
|
("./a", "Path contains a disallowed segment"),
|
|
("a/./b", "Path contains a disallowed segment"),
|
|
("../a", "Path contains a disallowed segment"),
|
|
("a/../b", "Path contains a disallowed segment"),
|
|
(".git/config", "Path contains a disallowed segment"),
|
|
("a/.git/b", "Path contains a disallowed segment"),
|
|
],
|
|
ids=[
|
|
"non-string", "non-string-int", "empty", "whitespace",
|
|
"nul-char", "cr-char", "lf-char", "backslash",
|
|
"absolute-unix", "windows-drive-prefix", "unc",
|
|
"dot-segment", "dot-segment-internal", "traversal", "traversal-internal",
|
|
"dot-git-segment", "dot-git-segment-internal",
|
|
]
|
|
)
|
|
def test_validate_path_rejects_invalid_inputs(mock_repo_root, invalid_path, match_error):
|
|
with pytest.raises(ValueError, match=re.escape(match_error)):
|
|
_validate_path_inside_repo(invalid_path)
|
|
|
|
|
|
def test_validate_path_rejects_symlink_escape(mock_repo_root):
|
|
"""Tests that a symlink pointing outside the repository root is rejected."""
|
|
outside_file = mock_repo_root.parent / "sensitive_file.txt"
|
|
outside_file.write_text("secret")
|
|
|
|
symlink_in_repo = mock_repo_root / "link_to_secret"
|
|
|
|
try:
|
|
os.symlink(outside_file, symlink_in_repo)
|
|
except (OSError, NotImplementedError) as e:
|
|
pytest.skip(f"Symlink creation skipped: {e}")
|
|
|
|
with pytest.raises(ValueError, match=re.escape("Path resolves outside the repository root.")):
|
|
_validate_path_inside_repo("link_to_secret")
|
|
|
|
@pytest.mark.parametrize("url, expected", [
|
|
("https://git.vauco.no/chris/OSVauco.git", True),
|
|
("git@git.vauco.no:chris/OSVauco.git", True),
|
|
("https://github.com/evil/repo.git", False),
|
|
])
|
|
def test_is_allowed_remote(url, expected):
|
|
assert _is_allowed_remote(url) is expected
|
|
|
|
def test_mask_secrets(mock_repo_root):
|
|
from server import SECRET_MASK_PATTERNS, _mask_secrets
|
|
test_secret = "TEST_SECRET_VALUE_123"
|
|
test_pattern = re.compile(re.escape(test_secret))
|
|
SECRET_MASK_PATTERNS.append(test_pattern)
|
|
try:
|
|
text = f"Token: gitea_abc123def456, Secret: {test_secret}"
|
|
masked = _mask_secrets(text)
|
|
assert "[MASKED_TOKEN]" in masked
|
|
assert "[MASKED_SECRET]" in masked
|
|
assert test_secret not in masked
|
|
finally:
|
|
SECRET_MASK_PATTERNS.remove(test_pattern)
|
|
|
|
@patch("server.subprocess.run")
|
|
def test_run_git_success(mock_run, mock_repo_root):
|
|
mock_run.return_value = MagicMock(stdout="OK", stderr="", returncode=0)
|
|
_run_git(["status", "--short"], timeout=30)
|
|
mock_run.assert_called_once_with(
|
|
["git", "status", "--short"],
|
|
cwd=mock_repo_root,
|
|
capture_output=True, text=True, timeout=30, check=True
|
|
)
|
|
|
|
def test_run_git_invalid_args():
|
|
with pytest.raises(ValueError):
|
|
_run_git(["status\nls"], timeout=30)
|
|
|
|
# --- Unit-tester for Read-only Verktøy ---
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("server._run_git")
|
|
async def test_get_git_status_clean(mock_run_git):
|
|
mock_run_git.return_value = MagicMock(stdout="")
|
|
result = await get_git_status({})
|
|
assert result == {"status": "clean", "details": ""}
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("server._run_git")
|
|
async def test_get_git_branch(mock_run_git):
|
|
mock_run_git.side_effect = [MagicMock(stdout="main\n"), MagicMock(stdout="* main\n")]
|
|
result = await get_git_branch({})
|
|
assert result["current_branch"] == "main"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_repo_files_no_path_uses_root(mock_repo_root):
|
|
"""Tests that calling list_repo_files with no 'path' argument lists the root."""
|
|
(mock_repo_root / "dir1").mkdir()
|
|
(mock_repo_root / "file.txt").touch()
|
|
|
|
result = await list_repo_files({})
|
|
|
|
assert result["path"] == "."
|
|
assert set(result["files"]) == {"dir1", "file.txt"}
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"invalid_path_value",
|
|
[".", "", None],
|
|
ids=["explicit-dot", "empty-string", "none-value"],
|
|
)
|
|
async def test_list_repo_files_rejects_explicit_invalid_paths(
|
|
mock_repo_root,
|
|
invalid_path_value,
|
|
):
|
|
"""Tests that list_repo_files rejects invalid explicit path values."""
|
|
with pytest.raises(ValueError):
|
|
await list_repo_files({"path": invalid_path_value})
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("server._mask_secrets")
|
|
async def test_read_repo_file(mock_mask_secrets, mock_repo_root):
|
|
(mock_repo_root / "my_file.txt").write_text("secret content")
|
|
mock_mask_secrets.return_value = "masked"
|
|
result = await read_repo_file({"path": "my_file.txt"})
|
|
assert result["content"] == "masked"
|
|
mock_mask_secrets.assert_called_once_with("secret content")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_repo_file_not_found(mock_repo_root):
|
|
with pytest.raises(FileNotFoundError):
|
|
await read_repo_file({"path": "non_existent.txt"})
|
|
|
|
# --- Integrasjonstester for Skrive-flyt ---
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_e2e_commit_flow_success(isolated_git_repo, monkeypatch):
|
|
local_repo_path, remote_repo_path = isolated_git_repo
|
|
|
|
remote_url = subprocess.run(
|
|
["git", "remote", "get-url", "origin"],
|
|
cwd=local_repo_path, check=True, capture_output=True, text=True
|
|
).stdout.strip()
|
|
monkeypatch.setattr("server._is_allowed_remote", lambda url: url == remote_url)
|
|
|
|
test_file = local_repo_path / "test.txt"
|
|
test_file.write_text("Hello, Git!")
|
|
head_sha_before = subprocess.run(
|
|
["git", "rev-parse", "HEAD"], cwd=local_repo_path, check=True, capture_output=True, text=True
|
|
).stdout.strip()
|
|
|
|
preview_result = await preview_git_change({"paths": ["test.txt"], "expected_head_sha": head_sha_before})
|
|
assert preview_result["status"] == "ok"
|
|
|
|
branch_name = subprocess.run(["git", "rev-parse", "--abbrev-ref", "HEAD"], cwd=local_repo_path, check=True, capture_output=True, text=True).stdout.strip()
|
|
commit_result = await commit_and_push_files({
|
|
"branch": branch_name, "paths": ["test.txt"],
|
|
"commit_message": "Test commit", "expected_head_sha": head_sha_before
|
|
})
|
|
assert commit_result["status"] == "success"
|
|
|
|
with tempfile.TemporaryDirectory() as clone_dir:
|
|
subprocess.run(["git", "clone", str(remote_repo_path), clone_dir], check=True)
|
|
assert (Path(clone_dir) / "test.txt").read_text() == "Hello, Git!"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_commit_flow_unrelated_changes_fails(isolated_git_repo):
|
|
"""Tester at preview feiler hvis det finnes urelaterte endringer."""
|
|
local_repo_path, _ = isolated_git_repo
|
|
(local_repo_path / "file_a.txt").write_text("a")
|
|
(local_repo_path / "file_b.txt").write_text("b")
|
|
|
|
# Stage one file to make the repo dirty
|
|
subprocess.run(["git", "add", "file_a.txt"], cwd=local_repo_path, check=True)
|
|
|
|
head_sha = subprocess.run(
|
|
["git", "rev-parse", "HEAD"], cwd=local_repo_path, check=True, capture_output=True, text=True
|
|
).stdout.strip()
|
|
|
|
with pytest.raises(RuntimeError, match="Repoet har endringer utenfor de angitte stiene"):
|
|
await preview_git_change({"paths": ["file_a.txt"], "expected_head_sha": head_sha})
|
|
|
|
# --- Tests for MCP Handler Dispatch Policy ---
|
|
|
|
@pytest.mark.asyncio
|
|
class TestMcpHandlerDispatch:
|
|
|
|
async def make_rpc_call(self, tool_name, token, params=None):
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
request_body = {
|
|
"jsonrpc": "2.0",
|
|
"id": "test-id-123",
|
|
"method": "tools/call",
|
|
"params": {"name": tool_name, "arguments": params or {}}
|
|
}
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://testserver") as client:
|
|
response = await client.post("/", json=request_body, headers=headers)
|
|
return response
|
|
|
|
@pytest.mark.parametrize(
|
|
"blocked_tool",
|
|
[
|
|
"commit_and_push_files",
|
|
"send_email",
|
|
"trigger_build",
|
|
"set_billing_budget",
|
|
"create_email_alias",
|
|
"get_emails",
|
|
"preview_git_change",
|
|
]
|
|
)
|
|
async def test_rejects_existing_but_blocked_tools(self, monkeypatch, mcp_secret, blocked_tool):
|
|
assert blocked_tool in TOOLS
|
|
assert blocked_tool not in WEB_AGENT_ALLOWED_TOOLS
|
|
|
|
mock_tool_func = AsyncMock()
|
|
_, description, schema = TOOLS[blocked_tool]
|
|
monkeypatch.setitem(TOOLS, blocked_tool, (mock_tool_func, description, schema))
|
|
|
|
response = await self.make_rpc_call(blocked_tool, mcp_secret)
|
|
|
|
assert response.status_code == 200
|
|
json_res = response.json()
|
|
assert json_res["error"]["code"] == -32601
|
|
assert json_res["error"]["message"] == "Tool not allowed or not found."
|
|
mock_tool_func.assert_not_awaited()
|
|
|
|
async def test_rejects_unknown_tool(self, mcp_secret):
|
|
response = await self.make_rpc_call("non_existent_tool", mcp_secret)
|
|
|
|
assert response.status_code == 200
|
|
json_res = response.json()
|
|
assert json_res["error"]["code"] == -32601
|
|
assert json_res["error"]["message"] == "Tool not allowed or not found."
|
|
|
|
@pytest.mark.parametrize(
|
|
"allowed_tool", sorted(WEB_AGENT_ALLOWED_TOOLS)
|
|
)
|
|
async def test_allows_and_dispatches_web_agent_tools(self, monkeypatch, mcp_secret, allowed_tool):
|
|
mock_tool_func = AsyncMock(return_value={"status": "mocked_success"})
|
|
_, description, schema = TOOLS[allowed_tool]
|
|
monkeypatch.setitem(TOOLS, allowed_tool, (mock_tool_func, description, schema))
|
|
|
|
test_args = {"param": "value"}
|
|
response = await self.make_rpc_call(allowed_tool, mcp_secret, params=test_args)
|
|
|
|
assert response.status_code == 200
|
|
json_res = response.json()
|
|
assert "result" in json_res
|
|
inner_result = json.loads(json_res["result"]["content"][0]["text"])
|
|
assert inner_result["status"] == "mocked_success"
|
|
|
|
mock_tool_func.assert_awaited_once_with(test_args)
|
|
|
|
async def test_auth_is_checked_before_policy(self, monkeypatch, mcp_secret):
|
|
mock_tool_func = AsyncMock()
|
|
allowed_tool = "get_health"
|
|
monkeypatch.setitem(TOOLS, allowed_tool, (mock_tool_func, "", {}))
|
|
|
|
response = await self.make_rpc_call(allowed_tool, "wrong-secret")
|
|
|
|
assert response.status_code == 401
|
|
mock_tool_func.assert_not_awaited()
|