Add integration tests for Async client (#326)

Author: IgorChvyrov-sm

.github/workflows/pull_request.yml

@@ -80,6 +80,18 @@ jobs:
             conductor-sdk-test:latest \
             /bin/sh -c "cd /package && COVERAGE_FILE=/package/${{ env.COVERAGE_DIR }}/.coverage.integration coverage run -m pytest -m v4 tests/integration -v"
 
+      - name: Run asyncio integration tests
+        id: integration_tests
+        continue-on-error: true
+        run: |
+          docker run --rm \
+            -e CONDUCTOR_AUTH_KEY=${{ env.CONDUCTOR_AUTH_KEY }} \
+            -e CONDUCTOR_AUTH_SECRET=${{ env.CONDUCTOR_AUTH_SECRET }} \
+            -e CONDUCTOR_SERVER_URL=${{ env.CONDUCTOR_SERVER_URL }} \
+            -v ${{ github.workspace }}/${{ env.COVERAGE_DIR }}:/package/${{ env.COVERAGE_DIR }}:rw \
+            conductor-sdk-test:latest \
+            /bin/sh -c "cd /package && COVERAGE_FILE=/package/${{ env.COVERAGE_DIR }}/.coverage.integration coverage run -m pytest -m v4 tests/integration -v"
+
       - name: Generate coverage report
         id: coverage_report
         continue-on-error: true

pyproject.toml

@@ -188,4 +188,4 @@ markers = [
     "v4_1_73: mark test to run for version 4.1.73",
     "v5_2_6: mark test to run for version 5.2.6",
     "v3_21_16: mark test to run for version 3.21.16"
-]
+]

src/conductor/asyncio_client/adapters/api/workflow_resource_api.py

@@ -75,3 +75,90 @@ async def update_workflow_state(
             response_data=response_data,
             response_types_map=_response_types_map,
         ).data
+
+    @validate_call
+    async def start_workflow1(
+        self,
+        name: StrictStr,
+        request_body: Dict[str, Any],
+        version: Optional[StrictInt] = None,
+        correlation_id: Optional[StrictStr] = None,
+        priority: Optional[StrictInt] = None,
+        x_idempotency_key: Optional[StrictStr] = None,
+        x_on_conflict: Optional[StrictStr] = None,
+        _request_timeout: Union[
+            None,
+            Annotated[StrictFloat, Field(gt=0)],
+            Tuple[
+                Annotated[StrictFloat, Field(gt=0)], Annotated[StrictFloat, Field(gt=0)]
+            ],
+        ] = None,
+        _request_auth: Optional[Dict[StrictStr, Any]] = None,
+        _content_type: Optional[StrictStr] = None,
+        _headers: Optional[Dict[StrictStr, Any]] = None,
+        _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0,
+    ) -> str:
+        """Start a new workflow. Returns the ID of the workflow instance that can be later used for tracking
+
+
+        :param name: (required)
+        :type name: str
+        :param request_body: (required)
+        :type request_body: Dict[str, object]
+        :param version:
+        :type version: int
+        :param correlation_id:
+        :type correlation_id: str
+        :param priority:
+        :type priority: int
+        :param x_idempotency_key:
+        :type x_idempotency_key: str
+        :param x_on_conflict:
+        :type x_on_conflict: str
+        :param _request_timeout: timeout setting for this request. If one
+                                 number provided, it will be total request
+                                 timeout. It can also be a pair (tuple) of
+                                 (connection, read) timeouts.
+        :type _request_timeout: int, tuple(int, int), optional
+        :param _request_auth: set to override the auth_settings for an a single
+                              request; this effectively ignores the
+                              authentication in the spec for a single request.
+        :type _request_auth: dict, optional
+        :param _content_type: force content-type for the request.
+        :type _content_type: str, Optional
+        :param _headers: set to override the headers for a single
+                         request; this effectively ignores the headers
+                         in the spec for a single request.
+        :type _headers: dict, optional
+        :param _host_index: set to override the host_index for a single
+                            request; this effectively ignores the host_index
+                            in the spec for a single request.
+        :type _host_index: int, optional
+        :return: Returns the result object.
+        """  # noqa: E501
+
+        _param = self._start_workflow1_serialize(
+            name=name,
+            request_body=request_body,
+            version=version,
+            correlation_id=correlation_id,
+            priority=priority,
+            x_idempotency_key=x_idempotency_key,
+            x_on_conflict=x_on_conflict,
+            _request_auth=_request_auth,
+            _content_type=_content_type,
+            _headers=_headers,
+            _host_index=_host_index,
+        )
+
+        _response_types_map: Dict[str, Optional[str]] = {
+            "200": "str",
+        }
+        response_data = await self.api_client.call_api(
+            *_param, _request_timeout=_request_timeout
+        )
+        await response_data.read()
+        return self.api_client.response_deserialize(
+            response_data=response_data,
+            response_types_map=_response_types_map,
+        ).data

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

@@ -19,6 +19,8 @@ class ExtendedTaskDefAdapter(ExtendedTaskDef):
         default=None, alias="outputSchema"
     )
     tags: Optional[List["TagAdapter"]] = None
+    timeout_seconds: Optional[int] = Field(alias="timeoutSeconds", default=None)
+    total_timeout_seconds: Optional[int] = Field(alias="totalTimeoutSeconds", default=None)
 
     @classmethod
     def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]:

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

@@ -27,6 +27,7 @@ class ExtendedWorkflowDefAdapter(ExtendedWorkflowDef):
     tags: Optional[List["TagAdapter"]] = None
     tasks: List["WorkflowTaskAdapter"]
     variables: Optional[Dict[str, Any]] = None
+    timeout_seconds: Optional[int] = Field(default=None, alias="timeoutSeconds")
 
     @classmethod
     def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]:
@@ -99,7 +100,9 @@ def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]:
 from conductor.asyncio_client.adapters.models.schema_def_adapter import (  # noqa: E402
     SchemaDefAdapter,
 )
-from conductor.asyncio_client.adapters.models.tag_adapter import TagAdapter  # noqa: E402
+from conductor.asyncio_client.adapters.models.tag_adapter import (
+    TagAdapter,
+)  # noqa: E402
 from conductor.asyncio_client.adapters.models.workflow_task_adapter import (  # noqa: E402
     WorkflowTaskAdapter,
 )

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

@@ -2,6 +2,7 @@
 
 from typing import Any, Dict, List, Optional
 
+from pydantic import ConfigDict, field_validator
 from typing_extensions import Self
 
 from conductor.asyncio_client.http.models import IntegrationDef
@@ -10,6 +11,34 @@
 class IntegrationDefAdapter(IntegrationDef):
     configuration: Optional[List["IntegrationDefFormFieldAdapter"]] = None
 
+    @field_validator("category")
+    def category_validate_enum(cls, value):
+        if value is None:
+            return value
+
+        if value not in set(
+            [
+                "API",
+                "AI_MODEL",
+                "VECTOR_DB",
+                "RELATIONAL_DB",
+                "MESSAGE_BROKER",
+                "GIT",
+                "EMAIL",
+                "MCP",
+            ]
+        ):
+            raise ValueError(
+                "must be one of enum values ('API', 'AI_MODEL', 'VECTOR_DB', 'RELATIONAL_DB', 'MESSAGE_BROKER', 'GIT', 'EMAIL', 'MCP')"
+            )
+        return value
+
+    model_config = ConfigDict(
+        populate_by_name=True,
+        validate_assignment=True,
+        protected_namespaces=(),
+    )
+
     @classmethod
     def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]:
         """Create an instance of IntegrationDef from a dict"""

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

@@ -1,9 +1,19 @@
 from __future__ import annotations
+from enum import Enum
 
 from typing import Any, Dict, Optional
 
 from conductor.asyncio_client.http.models import SchemaDef
 
 
+class SchemaType(str, Enum):
+    JSON = ("JSON",)
+    AVRO = ("AVRO",)
+    PROTOBUF = "PROTOBUF"
+
+    def __str__(self) -> str:
+        return self.name.__str__()
+
+
 class SchemaDefAdapter(SchemaDef):
     data: Optional[Dict[str, Any]] = None

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

@@ -18,6 +18,8 @@ class TaskDefAdapter(TaskDef):
     output_schema: Optional["SchemaDefAdapter"] = Field(
         default=None, alias="outputSchema"
     )
+    timeout_seconds: Optional[int] = Field(alias="timeoutSeconds", default=None)
+    total_timeout_seconds: Optional[int] = Field(alias="totalTimeoutSeconds", default=None)
 
     @classmethod
     def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]:

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

@@ -28,6 +28,8 @@ class WorkflowDefAdapter(WorkflowDef):
     rate_limit_config: Optional["RateLimitConfigAdapter"] = Field(
         default=None, alias="rateLimitConfig"
     )
+    timeout_seconds: Optional[int] = Field(default=None, alias="timeoutSeconds")
+
     __properties: ClassVar[List[str]] = [
         "createTime",
         "createdBy",