Coverage for /pythoncovmergedfiles/medio/medio/src/dag_fuzz.py: 20%

1###### Coverage stub 

2import atexit 

3import coverage 

4cov = coverage.coverage(data_file='.coverage', cover_pylib=True) 

5cov.start() 

6# Register an exist handler that will print coverage 

7def exit_handler(): 

8 cov.stop() 

9 cov.save() 

10atexit.register(exit_handler) 

11####### End of coverage stub 

12#!/usr/bin/python3 

13 

14# Copyright 2022 Google LLC 

15# 

16# Licensed under the Apache License, Version 2.0 (the "License"); 

17# you may not use this file except in compliance with the License. 

18# You may obtain a copy of the License at 

19# 

20# http://www.apache.org/licenses/LICENSE-2.0 

21# 

22# Unless required by applicable law or agreed to in writing, software 

23# distributed under the License is distributed on an "AS IS" BASIS, 

24# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

25# See the License for the specific language governing permissions and 

26# limitations under the License. 

27import sys 

28import atheris 

29import colorlog 

30from datetime import datetime, timedelta 

31 

32with atheris.instrument_imports(include=['airflow'], enable_loader_override=False): 

33 import airflow 

34 from airflow import DAG 

35 from airflow.exceptions import AirflowException 

36 

37 try: 

38 from airflow.providers.standard.operators.empty import EmptyOperator as DummyOperator 

39 from airflow.providers.standard.operators.python import PythonOperator 

40 except ImportError: 

41 try: 

42 from airflow.operators.empty import EmptyOperator as DummyOperator 

43 from airflow.operators.python import PythonOperator 

44 except ImportError: 

45 from airflow.operators.dummy_operator import DummyOperator 

46 from airflow.operators.python_operator import PythonOperator 

47 

48def py_func(): 

49 return 

50 

51def TestInput(input_bytes): 

52 fdp = atheris.FuzzedDataProvider(input_bytes) 

53 

54 default_args = { 

55 'owner': fdp.ConsumeString(8), 

56 'depends_on_past': fdp.ConsumeBool(), 

57 'start_date': datetime.now() - timedelta(days=fdp.ConsumeIntInRange(1,5)), 

58 'email': [fdp.ConsumeString(8)], 

59 'email_on_failure': fdp.ConsumeBool(), 

60 'email_on_retry': fdp.ConsumeBool(), 

61 'retries': fdp.ConsumeIntInRange(1,5), 

62 'retry_delay': timedelta(minutes=fdp.ConsumeIntInRange(1,5)), 

63 } 

64 

65 try: 

66 with DAG(fdp.ConsumeString(8), schedule_interval='@daily', default_args=default_args) as dag: 

67 dummy_task = DummyOperator(task_id=fdp.ConsumeString(8), retries=fdp.ConsumeIntInRange(1,5)) 

68 python_task = PythonOperator(task_id=fdp.ConsumeString(8), python_callable=py_func) 

69 

70 dummy_task >> python_task 

71 except (AirflowException, ValueError, TypeError): 

72 pass 

73 

74def main(): 

75 atheris.Setup(sys.argv, TestInput, enable_python_coverage=True) 

76 atheris.Fuzz() 

77 

78if __name__ == "__main__": 

79 main()