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

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 atheris 

28import colorlog 

29from datetime import datetime, timedelta 

30with atheris.instrument_imports(): 

31 import airflow 

32 from airflow import DAG 

33 from airflow.exceptions import AirflowException 

34 from airflow.operators.dummy_operator import DummyOperator 

35 from airflow.operators.python_operator import PythonOperator 

36 

37def py_func(): 

38 return 

39 

40def TestInput(input_bytes): 

41 fdp = atheris.FuzzedDataProvider(input_bytes) 

42 

43 default_args = { 

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

45 'depends_on_past': fdp.ConsumeBool(), 

46 'start_date': airflow.utils.dates.days_ago(fdp.ConsumeIntInRange(1,5)), 

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

48 'email_on_failure': fdp.ConsumeBool(), 

49 'email_on_retry': fdp.ConsumeBool(), 

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

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

52 } 

53 

54 try: 

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

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

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

58 

59 dummy_task >> python_task 

60 except AirflowException: 

61 pass 

62 

63def main(): 

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

65 atheris.Fuzz() 

66 

67if __name__ == "__main__": 

68 main()