COVERAGE BEFORE
===============
tasks.py 2,117 lines ZERO tests
gitea_client.py 539 lines ZERO tests (in this repo)
Total: 2,656 lines of orchestration with no safety net
COVERAGE AFTER
==============
test_tasks_core.py — 63 tests across 12 test classes:
TestExtractFirstJsonObject (10) — JSON parsing from noisy LLM output
Every @huey.task depends on this. Tested: clean JSON, markdown
fences, prose-wrapped, nested, malformed, arrays, unicode, empty
TestParseJsonOutput (4) — stdout/stderr fallback chain
TestNormalizeCandidateEntry (12) — knowledge graph data cleaning
Confidence clamping, status validation, deduplication, truncation
TestNormalizeTrainingExamples (5) — autolora training data prep
Fallback when empty, alternative field names, empty prompt/response
TestNormalizeRubricScores (3) — eval score clamping
TestReadJson (4) — defensive file reads
Missing files, corrupt JSON, deep-copy of defaults
TestWriteJson (3) — atomic writes with sorted keys
TestJsonlIO (9) — JSONL read/write/append/count
Missing files, blank lines, append vs overwrite
TestWriteText (3) — trailing newline normalization
TestPathUtilities (4) — newest/latest path resolution
TestFormatting (6) — batch IDs, profile summaries,
tweet prompts, checkpoint defaults
test_gitea_client_core.py — 22 tests across 9 test classes:
TestUserFromDict (3) — all from_dict() deserialization
TestLabelFromDict (1)
TestIssueFromDict (4) — null assignees/labels (THE bug)
TestCommentFromDict (2) — null body handling
TestPullRequestFromDict (3) — null head/base/merged
TestPRFileFromDict (1)
TestGiteaError (2) — error formatting
TestClientHelpers (1) — _repo_path formatting
TestFindUnassigned (3) — label/title/assignee filtering
TestFindAgentIssues (2) — case-insensitive matching
WHY THESE TESTS MATTER
======================
A bug in extract_first_json_object() corrupts every @huey.task
that processes LLM output — which is all of them. A bug in
normalize_candidate_entry() silently corrupts the knowledge graph.
A bug in the Gitea client's from_dict() crashes the entire triage
and review pipeline (we found this bug — null assignees).
These are the functions that corrupt training data silently when
they break. No one notices until the next autolora run produces
a worse model.
FULL SUITE: 108/108 pass, zero regressions.
Signed-off-by: gemini <gemini@hermes.local>
319 lines
11 KiB
Python
319 lines
11 KiB
Python
"""Tests for gitea_client.py — the typed, sovereign API client.
|
|
|
|
gitea_client.py is 539 lines with zero tests in this repo (there are
|
|
tests in hermes-agent, but not here where it's actually used).
|
|
|
|
These tests cover:
|
|
- All 6 dataclass from_dict() constructors (User, Label, Issue, etc.)
|
|
- Defensive handling of missing/null fields from Gitea API
|
|
- find_unassigned_issues() filtering logic
|
|
- find_agent_issues() case-insensitive matching
|
|
- GiteaError formatting
|
|
- _repo_path() formatting
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
# Import gitea_client directly via importlib to avoid any sys.modules mocking
|
|
# from test_tasks_core which stubs gitea_client as a MagicMock.
|
|
REPO_ROOT = Path(__file__).parent.parent
|
|
_spec = importlib.util.spec_from_file_location(
|
|
"gitea_client_real",
|
|
REPO_ROOT / "gitea_client.py",
|
|
)
|
|
_gc = importlib.util.module_from_spec(_spec)
|
|
sys.modules["gitea_client_real"] = _gc
|
|
_spec.loader.exec_module(_gc)
|
|
|
|
User = _gc.User
|
|
Label = _gc.Label
|
|
Issue = _gc.Issue
|
|
Comment = _gc.Comment
|
|
PullRequest = _gc.PullRequest
|
|
PRFile = _gc.PRFile
|
|
GiteaError = _gc.GiteaError
|
|
GiteaClient = _gc.GiteaClient
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# DATACLASS DESERIALIZATION
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
class TestUserFromDict:
|
|
def test_full_user(self):
|
|
u = User.from_dict({"id": 1, "login": "timmy", "full_name": "Timmy", "email": "t@t.com"})
|
|
assert u.id == 1
|
|
assert u.login == "timmy"
|
|
assert u.full_name == "Timmy"
|
|
assert u.email == "t@t.com"
|
|
|
|
def test_minimal_user(self):
|
|
"""Missing fields default to empty."""
|
|
u = User.from_dict({})
|
|
assert u.id == 0
|
|
assert u.login == ""
|
|
|
|
def test_extra_fields_ignored(self):
|
|
"""Unknown fields from Gitea are silently ignored."""
|
|
u = User.from_dict({"id": 1, "login": "x", "avatar_url": "http://..."})
|
|
assert u.login == "x"
|
|
|
|
|
|
class TestLabelFromDict:
|
|
def test_label(self):
|
|
lb = Label.from_dict({"id": 5, "name": "bug", "color": "#ff0000"})
|
|
assert lb.id == 5
|
|
assert lb.name == "bug"
|
|
assert lb.color == "#ff0000"
|
|
|
|
|
|
class TestIssueFromDict:
|
|
def test_full_issue(self):
|
|
issue = Issue.from_dict({
|
|
"number": 42,
|
|
"title": "Fix the bug",
|
|
"body": "Please fix it",
|
|
"state": "open",
|
|
"user": {"id": 1, "login": "reporter"},
|
|
"assignees": [{"id": 2, "login": "dev"}],
|
|
"labels": [{"id": 3, "name": "bug"}],
|
|
"comments": 5,
|
|
})
|
|
assert issue.number == 42
|
|
assert issue.user.login == "reporter"
|
|
assert len(issue.assignees) == 1
|
|
assert issue.assignees[0].login == "dev"
|
|
assert len(issue.labels) == 1
|
|
assert issue.comments == 5
|
|
|
|
def test_null_assignees_handled(self):
|
|
"""Gitea returns null for assignees sometimes — the exact bug
|
|
that crashed find_unassigned_issues() before the defensive fix."""
|
|
issue = Issue.from_dict({
|
|
"number": 1,
|
|
"title": "test",
|
|
"body": None,
|
|
"state": "open",
|
|
"user": {"id": 1, "login": "x"},
|
|
"assignees": None,
|
|
})
|
|
assert issue.assignees == []
|
|
assert issue.body == ""
|
|
|
|
def test_null_labels_handled(self):
|
|
"""Labels can also be null."""
|
|
issue = Issue.from_dict({
|
|
"number": 1,
|
|
"title": "test",
|
|
"state": "open",
|
|
"user": {},
|
|
"labels": None,
|
|
})
|
|
assert issue.labels == []
|
|
|
|
def test_missing_user_defaults(self):
|
|
"""Issue with no user field doesn't crash."""
|
|
issue = Issue.from_dict({"number": 1, "title": "t", "state": "open"})
|
|
assert issue.user.login == ""
|
|
|
|
|
|
class TestCommentFromDict:
|
|
def test_comment(self):
|
|
c = Comment.from_dict({
|
|
"id": 10,
|
|
"body": "LGTM",
|
|
"user": {"id": 1, "login": "reviewer"},
|
|
})
|
|
assert c.id == 10
|
|
assert c.body == "LGTM"
|
|
assert c.user.login == "reviewer"
|
|
|
|
def test_null_body(self):
|
|
c = Comment.from_dict({"id": 1, "body": None, "user": {}})
|
|
assert c.body == ""
|
|
|
|
|
|
class TestPullRequestFromDict:
|
|
def test_full_pr(self):
|
|
pr = PullRequest.from_dict({
|
|
"number": 99,
|
|
"title": "Add feature",
|
|
"body": "Description here",
|
|
"state": "open",
|
|
"user": {"id": 1, "login": "dev"},
|
|
"head": {"ref": "feature-branch"},
|
|
"base": {"ref": "main"},
|
|
"mergeable": True,
|
|
"merged": False,
|
|
"changed_files": 3,
|
|
})
|
|
assert pr.number == 99
|
|
assert pr.head_branch == "feature-branch"
|
|
assert pr.base_branch == "main"
|
|
assert pr.mergeable is True
|
|
|
|
def test_null_head_base(self):
|
|
"""Handles null head/base objects."""
|
|
pr = PullRequest.from_dict({
|
|
"number": 1, "title": "t", "state": "open",
|
|
"user": {}, "head": None, "base": None,
|
|
})
|
|
assert pr.head_branch == ""
|
|
assert pr.base_branch == ""
|
|
|
|
def test_null_merged(self):
|
|
"""merged can be null from Gitea."""
|
|
pr = PullRequest.from_dict({
|
|
"number": 1, "title": "t", "state": "open",
|
|
"user": {}, "merged": None,
|
|
})
|
|
assert pr.merged is False
|
|
|
|
|
|
class TestPRFileFromDict:
|
|
def test_pr_file(self):
|
|
f = PRFile.from_dict({
|
|
"filename": "src/main.py",
|
|
"status": "modified",
|
|
"additions": 10,
|
|
"deletions": 3,
|
|
})
|
|
assert f.filename == "src/main.py"
|
|
assert f.status == "modified"
|
|
assert f.additions == 10
|
|
assert f.deletions == 3
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# ERROR HANDLING
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
class TestGiteaError:
|
|
def test_error_formatting(self):
|
|
err = GiteaError(404, "not found", "http://example.com/api/v1/repos/x")
|
|
assert "404" in str(err)
|
|
assert "not found" in str(err)
|
|
|
|
def test_error_attributes(self):
|
|
err = GiteaError(500, "internal")
|
|
assert err.status == 500
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# CLIENT HELPER METHODS
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
class TestClientHelpers:
|
|
def test_repo_path(self):
|
|
"""_repo_path converts owner/name to API path."""
|
|
client = GiteaClient.__new__(GiteaClient)
|
|
assert client._repo_path("Timmy_Foundation/the-nexus") == "/repos/Timmy_Foundation/the-nexus"
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# FILTERING LOGIC — find_unassigned_issues, find_agent_issues
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
class TestFindUnassigned:
|
|
"""Tests for find_unassigned_issues() filtering logic.
|
|
|
|
These tests use pre-constructed Issue objects to test the filtering
|
|
without making any API calls.
|
|
"""
|
|
|
|
def _make_issue(self, number, assignees=None, labels=None, title="test"):
|
|
return Issue(
|
|
number=number, title=title, body="", state="open",
|
|
user=User(id=0, login=""),
|
|
assignees=[User(id=0, login=a) for a in (assignees or [])],
|
|
labels=[Label(id=0, name=lb) for lb in (labels or [])],
|
|
)
|
|
|
|
def test_filters_assigned_issues(self):
|
|
"""Issues with assignees are excluded."""
|
|
from unittest.mock import patch
|
|
|
|
issues = [
|
|
self._make_issue(1, assignees=["dev"]),
|
|
self._make_issue(2), # unassigned
|
|
]
|
|
|
|
client = GiteaClient.__new__(GiteaClient)
|
|
with patch.object(client, "list_issues", return_value=issues):
|
|
result = client.find_unassigned_issues("repo")
|
|
|
|
assert len(result) == 1
|
|
assert result[0].number == 2
|
|
|
|
def test_excludes_by_label(self):
|
|
"""Issues with excluded labels are filtered."""
|
|
from unittest.mock import patch
|
|
|
|
issues = [
|
|
self._make_issue(1, labels=["wontfix"]),
|
|
self._make_issue(2, labels=["bug"]),
|
|
]
|
|
|
|
client = GiteaClient.__new__(GiteaClient)
|
|
with patch.object(client, "list_issues", return_value=issues):
|
|
result = client.find_unassigned_issues("repo", exclude_labels=["wontfix"])
|
|
|
|
assert len(result) == 1
|
|
assert result[0].number == 2
|
|
|
|
def test_excludes_by_title_pattern(self):
|
|
"""Issues matching title patterns are filtered."""
|
|
from unittest.mock import patch
|
|
|
|
issues = [
|
|
self._make_issue(1, title="[PHASE] Research AI"),
|
|
self._make_issue(2, title="Fix login bug"),
|
|
]
|
|
|
|
client = GiteaClient.__new__(GiteaClient)
|
|
with patch.object(client, "list_issues", return_value=issues):
|
|
result = client.find_unassigned_issues(
|
|
"repo", exclude_title_patterns=["[PHASE]"]
|
|
)
|
|
|
|
assert len(result) == 1
|
|
assert result[0].number == 2
|
|
|
|
|
|
class TestFindAgentIssues:
|
|
"""Tests for find_agent_issues() case-insensitive matching."""
|
|
|
|
def test_case_insensitive_match(self):
|
|
from unittest.mock import patch
|
|
|
|
issues = [
|
|
Issue(number=1, title="t", body="", state="open",
|
|
user=User(0, ""), assignees=[User(0, "Timmy")], labels=[]),
|
|
]
|
|
|
|
client = GiteaClient.__new__(GiteaClient)
|
|
with patch.object(client, "list_issues", return_value=issues):
|
|
result = client.find_agent_issues("repo", "timmy")
|
|
|
|
assert len(result) == 1
|
|
|
|
def test_no_match_for_different_agent(self):
|
|
from unittest.mock import patch
|
|
|
|
issues = [
|
|
Issue(number=1, title="t", body="", state="open",
|
|
user=User(0, ""), assignees=[User(0, "Timmy")], labels=[]),
|
|
]
|
|
|
|
client = GiteaClient.__new__(GiteaClient)
|
|
with patch.object(client, "list_issues", return_value=issues):
|
|
result = client.find_agent_issues("repo", "claude")
|
|
|
|
assert len(result) == 0
|