Coverage for /pythoncovmergedfiles/medio/medio/src/fuzz_looker.py: 43%

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 2022 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. 

26"""Targetting the looker-sdk""" 

27 

28import os 

29import sys 

30import json 

31import mock 

32import atheris 

33import looker_sdk 

34import requests 

35from looker_sdk import models40 as models 

36 

37 

38# Sample configuration. We use mock to handle network communication so 

39# these fields are not important. 

40conf = """[Looker] 

41api_versions=3.1,4.0 

42base_url=https://localhost:19999 

43client_id=your_API3_client_id 

44client_secret=your_API3_client_secret 

45verify_ssl=true 

46timeout=2""" 

47return_dict = None 

48def transport_requests_mock( 

49 method, 

50 path, 

51 query_params=None, 

52 body=None, 

53 authenticator=None, 

54 transport_options=None 

55): 

56 """Returns an arbitrary dictionary string""" 

57 global return_dict 

58 r1 = looker_sdk.rtl.transport.Response(True, return_dict, 3, "utf-8") 

59 return r1 

60 

61@atheris.instrument_func 

62def TestOneInput(data): 

63 global return_dict 

64 fdp = atheris.FuzzedDataProvider(data) 

65 s1 = fdp.ConsumeString(fdp.ConsumeIntInRange(0, 1024)) 

66 try: 

67 fuzzed_dict = json.loads(s1) 

68 except: 

69 return 

70 if type(fuzzed_dict) is not dict: 

71 return 

72 

73 return_dict = str.encode(s1) 

74 

75 with open("looker.ini", "w") as f: 

76 f.write(conf) 

77 sdk = looker_sdk.init40("looker.ini") 

78 m1 = fdp.ConsumeUnicodeNoSurrogates(20) 

79 v1 = fdp.ConsumeUnicodeNoSurrogates(20) 

80 l1 = [fdp.ConsumeUnicodeNoSurrogates(20)] 

81 

82 patch = mock.patch( 

83 "looker_sdk.rtl.requests_transport.RequestsTransport.request", 

84 wraps=transport_requests_mock 

85 ) 

86 # Perform a set of operations 

87 op_count = fdp.ConsumeIntInRange(1, 10) 

88 with patch: 

89 for i in range(op_count): 

90 op = fdp.ConsumeIntInRange(1,3) 

91 if op == 1: 

92 try: 

93 query = sdk.create_query( 

94 body=models.WriteQuery(model=m1, view=v1, fields=l1) 

95 ) 

96 except looker_sdk.rtl.serialize.DeserializeError: 

97 pass 

98 elif op == 2: 

99 try: 

100 query = sdk.create_query_task( 

101 body = models.WriteCreateQueryTask( 

102 query_id=fdp.ConsumeUnicodeNoSurrogates(10), 

103 source=fdp.ConsumeUnicodeNoSurrogates(10), 

104 results_format=looker_sdk.sdk.api40.ResultFormat.csv 

105 ) 

106 ) 

107 except looker_sdk.rtl.serialize.DeserializeError: 

108 pass 

109 

110 

111def main(): 

112 atheris.instrument_all() 

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

114 atheris.Fuzz() 

115 

116if __name__ == "__main__": 

117 main()