Tests: automator, kafka_publish input

Author: IgorChvyrov-sm

src/conductor/asyncio_client/adapters/models/task_exec_log_adapter.py

@@ -1,4 +1,9 @@
+from __future__ import annotations
+
 from conductor.asyncio_client.http.models import TaskExecLog
+from typing import Optional, Any
+from pydantic import Field
 
 
-class TaskExecLogAdapter(TaskExecLog): ...
+class TaskExecLogAdapter(TaskExecLog):
+    created_time: Optional[Any] = Field(default=None, alias="createdTime")

src/conductor/asyncio_client/automator/task_runner.py

@@ -14,7 +14,7 @@
 from conductor.asyncio_client.adapters.models.task_result_adapter import \
     TaskResultAdapter
 from conductor.asyncio_client.configuration import Configuration
-from conductor.asyncio_client.http.api.task_resource_api import TaskResourceApi
+from conductor.asyncio_client.adapters.api.task_resource_api import TaskResourceApiAdapter
 from conductor.asyncio_client.http.api_client import ApiClient
 from conductor.asyncio_client.http.exceptions import UnauthorizedException
 from conductor.asyncio_client.telemetry.metrics_collector import \
@@ -43,7 +43,7 @@ def __init__(
         self.metrics_collector = None
         if metrics_settings is not None:
             self.metrics_collector = MetricsCollector(metrics_settings)
-        self.task_client = TaskResourceApi(ApiClient(configuration=self.configuration))
+        self.task_client = TaskResourceApiAdapter(ApiClient(configuration=self.configuration))
 
     async def run(self) -> None:
         if self.configuration is not None:

src/conductor/asyncio_client/configuration/configuration.py

@@ -154,6 +154,15 @@ def __init__(
             **kwargs,
         )
 
+        # Debug switch and logging setup
+        self.__debug = debug
+        if self.__debug:
+            self.__log_level = logging.DEBUG
+        else:
+            self.__log_level = logging.INFO
+        # Log format
+        self.__logger_format = "%(asctime)s %(name)-12s %(levelname)-8s %(message)s"
+        
         # Setup logging
         self.logger = logging.getLogger(__name__)
         if debug:
@@ -336,8 +345,10 @@ def debug(self, value: bool) -> None:
         self._http_config.debug = value
         if value:
             self.logger.setLevel(logging.DEBUG)
+            self.__log_level = logging.DEBUG
         else:
             self.logger.setLevel(logging.WARNING)
+            self.__log_level = logging.INFO
 
     @property
     def api_key(self) -> Dict[str, str]:
@@ -420,6 +431,21 @@ def retries(self, value: Optional[int]) -> None:
         """Set number of retries."""
         self._http_config.retries = value
 
+    @property
+    def logger_format(self) -> str:
+        """Get logger format."""
+        return self.__logger_format
+
+    @logger_format.setter
+    def logger_format(self, value: str) -> None:
+        """Set logger format."""
+        self.__logger_format = value
+
+    @property
+    def log_level(self) -> int:
+        """Get log level."""
+        return self.__log_level
+
     def apply_logging_config(self, log_format : Optional[str] = None, level = None):
         """Apply logging configuration for the application."""
         if log_format is None:

src/conductor/shared/workflow/models/chat_message.py

@@ -6,4 +6,4 @@ class ChatMessage(BaseModel):
     message: str = Field(..., alias="message")
 
     class Config:
-        allow_population_by_field_name = True
+        validate_by_name = True

src/conductor/shared/workflow/models/embedding_model.py

@@ -6,4 +6,4 @@ class EmbeddingModel(BaseModel):
     model: str = Field(..., alias="embeddingModel")
 
     class Config:
-        allow_population_by_field_name = True
+        validate_by_name = True

src/conductor/shared/workflow/models/http_input.py

@@ -18,6 +18,6 @@ class HttpInput(BaseModel):
     body: Optional[Any] = Field(None, alias="body")
 
     class Config:
-        allow_population_by_field_name = True
+        validate_by_name = True
         use_enum_values = True
         arbitrary_types_allowed = True

src/conductor/shared/workflow/models/http_poll_input.py

@@ -23,7 +23,7 @@ class HttpPollInput(BaseModel):
     polling_strategy: str = Field("FIXED", alias="pollingStrategy")
 
     class Config:
-        allow_population_by_field_name = True
+        validate_by_name = True
         use_enum_values = True
         arbitrary_types_allowed = True
         json_encoders: ClassVar[Dict[Type[Any], Callable[[Any], Any]]] = {

src/conductor/shared/workflow/models/kafka_publish_input.py

@@ -16,5 +16,5 @@ class KafkaPublishInput(BaseModel):
     topic: Optional[str] = Field(None, alias="topic")
 
     class Config:
-        allow_population_by_field_name = True
+        validate_by_name = True
         arbitrary_types_allowed = True

src/conductor/shared/workflow/models/prompt.py

@@ -8,4 +8,4 @@ class Prompt(BaseModel):
     variables: Dict[str, Any] = Field(..., alias="promptVariables")
 
     class Config:
-        allow_population_by_field_name = True
+        validate_by_name = True

tests/unit/automator/test_async_task_handler.py

@@ -0,0 +1,34 @@
+import multiprocessing
+
+import pytest
+
+from conductor.asyncio_client.automator.task_handler import TaskHandler
+from conductor.asyncio_client.automator.task_runner import AsyncTaskRunner
+from conductor.asyncio_client.configuration.configuration import Configuration
+from tests.unit.resources.workers import ClassWorker2
+
+
+def test_initialization_with_invalid_workers(mocker):
+    mocker.patch(
+        "conductor.asyncio_client.automator.task_handler._setup_logging_queue",
+        return_value=(None, None),
+    )
+    with pytest.raises(Exception, match="Invalid worker"):
+        TaskHandler(
+            configuration=Configuration("http://localhost:8080/api"),
+            workers=["invalid-worker"],
+        )
+
+
+def test_start_processes(mocker, valid_task_handler):
+    mocker.patch.object(AsyncTaskRunner, "run", return_value=None)
+    with valid_task_handler as task_handler:
+        task_handler.start_processes()
+        assert len(task_handler.task_runner_processes) == 1
+        for process in task_handler.task_runner_processes:
+            assert isinstance(process, multiprocessing.Process)
+
+
+@pytest.fixture
+def valid_task_handler():
+    return TaskHandler(configuration=Configuration(), workers=[ClassWorker2("task")])