@ -19,18 +19,86 @@
import logging
import re
from unittest . mock import patch
from typing import Set
from unittest . mock import patch
import pytest
from pydolphinscheduler . core . process_definition import ProcessDefinition
from pydolphinscheduler . core . task import Task , TaskRelation
from tests . testing . task import Task as t estTask
from tests . testing . task import Task as T estTask
from tests . testing . task import TaskWithCode
TEST_TASK_RELATION_SET = set ( )
TEST_TASK_RELATION_SIZE = 0
@pytest . mark . parametrize (
" addition, ignore, expect " ,
[
(
set ( ) ,
set ( ) ,
{
" local_params " ,
" resource_list " ,
" dependence " ,
" wait_start_timeout " ,
" condition_result " ,
} ,
) ,
(
set ( ) ,
{ " dependence " , " condition_result " , " not_exists " } ,
{
" local_params " ,
" resource_list " ,
" wait_start_timeout " ,
} ,
) ,
(
{
" not_exists_1 " ,
" not_exists_2 " ,
} ,
set ( ) ,
{
" not_exists_1 " ,
" not_exists_2 " ,
" local_params " ,
" resource_list " ,
" dependence " ,
" wait_start_timeout " ,
" condition_result " ,
} ,
) ,
# test addition and ignore conflict to see behavior
(
{
" not_exists " ,
} ,
{ " condition_result " , " not_exists " } ,
{
" not_exists " ,
" local_params " ,
" resource_list " ,
" dependence " ,
" wait_start_timeout " ,
} ,
) ,
] ,
)
def test__get_attr ( addition : Set , ignore : Set , expect : Set ) :
""" Test task function `_get_attr`. """
task = TestTask (
name = " test-get-attr " ,
task_type = " test " ,
)
task . _task_custom_attr = addition
task . _task_ignore_attr = ignore
assert task . _get_attr ( ) == expect
@pytest . mark . parametrize (
" attr, expect " ,
[
@ -72,7 +140,7 @@ TEST_TASK_RELATION_SIZE = 0
)
def test_property_task_params ( mock_resource , mock_user_name , attr , expect ) :
""" Test class task property. """
task = t estTask(
task = T estTask(
" test-property-task-params " ,
" test-task " ,
* * attr ,
@ -182,8 +250,8 @@ def test_two_tasks_shift(shift: str):
Here we test both ` >> ` and ` << ` bit operator .
"""
upstream = t estTask( name = " upstream " , task_type = shift )
downstream = t estTask( name = " downstream " , task_type = shift )
upstream = T estTask( name = " upstream " , task_type = shift )
downstream = T estTask( name = " downstream " , task_type = shift )
if shift == " << " :
downstream << upstream
elif shift == " >> " :
@ -219,10 +287,10 @@ def test_tasks_list_shift(dep_expr: str, flag: str):
" downstream " : " upstream " ,
}
task_type = " dep_task_and_tasks "
task = t estTask( name = " upstream " , task_type = task_type )
task = T estTask( name = " upstream " , task_type = task_type )
tasks = [
t estTask( name = " downstream1 " , task_type = task_type ) ,
t estTask( name = " downstream2 " , task_type = task_type ) ,
T estTask( name = " downstream1 " , task_type = task_type ) ,
T estTask( name = " downstream2 " , task_type = task_type ) ,
]
# Use build-in function eval to simply test case and reduce duplicate code