135 lines
5.2 KiB
Python
135 lines
5.2 KiB
Python
"""自动代码评审模块单测：cloud/review.py。
|
|
|
|
不跑真实模型、不联网：纯 sqlite 内存库 + 协议目标构造。
|
|
覆盖四种 ReviewTarget、非法组合、登记/队列/取单条/写回
|
|
状态机（findings 写回默认 completed、显式 failed 可覆盖）。
|
|
"""
|
|
|
|
import sqlite3
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from cloud.review import build_review_target
|
|
from cloud.review import create_review
|
|
from cloud.review import ensure_reviews_table
|
|
from cloud.review import get_review
|
|
from cloud.review import list_project_reviews
|
|
from cloud.review import list_reviews
|
|
from cloud.review import update_review
|
|
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
|
from cloud.review import build_review_target
|
|
from cloud.review import create_review
|
|
from cloud.review import ensure_reviews_table
|
|
from cloud.review import get_review
|
|
from cloud.review import list_project_reviews
|
|
from cloud.review import list_reviews
|
|
from cloud.review import update_review
|
|
|
|
|
|
class ReviewTargetTestCase(unittest.TestCase):
|
|
def test_four_targets(self):
|
|
self.assertEqual(
|
|
build_review_target("uncommittedChanges"),
|
|
{"type": "uncommittedChanges"})
|
|
self.assertEqual(
|
|
build_review_target("baseBranch", branch="main"),
|
|
{"type": "baseBranch", "branch": "main"})
|
|
self.assertEqual(
|
|
build_review_target("commit", sha="abc123"),
|
|
{"type": "commit", "sha": "abc123"})
|
|
self.assertEqual(
|
|
build_review_target("custom", instructions="look closely"),
|
|
{"type": "custom", "instructions": "look closely"})
|
|
|
|
def test_target_validation(self):
|
|
with self.assertRaises(ValueError):
|
|
build_review_target("baseBranch")
|
|
with self.assertRaises(ValueError):
|
|
build_review_target("baseBranch", branch=" ")
|
|
with self.assertRaises(ValueError):
|
|
build_review_target("commit")
|
|
with self.assertRaises(ValueError):
|
|
build_review_target("custom")
|
|
with self.assertRaises(ValueError):
|
|
build_review_target("nope")
|
|
|
|
def test_whitespace_trimmed(self):
|
|
self.assertEqual(
|
|
build_review_target("baseBranch", branch=" main "),
|
|
{"type": "baseBranch", "branch": "main"})
|
|
|
|
|
|
class ReviewStoreTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
self.conn = sqlite3.connect(":memory:")
|
|
ensure_reviews_table(self.conn)
|
|
|
|
def tearDown(self):
|
|
self.conn.close()
|
|
|
|
def test_create_and_list(self):
|
|
first = create_review(
|
|
self.conn, "task-a", "proj-a", "uncommittedChanges",
|
|
"owner-1")
|
|
self.assertEqual(first["status"], "requested")
|
|
self.assertIsNone(first["findings"])
|
|
self.assertEqual(first["task_id"], "task-a")
|
|
second = create_review(
|
|
self.conn, "task-a", "proj-a", "baseBranch", "owner-1",
|
|
branch="main")
|
|
self.assertEqual(second["target_type"], "baseBranch")
|
|
self.assertEqual(second["branch"], "main")
|
|
queue = list_reviews(self.conn, "task-a")
|
|
self.assertEqual(
|
|
[r["id"] for r in queue], [first["id"], second["id"]])
|
|
self.assertEqual(list_reviews(self.conn, "task-other"), [])
|
|
proj = list_project_reviews(self.conn, "proj-a")
|
|
self.assertEqual(len(proj), 2)
|
|
|
|
def test_create_invalid_target_writes_nothing(self):
|
|
with self.assertRaises(ValueError):
|
|
create_review(
|
|
self.conn, "task-a", "proj-a", "commit", "owner-1")
|
|
self.assertEqual(list_reviews(self.conn, "task-a"), [])
|
|
|
|
def test_get_missing_returns_none(self):
|
|
self.assertIsNone(get_review(self.conn, "no-such-id"))
|
|
|
|
def test_findings_writeback_defaults_completed(self):
|
|
record = create_review(
|
|
self.conn, "task-a", "proj-a", "custom", "owner-1",
|
|
instructions="check it")
|
|
updated = update_review(
|
|
self.conn, record["id"], findings="two issues found")
|
|
self.assertEqual(updated["status"], "completed")
|
|
self.assertEqual(updated["findings"], "two issues found")
|
|
self.assertEqual(
|
|
get_review(self.conn, record["id"])["status"], "completed")
|
|
|
|
def test_explicit_failed_status_kept(self):
|
|
record = create_review(
|
|
self.conn, "task-a", "proj-a", "uncommittedChanges",
|
|
"owner-1")
|
|
updated = update_review(
|
|
self.conn, record["id"], status="failed",
|
|
findings="reviewer crashed")
|
|
self.assertEqual(updated["status"], "failed")
|
|
|
|
def test_update_validation(self):
|
|
record = create_review(
|
|
self.conn, "task-a", "proj-a", "uncommittedChanges",
|
|
"owner-1")
|
|
with self.assertRaises(ValueError):
|
|
update_review(self.conn, record["id"], status="bogus")
|
|
with self.assertRaises(ValueError):
|
|
update_review(self.conn, record["id"], findings=123)
|
|
self.assertIsNone(
|
|
update_review(self.conn, "no-such-id", findings="x"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|