feat: add final build

Author: agredyaev

.gitignore

@@ -114,3 +114,4 @@ ruff_cache/
 output.*
 tmp*
 test_state
+test_file.*

src/scheduler/manager.py

@@ -16,61 +16,58 @@
 
 
 class Scheduler:
-    """Scheduler class."""
-
-    __slots__ = (
-        "_completed_tasks",
-        "_config",
-        "_context_manager",
-        "_failed_tasks",
-        "_lock",
-        "_state_manager",
-        "_tasks",
-    )
+    """Task scheduler."""
 
     def __init__(self, context_manager: ContextManagerProtocol, state_manager: StateManagerProtocol) -> None:
         self._context_manager = context_manager
         self._state_manager = state_manager
         self._config = settings.scheduler
         self._lock = RLock()
-        self._tasks: deque[BaseTask] = deque()
+        self._tasks: deque[BaseTask] = deque(maxlen=self._config.max_concurrent_tasks)
         self._completed_tasks: set[UUID] = set()
         self._failed_tasks: set[UUID] = set()
 
     def add_task(self, task: BaseTask) -> None:
         """Add task to the scheduler queue."""
         self._tasks.append(task)
+        context = next(self._context_manager.create_context())
+        next(self._context_manager.associate_task(task.task_id, context.id))
 
     def _can_execute(self, task: BaseTask) -> bool:
-        """Check if task can be executed."""
-        if task.config.start_time and task.config.start_time > get_current_timestamp():
-            return False
+        """Checks if all dependencies are completed and if start time is in the future."""
         for dependency in task.dependencies:
             if dependency in self._failed_tasks:
                 task.set_state(TaskState.FAILED)
+                self._failed_tasks.add(task.task_id)
                 return False
             if dependency not in self._completed_tasks:
                 return False
-        return True
+        return not (task.config.start_time and task.config.start_time > get_current_timestamp())
 
     def _process_task(self, task: BaseTask) -> Generator[None, None, None]:
-        """Process task execution."""
-        yield None
+        """Process task execution. Moves task to completed or failed states."""
+        context = self._context_manager.get_context(task.task_id)
+        yield from task.execute(context)
+        match task.state:
+            case TaskState.COMPLETED:
+                self._completed_tasks.add(task.task_id)
+            case TaskState.FAILED:
+                self._failed_tasks.add(task.task_id)
 
     def run(self) -> Generator[None, None, None]:
         """Run event loop."""
         while self._tasks:
             task = self._tasks.popleft()
             if self._can_execute(task):
                 yield from self._process_task(task)
-            else:
+            elif task.state != TaskState.FAILED:
                 self._tasks.append(task)
 
     def __enter__(self) -> "Scheduler":
         """Logic when the scheduler starts."""
         logger.info("Starting scheduler.")
         with self._lock:
-            self._state_manager.load()
+            next(self._state_manager.load())
         logger.info("Loaded state from storage.")
         return self
 
@@ -79,7 +76,7 @@ def __exit__(
     ) -> None:
         """Logic when the scheduler exits."""
         with self._lock:
-            self._state_manager.save()
+            next(self._state_manager.save())
         logger.info("Saved state to storage.")
         logger.info("Scheduler stopped")
         if exc_val:

src/schemas/task.py

@@ -4,7 +4,7 @@
 from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 from src.helpers import get_current_timestamp
-from src.schemas.enums import FileOperation, TaskPriority, TaskType
+from src.schemas.enums import FileOperation, HttpOperation, TaskPriority, TaskType
 from src.schemas.mixins import CreatedAtMixin, UpdatedAtMixin, UUIDMixin
 
 
@@ -48,19 +48,12 @@ class FileTaskConfig(TaskConfig):
     file_path: str = Field(..., min_length=1, description="Target file path")
     content: str | None = Field(None, description="Content for write operations")
 
-    @model_validator(mode="after")
-    def validate_content_required(self) -> "FileTaskConfig":
-        """Validates content presence for write operations."""
-        if self.operation in [FileOperation.WRITE, FileOperation.APPEND] and self.content is None:
-            raise ValueError("Content is required for write and append operations")
-        return self
-
 
 class HttpTaskConfig(TaskConfig):
     """Configuration for HTTP request tasks."""
 
     task_type: TaskType = Field(TaskType.HTTP, frozen=True)
     url: str = Field(..., description="Target URL for HTTP request")
-    method: str = Field(default="GET", description="HTTP method")
+    method: HttpOperation = Field(default=HttpOperation.GET, description="HTTP method")
     headers: dict[str, str] = Field(default_factory=dict, description="HTTP headers")
     timeout: float = Field(default=30.0, ge=0, description="HTTP request timeout")

tests/test_context_manager.py

@@ -1,5 +1,4 @@
-from idlelib.configdialog import changes
-from uuid import UUID, uuid4
+from uuid import uuid4
 
 import pytest
 
@@ -9,7 +8,7 @@
 from src.context import ContextManager
 from src.core.exceptions import ContextNotFoundError, ContextVersionError
 from src.helpers import get_current_timestamp
-from src.schemas import Context, TaskConfig
+from src.schemas import Context
 
 
 class BaseFactory:

tests/test_scheduler.py

@@ -0,0 +1,120 @@
+from collections.abc import Generator
+from datetime import timedelta
+from uuid import UUID, uuid4
+
+import pytest
+
+from polyfactory.factories.pydantic_factory import ModelFactory
+from pytest_mock import MockerFixture
+
+from src.helpers import get_current_timestamp
+from src.scheduler import Scheduler
+from src.schemas import Context, FileTaskConfig, TaskState
+from src.task import BaseTask
+
+
+class BaseFactory:
+    __random_seed__ = 1
+
+
+class BaseTaskFactory(BaseFactory, ModelFactory[FileTaskConfig]): ...
+
+
+@pytest.fixture
+def mock_context_manager(mocker: MockerFixture):
+    return mocker.Mock()
+
+
+@pytest.fixture
+def mock_state_manager(mocker: MockerFixture):
+    return mocker.Mock()
+
+
+@pytest.fixture
+def scheduler(mock_context_manager, mock_state_manager):
+    return Scheduler(mock_context_manager, mock_state_manager)
+
+
+class FBaseTask(BaseTask):
+    def _do_execute(self, context: Context) -> Generator[None, None, None]:
+        """Simulate a task that always fails."""
+        yield None
+
+
+class TestScheduler:
+    def test_add_task(self, scheduler, mocker: MockerFixture):
+        config = BaseTaskFactory.build()
+        task = FBaseTask(config)
+
+        mock_context = mocker.Mock()
+        scheduler._context_manager.create_context.return_value = iter([mock_context])
+        scheduler._context_manager.associate_task.return_value = iter([None])
+
+        scheduler.add_task(task)
+
+        assert len(scheduler._tasks) == 1, "Task was not added to the queue"
+        assert scheduler._tasks[0] == task, "Added task is not at the front of the queue"
+
+    def test_can_execute_start_time_not_reached(self, scheduler, mocker: MockerFixture):
+        future_time = get_current_timestamp() + timedelta(hours=1)
+        config = BaseTaskFactory.build(start_time=future_time.timestamp())
+        task = FBaseTask(config)
+
+        mocker.patch("src.helpers.get_current_timestamp", return_value=get_current_timestamp().timestamp())
+
+        assert not scheduler._can_execute(task), "Task should not be executable before start time"
+
+    def test_can_execute_dependencies_completed(self, scheduler):
+        dependency = uuid4()
+        config = BaseTaskFactory.build(dependencies=[dependency])
+        task = FBaseTask(config)
+        scheduler._completed_tasks.add(dependency)
+
+        assert scheduler._can_execute(task), "Task should be executable if all dependencies are completed"
+
+    def test_can_execute_dependencies_failed(self, scheduler, mocker: MockerFixture):
+        from datetime import datetime, timedelta
+        from zoneinfo import ZoneInfo
+
+        future_time = datetime.now(ZoneInfo("UTC")) + timedelta(hours=1)
+
+        dependency = uuid4()
+        config = BaseTaskFactory.build(dependencies=[dependency], start_time=future_time)
+        task = FBaseTask(config)
+        scheduler._failed_tasks.add(dependency)
+
+        mock_context = mocker.Mock()
+        scheduler._context_manager.create_context.return_value = iter([mock_context])
+        scheduler._context_manager.associate_task.return_value = iter([None])
+
+        scheduler.add_task(task)
+        list(scheduler.run())
+
+        assert task.task_id in scheduler._failed_tasks, "Task should be added to failed tasks"
+        assert task not in scheduler._tasks, "Task should be removed from the queue"
+        assert task.state == TaskState.FAILED, "Task state should be set to FAILED"
+
+    def test_run(self, scheduler, mocker: MockerFixture):
+        config1 = BaseTaskFactory.build(start_time=None)
+        task1 = FBaseTask(config1)
+        config2 = BaseTaskFactory.build(start_time=None, dependencies=[task1.task_id])
+        task2 = FBaseTask(config2)
+        scheduler._tasks.extend([task1, task2])
+
+        mocker.patch.object(scheduler, "_can_execute", side_effect=[True, False, True])
+        mocker.patch.object(scheduler, "_process_task", return_value=iter([None]))
+
+        list(scheduler.run())
+
+        assert len(scheduler._tasks) == 0, "Tasks should be removed from the queue"
+        assert scheduler._process_task.call_count == 2, "Task should be processed twice"
+
+    def test_context_manager(self, scheduler, mocker: MockerFixture):
+        mock_load = mocker.patch.object(scheduler._state_manager, "load", return_value=iter([None]))
+        mock_save = mocker.patch.object(scheduler._state_manager, "save", return_value=iter([None]))
+
+        with scheduler:
+            pass
+
+        mock_load.assert_called_once(), "State manager should be loaded"
+        mock_save.assert_called_once(), "State manager should be saved"

tests/test_state_manager.py

@@ -1,15 +1,13 @@
-import os
-
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
-from uuid import UUID, uuid4
+from uuid import uuid4
 
 import pytest
 
 from polyfactory.factories.pydantic_factory import ModelFactory
 from pytest_mock import MockerFixture
 
-from src.core.exceptions import StateLoadError, StateLockError, StateNotFoundError, StateSaveError
+from src.core.exceptions import StateLoadError, StateLockError, StateNotFoundError
 from src.schemas import StateData, TaskState, TaskStateData, TaskStates