test_jobs.py
3.25 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
"""Hermes Jobs - DB layer tests (TDD: RED phase first)"""
import sys, os
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import sqlite3
from pathlib import Path
# Use a test DB so we don't pollute production
TEST_DB = "/tmp/test_hermes_jobs.db"
def setup():
"""Fresh test DB each run"""
if Path(TEST_DB).exists():
Path(TEST_DB).unlink()
# Override get_conn to use test DB
import local_db
local_db._TEST_DB = TEST_DB
local_db.init_db()
def teardown():
if Path(TEST_DB).exists():
Path(TEST_DB).unlink()
# ============================================================
# RED: These should fail until we make the CRUD work properly
# ============================================================
def test_job_add():
"""INSERT a job and verify it's stored correctly"""
import local_db
task_id = local_db.job_add("test_task_1", "article_summary", "Test Article Title", "/tmp/test.log")
assert task_id > 0, f"Expected positive ID, got {task_id}"
job = local_db.job_get("test_task_1")
assert job is not None, "job_get returned None"
assert job["task_id"] == "test_task_1"
assert job["job_type"] == "article_summary"
assert job["job_name"] == "Test Article Title"
assert job["status"] == "running"
assert job["started_at"] is not None
assert job["completed_at"] is None
print(" PASS test_job_add")
def test_job_update_status():
"""Update status to completed sets completed_at"""
import local_db
local_db.job_add("test_task_2", "daily_report", "Group Report")
local_db.job_update_status("test_task_2", "completed")
job = local_db.job_get("test_task_2")
assert job["status"] == "completed"
assert job["completed_at"] is not None, "completed_at should be set"
print(" PASS test_job_update_status")
def test_job_list():
"""List returns all jobs in descending order"""
import local_db
local_db.job_add("task_a", "article_summary", "First")
local_db.job_add("task_b", "daily_report", "Second")
local_db.job_add("task_c", "article_image", "Third")
jobs = local_db.job_list()
assert len(jobs) >= 3, f"Expected >= 3 jobs, got {len(jobs)}"
# Most recent first
assert jobs[0]["task_id"] == "task_c", f"Expected task_c first, got {jobs[0]['task_id']}"
print(" PASS test_job_list")
def test_job_delete():
"""Delete removes job from DB"""
import local_db
local_db.job_add("to_delete", "article_summary", "Delete Me")
assert local_db.job_get("to_delete") is not None
local_db.job_delete("to_delete")
assert local_db.job_get("to_delete") is None
print(" PASS test_job_delete")
# ============================================================
# But wait - local_db uses config.LOCAL_DB, not our TEST_DB!
# These will FAIL because we haven't made local_db testable.
# That's RED — exactly what TDD wants.
# ============================================================
if __name__ == "__main__":
try:
setup()
test_job_add()
test_job_update_status()
test_job_list()
test_job_delete()
print("\nALL TESTS PASSED")
except Exception as e:
print(f"\nTEST FAILED: {e}")
import traceback
traceback.print_exc()
finally:
teardown()