File size: 2,614 Bytes
6a0ff33 | 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 | """
Tests for SQLite database initialization and models.
"""
import os
from sqlalchemy import create_engine, inspect
from sqlalchemy.orm import sessionmaker
from backend.db.database import Base
from backend.db.models import User, Workflow, Task, Attempt, Event
def test_tables_are_created():
"""Verify all expected tables exist after init."""
engine = create_engine(
"sqlite:///data/test_autoforge.db",
connect_args={"check_same_thread": False},
)
Base.metadata.create_all(bind=engine)
inspector = inspect(engine)
tables = inspector.get_table_names()
assert "users" in tables
assert "workflows" in tables
assert "tasks" in tables
assert "attempts" in tables
assert "events" in tables
Base.metadata.drop_all(bind=engine)
def test_user_model_creation():
"""Create a user and verify fields."""
engine = create_engine(
"sqlite:///data/test_autoforge.db",
connect_args={"check_same_thread": False},
)
Base.metadata.create_all(bind=engine)
Session = sessionmaker(bind=engine)
session = Session()
user = User(username="dbuser", email="db@test.com", password_hash="fakehash")
session.add(user)
session.commit()
session.refresh(user)
assert user.id is not None
assert user.username == "dbuser"
assert user.created_at is not None
session.close()
Base.metadata.drop_all(bind=engine)
def test_workflow_task_relationship():
"""Verify workflow → task → attempt relationships."""
engine = create_engine(
"sqlite:///data/test_autoforge.db",
connect_args={"check_same_thread": False},
)
Base.metadata.create_all(bind=engine)
Session = sessionmaker(bind=engine)
session = Session()
user = User(username="reluser", email="rel@test.com", password_hash="hash")
session.add(user)
session.commit()
wf = Workflow(user_id=user.id, prompt="test prompt")
session.add(wf)
session.commit()
task = Task(workflow_id=wf.id, name="code_gen", type="code")
session.add(task)
session.commit()
attempt = Attempt(task_id=task.id, attempt_number=1, status="pending")
session.add(attempt)
session.commit()
event = Event(workflow_id=wf.id, event_type="test", payload={"key": "value"})
session.add(event)
session.commit()
# Verify relationships
session.refresh(wf)
assert len(wf.tasks) == 1
assert len(wf.events) == 1
assert wf.tasks[0].name == "code_gen"
assert wf.tasks[0].attempts[0].attempt_number == 1
session.close()
Base.metadata.drop_all(bind=engine)
|