File size: 1,524 Bytes
85b7ac8 c8f3b98 85b7ac8 c8f3b98 85b7ac8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 | """Task registry for the environment."""
from typing import Dict, Type
from server.tasks.base import BaseTask
from server.tasks.task_1_build_errors import DockerfileSyntaxTask
from server.tasks.task_2_docker_runtime import DockerfileRuntimeTask
from server.tasks.task_3_workflow_syntax import WorkflowSyntaxStructureTask
from server.tasks.task_4_workflow_secrets_permissions import WorkflowSecretsPermissionsTask
from server.tasks.task_5_ci_docker_integration import CIDockerIntegrationTask
from server.tasks.task_6_multi_stage_matrix import MultiStageMatrixTask
from server.tasks.k8s_pod import K8sPodTask
from server.tasks.k8s_networking import K8sNetworkingTask
from server.tasks.pipeline_build_deploy import PipelineBuildDeployTask
from server.tasks.pipeline_full import PipelineFullTask
TASK_REGISTRY: Dict[str, Type[BaseTask]] = {
"dockerfile_syntax": DockerfileSyntaxTask,
"dockerfile_runtime": DockerfileRuntimeTask,
"workflow_syntax_structure": WorkflowSyntaxStructureTask,
"workflow_secrets_permissions": WorkflowSecretsPermissionsTask,
"ci_docker_integration": CIDockerIntegrationTask,
"multi_stage_pipeline_matrix": MultiStageMatrixTask,
"k8s_pod_failures": K8sPodTask,
"k8s_networking": K8sNetworkingTask,
"pipeline_build_deploy": PipelineBuildDeployTask,
"pipeline_full_stack": PipelineFullTask,
}
def get_task(task_id: str) -> BaseTask:
if task_id not in TASK_REGISTRY:
raise ValueError(f"Unknown task: {task_id}")
return TASK_REGISTRY[task_id]()
|