Coverage for /pythoncovmergedfiles/medio/medio/src/fuzz_avroparser.py: 33%

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# Copyright 2023 Google LLC 

14# 

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

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

17# You may obtain a copy of the License at 

18# 

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

20# 

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

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

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

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

25# limitations under the License. 

26import sys 

27import atheris 

28 

29from google.cloud.bigquery_storage_v1.types import ReadSession 

30from google.protobuf.json_format import Parse 

31 

32 

33def TestOneInput(data): 

34 """Fuzzer that creates a random read session and parses 

35 the attached avro session. The main point is to ensure 

36 none of the parsing routines have insecure calls and 

37 that the parsing routines can work with arbitrary ReadSessions. 

38 """ 

39 fdp = atheris.FuzzedDataProvider(data) 

40 try: 

41 # Create a random session 

42 msg = Parse(fdp.ConsumeUnicodeNoSurrogates(sys.maxsize), ReadSession()) 

43 except Exception: 

44 return 

45 

46 avro_parser = _AvroStreamParser(msg) 

47 avro_parser.to_dataframe() 

48 

49 

50def main(): 

51 atheris.instrument_all() 

52 atheris.Setup(sys.argv, TestOneInput, enable_python_coverage=True) 

53 atheris.Fuzz() 

54 

55 

56if __name__ == "__main__": 

57 main()