test_task_create.py
4.99 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
"""U-009 tests: c_task_create. REQ-01's boundary and failure cases are fully
covered here; the happy case (actually triggering planner/coder/validator)
is not achievable until U-010 -- see task_create.py's module docstring.
Run: python3 -m unittest src.test_task_create -v
"""
import sqlite3
import unittest
from src import project_scaffold, schema, task_create
class TaskCreateTestBase(unittest.TestCase):
def setUp(self):
self.conn = schema.connect(":memory:")
schema.create_schema(self.conn)
self.project_id = self._make_project()
def tearDown(self):
self.conn.close()
def _make_project(self, path="/tmp/graphlemon-u009-project"):
self.conn.execute(
"INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
"VALUES ('proj-1', 'P', ?, 1, 'active')",
(path,),
)
self.conn.commit()
return "proj-1"
class CreateTaskTest(TaskCreateTestBase):
def test_task_lands_queued_with_expected_fields(self):
result = task_create.create_task(self.conn, self.project_id, "实现贪吃蛇游戏")
self.assertEqual(result["status"], "queued")
row = self.conn.execute(
"SELECT project_id, goal_text, task_type, created_via, status, loop_round, assigned_org_node "
"FROM task WHERE task_id = ?",
(result["task_id"],),
).fetchone()
self.assertEqual(row, (self.project_id, "实现贪吃蛇游戏", "goal", "api", "queued", 0, "planner"))
def test_writes_task_status_changed_event_in_same_transaction(self):
result = task_create.create_task(self.conn, self.project_id, "goal text")
event = self.conn.execute(
"SELECT project_id, task_id, event_type FROM execution_event WHERE task_id = ?",
(result["task_id"],),
).fetchone()
self.assertEqual(event, (self.project_id, result["task_id"], "task_status_changed"))
def test_created_via_human_console_honored(self):
result = task_create.create_task(self.conn, self.project_id, "goal text", created_via="human_console")
created_via = self.conn.execute(
"SELECT created_via FROM task WHERE task_id = ?", (result["task_id"],)
).fetchone()[0]
self.assertEqual(created_via, "human_console")
def test_each_call_creates_a_new_task_not_idempotent(self):
first = task_create.create_task(self.conn, self.project_id, "goal text")
second = task_create.create_task(self.conn, self.project_id, "goal text")
self.assertNotEqual(first["task_id"], second["task_id"])
count = self.conn.execute("SELECT COUNT(*) FROM task").fetchone()[0]
self.assertEqual(count, 2)
class InvalidParamTest(TaskCreateTestBase):
def test_empty_goal_text_rejected(self):
with self.assertRaises(task_create.InvalidParamError):
task_create.create_task(self.conn, self.project_id, "")
self.assertEqual(self.conn.execute("SELECT COUNT(*) FROM task").fetchone()[0], 0)
def test_nonexistent_project_id_rejected(self):
with self.assertRaises(task_create.InvalidParamError):
task_create.create_task(self.conn, "no-such-project", "goal text")
self.assertEqual(self.conn.execute("SELECT COUNT(*) FROM task").fetchone()[0], 0)
def test_non_goal_task_type_rejected(self):
with self.assertRaises(task_create.InvalidParamError):
task_create.create_task(self.conn, self.project_id, "goal text", task_type="bug_fix")
self.assertEqual(self.conn.execute("SELECT COUNT(*) FROM task").fetchone()[0], 0)
class BoundaryConcurrencyLimitTest(TaskCreateTestBase):
"""REQ-01 boundary: max_concurrent_goals=1 doesn't block task *creation*
-- it only matters for whoever later decides to trigger org nodes
(03-C §3.1 side_effects: "任务落 queued 但不触发调度")."""
def test_second_task_still_created_as_queued_under_concurrency_limit_of_one(self):
self.conn.execute(
"INSERT INTO org_concurrency_config (config_id, max_concurrent_goals, updated_by) "
"VALUES ('default', 1, 'test')"
)
self.conn.commit()
first = task_create.create_task(self.conn, self.project_id, "goal one")
second = task_create.create_task(self.conn, self.project_id, "goal two")
statuses = {
row[0]
for row in self.conn.execute(
"SELECT status FROM task WHERE task_id IN (?, ?)", (first["task_id"], second["task_id"])
).fetchall()
}
self.assertEqual(statuses, {"queued"})
class TransactionAtomicityTest(TaskCreateTestBase):
def test_failure_between_the_two_inserts_rolls_back_the_task_row(self):
self.conn.execute("DROP TABLE execution_event")
with self.assertRaises(sqlite3.OperationalError):
task_create.create_task(self.conn, self.project_id, "goal text")
self.assertEqual(self.conn.execute("SELECT COUNT(*) FROM task").fetchone()[0], 0)
if __name__ == "__main__":
unittest.main()