Coverage for /pythoncovmergedfiles/medio/medio/src/fuzz_uploader.py: 60%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
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.
27import atheris
28import sys
29import io
31# All code is instrumented by calling atheris.instrument_all below
32from google.resumable_media import _upload
34def test_simple(data):
35 upload = _upload.SimpleUpload("localhost:8008/index.html")
36 upload._prepare_request(data, None)
38def test_multipart(data):
39 fdp = atheris.FuzzedDataProvider(data)
40 headers = {"spin": "doctors"}
41 upload = _upload.MultipartUpload(
42 "localhost:8008/index.html",
43 headers=headers,
44 checksum="md5"
45 )
46 try:
47 upload._prepare_request(
48 fdp.ConsumeBytes(200),
49 {},
50 fdp.ConsumeString(200)
51 )
52 except UnicodeEncodeError:
53 pass
55def _upload_in_flight(data, headers=None, checksum=None):
56 upload = _upload.ResumableUpload(
57 "localhost:8008://heyo",
58 1024 * 1024,
59 headers=headers,
60 checksum=checksum
61 )
62 upload._stream = io.BytesIO(data)
63 upload._content_type = "text/plain"
64 upload._total_bytes = len(data)
65 upload._resumable_url = "localhost:8009/test.invalid?upload_id=not-none"
66 return upload
69def test_checksum(data):
70 fdp = atheris.FuzzedDataProvider(data)
71 for cs in ["md5", "crc32"]:
72 upload = _upload_in_flight(fdp.ConsumeBytes(200), checksum=cs)
73 try:
74 start_byte, payload, _ = _upload.get_next_chunk(
75 upload._stream,
76 fdp.ConsumeIntInRange(1, 250),
77 fdp.ConsumeIntInRange(1, 250)
78 )
79 except ValueError:
80 pass
83def TestOneInput(data):
84 test_simple(data)
85 test_multipart(data)
86 test_checksum(data)
88def main():
89 atheris.instrument_all()
90 atheris.Setup(sys.argv, TestOneInput, enable_python_coverage=True)
91 atheris.Fuzz()
93if __name__ == "__main__":
94 main()