64 lines
2.3 KiB
Python
64 lines
2.3 KiB
Python
import asyncio
|
|
import os
|
|
import tempfile
|
|
from db.database import init_db
|
|
from db.repository import Repository
|
|
|
|
async def run_tests():
|
|
with tempfile.NamedTemporaryFile(suffix=".db") as tmp:
|
|
db_path = tmp.name
|
|
print(f"Testing DB on {db_path}...")
|
|
await init_db(db_path)
|
|
repo = Repository(db_path)
|
|
|
|
# 1. Test Sources
|
|
s_id = await repo.add_source(-1001234567890, "Source Tech", "source_tech")
|
|
sources = await repo.get_active_sources()
|
|
assert len(sources) == 1
|
|
assert sources[0].channel_id == -1001234567890
|
|
|
|
# 2. Test Targets
|
|
t_id = await repo.add_target(-1009876543210, "Target Channel", "target_chan", post_interval_min=15)
|
|
targets = await repo.get_active_targets()
|
|
assert len(targets) == 1
|
|
assert targets[0].post_interval_min == 15
|
|
|
|
# 3. Test Raw Post & Deduplication
|
|
post_id = await repo.create_raw_post(
|
|
source_channel_id=-1001234567890,
|
|
source_message_id=101,
|
|
raw_text="Breaking news: AI update released!",
|
|
content_hash="hash_12345",
|
|
is_duplicate=False
|
|
)
|
|
assert post_id is not None
|
|
|
|
# Check duplicate lookup
|
|
dup_match = await repo.find_duplicate_post("hash_12345")
|
|
assert dup_match is not None
|
|
assert dup_match.id == post_id
|
|
|
|
# 4. Test AI update & Approval flow
|
|
await repo.update_ai_result(post_id, subject="AI/Tech", ai_text="Rewritten: AI update is now live!", suggested_target_id=t_id)
|
|
pending_review = await repo.get_posts_by_status("pending_review")
|
|
assert len(pending_review) == 1
|
|
assert pending_review[0].subject == "AI/Tech"
|
|
|
|
await repo.approve_post(post_id, target_channel_id=t_id)
|
|
approved_post = await repo.get_next_approved_post_for_target(t_id)
|
|
assert approved_post is not None
|
|
assert approved_post.id == post_id
|
|
|
|
await repo.mark_post_published(post_id)
|
|
assert await repo.get_next_approved_post_for_target(t_id) is None
|
|
|
|
# 5. Settings
|
|
await repo.set_setting("system_prompt", "You are a professional editor.")
|
|
val = await repo.get_setting("system_prompt")
|
|
assert val == "You are a professional editor."
|
|
|
|
print("All database tests passed successfully!")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(run_tests())
|