57 lines
2.1 KiB
Python
57 lines
2.1 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()
|
|
result = 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
|
|
# Note: output_path may not exist in test environment without actual fio
|
|
|
|
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()
|
|
result = asyncio.get_event_loop().run_until_complete(
|
|
orchestrator.run_fio("/dev/sda")
|
|
)
|
|
# result is the output path, job_id is derived from it
|
|
job_id = f"fio_{result.stem.split('_')[-1]}"
|
|
# Get the actual job_id from the jobs dict
|
|
jobs = orchestrator.list_jobs()
|
|
job_id = list(jobs.keys())[0]
|
|
|
|
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 |