"""
This type stub file was generated by pyright.
"""

import temporalio.bridge.testing
import temporalio.client
import temporalio.common
import temporalio.converter
import temporalio.runtime
import temporalio.worker
from contextlib import asynccontextmanager, contextmanager
from datetime import datetime, timedelta
from typing import Any, AsyncIterator, Iterator, Mapping, Optional, Sequence, Type, Union

"""
This type stub file was generated by pyright.
"""
logger = ...
class WorkflowEnvironment:
    """Workflow environment for testing workflows.

    Most developers will want to use the static :py:meth:`start_time_skipping`
    to start a test server process that automatically skips time as needed.
    Alternatively, :py:meth:`start_local` may be used for a full, local Temporal
    server with more features. To use an existing server, use
    :py:meth:`from_client`.

    This environment is an async context manager, so it can be used with
    ``async with`` to make sure it shuts down properly. Otherwise,
    :py:meth:`shutdown` can be manually called.

    To use the environment, simply use the :py:attr:`client` on it.

    Workflows invoked on the workflow environment are automatically configured
    to have ``assert`` failures fail the workflow with the assertion error.
    """
    @staticmethod
    def from_client(client: temporalio.client.Client) -> WorkflowEnvironment:
        """Create a workflow environment from the given client.

        :py:attr:`supports_time_skipping` will always return ``False`` for this
        environment. :py:meth:`sleep` will sleep the actual amount of time and
        :py:meth:`get_current_time` will return the current time.

        Args:
            client: The client to use for the environment.

        Returns:
            The workflow environment that runs against the given client.
        """
        ...
    
    @staticmethod
    async def start_local(*, namespace: str = ..., data_converter: temporalio.converter.DataConverter = ..., interceptors: Sequence[temporalio.client.Interceptor] = ..., default_workflow_query_reject_condition: Optional[temporalio.common.QueryRejectCondition] = ..., retry_config: Optional[temporalio.client.RetryConfig] = ..., rpc_metadata: Mapping[str, str] = ..., identity: Optional[str] = ..., tls: bool | temporalio.client.TLSConfig = ..., ip: str = ..., port: Optional[int] = ..., download_dest_dir: Optional[str] = ..., ui: bool = ..., runtime: Optional[temporalio.runtime.Runtime] = ..., dev_server_existing_path: Optional[str] = ..., dev_server_database_filename: Optional[str] = ..., dev_server_log_format: str = ..., dev_server_log_level: Optional[str] = ..., dev_server_download_version: str = ..., dev_server_extra_args: Sequence[str] = ...) -> WorkflowEnvironment:
        """Start a full Temporal server locally, downloading if necessary.

        This environment is good for testing full server capabilities, but does
        not support time skipping like :py:meth:`start_time_skipping` does.
        :py:attr:`supports_time_skipping` will always return ``False`` for this
        environment. :py:meth:`sleep` will sleep the actual amount of time and
        :py:meth:`get_current_time` will return the current time.

        Internally, this uses the Temporal CLI dev server from
        https://github.com/temporalio/cli. This is a self-contained binary for
        Temporal using Sqlite persistence. This call will download the CLI to a
        temporary directory by default if it has not already been downloaded
        before and ``dev_server_existing_path`` is not set.

        In the future, the dev server implementation may be changed to another
        implementation. Therefore, all ``dev_server_`` prefixed parameters are
        dev-server specific and may not apply to newer versions.

        Args:
            namespace: Namespace name to use for this environment.
            data_converter: See parameter of the same name on
                :py:meth:`temporalio.client.Client.connect`.
            interceptors: See parameter of the same name on
                :py:meth:`temporalio.client.Client.connect`.
            default_workflow_query_reject_condition: See parameter of the same
                name on :py:meth:`temporalio.client.Client.connect`.
            retry_config: See parameter of the same name on
                :py:meth:`temporalio.client.Client.connect`.
            rpc_metadata: See parameter of the same name on
                :py:meth:`temporalio.client.Client.connect`.
            identity: See parameter of the same name on
                :py:meth:`temporalio.client.Client.connect`.
            tls: See parameter of the same name on
                :py:meth:`temporalio.client.Client.connect`.
            ip: IP address to bind to, or 127.0.0.1 by default.
            port: Port number to bind to, or an OS-provided port by default.
            download_dest_dir: Directory to download binary to if a download is
                needed. If unset, this is the system's temporary directory.
            ui: If ``True``, will start a UI in the dev server.
            runtime: Specific runtime to use or default if unset.
            dev_server_existing_path: Existing path to the CLI binary.
                If present, no download will be attempted to fetch the binary.
            dev_server_database_filename: Path to the Sqlite database to use
                for the dev server. Unset default means only in-memory Sqlite
                will be used.
            dev_server_log_format: Log format for the dev server.
            dev_server_log_level: Log level to use for the dev server. Default
                is ``warn``, but if set to ``None`` this will translate the
                Python logger's level to a dev server log level.
            dev_server_download_version: Specific CLI version to download.
                Defaults to ``default`` which downloads the version known to
                work best with this SDK.
            dev_server_extra_args: Extra arguments for the CLI binary.

        Returns:
            The started CLI dev server workflow environment.
        """
        ...
    
    @staticmethod
    async def start_time_skipping(*, data_converter: temporalio.converter.DataConverter = ..., interceptors: Sequence[temporalio.client.Interceptor] = ..., default_workflow_query_reject_condition: Optional[temporalio.common.QueryRejectCondition] = ..., retry_config: Optional[temporalio.client.RetryConfig] = ..., rpc_metadata: Mapping[str, str] = ..., identity: Optional[str] = ..., port: Optional[int] = ..., download_dest_dir: Optional[str] = ..., runtime: Optional[temporalio.runtime.Runtime] = ..., test_server_existing_path: Optional[str] = ..., test_server_download_version: str = ..., test_server_extra_args: Sequence[str] = ...) -> WorkflowEnvironment:
        """Start a time skipping workflow environment.

        By default, this environment will automatically skip to the next events
        in time when a workflow's
        :py:meth:`temporalio.client.WorkflowHandle.result` is awaited on (which
        includes :py:meth:`temporalio.client.Client.execute_workflow`). Before
        the result is awaited on, time can be manually skipped forward using
        :py:meth:`sleep`. The currently known time can be obtained via
        :py:meth:`get_current_time`.

        Internally, this environment lazily downloads a test-server binary for
        the current OS/arch into the temp directory if it is not already there.
        Then the executable is started and will be killed when
        :py:meth:`shutdown` is called (which is implicitly done if this is
        started via
        ``async with await WorkflowEnvironment.start_time_skipping()``).

        Users can reuse this environment for testing multiple independent
        workflows, but not concurrently. Time skipping, which is automatically
        done when awaiting a workflow result and manually done on
        :py:meth:`sleep`, is global to the environment, not to the workflow
        under test.

        In the future, the test server implementation may be changed to another
        implementation. Therefore, all ``test_server_`` prefixed parameters are
        test server specific and may not apply to newer versions.

        Args:
            data_converter: See parameter of the same name on
                :py:meth:`temporalio.client.Client.connect`.
            interceptors: See parameter of the same name on
                :py:meth:`temporalio.client.Client.connect`.
            default_workflow_query_reject_condition: See parameter of the same
                name on :py:meth:`temporalio.client.Client.connect`.
            retry_config: See parameter of the same name on
                :py:meth:`temporalio.client.Client.connect`.
            rpc_metadata: See parameter of the same name on
                :py:meth:`temporalio.client.Client.connect`.
            identity: See parameter of the same name on
                :py:meth:`temporalio.client.Client.connect`.
            port: Port number to bind to, or an OS-provided port by default.
            download_dest_dir: Directory to download binary to if a download is
                needed. If unset, this is the system's temporary directory.
            runtime: Specific runtime to use or default if unset.
            test_server_existing_path: Existing path to the test server binary.
                If present, no download will be attempted to fetch the binary.
            test_server_download_version: Specific test server version to
                download. Defaults to ``default`` which downloads the version
                known to work best with this SDK.
            test_server_extra_args: Extra arguments for the test server binary.

        Returns:
            The started workflow environment with time skipping.
        """
        ...
    
    def __init__(self, client: temporalio.client.Client) -> None:
        """Create a workflow environment from a client.

        Most users would use a static method instead.
        """
        ...
    
    async def __aenter__(self) -> WorkflowEnvironment:
        """Noop for ``async with`` support."""
        ...
    
    async def __aexit__(self, *args) -> None:
        """For ``async with`` support to just call :py:meth:`shutdown`."""
        ...
    
    @property
    def client(self) -> temporalio.client.Client:
        """Client to this environment."""
        ...
    
    async def shutdown(self) -> None:
        """Shut down this environment."""
        ...
    
    async def sleep(self, duration: Union[timedelta, float]) -> None:
        """Sleep in this environment.

        This awaits a regular :py:func:`asyncio.sleep` in regular environments,
        or manually skips time in time-skipping environments.

        Args:
            duration: Amount of time to sleep.
        """
        ...
    
    async def get_current_time(self) -> datetime:
        """Get the current time known to this environment.

        For non-time-skipping environments this is simply the system time. For
        time-skipping environments this is whatever time has been skipped to.
        """
        ...
    
    @property
    def supports_time_skipping(self) -> bool:
        """Whether this environment supports time skipping."""
        ...
    
    @contextmanager
    def auto_time_skipping_disabled(self) -> Iterator[None]:
        """Disable any automatic time skipping if this is a time-skipping
        environment.

        This is a context manager for use via ``with``. Usually in time-skipping
        environments, waiting on a workflow result causes time to automatically
        skip until the next event. This can disable that. However, this only
        applies to results awaited inside this context. This will not disable
        automatic time skipping on previous results.

        This has no effect on non-time-skipping environments.
        """
        ...
    


class _EphemeralServerWorkflowEnvironment(WorkflowEnvironment):
    def __init__(self, client: temporalio.client.Client, server: temporalio.bridge.testing.EphemeralServer) -> None:
        ...
    
    async def shutdown(self) -> None:
        ...
    
    async def sleep(self, duration: Union[timedelta, float]) -> None:
        ...
    
    async def get_current_time(self) -> datetime:
        ...
    
    @property
    def supports_time_skipping(self) -> bool:
        ...
    
    @contextmanager
    def auto_time_skipping_disabled(self) -> Iterator[None]:
        ...
    
    @asynccontextmanager
    async def time_skipping_unlocked(self) -> AsyncIterator[None]:
        ...
    


class _AssertionErrorInterceptor(temporalio.client.Interceptor, temporalio.worker.Interceptor):
    def workflow_interceptor_class(self, input: temporalio.worker.WorkflowInterceptorClassInput) -> Optional[Type[temporalio.worker.WorkflowInboundInterceptor]]:
        ...
    


class _AssertionErrorWorkflowInboundInterceptor(temporalio.worker.WorkflowInboundInterceptor):
    async def execute_workflow(self, input: temporalio.worker.ExecuteWorkflowInput) -> Any:
        ...
    
    async def handle_signal(self, input: temporalio.worker.HandleSignalInput) -> None:
        ...
    
    @contextmanager
    def assert_error_as_app_error(self) -> Iterator[None]:
        ...
    


class _TimeSkippingClientInterceptor(temporalio.client.Interceptor):
    def __init__(self, env: _EphemeralServerWorkflowEnvironment) -> None:
        ...
    
    def intercept_client(self, next: temporalio.client.OutboundInterceptor) -> temporalio.client.OutboundInterceptor:
        ...
    


class _TimeSkippingClientOutboundInterceptor(temporalio.client.OutboundInterceptor):
    def __init__(self, next: temporalio.client.OutboundInterceptor, env: _EphemeralServerWorkflowEnvironment) -> None:
        ...
    
    async def start_workflow(self, input: temporalio.client.StartWorkflowInput) -> temporalio.client.WorkflowHandle[Any, Any]:
        ...
    


class _TimeSkippingWorkflowHandle(temporalio.client.WorkflowHandle):
    env: _EphemeralServerWorkflowEnvironment
    async def result(self, *, follow_runs: bool = ..., rpc_metadata: Mapping[str, str] = ..., rpc_timeout: Optional[timedelta] = ...) -> Any:
        ...
    


