Git-Mirrors/onionshare
1
0
Fork 0
mirror of https://github.com/onionshare/onionshare.git synced 2024-12-24 06:49:44 -05:00
Code Issues Packages Projects Releases Wiki Activity

Fix flake8 issues found in cli

Browse Source
This commit is contained in:
Micah Lee 2021-04-29 17:13:05 -07:00
parent 343e1be3d0
commit 0a6056e5e6
No known key found for this signature in database
GPG Key ID: 403C2657CD994F73
14 changed files with 146 additions and 128 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
cli/onionshare_cli/__init__.py
View File

@ -18,13 +18,22 @@ You should have received a copy of the GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
"""
import os, sys, time, argparse, threading
import os
import sys
import time
import argparse
import threading
from datetime import datetime
from datetime import timedelta
from .common import Common, CannotFindTor
from .web import Web
from .onion import *
from .onion import (
TorErrorProtocolError,
TorTooOldEphemeral,
TorTooOldStealth,
Onion,
)
from .onionshare import OnionShare
from .mode_settings import ModeSettings
@ -310,7 +319,7 @@ def main(cwd=None):
# In receive mode, you must allows either text, files, or both
if mode == "receive" and disable_text and disable_files:
print(f"You cannot disable both text and files")
print("You cannot disable both text and files")
sys.exit()
# Create the Web object
@ -336,7 +345,7 @@ def main(cwd=None):
except KeyboardInterrupt:
print("")
sys.exit()
except Exception as e:
except Exception:
sys.exit()
# Start the onionshare app
@ -380,7 +389,9 @@ def main(cwd=None):
)
print("")
print(
"Warning: Receive mode lets people upload files to your computer. Some files can potentially take control of your computer if you open them. Only open things from people you trust, or if you know what you are doing."
"Warning: Receive mode lets people upload files to your computer. Some files can potentially take "
"control of your computer if you open them. Only open things from people you trust, or if you know "
"what you are doing."
)
print("")
if mode_settings.get("general", "client_auth"):
@ -474,7 +485,9 @@ def main(cwd=None):
)
print("")
print(
"Warning: Receive mode lets people upload files to your computer. Some files can potentially take control of your computer if you open them. Only open things from people you trust, or if you know what you are doing."
"Warning: Receive mode lets people upload files to your computer. Some files can potentially take "
"control of your computer if you open them. Only open things from people you trust, or if you know "
"what you are doing."
)
print("")

4
cli/onionshare_cli/common.py
View File

@ -383,7 +383,7 @@ class Common:
try:
xdg_config_home = os.environ["XDG_CONFIG_HOME"]
onionshare_data_dir = f"{xdg_config_home}/onionshare"
except:
except Exception:
onionshare_data_dir = os.path.expanduser("~/.config/onionshare")
elif self.platform == "Darwin":
onionshare_data_dir = os.path.expanduser(
@ -393,7 +393,7 @@ class Common:
try:
xdg_config_home = os.environ["XDG_CONFIG_HOME"]
onionshare_data_dir = f"{xdg_config_home}/onionshare"
except:
except Exception:
onionshare_data_dir = os.path.expanduser("~/.config/onionshare")
# Modify the data dir if running tests

2
cli/onionshare_cli/mode_settings.py
View File

@ -129,7 +129,7 @@ class ModeSettings:
self.fill_in_defaults()
self.common.log("ModeSettings", "load", f"loaded {self.filename}")
return
except:
except Exception:
pass
# If loading settings didn't work, create the settings file

34
cli/onionshare_cli/onion.py
View File

@ -222,7 +222,7 @@ class Onion(object):
)
try:
self.tor_socks_port = self.common.get_available_port(1000, 65535)
except:
except Exception:
print("OnionShare port not available")
raise PortNotAvailable()
self.tor_torrc = os.path.join(self.tor_data_directory_name, "torrc")
@ -244,7 +244,7 @@ class Onion(object):
proc.terminate()
proc.wait()
break
except:
except Exception:
pass
if self.common.platform == "Windows" or self.common.platform == "Darwin":
@ -255,7 +255,7 @@ class Onion(object):
torrc_template += "ControlPort {{control_port}}\n"
try:
self.tor_control_port = self.common.get_available_port(1000, 65535)
except:
except Exception:
print("OnionShare port not available")
raise PortNotAvailable()
self.tor_control_socket = None
@ -428,7 +428,7 @@ class Onion(object):
try:
self.c = Controller.from_port(port=int(env_port))
found_tor = True
except:
except Exception:
pass
else:
@ -438,7 +438,7 @@ class Onion(object):
for port in ports:
self.c = Controller.from_port(port=port)
found_tor = True
except:
except Exception:
pass
# If this still didn't work, try guessing the default socket file path
@ -452,7 +452,7 @@ class Onion(object):
self.c = Controller.from_socket_file(path=socket_file_path)
found_tor = True
except:
except Exception:
pass
# If connecting to default control ports failed, so let's try
@ -474,14 +474,14 @@ class Onion(object):
self.c = Controller.from_socket_file(path=socket_file_path)
except:
except Exception:
print(automatic_error)
raise TorErrorAutomatic()
# Try authenticating
try:
self.c.authenticate()
except:
except Exception:
print(automatic_error)
raise TorErrorAutomatic()
@ -504,7 +504,7 @@ class Onion(object):
print(invalid_settings_error)
raise TorErrorInvalidSetting()
except:
except Exception:
if self.settings.get("connection_type") == "control_port":
print(
"Can't connect to the Tor controller at {}:{}.".format(
@ -582,7 +582,7 @@ class Onion(object):
tmp_service_id = res.service_id
self.c.remove_ephemeral_hidden_service(tmp_service_id)
self.supports_stealth = True
except:
except Exception:
# ephemeral stealth onion services are not supported
self.supports_stealth = False
@ -708,7 +708,7 @@ class Onion(object):
self.c.remove_ephemeral_hidden_service(
mode_settings.get("general", "service_id")
)
except:
except Exception:
self.common.log(
"Onion", "stop_onion_service", f"failed to remove {onion_host}"
)
@ -729,12 +729,12 @@ class Onion(object):
"Onion", "cleanup", f"trying to remove onion {onion_host}"
)
self.c.remove_ephemeral_hidden_service(service_id)
except:
except Exception:
self.common.log(
"Onion", "cleanup", f"failed to remove onion {onion_host}"
)
pass
except:
except Exception:
pass
if stop_tor:
@ -777,7 +777,7 @@ class Onion(object):
)
symbols_i = (symbols_i + 1) % len(symbols)
time.sleep(1)
except:
except Exception:
pass
self.tor_proc.terminate()
@ -797,7 +797,7 @@ class Onion(object):
"cleanup",
"Tried to kill tor process but it's still running",
)
except:
except Exception:
self.common.log(
"Onion", "cleanup", "Exception while killing tor process"
)
@ -810,7 +810,7 @@ class Onion(object):
# Delete the temporary tor data directory
if self.use_tmp_dir:
self.tor_data_directory.cleanup()
except:
except Exception:
pass
def get_tor_socks_port(self):
@ -835,5 +835,5 @@ class Onion(object):
key = RSA.importKey(base64.b64decode(key))
# Is this a v2 Onion key? (1024 bits) If so, we should keep using it.
return key.n.bit_length() == 1024
except:
except Exception:
return False

7
cli/onionshare_cli/onionshare.py
View File

@ -18,7 +18,8 @@ You should have received a copy of the GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
"""
import os, shutil
import os
import shutil
from .common import AutoStopTimer
@ -57,7 +58,7 @@ class OnionShare(object):
"""
try:
self.port = self.common.get_available_port(17600, 17650)
except:
except Exception:
raise OSError("Cannot find an available OnionShare port")
def start_onion_service(self, mode, mode_settings, await_publication=True):
@ -102,7 +103,7 @@ class OnionShare(object):
os.remove(filename)
elif os.path.isdir(filename):
shutil.rmtree(filename)
except:
except Exception:
# Don't crash if file is still in use
pass
self.cleanup_filenames = []

12
cli/onionshare_cli/settings.py
View File

@ -22,12 +22,6 @@ import json
import os
import locale
try:
# We only need pwd module in macOS, and it's not available in Windows
import pwd
except:
pass
class Settings(object):
"""
@ -166,13 +160,13 @@ class Settings(object):
with open(self.filename, "r") as f:
self._settings = json.load(f)
self.fill_in_defaults()
except:
except Exception:
pass
# Make sure data_dir exists
try:
os.makedirs(self.get("data_dir"), exist_ok=True)
except:
except Exception:
pass
def save(self):
@ -191,7 +185,7 @@ class Settings(object):
if key == "control_port_port" or key == "socks_port":
try:
val = int(val)
except:
except Exception:
if key == "control_port_port":
val = self.default_settings["control_port_port"]
elif key == "socks_port":

10
cli/onionshare_cli/web/receive_mode.py
View File

@ -33,8 +33,6 @@ uploads files:
- new ReceiveModeRequest object is created
- creates a directory based on the timestamp
-
"""
@ -303,7 +301,7 @@ class ReceiveModeFile(object):
self.upload_error = False
try:
self.f = open(self.filename_in_progress, "wb+")
except:
except Exception:
# This will only happen if someone is messing with the data dir while
# OnionShare is running, but if it does make sure to throw an error
self.upload_error = True
@ -351,7 +349,7 @@ class ReceiveModeFile(object):
bytes_written = self.f.write(b)
self.onionshare_write_func(self.onionshare_filename, bytes_written)
except:
except Exception:
self.upload_error = True
def close(self):
@ -365,7 +363,7 @@ class ReceiveModeFile(object):
# Rename the in progress file to the final filename
os.rename(self.filename_in_progress, self.filename)
except:
except Exception:
self.upload_error = True
self.onionshare_close_func(self.onionshare_filename, self.upload_error)
@ -465,7 +463,7 @@ class ReceiveModeRequest(Request):
# Figure out the content length
try:
self.content_length = int(self.headers["Content-Length"])
except:
except Exception:
self.content_length = 0
date_str = datetime.now().strftime("%b %d, %I:%M%p")

2
cli/onionshare_cli/web/send_base_mode.py
View File

@ -242,7 +242,7 @@ class SendBaseModeWeb:
},
)
done = False
except:
except Exception:
# Looks like the download was canceled
done = True

90
cli/onionshare_cli/web/share_mode.py
View File

@ -44,7 +44,7 @@ def make_etag(data):
else:
break
hash_value = binascii.hexlify(hasher.digest()).decode('utf-8')
hash_value = binascii.hexlify(hasher.digest()).decode("utf-8")
return '"sha256:{}"'.format(hash_value)
@ -53,13 +53,13 @@ def parse_range_header(range_header: str, target_size: int) -> list:
if range_header is None:
return [(0, end_index)]
bytes_ = 'bytes='
bytes_ = "bytes="
if not range_header.startswith(bytes_):
abort(416)
ranges = []
for range_ in range_header[len(bytes_):].split(','):
split = range_.split('-')
for range_ in range_header[len(bytes_) :].split(","):
split = range_.split("-")
if len(split) == 1:
try:
start = int(split[0])
@ -194,13 +194,17 @@ class ShareModeWeb(SendBaseModeWeb):
etag = self.download_etag
# for range requests
range_, status_code = self.get_range_and_status_code(self.filesize, etag, self.last_modified)
range_, status_code = self.get_range_and_status_code(
self.filesize, etag, self.last_modified
)
# Tell GUI the download started
history_id = self.cur_history_id
self.cur_history_id += 1
self.web.add_request(
self.web.REQUEST_STARTED, request_path, {"id": history_id, "use_gzip": use_gzip}
self.web.REQUEST_STARTED,
request_path,
{"id": history_id, "use_gzip": use_gzip},
)
basename = os.path.basename(self.download_filename)
@ -209,32 +213,41 @@ class ShareModeWeb(SendBaseModeWeb):
r = Response()
else:
r = Response(
self.generate(shutdown_func, range_, file_to_download, request_path,
history_id, self.filesize))
self.generate(
shutdown_func,
range_,
file_to_download,
request_path,
history_id,
self.filesize,
)
)
if use_gzip:
r.headers.set('Content-Encoding', 'gzip')
r.headers.set("Content-Encoding", "gzip")
r.headers.set('Content-Length', range_[1] - range_[0] + 1)
r.headers.set("Content-Length", range_[1] - range_[0] + 1)
filename_dict = {
"filename": unidecode(basename),
"filename*": "UTF-8''%s" % url_quote(basename),
}
r.headers.set('Content-Disposition', 'attachment', **filename_dict)
r.headers.set("Content-Disposition", "attachment", **filename_dict)
r = self.web.add_security_headers(r)
# guess content type
(content_type, _) = mimetypes.guess_type(basename, strict=False)
if content_type is not None:
r.headers.set('Content-Type', content_type)
r.headers.set('Accept-Ranges', 'bytes')
r.headers.set('ETag', etag)
r.headers.set('Last-Modified', http_date(self.last_modified))
r.headers.set("Content-Type", content_type)
r.headers.set("Accept-Ranges", "bytes")
r.headers.set("ETag", etag)
r.headers.set("Last-Modified", http_date(self.last_modified))
# we need to set this for range requests
r.headers.set('Vary', 'Accept-Encoding')
r.headers.set("Vary", "Accept-Encoding")
if status_code == 206:
r.headers.set('Content-Range',
'bytes {}-{}/{}'.format(range_[0], range_[1], self.filesize))
r.headers.set(
"Content-Range",
"bytes {}-{}/{}".format(range_[0], range_[1], self.filesize),
)
r.status_code = status_code
@ -244,17 +257,19 @@ class ShareModeWeb(SendBaseModeWeb):
def get_range_and_status_code(cls, dl_size, etag, last_modified):
use_default_range = True
status_code = 200
range_header = request.headers.get('Range')
range_header = request.headers.get("Range")
# range requests are only allowed for get
if request.method == 'GET':
if request.method == "GET":
ranges = parse_range_header(range_header, dl_size)
if not (len(ranges) == 1 and ranges[0][0] == 0 and ranges[0][1] == dl_size - 1):
if not (
len(ranges) == 1 and ranges[0][0] == 0 and ranges[0][1] == dl_size - 1
):
use_default_range = False
status_code = 206
if range_header:
if_range = request.headers.get('If-Range')
if_range = request.headers.get("If-Range")
if if_range and if_range != etag:
use_default_range = True
status_code = 200
@ -266,11 +281,11 @@ class ShareModeWeb(SendBaseModeWeb):
abort(416) # We don't support multipart range requests yet
range_ = ranges[0]
etag_header = request.headers.get('ETag')
etag_header = request.headers.get("ETag")
if etag_header is not None and etag_header != etag:
abort(412)
if_unmod = request.headers.get('If-Unmodified-Since')
if_unmod = request.headers.get("If-Unmodified-Since")
if if_unmod:
if_date = parse_date(if_unmod)
if if_date and if_date > last_modified:
@ -280,10 +295,12 @@ class ShareModeWeb(SendBaseModeWeb):
return range_, status_code
def generate(self, shutdown_func, range_, file_to_download, path, history_id, filesize):
def generate(
self, shutdown_func, range_, file_to_download, path, history_id, filesize
):
# The user hasn't canceled the download
self.client_cancel = False
# Starting a new download
if self.web.settings.get("share", "autostop_sharing"):
self.download_in_progress = True
@ -326,9 +343,7 @@ class ShareModeWeb(SendBaseModeWeb):
):
sys.stdout.write(
"\r{0:s}, {1:.2f}% ".format(
self.common.human_readable_filesize(
downloaded_bytes
),
self.common.human_readable_filesize(downloaded_bytes),
percent,
)
)
@ -337,10 +352,14 @@ class ShareModeWeb(SendBaseModeWeb):
self.web.add_request(
self.web.REQUEST_PROGRESS,
path,
{"id": history_id, "bytes": downloaded_bytes, 'total_bytes': filesize,},
{
"id": history_id,
"bytes": downloaded_bytes,
"total_bytes": filesize,
},
)
self.web.done = False
except:
except Exception:
# looks like the download was canceled
self.web.done = True
canceled = True
@ -367,10 +386,9 @@ class ShareModeWeb(SendBaseModeWeb):
if shutdown_func is None:
raise RuntimeError("Not running with the Werkzeug Server")
shutdown_func()
except:
except Exception:
pass
def directory_listing_template(
self, path, files, dirs, breadcrumbs, breadcrumbs_leaf
):
@ -466,7 +484,7 @@ class ShareModeWeb(SendBaseModeWeb):
if len(self.file_info["files"]) == 1 and len(self.file_info["dirs"]) == 0:
self.download_filename = self.file_info["files"][0]["filename"]
self.download_filesize = self.file_info["files"][0]["size"]
with open(self.download_filename, 'rb') as f:
with open(self.download_filename, "rb") as f:
self.download_etag = make_etag(f)
# Compress the file with gzip now, so we don't have to do it on each request
@ -475,7 +493,7 @@ class ShareModeWeb(SendBaseModeWeb):
self.download_filename, self.gzip_filename, 6, processed_size_callback
)
self.gzip_filesize = os.path.getsize(self.gzip_filename)
with open(self.gzip_filename, 'rb') as f:
with open(self.gzip_filename, "rb") as f:
self.gzip_etag = make_etag(f)
# Make sure the gzip file gets cleaned up when onionshare stops
@ -502,7 +520,7 @@ class ShareModeWeb(SendBaseModeWeb):
self.zip_writer.close()
self.download_filesize = os.path.getsize(self.download_filename)
with open(self.download_filename, 'rb') as f:
with open(self.download_filename, "rb") as f:
self.download_etag = make_etag(f)
# Make sure the zip file gets cleaned up when onionshare stops

7
cli/onionshare_cli/web/web.py
View File

@ -41,6 +41,7 @@ from .receive_mode import ReceiveModeWeb, ReceiveModeWSGIMiddleware, ReceiveMode
from .website_mode import WebsiteModeWeb
from .chat_mode import ChatModeWeb
# Stub out flask's show_server_banner function, to avoiding showing warnings that
# are not applicable to OnionShare
def stubbed_show_server_banner(env, debug, app_import_path, eager_loading):
@ -49,7 +50,7 @@ def stubbed_show_server_banner(env, debug, app_import_path, eager_loading):
try:
flask.cli.show_server_banner = stubbed_show_server_banner
except:
except Exception:
pass
@ -327,7 +328,7 @@ class Web:
def generate_password(self, saved_password=None):
self.common.log("Web", "generate_password", f"saved_password={saved_password}")
if saved_password != None and saved_password != "":
if saved_password is not None and saved_password != "":
self.password = saved_password
self.common.log(
"Web",
@ -363,7 +364,7 @@ class Web:
if func is None and self.mode != "chat":
raise RuntimeError("Not running with the Werkzeug Server")
func()
except:
except Exception:
pass
self.running = False

7
cli/setup.py
View File

@ -25,7 +25,12 @@ version = "2.3.1"
setuptools.setup(
name="onionshare-cli",
version=version,
description="OnionShare lets you securely and anonymously send and receive files. It works by starting a web server, making it accessible as a Tor onion service, and generating an unguessable web address so others can download files from you, or upload files to you. It does _not_ require setting up a separate server or using a third party file-sharing service.",
description=(
"OnionShare lets you securely and anonymously send and receive files. It works by starting a web server, "
"making it accessible as a Tor onion service, and generating an unguessable web address so others can "
"download files from you, or upload files to you. It does _not_ require setting up a separate server or "
"using a third party file-sharing service."
),
author="Micah Lee",
author_email="micah@micahflee.com",
maintainer="Micah Lee",

27
cli/tests/conftest.py
View File

@ -1,11 +1,4 @@
import sys
# Force tests to look for resources in the source code tree
sys.onionshare_dev_mode = True
# Let OnionShare know the tests are running, to avoid colliding with settings files
sys.onionshare_test_mode = True
import os
import shutil
import tempfile
@ -14,6 +7,11 @@ import pytest
from onionshare_cli import common, web
# Force tests to look for resources in the source code tree
sys.onionshare_dev_mode = True
# Let OnionShare know the tests are running, to avoid colliding with settings files
sys.onionshare_test_mode = True
# The temporary directory for CLI tests
test_temp_dir = None
@ -45,7 +43,7 @@ def temp_dir():
@pytest.fixture
def temp_dir_1024(temp_dir):
""" Create a temporary directory that has a single file of a
"""Create a temporary directory that has a single file of a
particular size (1024 bytes).
"""
@ -59,7 +57,7 @@ def temp_dir_1024(temp_dir):
# pytest > 2.9 only needs @pytest.fixture
@pytest.yield_fixture
def temp_dir_1024_delete(temp_dir):
""" Create a temporary directory that has a single file of a
"""Create a temporary directory that has a single file of a
particular size (1024 bytes). The temporary directory (including
the file inside) will be deleted after fixture usage.
"""
@ -73,7 +71,7 @@ def temp_dir_1024_delete(temp_dir):
@pytest.fixture
def temp_file_1024(temp_dir):
""" Create a temporary file of a particular size (1024 bytes). """
"""Create a temporary file of a particular size (1024 bytes)."""
with tempfile.NamedTemporaryFile(delete=False, dir=temp_dir) as tmp_file:
tmp_file.write(b"*" * 1024)
@ -117,7 +115,7 @@ def default_zw():
tmp_dir = os.path.dirname(zw.zip_filename)
try:
shutil.rmtree(tmp_dir, ignore_errors=True)
except:
except Exception:
pass
@ -189,10 +187,3 @@ def time_strftime(monkeypatch):
@pytest.fixture
def common_obj():
return common.Common()
@pytest.fixture
def settings_obj(sys_onionshare_dev_mode, platform_linux):
_common = common.Common()
_common.version = "DUMMY_VERSION_1.2.3"
return settings.Settings(_common)

14
cli/tests/test_cli_web.py
View File

@ -42,7 +42,7 @@ RANDOM_STR_REGEX = re.compile(r"^[a-z2-7]+$")
def web_obj(temp_dir, common_obj, mode, num_files=0):
""" Creates a Web object, in either share mode or receive mode, ready for testing """
"""Creates a Web object, in either share mode or receive mode, ready for testing"""
common_obj.settings = Settings(common_obj)
mode_settings = ModeSettings(common_obj)
web = Web(common_obj, False, mode_settings, mode)
@ -100,7 +100,7 @@ class TestWeb:
web = web_obj(temp_dir, common_obj, "share", 3)
web.settings.set("share", "autostop_sharing", True)
assert web.running == True
assert web.running is True
with web.app.test_client() as c:
# Download the first time
@ -112,7 +112,7 @@ class TestWeb:
or res.mimetype == "application/x-zip-compressed"
)
assert web.running == False
assert web.running is False
def test_share_mode_autostop_sharing_off(
self, temp_dir, common_obj, temp_file_1024
@ -120,7 +120,7 @@ class TestWeb:
web = web_obj(temp_dir, common_obj, "share", 3)
web.settings.set("share", "autostop_sharing", False)
assert web.running == True
assert web.running is True
with web.app.test_client() as c:
# Download the first time
@ -131,7 +131,7 @@ class TestWeb:
res.mimetype == "application/zip"
or res.mimetype == "application/x-zip-compressed"
)
assert web.running == True
assert web.running is True
def test_receive_mode(self, temp_dir, common_obj):
web = web_obj(temp_dir, common_obj, "receive")
@ -183,7 +183,7 @@ class TestWeb:
assert res.status_code == 200
assert webhook_url == "http://127.0.0.1:1337/example"
assert webhook_data == "1 file uploaded to OnionShare"
assert webhook_data == "1 file submitted to OnionShare"
def test_public_mode_on(self, temp_dir, common_obj):
web = web_obj(temp_dir, common_obj, "receive")
@ -192,7 +192,7 @@ class TestWeb:
with web.app.test_client() as c:
# Loading / should work without auth
res = c.get("/")
data1 = res.get_data()
res.get_data()
assert res.status_code == 200
def test_public_mode_off(self, temp_dir, common_obj):

33
cli/tests/test_range_request.py
View File

@ -1,7 +1,4 @@
import pytest
import subprocess
from tempfile import NamedTemporaryFile
from werkzeug.exceptions import RequestedRangeNotSatisfiable
from onionshare_cli.web.share_mode import parse_range_header
@ -9,24 +6,24 @@ from onionshare_cli.web.share_mode import parse_range_header
VALID_RANGES = [
(None, 500, [(0, 499)]),
('bytes=0', 500, [(0, 499)]),
('bytes=100', 500, [(100, 499)]),
('bytes=100-', 500, [(100, 499)]), # not in the RFC, but how curl sends
('bytes=0-99', 500, [(0, 99)]),
('bytes=0-599', 500, [(0, 499)]),
('bytes=0-0', 500, [(0, 0)]),
('bytes=-100', 500, [(400, 499)]),
('bytes=0-99,100-199', 500, [(0, 199)]),
('bytes=0-100,100-199', 500, [(0, 199)]),
('bytes=0-99,101-199', 500, [(0, 99), (101, 199)]),
('bytes=0-199,100-299', 500, [(0, 299)]),
('bytes=0-99,200-299', 500, [(0, 99), (200, 299)]),
("bytes=0", 500, [(0, 499)]),
("bytes=100", 500, [(100, 499)]),
("bytes=100-", 500, [(100, 499)]), # not in the RFC, but how curl sends
("bytes=0-99", 500, [(0, 99)]),
("bytes=0-599", 500, [(0, 499)]),
("bytes=0-0", 500, [(0, 0)]),
("bytes=-100", 500, [(400, 499)]),
("bytes=0-99,100-199", 500, [(0, 199)]),
("bytes=0-100,100-199", 500, [(0, 199)]),
("bytes=0-99,101-199", 500, [(0, 99), (101, 199)]),
("bytes=0-199,100-299", 500, [(0, 299)]),
("bytes=0-99,200-299", 500, [(0, 99), (200, 299)]),
]
INVALID_RANGES = [
'bytes=200-100',
'bytes=0-100,300-200',
"bytes=200-100",
"bytes=0-100,300-200",
]
@ -38,4 +35,4 @@ def test_parse_ranges():
for invalid in INVALID_RANGES:
with pytest.raises(RequestedRangeNotSatisfiable):
parse_range_header(invalid, 500)
parse_range_header(invalid, 500)