tests/test_iworkflow/test_iworkflow1.py

0001 #!/usr/bin/env python
0002 #
0003 # Licensed under the Apache License, Version 2.0 (the "License");
0004 # You may not use this file except in compliance with the License.
0005 # You may obtain a copy of the License at
0006 # http://www.apache.org/licenses/LICENSE-2.0OA
0007 #
0008 # Authors:
0009 # - Wen Guan, <wen.guan@cern.ch>, 2024
0010
0011
0012 """
0013 Test workflow.
0014 """
0015
0016 import datetime
0017 import inspect
0018
0019 # import unittest2 as unittest
0020 # from nose.tools import assert_equal
0021 from idds.common.utils import setup_logging
0022
0023
0024 from idds.iworkflow.work import work
0025 from idds.iworkflow.workflow import workflow
0026
0027 # from idds.iworkflow.utils import perform_workflow, run_workflow
0028
0029 from idds.common.utils import json_dumps, run_process, encode_base64
0030
0031
0032 setup_logging(__name__)
0033
0034
0035 @work
0036 def test_func(name):
0037     print('test_func starts')
0038     print(name)
0039     print('test_func ends')
0040     return 'test result: %s' % name
0041
0042
0043 def test_workflow():
0044     print("test workflow starts")
0045     ret = test_func(name='idds')
0046     print(ret)
0047     print("test workflow ends")
0048
0049
0050 @work
0051 def get_params():
0052     list_params = [i for i in range(10)]
0053     return list_params
0054
0055
0056 def test_workflow_mulitple_work():
0057     print("test workflow multiple work starts")
0058     list_params = get_params()
0059
0060     ret = test_func(list_params)
0061     print(ret)
0062     print("test workflow multiple work ends")
0063
0064
0065 def submit_workflow(workflow):
0066     req_id = workflow.submit()
0067     print("req id: %s" % req_id)
0068
0069
0070 def perform_workflow_wrapper(workflow):
0071     # ret = perform_workflow(workflow)
0072     # print(ret)
0073     setup = workflow.setup()
0074     cmd = setup + ";"
0075     return cmd
0076
0077
0078 def run_workflow_wrapper(workflow):
0079     setup = workflow.setup()
0080
0081     cmd = setup + "; perform_workflow --workflow " + encode_base64(json_dumps(workflow))
0082     print(f'To run workflow: {cmd}')
0083     exit_code = run_process(cmd)
0084     print(f'Run workflow finished with exit code: {exit_code}')
0085     return exit_code
0086
0087
0088 @workflow
0089 def test_workflow1():
0090     print("test workflow starts")
0091     # ret = test_func(name='idds')
0092     # print(ret)
0093     print("test workflow ends")
0094
0095
0096 def test_workflow2():
0097     print("test workflow2 starts")
0098     # ret = test_func(name='idds')
0099     # print(ret)
0100     print("test workflow2 ends")
0101
0102
0103 if __name__ == '__main__':
0104     print("datetime.datetime object")
0105     f = datetime.datetime(2019, 10, 10, 10, 10)
0106     print(dir(f))
0107     print(inspect.getmodule(f))
0108     # print(inspect.getfile(f))
0109     # print(inspect.getmodule(f).__name__)
0110     # print(inspect.getmodule(f).__file__)
0111     # print(inspect.signature(f))
0112
0113     print("datetime.datetime function")
0114     f = datetime.datetime
0115     print(dir(f))
0116     print(f.__module__)
0117     print(f.__name__)
0118     print(inspect.getmodule(f))
0119     print(inspect.getfile(f))
0120     print(inspect.getmodule(f).__name__)
0121     print(inspect.getmodule(f).__file__)
0122     # print(inspect.signature(f))
0123
0124     print("test_workflow function")
0125     f = test_workflow
0126     print(dir(f))
0127     print(f.__module__)
0128     print(f.__name__)
0129     print(inspect.getmodule(f))
0130     print(inspect.getfile(f))
0131     print(inspect.getmodule(f).__name__)
0132     print(inspect.getmodule(f).__file__)
0133     print(inspect.signature(f))
0134
0135     print("test_workflow1")
0136     test_workflow1()