fixtures.py

from collections.abc import Generator
from typing import Any
from uuid import uuid4

import pytest
from orchestrator import step, workflow
from orchestrator.config.assignee import Assignee
from orchestrator.forms import FormPage
from orchestrator.types import UUIDstr
from orchestrator.workflow import done, init, inputstep
from pydantic_forms.types import FormGenerator
from pydantic_forms.validators import Choice

from test.workflows import WorkflowInstanceForTests


@pytest.fixture()
def test_workflow(generic_subscription_1: UUIDstr, generic_product_type_1) -> Generator:
    _, generic_product_one = generic_product_type_1

    @step("Insert UUID in state")
    def insert_object():
        return {"subscription_id": str(uuid4()), "model": generic_product_one.from_subscription(generic_subscription_1)}

    @step("Test that it is a string now")
    def check_object(subscription_id: Any, model: dict) -> None:
        # This is actually a test. It would be nicer to have this in a proper test but that takes to much setup that
        # already happens here. So we hijack this fixture and run this test for all tests that use this fixture
        # (which should not be an issue)
        assert isinstance(subscription_id, str)
        assert isinstance(model, dict)

    @inputstep("Modify", assignee=Assignee.CHANGES)
    def modify(subscription_id: UUIDstr) -> FormGenerator:
        class TestChoice(Choice):
            A = "A"
            B = "B"
            C = "C"

        class TestForm(FormPage):
            generic_select: TestChoice

        user_input = yield TestForm
        return user_input.model_dump()

    @workflow("Workflow")
    def workflow_for_testing_processes_py():
        return init >> insert_object >> check_object >> modify >> done

    with WorkflowInstanceForTests(workflow_for_testing_processes_py, "workflow_for_testing_processes_py") as wf:
        yield wf