Coverage for /pythoncovmergedfiles/medio/medio/src/fuzz_bn.py: 2%

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"""Scalar equivalene checker""" 

27 

28import os 

29import sys 

30import atheris 

31 

32import numpy as np 

33from numpy.testing import assert_array_almost_equal 

34import bottleneck as bn 

35 

36 

37def gen_random_array(data, fdp = None): 

38 if fdp is None: 

39 fdp = atheris.FuzzedDataProvider(data) 

40 l1 = list() 

41 for i in range(fdp.ConsumeIntInRange(5, 3000)): 

42 l1.append(fdp.ConsumeIntInRange(1,10000)) 

43 a = np.array(l1) 

44 return a 

45 

46def TestOneInput(data): 

47 """Tests scalar equivalence and also move operations""" 

48 fdp = atheris.FuzzedDataProvider(data) 

49 a = gen_random_array(data, fdp) 

50 

51 func_pairs = [ 

52 (bn.nansum, bn.slow.nansum), 

53 (bn.nanmean, bn.slow.nanmean), 

54 (bn.nanstd, bn.slow.nanstd), 

55 (bn.nanvar, bn.slow.nanvar), 

56 (bn.nanmin, bn.slow.nanmin), 

57 (bn.median, bn.slow.median), 

58 (bn.nanmedian, bn.slow.nanmedian), 

59 (bn.ss, bn.slow.ss), 

60 (bn.nanargmin, bn.slow.nanargmin), 

61 (bn.nanargmax, bn.slow.nanargmax), 

62 (bn.anynan, bn.slow.anynan), 

63 (bn.allnan, bn.slow.allnan), 

64 ] 

65 

66 idx = 0 

67 for func0, func1 in func_pairs: 

68 idx = idx + 1 

69 actual = func0(a) 

70 desired = func1(a) 

71 assert_array_almost_equal( 

72 actual, 

73 desired, 

74 err_msg="Failed scalar equivalence" 

75 ) 

76 

77 

78 # Test move operations 

79 window = fdp.ConsumeIntInRange(2, 50) 

80 min_count = fdp.ConsumeIntInRange(1, window) 

81 try: 

82 actual = bn.move_median( 

83 a, 

84 window=window, 

85 min_count = fdp.ConsumeIntInRange(1,100) 

86 ) 

87 except ValueError: 

88 return 

89 try: 

90 desired = bn.slow.move_median( 

91 a, 

92 window=window, 

93 min_count=fdp.ConsumeIntInRange(1, 100) 

94 ) 

95 except ValueError: 

96 return 

97 assert_array_almost_equal( 

98 actual, 

99 desired, 

100 err_msg="Failed move operation" 

101 ) 

102 

103 

104def main(): 

105 atheris.instrument_all() 

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

107 atheris.Fuzz() 

108 

109if __name__ == "__main__": 

110 main()