fix: add update_gitea_avatar capability (#368)
Co-authored-by: Kimi Agent <kimi@timmy.local> Co-committed-by: Kimi Agent <kimi@timmy.local>
This commit was merged in pull request #368.
This commit is contained in:
@@ -29,6 +29,8 @@ from contextlib import closing
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
|
||||
from config import settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -268,6 +270,140 @@ async def create_gitea_issue_via_mcp(title: str, body: str = "", labels: str = "
|
||||
return f"Failed to create issue via MCP: {exc}"
|
||||
|
||||
|
||||
def _generate_avatar_image() -> bytes:
|
||||
"""Generate a Timmy-themed avatar image using Pillow.
|
||||
|
||||
Creates a 512x512 wizard-themed avatar with emerald/purple/gold palette.
|
||||
Returns raw PNG bytes. Falls back to a minimal solid-color image if
|
||||
Pillow drawing primitives fail.
|
||||
"""
|
||||
from PIL import Image, ImageDraw
|
||||
|
||||
size = 512
|
||||
img = Image.new("RGB", (size, size), (15, 25, 20))
|
||||
draw = ImageDraw.Draw(img)
|
||||
|
||||
# Background gradient effect — concentric circles
|
||||
for i in range(size // 2, 0, -4):
|
||||
g = int(25 + (i / (size // 2)) * 30)
|
||||
draw.ellipse(
|
||||
[size // 2 - i, size // 2 - i, size // 2 + i, size // 2 + i],
|
||||
fill=(10, g, 20),
|
||||
)
|
||||
|
||||
# Wizard hat (triangle)
|
||||
hat_color = (100, 50, 160) # purple
|
||||
draw.polygon(
|
||||
[(256, 40), (160, 220), (352, 220)],
|
||||
fill=hat_color,
|
||||
outline=(180, 130, 255),
|
||||
)
|
||||
|
||||
# Hat brim
|
||||
draw.ellipse([140, 200, 372, 250], fill=hat_color, outline=(180, 130, 255))
|
||||
|
||||
# Face circle
|
||||
draw.ellipse([190, 220, 322, 370], fill=(60, 180, 100), outline=(80, 220, 120))
|
||||
|
||||
# Eyes
|
||||
draw.ellipse([220, 275, 248, 310], fill=(255, 255, 255))
|
||||
draw.ellipse([264, 275, 292, 310], fill=(255, 255, 255))
|
||||
draw.ellipse([228, 285, 242, 300], fill=(30, 30, 60))
|
||||
draw.ellipse([272, 285, 286, 300], fill=(30, 30, 60))
|
||||
|
||||
# Smile
|
||||
draw.arc([225, 300, 287, 355], start=10, end=170, fill=(30, 30, 60), width=3)
|
||||
|
||||
# Stars around the hat
|
||||
gold = (220, 190, 50)
|
||||
star_positions = [(120, 100), (380, 120), (100, 300), (400, 280), (256, 10)]
|
||||
for sx, sy in star_positions:
|
||||
r = 8
|
||||
draw.polygon(
|
||||
[
|
||||
(sx, sy - r),
|
||||
(sx + r // 3, sy - r // 3),
|
||||
(sx + r, sy),
|
||||
(sx + r // 3, sy + r // 3),
|
||||
(sx, sy + r),
|
||||
(sx - r // 3, sy + r // 3),
|
||||
(sx - r, sy),
|
||||
(sx - r // 3, sy - r // 3),
|
||||
],
|
||||
fill=gold,
|
||||
)
|
||||
|
||||
# "T" monogram on the hat
|
||||
draw.text((243, 100), "T", fill=gold)
|
||||
|
||||
# Robe / body
|
||||
draw.polygon(
|
||||
[(180, 370), (140, 500), (372, 500), (332, 370)],
|
||||
fill=(40, 100, 70),
|
||||
outline=(60, 160, 100),
|
||||
)
|
||||
|
||||
import io
|
||||
|
||||
buf = io.BytesIO()
|
||||
img.save(buf, format="PNG")
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
async def update_gitea_avatar() -> str:
|
||||
"""Generate and upload a unique avatar to Timmy's Gitea profile.
|
||||
|
||||
Creates a wizard-themed avatar image using Pillow drawing primitives,
|
||||
base64-encodes it, and POSTs to the Gitea user avatar API endpoint.
|
||||
|
||||
Returns:
|
||||
Success or failure message string.
|
||||
"""
|
||||
if not settings.gitea_enabled or not settings.gitea_token:
|
||||
return "Gitea integration is not configured (no token or disabled)."
|
||||
|
||||
try:
|
||||
from PIL import Image # noqa: F401 — availability check
|
||||
except ImportError:
|
||||
return "Pillow is not installed — cannot generate avatar image."
|
||||
|
||||
try:
|
||||
import base64
|
||||
|
||||
# Step 1: Generate the avatar image
|
||||
png_bytes = _generate_avatar_image()
|
||||
logger.info("Generated avatar image (%d bytes)", len(png_bytes))
|
||||
|
||||
# Step 2: Base64-encode (raw, no data URI prefix)
|
||||
b64_image = base64.b64encode(png_bytes).decode("ascii")
|
||||
|
||||
# Step 3: POST to Gitea
|
||||
async with httpx.AsyncClient(timeout=15) as client:
|
||||
resp = await client.post(
|
||||
f"{settings.gitea_url}/api/v1/user/avatar",
|
||||
headers={
|
||||
"Authorization": f"token {settings.gitea_token}",
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
json={"image": b64_image},
|
||||
)
|
||||
|
||||
# Gitea returns empty body on success (204 or 200)
|
||||
if resp.status_code in (200, 204):
|
||||
logger.info("Gitea avatar updated successfully")
|
||||
return "Avatar updated successfully on Gitea."
|
||||
|
||||
logger.warning("Gitea avatar update failed: %s %s", resp.status_code, resp.text[:200])
|
||||
return f"Gitea avatar update failed (HTTP {resp.status_code}): {resp.text[:200]}"
|
||||
|
||||
except (httpx.ConnectError, httpx.ReadError, ConnectionError) as exc:
|
||||
logger.warning("Gitea connection failed during avatar update: %s", exc)
|
||||
return f"Could not connect to Gitea: {exc}"
|
||||
except Exception as exc:
|
||||
logger.error("Avatar update failed: %s", exc)
|
||||
return f"Avatar update failed: {exc}"
|
||||
|
||||
|
||||
async def close_mcp_sessions() -> None:
|
||||
"""Close any open MCP sessions. Called during app shutdown."""
|
||||
global _issue_session
|
||||
|
||||
@@ -586,6 +586,13 @@ def _register_introspection_tools(toolkit: Toolkit) -> None:
|
||||
logger.warning("Tool execution failed (Introspection tools registration): %s", exc)
|
||||
logger.debug("Introspection tools not available")
|
||||
|
||||
try:
|
||||
from timmy.mcp_tools import update_gitea_avatar
|
||||
|
||||
toolkit.register(update_gitea_avatar, name="update_gitea_avatar")
|
||||
except (ImportError, AttributeError) as exc:
|
||||
logger.debug("update_gitea_avatar tool not available: %s", exc)
|
||||
|
||||
try:
|
||||
from timmy.session_logger import session_history
|
||||
|
||||
@@ -867,6 +874,11 @@ def _introspection_tool_catalog() -> dict:
|
||||
"description": "Query Timmy's own thought history for past reflections and insights",
|
||||
"available_in": ["orchestrator"],
|
||||
},
|
||||
"update_gitea_avatar": {
|
||||
"name": "Update Gitea Avatar",
|
||||
"description": "Generate and upload a wizard-themed avatar to Timmy's Gitea profile",
|
||||
"available_in": ["orchestrator"],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -6,11 +6,13 @@ import pytest
|
||||
|
||||
from timmy.mcp_tools import (
|
||||
_bridge_to_work_order,
|
||||
_generate_avatar_image,
|
||||
_parse_command,
|
||||
close_mcp_sessions,
|
||||
create_filesystem_mcp_tools,
|
||||
create_gitea_issue_via_mcp,
|
||||
create_gitea_mcp_tools,
|
||||
update_gitea_avatar,
|
||||
)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -302,3 +304,122 @@ def test_mcp_tools_classified_in_safety():
|
||||
assert not requires_confirmation("issue_write")
|
||||
assert not requires_confirmation("list_directory")
|
||||
assert requires_confirmation("write_file")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# update_gitea_avatar
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_generate_avatar_image_returns_png():
|
||||
"""_generate_avatar_image returns valid PNG bytes."""
|
||||
pytest.importorskip("PIL")
|
||||
data = _generate_avatar_image()
|
||||
assert isinstance(data, bytes)
|
||||
assert len(data) > 0
|
||||
# PNG magic bytes
|
||||
assert data[:4] == b"\x89PNG"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_avatar_not_configured():
|
||||
"""update_gitea_avatar returns message when Gitea is disabled."""
|
||||
with patch("timmy.mcp_tools.settings") as mock_settings:
|
||||
mock_settings.gitea_enabled = False
|
||||
mock_settings.gitea_token = ""
|
||||
result = await update_gitea_avatar()
|
||||
assert "not configured" in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_avatar_success():
|
||||
"""update_gitea_avatar uploads avatar and returns success."""
|
||||
import sys
|
||||
|
||||
import timmy.mcp_tools as mcp_mod
|
||||
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 204
|
||||
mock_response.text = ""
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.post = AsyncMock(return_value=mock_response)
|
||||
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
||||
mock_client.__aexit__ = AsyncMock(return_value=False)
|
||||
|
||||
# Ensure PIL import check passes even if Pillow isn't installed
|
||||
pil_stub = MagicMock()
|
||||
with (
|
||||
patch("timmy.mcp_tools.settings") as mock_settings,
|
||||
patch.object(mcp_mod.httpx, "AsyncClient", return_value=mock_client),
|
||||
patch("timmy.mcp_tools._generate_avatar_image", return_value=b"\x89PNG fake"),
|
||||
patch.dict(sys.modules, {"PIL": pil_stub, "PIL.Image": pil_stub}),
|
||||
):
|
||||
mock_settings.gitea_enabled = True
|
||||
mock_settings.gitea_token = "tok123"
|
||||
mock_settings.gitea_url = "http://localhost:3000"
|
||||
result = await update_gitea_avatar()
|
||||
|
||||
assert "successfully" in result
|
||||
mock_client.post.assert_awaited_once()
|
||||
call_args = mock_client.post.call_args
|
||||
assert "/api/v1/user/avatar" in call_args[0][0]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_avatar_api_failure():
|
||||
"""update_gitea_avatar handles HTTP error gracefully."""
|
||||
import sys
|
||||
|
||||
import timmy.mcp_tools as mcp_mod
|
||||
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 400
|
||||
mock_response.text = "bad request"
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.post = AsyncMock(return_value=mock_response)
|
||||
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
||||
mock_client.__aexit__ = AsyncMock(return_value=False)
|
||||
|
||||
pil_stub = MagicMock()
|
||||
with (
|
||||
patch("timmy.mcp_tools.settings") as mock_settings,
|
||||
patch.object(mcp_mod.httpx, "AsyncClient", return_value=mock_client),
|
||||
patch("timmy.mcp_tools._generate_avatar_image", return_value=b"\x89PNG fake"),
|
||||
patch.dict(sys.modules, {"PIL": pil_stub, "PIL.Image": pil_stub}),
|
||||
):
|
||||
mock_settings.gitea_enabled = True
|
||||
mock_settings.gitea_token = "tok123"
|
||||
mock_settings.gitea_url = "http://localhost:3000"
|
||||
result = await update_gitea_avatar()
|
||||
|
||||
assert "failed" in result.lower()
|
||||
assert "400" in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_avatar_connection_error():
|
||||
"""update_gitea_avatar handles connection errors gracefully."""
|
||||
import sys
|
||||
|
||||
import timmy.mcp_tools as mcp_mod
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_client.post = AsyncMock(side_effect=mcp_mod.httpx.ConnectError("refused"))
|
||||
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
||||
mock_client.__aexit__ = AsyncMock(return_value=False)
|
||||
|
||||
pil_stub = MagicMock()
|
||||
with (
|
||||
patch("timmy.mcp_tools.settings") as mock_settings,
|
||||
patch.object(mcp_mod.httpx, "AsyncClient", return_value=mock_client),
|
||||
patch("timmy.mcp_tools._generate_avatar_image", return_value=b"\x89PNG fake"),
|
||||
patch.dict(sys.modules, {"PIL": pil_stub, "PIL.Image": pil_stub}),
|
||||
):
|
||||
mock_settings.gitea_enabled = True
|
||||
mock_settings.gitea_token = "tok123"
|
||||
mock_settings.gitea_url = "http://localhost:3000"
|
||||
result = await update_gitea_avatar()
|
||||
|
||||
assert "connect" in result.lower()
|
||||
|
||||
Reference in New Issue
Block a user