Coverage for /pythoncovmergedfiles/medio/medio/src/dulwich/fuzzing/fuzz-targets/fuzz_object_store.py: 66%

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# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later 

13 

14import stat 

15import sys 

16from io import BytesIO 

17from typing import Optional 

18 

19import atheris 

20 

21with atheris.instrument_imports(): 

22 # We instrument `test_utils` as well, so it doesn't block coverage analysis in Fuzz Introspector: 

23 from test_utils import EnhancedFuzzedDataProvider, is_expected_exception 

24 

25 from dulwich.errors import ObjectFormatException 

26 from dulwich.objects import S_IFGITLINK, Blob, Commit, Tree 

27 from dulwich.patch import write_tree_diff 

28 from dulwich.repo import ( 

29 InvalidUserIdentity, 

30 MemoryRepo, 

31 ) 

32 

33 

34def TestOneInput(data) -> Optional[int]: 

35 fdp = EnhancedFuzzedDataProvider(data) 

36 repo = MemoryRepo() 

37 blob = Blob.from_string(fdp.ConsumeRandomBytes()) 

38 tree = Tree() 

39 tree.add( 

40 fdp.ConsumeRandomBytes(), 

41 fdp.PickValueInList([stat.S_IFREG, stat.S_IFLNK, stat.S_IFDIR, S_IFGITLINK]), 

42 blob.id, 

43 ) 

44 commit = Commit() 

45 commit.tree = tree.id 

46 commit.author = fdp.ConsumeRandomBytes() 

47 commit.committer = fdp.ConsumeRandomBytes() 

48 commit.commit_time = fdp.ConsumeRandomInt() 

49 commit.commit_timezone = fdp.ConsumeRandomInt() 

50 commit.author_time = fdp.ConsumeRandomInt() 

51 commit.author_timezone = fdp.ConsumeRandomInt() 

52 commit.message = fdp.ConsumeRandomBytes() 

53 

54 object_store = repo.object_store 

55 

56 try: 

57 object_store.add_object(blob) 

58 object_store.add_object(tree) 

59 object_store.add_object(commit) 

60 except (InvalidUserIdentity, ObjectFormatException): 

61 return -1 

62 except ValueError as e: 

63 expected_exceptions = [ 

64 "subsection not found", 

65 "Unable to handle non-minute offset", 

66 ] 

67 if is_expected_exception(expected_exceptions, e): 

68 return -1 

69 else: 

70 raise e 

71 

72 commit2 = Commit() 

73 commit2.tree = tree.id 

74 commit2.parents = [commit.id] 

75 commit2.author = commit.author 

76 commit2.committer = commit.committer 

77 commit2.commit_time = fdp.ConsumeRandomInt() 

78 commit2.commit_timezone = fdp.ConsumeRandomInt() 

79 commit2.author_time = fdp.ConsumeRandomInt() 

80 commit2.author_timezone = fdp.ConsumeRandomInt() 

81 commit2.message = fdp.ConsumeRandomBytes() 

82 

83 try: 

84 blob.data = fdp.ConsumeRandomBytes() 

85 repo.object_store.add_object(blob) 

86 repo.object_store.add_object(tree) 

87 repo.object_store.add_object(commit2) 

88 out = BytesIO() 

89 write_tree_diff(out, repo.object_store, commit.tree, tree.id) 

90 except (InvalidUserIdentity, ObjectFormatException): 

91 return -1 

92 except ValueError as e: 

93 expected_exceptions = [ 

94 "Unable to handle non-minute offset", 

95 ] 

96 if is_expected_exception(expected_exceptions, e): 

97 return -1 

98 else: 

99 raise e 

100 

101 

102def main() -> None: 

103 atheris.Setup(sys.argv, TestOneInput) 

104 atheris.Fuzz() 

105 

106 

107if __name__ == "__main__": 

108 main()