REGISTERED = 'REGISTERED' class Domain(BaseModel):
    """Simple Workflow Domain wrapper

    :param      name: Name of the domain to register (unique)
    :type       name: string

    :param      retention_period: Domain's workflow executions records retention in days
    :type       retention_period: Integer

    :param      status: Specifies the registration status of the
                        workflow types to list. Valid values are:
                        * swf.constants.REGISTERED
                        * swf.constants.DEPRECATED
    :type       status: string

    :param      description: Textual description of the domain
    :type       description: string
    """

    __slots__ = [
        'name',
        'status',
        'description',
        'retention_period',
    ]

    def __init__(self, name,
                 status=REGISTERED,
                 description=None,
                 retention_period=30, *args, **kwargs):
        self.name = name
        self.status = status
        self.description = description
        self.retention_period = retention_period

        # immutable decorator rebinds class name,
        # so have to use generice self.__class__
        super(self.__class__, self).__init__(*args, **kwargs)

    @classmethod
    def check(cls, domain):
        """
        Ensures *domain* is a :class:`Domain` object and has a *name* attribute
        as a string.

        """
        if (not isinstance(domain, cls) or
                not hasattr(domain, 'name') or
                not isinstance(domain.name, basestring)):
            raise TypeError('invalid type {} for domain'.format(type(domain)))

    def _diff(self):
        """Checks for differences between Domain instance
        and upstream version

        :returns: A list of swf.models.base.ModelDiff namedtuple describing
                  differences
        :rtype: list
        """
        try:
            description = self.connection.describe_domain(self.name)
        except SWFResponseError as e:
            if e.error_code == 'UnknownResourceFault':
                raise DoesNotExistError("Remote Domain does not exist")

            raise ResponseError(e.body['message'])

        domain_info = description['domainInfo']
        domain_config = description['configuration']

        return ModelDiff(
            ('name', self.name, domain_info['name']),
            ('status', self.status, domain_info['status']),
            ('description', self.description, domain_info.get('description')),
            ('retention_period', self.retention_period, domain_config['workflowExecutionRetentionPeriodInDays']),
        )

    @property
    @exceptions.translate(SWFResponseError, to=ResponseError)
    @exceptions.is_not(DomainDoesNotExist)
    @exceptions.when(SWFResponseError,
                     raises(DomainDoesNotExist,
                            when=exceptions.is_unknown('domain'),
                            extract=exceptions.extract_resource))
    def exists(self):
        """Checks if the Domain exists amazon-side

        :rtype: bool
        """
        self.connection.describe_domain(self.name)
        return True

    def save(self):
        """Creates the domain amazon side"""
        try:
            self.connection.register_domain(self.name,
                                            str(self.retention_period),
                                            self.description)
        except SWFDomainAlreadyExistsError:
            raise AlreadyExistsError("Domain %s already exists amazon-side" % self.name)

    @exceptions.translate(SWFResponseError,
                          to=ResponseError)
    @exceptions.when(SWFResponseError,
                     raises(DomainDoesNotExist,
                            when=exceptions.is_unknown('domain'),
                            extract=exceptions.extract_resource))
    def delete(self):
        """Deprecates the domain amazon side"""
        self.connection.deprecate_domain(self.name)

    def upstream(self):
        from swf.querysets.domain import DomainQuerySet
        qs = DomainQuerySet()
        return qs.get(self.name)

    def workflows(self, status=REGISTERED):
        """Lists the current domain's workflow types

        :param      status: Specifies the registration status of the
                            workflow types to list. Valid values are:
                            * swf.constants.REGISTERED
                            * swf.constants.DEPRECATED
        :type       status: string
        """
        from swf.querysets.workflow import WorkflowTypeQuerySet
        qs = WorkflowTypeQuerySet(self)
        return qs.all(registration_status=status)

    def activities(self, status=REGISTERED):
        """Lists the current domain's activity types

        :param      status: Specifies the registration status of the
                            workflow types to list. Valid values are:
                            * swf.constants.REGISTERED
                            * swf.constants.DEPRECATED
        :type       status: string
        """
        from swf.querysets.activity import ActivityTypeQuerySet
        qs = ActivityTypeQuerySet(self)
        return qs.all(registration_status=status)

    @property
    def executions(self):
        pass

    def __repr__(self):
        return '<{} name={} status={}>'.format(
               self.__class__.__name__, self.name, self.status) class WorkflowType(BaseModel):
    """Simple Workflow Type wrapper

    :param  domain: Domain the workflow type should be registered in
    :type   domain: swf.models.Domain

    :param  name: name of the workflow type
    :type   name: String

    :param  version: workflow type version
    :type   version: String

    :param  status: workflow type status
    :type   status: swf.core.ConnectedSWFObject.{REGISTERED, DEPRECATED}

    :param   creation_date: creation date of the current WorkflowType
    :type    creation_date: float (timestamp)

    :param   deprecation_date: deprecation date of WorkflowType
    :type    deprecation_date: float (timestamp)

    :param  task_list: task list to use for scheduling decision tasks for executions
                       of this workflow type
    :type   task_list: String

    :param  child_policy: policy to use for the child workflow executions
                          when a workflow execution of this type is terminated
    :type   child_policy: CHILD_POLICIES.{TERMINATE |
                                          REQUEST_CANCEL |
                                          ABANDON}

    :param  execution_timeout: maximum duration for executions of this workflow type
    :type   execution_timeout: String

    :param  decision_tasks_timeout: maximum duration of decision tasks for this workflow type
    :type   decision_tasks_timeout: String

    :param  description: Textual description of the workflow type
    :type   description: String
    """
    __slots__ = [
        'domain',
        'name',
        'version',
        'status',
        'creation_date',
        'deprecation_date',
        'task_list',
        'child_policy',
        'execution_timeout',
        'decision_tasks_timeout',
        'description',
    ]

    def __init__(self, domain, name, version,
                 status=REGISTERED,
                 creation_date=0.0,
                 deprecation_date=0.0,
                 task_list=None,
                 child_policy=CHILD_POLICIES.TERMINATE,
                 execution_timeout='300',
                 decision_tasks_timeout='300',
                 description=None, *args, **kwargs):
        self.domain = domain
        self.name = name
        self.version = version
        self.status = status
        self.creation_date = creation_date
        self.deprecation_date = deprecation_date

        self.task_list = task_list
        self.execution_timeout = execution_timeout
        self.decision_tasks_timeout = decision_tasks_timeout
        self.description = description

        # Explicitly call child_policy setter
        # to validate input value
        self.set_child_policy(child_policy)

        # immutable decorator rebinds class name,
        # so have to use generice self.__class__
        super(self.__class__, self).__init__(*args, **kwargs)

    def set_child_policy(self, policy):
        if not policy in CHILD_POLICIES:
            raise ValueError("invalid child policy value: {}".format(policy))

        self.child_policy = policy

    def _diff(self):
        """Checks for differences between WorkflowType instance
        and upstream version

        :returns: A list of swf.models.base.ModelDiff namedtuple describing
                  differences
        :rtype: list
        """
        try:
            description = self.connection.describe_workflow_type(
                self.domain.name,
                self.name,
                self.version
            )
        except SWFResponseError as e:
            if e.error_code == 'UnknownResourceFault':
                raise DoesNotExistError("Remote Domain does not exist")

            raise ResponseError(e.body['message'])

        workflow_info = description['typeInfo']
        workflow_config = description['configuration']

        return ModelDiff(
            ('name', self.name, workflow_info['workflowType']['name']),
            ('version', self.version, workflow_info['workflowType']['version']),
            ('status', self.status, workflow_info['status']),
            ('creation_date', self.creation_date, workflow_info['creationDate']),
            ('deprecation_date', self.deprecation_date, workflow_info['deprecationDate']),
            ('task_list', self.task_list, workflow_config['defaultTaskList']['name']),
            ('child_policy', self.child_policy, workflow_config['defaultChildPolicy']),
            ('execution_timeout', self.execution_timeout, workflow_config['defaultExecutionStartToCloseTimeout']),
            ('decision_tasks_timout', self.decision_tasks_timeout, workflow_config['defaultTaskStartToCloseTimeout']),
            ('description', self.description, workflow_info['description']),
        )

    @property
    @exceptions.translate(SWFResponseError, to=ResponseError)
    @exceptions.is_not(WorkflowTypeDoesNotExist)
    @exceptions.when(SWFResponseError,
                     raises(WorkflowTypeDoesNotExist,
                            when=exceptions.is_unknown('WorkflowType'),
                            extract=exceptions.extract_resource))
    def exists(self):
        """Checks if the WorkflowType exists amazon-side

        :rtype: bool
        """
        self.connection.describe_workflow_type(
            self.domain.name,
            self.name,
            self.version
        )
        return True

    def save(self):
        """Creates the workflow type amazon side"""
        try:
            self.connection.register_workflow_type(
                self.domain.name,
                self.name,
                self.version,
                task_list=str(self.task_list),
                default_child_policy=str(self.child_policy),
                default_execution_start_to_close_timeout=str(self.execution_timeout),
                default_task_start_to_close_timeout=str(self.decision_tasks_timeout),
                description=self.description
            )
        except SWFTypeAlreadyExistsError:
            raise AlreadyExistsError("Workflow type %s already exists amazon-side" % self.name)
        except SWFResponseError as e:
            if e.error_code == 'UnknownResourceFault':
                raise DoesNotExistError(e.body['message'])

    def delete(self):
        """Deprecates the workflow type amazon-side"""
        try:
            self.connection.deprecate_workflow_type(self.domain.name, self.name, self.version)
        except SWFResponseError as e:
            if e.error_code in ['UnknownResourceFault', 'TypeDeprecatedFault']:
                raise DoesNotExistError(e.body['message'])

    def upstream(self):
        from swf.querysets.workflow import WorkflowTypeQuerySet
        qs = WorkflowTypeQuerySet(self.domain)
        return qs.get(self.name, self.version)

    def start_execution(self, workflow_id=None, task_list=None,
                        child_policy=None, execution_timeout=None,
                        input=None, tag_list=None, decision_tasks_timeout=None):
        """Starts a Workflow execution of current workflow type

        :param  workflow_id: The user defined identifier associated with the workflow execution
        :type   workflow_id: String

        :param  task_list: task list to use for scheduling decision tasks for execution
                        of this workflow
        :type   task_list: String

        :param  child_policy: policy to use for the child workflow executions
                              of this workflow execution.
        :type   child_policy: CHILD_POLICIES.{TERMINATE |
                                              REQUEST_CANCEL |
                                        ABANDON}

        :param  execution_timeout: maximum duration for the workflow execution
        :type   execution_timeout: String

        :param  input: Input of the workflow execution
        :type   input: dict

        :param  tag_list: Tags associated with the workflow execution
        :type   tag_list: String

        :param  decision_tasks_timeout: maximum duration of decision tasks
                                        for this workflow execution
        :type   decision_tasks_timeout: String
        """
        workflow_id = workflow_id or '%s-%s-%i' % (self.name, self.version, time.time())
        task_list = task_list or self.task_list
        child_policy = child_policy or self.child_policy
        input = json.dumps(input) or None

        run_id = self.connection.start_workflow_execution(
            self.domain.name,
            workflow_id,
            self.name,
            self.version,
            task_list=task_list,
            child_policy=child_policy,
            execution_start_to_close_timeout=execution_timeout,
            input=input,
            tag_list=tag_list,
            task_start_to_close_timeout=decision_tasks_timeout,
        )['runId']

        return WorkflowExecution(self.domain, workflow_id, run_id=run_id)

    def __repr__(self):
        return '<{} domain={} name={} version={} status={}>'.format(
               self.__class__.__name__,
               self.domain.name,
               self.name,
               self.version,
               self.status) class WorkflowExecution(BaseModel):
    """Simple Workflow execution wrapper

    :param  domain: Domain the workflow execution should be registered in
    :type   domain: swf.models.domain.Domain

    :param  workflow_type: The WorkflowType associated with the workflow execution
                           is associated with
    :type   workflow_type: String

    :param  workflow_id: The user defined identifier associated with the workflow execution
    :type   workflow_id: String

    :param  run_id: The Amazon defined identifier associated with the workflow execution
    :type   run_id: String

    :param  status: Whether the WorkflowExecution instance represents an opened or
                    closed execution
    :type   status: String constant

    :param  task_list: The task list to use for the decision tasks generated
                       for this workflow execution.
    :type   task_list: string

    :param  input: input data of the execution, which will be passed around using
                   serialized json
    :type   input: dict
    """
    STATUS_OPEN = "OPEN"
    STATUS_CLOSED = "CLOSED"

    CLOSE_STATUS_COMPLETED = "COMPLETED"
    CLOSE_STATUS_FAILED = "FAILED"
    CLOSE_STATUS_CANCELED = "CANCELED"
    CLOSE_STATUS_TERMINATED = "TERMINATED"
    CLOSE_STATUS_CONTINUED_AS_NEW = "CLOSE_STATUS_CONTINUED_AS_NEW"
    CLOSE_TIMED_OUT = "TIMED_OUT"

    kind = 'execution'

    __slots__ = [
        'domain',
        'workflow_id',
        'run_id',
        'status',
        'workflow_type',
        'task_list',
        'child_policy',
        'close_status',
        'execution_timeout',
        'input',
        'tag_list',
        'decision_tasks_timeout',
        'close_timestamp',
        'start_timestamp',
        'cancel_requested',
        'latest_execution_context',
        'latest_activity_task_timestamp',
        'open_counts',
        'parent',
    ]

    def __init__(self, domain, workflow_id, run_id=None,
                 status=STATUS_OPEN, workflow_type=None,
                 task_list=None, child_policy=None,
                 close_status=None, execution_timeout=None,
                 input=None, tag_list=None,
                 decision_tasks_timeout=None,
                 close_timestamp=None,
                 start_timestamp=None,
                 cancel_requested=None,
                 latest_execution_context=None,
                 latest_activity_task_timestamp=None,
                 open_counts=None,
                 parent=None,
                 *args, **kwargs):
        Domain.check(domain)
        self.domain = domain
        self.workflow_id = workflow_id
        self.run_id = run_id
        self.status = status
        self.workflow_type = workflow_type
        self.task_list = task_list
        self.child_policy = child_policy
        self.close_status = close_status
        self.execution_timeout = execution_timeout
        self.input = input
        self.tag_list = tag_list or []
        self.decision_tasks_timeout = decision_tasks_timeout
        self.close_timestamp = close_timestamp
        self.start_timestamp = start_timestamp
        self.cancel_requested = cancel_requested
        self.latest_execution_context = latest_execution_context
        self.latest_activity_task_timestamp = latest_activity_task_timestamp
        self.open_counts = open_counts or {} # so we can query keys in any case
        self.parent = parent or {} # so we can query keys in any case

        # immutable decorator rebinds class name,
        # so have to use generice self.__class__
        super(self.__class__, self).__init__(*args, **kwargs)

    def _diff(self):
        """Checks for differences between WorkflowExecution instance
        and upstream version

        :returns: A list of swf.models.base.Diff namedtuple describing
                  differences
        :rtype: list
        """
        try:
            description = self.connection.describe_workflow_execution(
                self.domain.name,
                self.run_id,
                self.workflow_id
            )
        except SWFResponseError as e:
            if e.error_code == 'UnknownResourceFault':
                raise DoesNotExistError("Remote Domain does not exist")

            raise ResponseError(e.body['message'])

        execution_info = description['executionInfo']
        execution_config = description['executionConfiguration']

        return ModelDiff(
            ('workflow_id', self.workflow_id, execution_info['execution']['workflowId']),
            ('run_id', self.run_id, execution_info['execution']['runId']),
            ('status', self.status, execution_info['executionStatus']),
            ('task_list', self.task_list, execution_config['taskList']['name']),
            ('child_policy', self.child_policy, execution_config['childPolicy']),
            ('execution_timeout', self.execution_timeout, execution_config['executionStartToCloseTimeout']),
            ('tag_list', self.tag_list, execution_info.get('tagList')),
            ('decision_tasks_timeout', self.decision_tasks_timeout, execution_config['taskStartToCloseTimeout']),
        )

    @property
    @exceptions.translate(SWFResponseError, to=ResponseError)
    @exceptions.is_not(WorkflowExecutionDoesNotExist)
    @exceptions.when(SWFResponseError,
                     raises(WorkflowExecutionDoesNotExist,
                            when=exceptions.is_unknown('WorkflowExecution'),
                            extract=exceptions.extract_resource))
    def exists(self):
        """Checks if the WorkflowExecution exists amazon-side

        :rtype: bool
        """
        self.connection.describe_workflow_execution(
            self.domain.name,
            self.run_id,
            self.workflow_id
        )
        return True

    def upstream(self):
        from swf.querysets.workflow import WorkflowExecutionQuerySet
        qs = WorkflowExecutionQuerySet(self.domain)
        return qs.get(self.workflow_id, self.run_id)

    def history(self, *args, **kwargs):
        """Returns workflow execution history report

        :returns: The workflow execution complete events history
        :rtype: swf.models.event.History
        """
        domain = kwargs.pop('domain', self.domain)
        if not isinstance(domain, basestring):
            domain = domain.name

        response = self.connection.get_workflow_execution_history(
            self.domain.name,
            self.run_id,
            self.workflow_id,
            **kwargs
        )

        events = response['events']
        next_page = response.get('nextPageToken')
        while next_page is not None:
            response = self.connection.get_workflow_execution_history(
                self.domain.name,
                self.run_id,
                self.workflow_id,
                next_page_token=next_page,
                **kwargs
            )

            events.extend(response['events'])
            next_page = response.get('nextPageToken')

        return History.from_event_list(events)

    @exceptions.translate(SWFResponseError,
                          to=ResponseError)
    @exceptions.when(SWFResponseError,
                     raises(WorkflowExecutionDoesNotExist,
                            when=exceptions.is_unknown('WorkflowExecution'),
                            extract=exceptions.extract_resource))
    def signal(self, signal_name, input=None, *args, **kwargs):
        """Records a signal event in the workflow execution history and
        creates a decision task.

        The signal event is recorded with the specified user defined
        ``signal_name`` and ``input`` (if provided).

        :param  signal_name: The name of the signal. This name must be
                             meaningful to the target workflow.
        :type   signal_name: str

        :param  input: Data to attach to the WorkflowExecutionSignaled
                       event in the target workflow execution’s history.
        :type   input: dict
        """
        if input is None:
            input = {}
        self.connection.signal_workflow_execution(
            self.domain.name,
            signal_name,
            self.workflow_id,
            input=json.dumps(input),
            run_id=self.run_id)

    @exceptions.translate(SWFResponseError,
                          to=ResponseError)
    @exceptions.when(SWFResponseError,
                     raises(WorkflowExecutionDoesNotExist,
                            when=exceptions.is_unknown('domain'),
                            extract=exceptions.extract_resource))
    def request_cancel(self, *args, **kwargs):
        """Requests the workflow execution cancel"""
        self.connection.request_cancel_workflow_execution(
            self.domain.name,
            self.workflow_id,
            run_id=self.run_id)

    @exceptions.translate(SWFResponseError,
                          to=ResponseError)
    @exceptions.when(SWFResponseError,
                     raises(WorkflowExecutionDoesNotExist,
                            when=exceptions.is_unknown('domain'),
                            extract=exceptions.extract_resource))
    def terminate(self, *args, **kwargs):
        """Terminates the workflow execution"""
        self.connection.terminate_workflow_execution(
            self.domain.name,
            self.workflow_id,
            run_id=self.run_id
        ) class BaseWorkflowQuerySet(BaseQuerySet):
    """Base domain bounded workflow queryset objects

    Amazon workflows types and executions are always bounded
    to a specific domain: so any queryset which means to deal
    with workflows has to be built against a `domain`

    :param      domain: domain the inheriting queryset belongs to
    :type       domain: swf.model.domain.Domain
    """
    # Amazon response section corresponding
    # to current queryset informations
    _infos = 'typeInfo'
    _infos_plural = 'typeInfos'

    def __init__(self, domain, *args, **kwargs):
        super(BaseWorkflowQuerySet, self).__init__(*args, **kwargs)
        Domain.check(domain)
        self.domain = domain

    @property
    def domain(self):
        if not hasattr(self, '_domain'):
            self._domain = None
        return self._domain

    @domain.setter
    def domain(self, value):
        # Avoiding circular import
        from swf.models.domain import Domain

        if not isinstance(value, Domain):
            err = "domain property has to be of"\
                  "swf.model.domain.Domain type, not %r"\
                  % type(value)
            raise TypeError(err)
        self._domain = value

    def _list(self, *args, **kwargs):
        raise NotImplementedError

    def _list_items(self, *args, **kwargs):
        response = {'nextPageToken': None}
        while 'nextPageToken' in response:
            response = self._list(
                *args,
                next_page_token=response['nextPageToken'],
                **kwargs
            )

            for item in response[self._infos_plural]:
                yield item class WorkflowTypeQuerySet(BaseWorkflowQuerySet):

    # Explicit is better than implicit, keep zen
    _infos = 'typeInfo'
    _infos_plural = 'typeInfos'

    def to_WorkflowType(self, domain, workflow_info, **kwargs):
        # Not using get_subkey in order for it to explictly
        # raise when workflowType name doesn't exist for example
        return WorkflowType(
            domain,
            workflow_info['workflowType']['name'],
            workflow_info['workflowType']['version'],
            status=workflow_info['status'],
            **kwargs
        )

    def get(self, name, version, *args, **kwargs):
        """Fetches the Workflow Type with `name` and `version`

        :param  name: name of the workflow type
        :type   name: String

        :param  version: workflow type version
        :type   version: String

        :returns: matched workflow type instance
        :rtype: swf.core.model.workflow.WorkflowType

        A typical Amazon response looks like:

        .. code-block:: json

            {
                "configuration": {
                    "defaultExecutionStartToCloseTimeout": "300",
                    "defaultTaskStartToCloseTimeout": "300",
                    "defaultTaskList": {
                        "name": "None"
                    },
                    "defaultChildPolicy": "TERMINATE"
                },
                "typeInfo": {
                    "status": "REGISTERED",
                    "creationDate": 1364492094.968,
                    "workflowType": {
                        "version": "1",
                        "name": "testW"
                    }
                }
            }
        """
        try:
            response = self.connection.describe_workflow_type(self.domain.name, name, version)
        except SWFResponseError as e:
            if e.error_code == 'UnknownResourceFault':
                raise DoesNotExistError(e.body['message'])

            raise ResponseError(e.body['message'])

        wt_info = response[self._infos]
        wt_config = response['configuration']

        task_list = kwargs.get('task_list')
        if task_list is None:
            task_list = get_subkey(wt_config, ['defaultTaskList', 'name'])

        child_policy = kwargs.get('child_policy')
        if child_policy is None:
            child_policy = wt_config.get('defaultChildPolicy')

        decision_task_timeout = kwargs.get('decision_task_timeout')
        if decision_task_timeout is None:
            decision_task_timeout = wt_config.get(
                'defaultTaskStartToCloseTimeout')

        execution_timeout = kwargs.get('execution_timeout')
        if execution_timeout is None:
            execution_timeout = wt_config.get(
                'defaultExecutionStartToCloseTimeout')

        decision_tasks_timeout = kwargs.get('decision_tasks_timeout')
        if decision_tasks_timeout is None:
            decision_tasks_timeout = wt_config.get(
                'defaultTaskStartToCloseTimeout')

        return self.to_WorkflowType(
            self.domain,
            wt_info,
            task_list=task_list,
            child_policy=child_policy,
            execution_timeout=execution_timeout,
            decision_tasks_timeout=decision_tasks_timeout,
        )

    def get_or_create(self, name, version,
                      status=REGISTERED,
                      creation_date=0.0,
                      deprecation_date=0.0,
                      task_list=None,
                      child_policy=CHILD_POLICIES.TERMINATE,
                      execution_timeout='300',
                      decision_tasks_timeout='300',
                      description=None,
                      *args, **kwargs):
        """Fetches, or creates the ActivityType with ``name`` and ``version``

        When fetching trying to fetch a matching workflow type, only
        name and version parameters are taken in account.
        Anyway, If you'd wanna make sure that in case the workflow type
        has to be created it is made with specific values, just provide it.

        :param  name: name of the workflow type
        :type   name: String

        :param  version: workflow type version
        :type   version: String

        :param  status: workflow type status
        :type   status: swf.core.ConnectedSWFObject.{REGISTERED, DEPRECATED}

        :param   creation_date: creation date of the current WorkflowType
        :type    creation_date: float (timestamp)

        :param   deprecation_date: deprecation date of WorkflowType
        :type    deprecation_date: float (timestamp)

        :param  task_list: task list to use for scheduling decision tasks for executions
                           of this workflow type
        :type   task_list: String

        :param  child_policy: policy to use for the child workflow executions
                              when a workflow execution of this type is terminated
        :type   child_policy: CHILD_POLICIES.{TERMINATE |
                                              REQUEST_CANCEL |
                                              ABANDON}

        :param  execution_timeout: maximum duration for executions of this workflow type
        :type   execution_timeout: String

        :param  decision_tasks_timeout: maximum duration of decision tasks for this workflow type
        :type   decision_tasks_timeout: String

        :param  description: Textual description of the workflow type
        :type   description: String

        :returns: Fetched or created WorkflowType model object
        :rtype: WorkflowType
        """
        try:
            return self.get(name,
                            version,
                            task_list=task_list,
                            child_policy=child_policy,
                            execution_timeout=execution_timeout,
                            decision_tasks_timeout=decision_tasks_timeout)

        except DoesNotExistError:
            try:
                return self.create(
                    name,
                    version,
                    status=status,
                    creation_date=creation_date,
                    deprecation_date=deprecation_date,
                    task_list=task_list,
                    child_policy=child_policy,
                    execution_timeout=execution_timeout,
                    decision_tasks_timeout=decision_tasks_timeout,
                    description=description,
                )
            # race conditon could happen if two workflows trying to register the same type
            except AlreadyExistsError:
                return self.get(name,
                            version,
                            task_list=task_list,
                            child_policy=child_policy,
                            execution_timeout=execution_timeout,
                            decision_tasks_timeout=decision_tasks_timeout)

    def _list(self, *args, **kwargs):
        return self.connection.list_workflow_types(*args, **kwargs)

    def filter(self, domain=None,
               registration_status=REGISTERED,
               name=None,
               *args, **kwargs):
        """Filters workflows based on the ``domain`` they belong to,
        their ``status``, and/or their ``name``

        :param      domain: domain the workflow type belongs to
        :type       domain: swf.models.domain.Domain

        :param      registration_status: workflow type registration status to match,
                                         Valid values are:
                                         * ``swf.constants.REGISTERED``
                                         * ``swf.constants.DEPRECATED``

        :type       registration_status: string

        :param      name: workflow type name to match
        :type       name: string

        :returns: list of matched WorkflowType models objects
        :rtype: list
        """
        # As WorkflowTypeQuery has to be built against a specific domain
        # name, domain filter is disposable, but not mandatory.
        domain = domain or self.domain
        return [self.to_WorkflowType(domain, wf) for wf in
                self._list_items(domain.name, registration_status, name=name)]

    def all(self, registration_status=REGISTERED, *args, **kwargs):
        """Retrieves every Workflow types

        :param      registration_status: workflow type registration status to match,
                                 Valid values are:
                                 * ``swf.constants.REGISTERED``
                                 * ``swf.constants.DEPRECATED``

        :type       registration_status: string

        A typical Amazon response looks like:

        .. code-block:: json

            {
                "typeInfos": [
                    {
                        "status": "REGISTERED",
                        "creationDate": 1364293450.67,
                        "description": "",
                        "workflowType": {
                            "version": "1",
                            "name": "Crawl"
                        }
                    },
                    {
                        "status": "REGISTERED",
                        "creationDate": 1364492094.968,
                        "workflowType": {
                            "version": "1",
                            "name": "testW"
                        }
                    }
                ]
            }
        """
        return self.filter(registration_status=registration_status)

    def create(self, name, version,
               status=REGISTERED,
               creation_date=0.0,
               deprecation_date=0.0,
               task_list=None,
               child_policy=CHILD_POLICIES.TERMINATE,
               execution_timeout='300',
               decision_tasks_timeout='300',
               description=None,
               *args, **kwargs):
        """Creates a new remote workflow type and returns the
        created WorkflowType model instance.

        :param  name: name of the workflow type
        :type   name: String

        :param  version: workflow type version
        :type   version: String

        :param  status: workflow type status
        :type   status: swf.core.ConnectedSWFObject.{REGISTERED, DEPRECATED}

        :param   creation_date: creation date of the current WorkflowType
        :type    creation_date: float (timestamp)

        :param   deprecation_date: deprecation date of WorkflowType
        :type    deprecation_date: float (timestamp)

        :param  task_list: task list to use for scheduling decision tasks for executions
                           of this workflow type
        :type   task_list: String

        :param  child_policy: policy to use for the child workflow executions
                              when a workflow execution of this type is terminated
        :type   child_policy: CHILD_POLICIES.{TERMINATE |
                                              REQUEST_CANCEL |
                                              ABANDON}

        :param  execution_timeout: maximum duration for executions of this workflow type
        :type   execution_timeout: String

        :param  decision_tasks_timeout: maximum duration of decision tasks for this workflow type
        :type   decision_tasks_timeout: String

        :param  description: Textual description of the workflow type
        :type   description: String
        """
        workflow_type = WorkflowType(
            self.domain,
            name,
            version,
            status=status,
            creation_date=creation_date,
            deprecation_date=deprecation_date,
            task_list=task_list,
            child_policy=child_policy,
            execution_timeout=execution_timeout,
            decision_tasks_timeout=decision_tasks_timeout,
            description=description
        )
        workflow_type.save()

        return workflow_type class WorkflowExecutionQuerySet(BaseWorkflowQuerySet):
    """Fetches Workflow executions"""

    _infos = 'executionInfo'
    _infos_plural = 'executionInfos'

    def _is_valid_status_param(self, status, param):
        statuses = {
            WorkflowExecution.STATUS_OPEN: set([
                'oldest_date',
                'latest_date'],
            ),
            WorkflowExecution.STATUS_CLOSED: set([
                'start_latest_date',
                'start_oldest_date',
                'close_latest_date',
                'close_oldest_date',
                'close_status'
            ]),
        }
        return param in statuses.get(status, set())

    def _validate_status_parameters(self, status, params):
        return [param for param in params if
                not self._is_valid_status_param(status, param)]

    def list_workflow_executions(self, status, *args, **kwargs):
        statuses = {
            WorkflowExecution.STATUS_OPEN: 'open',
            WorkflowExecution.STATUS_CLOSED: 'closed',
        }

        # boto.swf.list_closed_workflow_executions awaits a `start_oldest_date`
        # MANDATORY kwarg, when boto.swf.list_open_workflow_executions awaits a
        # `oldest_date` mandatory arg.
        if status == WorkflowExecution.STATUS_OPEN:
            kwargs['oldest_date'] = kwargs.pop('start_oldest_date')

        try:
            method = 'list_{}_workflow_executions'.format(statuses[status])
            return getattr(self.connection, method)(*args, **kwargs)
        except KeyError:
            raise ValueError("Unknown status provided: %s" % status)

    def get_workflow_type(self, execution_info):
        workflow_type = execution_info['workflowType']
        workflow_type_qs = WorkflowTypeQuerySet(self.domain)

        return workflow_type_qs.get(
            workflow_type['name'],
            workflow_type['version'],
        )

    def to_WorkflowExecution(self, domain, execution_info, **kwargs):
        workflow_type = WorkflowType(
            self.domain,
            execution_info['workflowType']['name'],
            execution_info['workflowType']['version']
        )

        return WorkflowExecution(
            domain,
            get_subkey(execution_info, ['execution', 'workflowId']),  # workflow_id
            run_id=get_subkey(execution_info, ['execution', 'runId']),
            workflow_type=workflow_type,
            status=execution_info.get('executionStatus'),
            close_status=execution_info.get('closeStatus'),
            tag_list=execution_info.get('tagList'),
            start_timestamp=execution_info.get('startTimestamp'),
            close_timestamp=execution_info.get('closeTimestamp'),
            cancel_requested=execution_info.get('cancelRequested'),
            parent=execution_info.get('parent'),
            **kwargs
        )

    def get(self, workflow_id, run_id, *args, **kwargs):
        """ """
        try:
            response = self.connection.describe_workflow_execution(
                self.domain.name,
                run_id,
                workflow_id)
        except SWFResponseError as e:
            if e.error_code == 'UnknownResourceFault':
                raise DoesNotExistError(e.body['message'])

            raise ResponseError(e.body['message'])

        execution_info = response[self._infos]
        execution_config = response['executionConfiguration']

        return self.to_WorkflowExecution(
            self.domain,
            execution_info,
            task_list=get_subkey(execution_config, ['taskList', 'name']),
            child_policy=execution_config.get('childPolicy'),
            execution_timeout=execution_config.get('executionStartToCloseTimeout'),
            decision_tasks_timeout=execution_config.get('taskStartToCloseTimeout'),
            latest_activity_task_timestamp=response.get('latestActivityTaskTimestamp'),
            latest_execution_context=response.get('latestExecutionContext'),
            open_counts=response['openCounts'],
        )

    def filter(self,
               status=WorkflowExecution.STATUS_OPEN, tag=None,
               workflow_id=None, workflow_type_name=None,
               workflow_type_version=None,
               *args, **kwargs):
        """Filters workflow executions based on kwargs provided criteras

        :param  status: workflow executions with provided status will be kept.
                        Valid values are:
                        * ``swf.models.WorkflowExecution.STATUS_OPEN``
                        * ``swf.models.WorkflowExecution.STATUS_CLOSED``
        :type   status: string

        :param  tag: workflow executions containing the tag will be kept
        :type   tag: String

        :param  workflow_id: workflow executions attached to the id will be kept
        :type   workflow_id: String

        :param  workflow_type_name: workflow executions attached to the workflow type
                                    with provided name will be kept
        :type   workflow_type_name: String

        :param  workflow_type_version: workflow executions attached to the workflow type
                                       of the provided version will be kept
        :type   workflow_type_version: String

        **Be aware that** querying over status allows the usage of statuses specific
        kwargs

        * STATUS_OPEN

            :param start_latest_date: latest start or close date and time to return (in days)
            :type  start_latest_date: int

        * STATUS_CLOSED

            :param  start_latest_date: workflow executions that meet the start time criteria
                                       of the filter are kept (in days)
            :type   start_latest_date: int

            :param  start_oldest_date: workflow executions that meet the start time criteria
                                       of the filter are kept (in days)
            :type   start_oldest_date: int

            :param  close_latest_date: workflow executions that meet the close time criteria
                                       of the filter are kept (in days)
            :type   close_latest_date: int

            :param  close_oldest_date: workflow executions that meet the close time criteria
                                       of the filter are kept (in days)
            :type   close_oldest_date: int

            :param  close_status: must match the close status of an execution for it
                                  to meet the criteria of this filter.
                                  Valid values are:
                                  * ``CLOSE_STATUS_COMPLETED``
                                  * ``CLOSE_STATUS_FAILED``
                                  * ``CLOSE_STATUS_CANCELED``
                                  * ``CLOSE_STATUS_TERMINATED``
                                  * ``CLOSE_STATUS_CONTINUED_AS_NEW``
                                  * ``CLOSE_TIMED_OUT``
            :type   close_status: string

            :returns: workflow executions objects list
            :rtype: list
        """
        # As WorkflowTypeQuery has to be built against a specific domain
        # name, domain filter is disposable, but not mandatory.
        invalid_kwargs = self._validate_status_parameters(status, kwargs)

        if invalid_kwargs:
            err_msg = 'Invalid keyword arguments supplied: {}'.format(
                      ', '.join(invalid_kwargs))
            raise InvalidKeywordArgumentError(err_msg)

        if status == WorkflowExecution.STATUS_OPEN:
            oldest_date = kwargs.pop('oldest_date', 30)
        else:
            # The SWF docs on ListClosedWorkflowExecutions state that:
            #
            #   "startTimeFilter and closeTimeFilter are mutually exclusive"
            #
            # so we must figure out if we have to add a default value for
            # start_oldest_date or not.
            if "close_latest_date" in kwargs or "close_oldest_date" in kwargs:
                default_oldest_date = None
            else:
                default_oldest_date = 30
            oldest_date = kwargs.pop('start_oldest_date', default_oldest_date)

        # Compute a timestamp from the delta in days we got from params
        # If oldest_date is blank at this point, it's because we didn't want
        # it, so let's leave it blank and assume the user provided an other
        # time filter.
        if oldest_date:
            start_oldest_date = int(datetime_timestamp(past_day(oldest_date)))
        else:
            start_oldest_date = None

        return [self.to_WorkflowExecution(self.domain, wfe) for wfe in
                self._list_items(
                    *args,
                    domain=self.domain.name,
                    status=status,
                    workflow_id=workflow_id,
                    workflow_name=workflow_type_name,
                    workflow_version=workflow_type_version,
                    start_oldest_date=start_oldest_date,
                    tag=tag,
                    **kwargs
                )]

    def _list(self, *args, **kwargs):
        return self.list_workflow_executions(*args, **kwargs)

    def all(self, status=WorkflowExecution.STATUS_OPEN,
            start_oldest_date=30,
            *args, **kwargs):
        """Fetch every workflow executions during the last `start_oldest_date`
        days, with `status`

        :param  status: Workflow executions status filter
        :type   status: swf.models.WorkflowExecution.{STATUS_OPEN, STATUS_CLOSED}

        :param  start_oldest_date: Specifies the oldest start/close date to return.
        :type   start_oldest_date: integer (days)

        :returns: workflow executions objects list
        :rtype: list

        A typical amazon response looks like:

        .. code-block:: json

            {
                "executionInfos": [
                    {
                        "cancelRequested": "boolean",
                        "closeStatus": "string",
                        "closeTimestamp": "number",
                        "execution": {
                            "runId": "string",
                            "workflowId": "string"
                        },
                        "executionStatus": "string",
                        "parent": {
                            "runId": "string",
                            "workflowId": "string"
                        },
                        "startTimestamp": "number",
                        "tagList": [
                            "string"
                        ],
                        "workflowType": {
                            "name": "string",
                            "version": "string"
                        }
                    }
                ],
                "nextPageToken": "string"
            }
        """
        start_oldest_date = datetime_timestamp(past_day(start_oldest_date))

        return [self.to_WorkflowExecution(self.domain, wfe) for wfe
                in self._list_items(
                    status,
                    self.domain.name,
                    start_oldest_date=int(start_oldest_date))] class DoesNotExistError(SWFError):
    pass class ResponseError(SWFError):
    pass class Enum(object):
class _CachedProperty(property):
    class immutableclass(mutableclass):
    def __init__(self, *sequential, **named):
    def __contains__(self, key):
    def has_member(self, member):
def get_subkey(d, key_path):
    def __init__(self, fget, fset=None, fdel=None, doc=None):
    def _wrap_fget(self, fget):
        def do_fget(obj):
    def _wrap_fset(self, fset):
        def do_fset(obj, value):
    def _wrap_fdel(self, fdel):
        def do_fdel(obj):
def immutable(mutableclass):
        def __new__(cls, *args, **kw):
        def __init__(self, *args, **kw):    # Prevent re-init after __new__
def camel_to_underscore(string):
def underscore_to_camel(string): def mock_describe_workflow_type(*args, **kwargs):
    override_data = kwargs.pop('override_data', {})

    response = {
        "configuration": {
            "defaultChildPolicy": CHILD_POLICIES.TERMINATE,
            "defaultExecutionStartToCloseTimeout": "300",
            "defaultTaskList": {
                "name": "mocked-tasklist"
            },
            "defaultTaskStartToCloseTimeout": "300"
        },
        "typeInfo": {
            "creationDate": datetime_timestamp(datetime.now()),
            "deprecationDate": datetime_timestamp(datetime.now()),
            "description": "mocked-workflow-type",
            "status": REGISTERED,
            "workflowType": {
                "name": "mocked-workflow-type",
                "version": "0.1"
            }
        }
    }

    response.update(override_data)

    return response def mock_list_workflow_types(*args, **kwargs):
    override_data = kwargs.pop('override_data', {})

    response = {
        "typeInfos": [
            {
                "creationDate": datetime_timestamp(datetime.now()),
                "deprecationDate": datetime_timestamp(datetime.now()),
                "description": "mocked workflow type",
                "status": REGISTERED,
                "workflowType": {
                    "name": "mocked-workflow type",
                    "version": "0.1",
                }
            }
        ]
    }

    response.update(override_data)

    return response def mock_list_open_workflow_executions(*args, **kwargs):
    override_data = kwargs.pop('override_data', {})

    response = {
        "executionInfos": [
            {
                "cancelRequested": False,
                "closeStatus": "mocked",
                "closeTimestamp": datetime_timestamp(datetime.now()),
                "execution": {
                    "runId": "mocked-run-id",
                    "workflowId": "mocked-workflow-id"
                },
                "executionStatus": WorkflowExecution.STATUS_OPEN,
                "parent": {
                    "runId": "mocked-parent-run-id",
                    "workflowId": "mocked-parent-workflow-id"
                },
                "startTimestamp": datetime_timestamp(datetime.now()),
                "tagList": [
                    "mocked-tag-1",
                    "mocked-tag-2",
                    "mocked-tag-3"
                ],
                "workflowType": {
                    "name": "mocked-workflow-type",
                    "version": "0.1"
                }
            }
        ],
    }

    response.update(override_data)

    return response def mock_list_closed_workflow_executions(*args, **kwargs):
    override_data = kwargs.pop('override_data', {})

    response = {
        "executionInfos": [
            {
                "cancelRequested": False,
                "closeStatus": "mocked",
                "closeTimestamp": datetime_timestamp(datetime.now()),
                "execution": {
                    "runId": "mocked-run-id",
                    "workflowId": "mocked-workflow-id"
                },
                "executionStatus": WorkflowExecution.STATUS_CLOSED,
                "parent": {
                    "runId": "mocked-parent-run-id",
                    "workflowId": "mocked-parent-workflow-id"
                },
                "startTimestamp": datetime_timestamp(datetime.now()),
                "tagList": [
                    "mocked-tag-1",
                    "mocked-tag-2"
                ],
                "workflowType": {
                    "name": "mocked-workflow-type",
                    "version": "0.1"
                }
            }
        ],
    }

    response.update(override_data)

    return response def mock_describe_workflow_execution(*args, **kwargs):
    override_data = kwargs.pop('override_data', {})

    response = {
        "executionConfiguration": {
            "childPolicy": CHILD_POLICIES.TERMINATE,
            "executionStartToCloseTimeout": "300",
            "taskList": {
                "name": "mocked-task-list"
            },
            "taskStartToCloseTimeout": "300"
        },
        "executionInfo": {
            "cancelRequested": False,
            "closeStatus": WorkflowExecution.STATUS_CLOSED,
            "closeTimestamp": datetime_timestamp(datetime.now()),
            "execution": {
                "runId": "mocked-run-id",
                "workflowId": "mocked-workflow-id"
            },
            "executionStatus": WorkflowExecution.STATUS_OPEN,
            "parent": {
                "runId": "mocked-run-id",
                "workflowId": "mocked-workflow-id"
            },
            "startTimestamp": datetime_timestamp(datetime.now()),
            "tagList": [
                "mocked-tag-1"
            ],
            "workflowType": {
                "name": "mocked-workflow-type",
                "version": "0.1"
            }
        },
        "latestActivityTaskTimestamp": datetime_timestamp(datetime.now()),
        "latestExecutionContext": "string",
        "openCounts": {
            "openActivityTasks": 12,
            "openChildWorkflowExecutions": 3,
            "openDecisionTasks": 4,
            "openTimers": 5
        }
    }

    response.update(override_data)

    return response
import unittest
from mock import patch, Mock
from boto.swf.layer1 import Layer1
from boto.exception import SWFResponseError
from swf.constants import REGISTERED
from swf.models.domain import Domain
from swf.models.workflow import WorkflowType, WorkflowExecution
from swf.querysets.workflow import BaseWorkflowQuerySet,\
                                   WorkflowTypeQuerySet,\
                                   WorkflowExecutionQuerySet
from swf.exceptions import DoesNotExistError, ResponseError
from swf.utils import datetime_timestamp, past_day
from ..mocks.workflow import mock_describe_workflow_type,\
                             mock_list_workflow_types,\
                             mock_list_open_workflow_executions,\
                             mock_list_closed_workflow_executions,\
                             mock_describe_workflow_execution
        ):
            types = self.wtq.filter(registration_status=REGISTERED)
            self.assertIsNotNone(types)
            self.assertIsInstance(types, list)

            for wt in types:
                self.assertIsInstance(wt, WorkflowType)
                self.assertEqual(wt.status, REGISTERED)

    def test_create_workflow_type(self):
        with patch.object(Layer1, 'register_workflow_type'):
            new_wt = self.wtq.create(
                self.domain,
                "TestWorkflowType",
                "0.test",
            )

            self.assertIsNotNone(new_wt)
            self.assertIsInstance(new_wt, WorkflowType)


class TestWorkflowExecutionQuerySet(unittest.TestCase):

    def setUp(self):
        self.domain = Domain("TestDomain")
        self.wt = WorkflowType(self.domain, "TestType", "0.1")
        self.weq = WorkflowExecutionQuerySet(self.domain)

    def tearDown(self):
        pass

    def test__is_valid_open_status_param(self):
        status = WorkflowExecution.STATUS_OPEN

        self.assertTrue(
            self.weq._is_valid_status_param(status, 'oldest_date'),
        )

    def test__is_invalid_open_status_param(self):
        status = WorkflowExecution.STATUS_OPEN

        self.assertFalse(
            self.weq._is_valid_status_param(status, 'start_oldest_date'),
        )

    def test__is_valid_closed_status_param(self):
        status = WorkflowExecution.STATUS_CLOSED

        self.assertTrue(
            self.weq._is_valid_status_param(status, 'start_oldest_date'),
        )

    def test__is_invalid_closed_status_param(self):
        status = WorkflowExecution.STATUS_CLOSED

        self.assertFalse(
            self.weq._is_valid_status_param(status, 'oldest_date'),
        )

    def test_validate_valid_open_status_parameters(self):
        params = ['oldest_date', 'latest_date']
        status = WorkflowExecution.STATUS_OPEN

        self.assertEqual(
            self.weq._validate_status_parameters(status, params),
            []
        )

    def test_validate_invalid_open_status_parameters(self):
        params = ['oldest_date', 'start_latest_date']
        status = WorkflowExecution.STATUS_OPEN

        self.assertEqual(
            self.weq._validate_status_parameters(status, params),
            ['start_latest_date']
        )

    def test_validate_valid_closed_status_parameters(self):
        params = ['start_oldest_date', 'start_latest_date']
        status = WorkflowExecution.STATUS_CLOSED

        self.assertEqual(
            self.weq._validate_status_parameters(status, params),
            []
        )

    def test_validate_invalid_closed_status_parameters(self):
        params = ['oldest_date', 'start_latest_date']
        status = WorkflowExecution.STATUS_CLOSED

        self.assertEqual(
            self.weq._validate_status_parameters(status, params),
            ['oldest_date']
        )

    def test_list_open_workflows_executions_with_start_oldest_date(self):
        with patch.object(
            self.weq.connection,
            'list_open_workflow_executions',
            mock_list_open_workflow_executions
        ):
            we = self.weq.list_workflow_executions(
                WorkflowExecution.STATUS_OPEN,
                self.domain.name,
                start_oldest_date=int(datetime_timestamp(past_day(3)))
            )
            self.assertIsNotNone(we)
            self.assertIsInstance(we, dict)

            self.assertTrue(we['executionInfos'][0]['executionStatus'] == WorkflowExecution.STATUS_OPEN)

    def test_list_closed_workflows_executions(self):
        with patch.object(
            self.weq.connection,
            'list_closed_workflow_executions',
            mock_list_closed_workflow_executions
        ):
            we = self.weq.list_workflow_executions(
                WorkflowExecution.STATUS_CLOSED,
                self.domain.name,

                start_oldest_date=int(datetime_timestamp(past_day(3)))