File size: 3,143 Bytes
43dd6eb ae74af5 43dd6eb | 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 | import aiosqlite
import pytest
from app.database import Database
from app.models import JobStatus
@pytest.mark.asyncio
async def test_create_and_get_job(temp_db_path, sample_job_request):
db = Database(temp_db_path)
await db.init()
job_id = await db.create_job(sample_job_request)
assert isinstance(job_id, str)
assert len(job_id) > 0
job = await db.get_job(job_id)
assert job.id == job_id
assert job.status == JobStatus.QUEUED
assert job.request.aoi.name == "Khartoum North"
assert job.request.indicator_ids == ["fires", "cropland"]
@pytest.mark.asyncio
async def test_update_job_status(temp_db_path, sample_job_request):
db = Database(temp_db_path)
await db.init()
job_id = await db.create_job(sample_job_request)
await db.update_job_status(job_id, JobStatus.PROCESSING)
job = await db.get_job(job_id)
assert job.status == JobStatus.PROCESSING
@pytest.mark.asyncio
async def test_update_job_progress(temp_db_path, sample_job_request):
db = Database(temp_db_path)
await db.init()
job_id = await db.create_job(sample_job_request)
await db.update_job_progress(job_id, "fires", "complete")
job = await db.get_job(job_id)
assert job.progress["fires"] == "complete"
@pytest.mark.asyncio
async def test_get_next_queued_job(temp_db_path, sample_job_request):
db = Database(temp_db_path)
await db.init()
job_id = await db.create_job(sample_job_request)
next_job = await db.get_next_queued_job()
assert next_job is not None
assert next_job.id == job_id
@pytest.mark.asyncio
async def test_get_next_queued_returns_none_when_empty(temp_db_path):
db = Database(temp_db_path)
await db.init()
next_job = await db.get_next_queued_job()
assert next_job is None
@pytest.mark.asyncio
async def test_get_unknown_job_returns_none(temp_db_path):
db = Database(temp_db_path)
await db.init()
job = await db.get_job("nonexistent-id")
assert job is None
@pytest.mark.asyncio
async def test_create_job_stores_email(temp_db_path, sample_job_request):
db = Database(temp_db_path)
await db.init()
job_id = await db.create_job(sample_job_request)
# Read raw row to confirm email column
async with aiosqlite.connect(temp_db_path) as conn:
cur = await conn.execute("SELECT email FROM jobs WHERE id = ?", (job_id,))
row = await cur.fetchone()
assert row[0] == "test@example.com"
@pytest.mark.asyncio
async def test_get_jobs_by_email(temp_db_path, sample_job_request):
db = Database(temp_db_path)
await db.init()
id1 = await db.create_job(sample_job_request)
id2 = await db.create_job(sample_job_request)
jobs = await db.get_jobs_by_email("test@example.com")
assert len(jobs) == 2
# Most recent first
assert jobs[0].id == id2
assert jobs[1].id == id1
@pytest.mark.asyncio
async def test_get_jobs_by_email_filters(temp_db_path, sample_job_request):
db = Database(temp_db_path)
await db.init()
await db.create_job(sample_job_request)
jobs = await db.get_jobs_by_email("other@example.com")
assert len(jobs) == 0
|