52 lines
1.8 KiB
Python
52 lines
1.8 KiB
Python
"""Tests for the orchestrator module."""
|
|
|
|
import asyncio
|
|
import pytest
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from disk_checker.core.orchestrator import Orchestrator, ToolStatus, JobResult
|
|
|
|
|
|
class TestOrchestrator:
|
|
"""Tests for Orchestrator class."""
|
|
|
|
@pytest.fixture
|
|
def orchestrator(self, tmp_path):
|
|
"""Create an orchestrator instance."""
|
|
return Orchestrator(work_dir=tmp_path)
|
|
|
|
def test_initial_state(self, orchestrator):
|
|
"""Test initial orchestrator state."""
|
|
assert orchestrator.list_jobs() == {}
|
|
|
|
def test_run_fio_creates_job(self, orchestrator, tmp_path):
|
|
"""Test that run_fio creates a job entry."""
|
|
with patch("asyncio.create_subprocess_exec") as mock_exec:
|
|
mock_exec.return_value = AsyncMock()
|
|
asyncio.get_event_loop().run_until_complete(
|
|
orchestrator.run_fio("/dev/sda")
|
|
)
|
|
|
|
jobs = orchestrator.list_jobs()
|
|
assert len(jobs) == 1
|
|
job = list(jobs.values())[0]
|
|
assert job.status == ToolStatus.COMPLETED
|
|
assert job.output_path.exists()
|
|
|
|
def test_get_job_status(self, orchestrator, tmp_path):
|
|
"""Test retrieving job status."""
|
|
with patch("asyncio.create_subprocess_exec") as mock_exec:
|
|
mock_exec.return_value = AsyncMock()
|
|
job_id = asyncio.get_event_loop().run_until_complete(
|
|
orchestrator.run_fio("/dev/sda")
|
|
)
|
|
|
|
job = orchestrator.get_job_status(job_id)
|
|
assert job is not None
|
|
assert job.status == ToolStatus.COMPLETED
|
|
|
|
def test_get_nonexistent_job(self, orchestrator):
|
|
"""Test getting status of non-existent job."""
|
|
job = orchestrator.get_job_status("nonexistent")
|
|
assert job is None |