| 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) |
| |
| 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 |
| |
| 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 |
|
|
