mirror of
https://github.com/onionshare/onionshare.git
synced 2025-03-06 13:15:49 -05:00
Clean up CLI test use of temporary files
This commit is contained in:
parent
b681b53420
commit
78d5f4ff50
@ -15,6 +15,10 @@ import pytest
|
||||
from onionshare import common, web, settings, strings
|
||||
|
||||
|
||||
# The temporary directory for CLI tests
|
||||
test_temp_dir = None
|
||||
|
||||
|
||||
def pytest_addoption(parser):
|
||||
parser.addoption(
|
||||
"--rungui", action="store_true", default=False, help="run GUI tests"
|
||||
@ -41,51 +45,60 @@ def pytest_collection_modifyitems(config, items):
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_dir_1024():
|
||||
def temp_dir():
|
||||
"""Creates a persistent temporary directory for the CLI tests to use"""
|
||||
global test_temp_dir
|
||||
if not test_temp_dir:
|
||||
test_temp_dir = tempfile.mkdtemp()
|
||||
return test_temp_dir
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_dir_1024(temp_dir):
|
||||
""" Create a temporary directory that has a single file of a
|
||||
particular size (1024 bytes).
|
||||
"""
|
||||
|
||||
tmp_dir = tempfile.mkdtemp()
|
||||
tmp_file, tmp_file_path = tempfile.mkstemp(dir=tmp_dir)
|
||||
new_temp_dir = tempfile.mkdtemp(dir=temp_dir)
|
||||
tmp_file, tmp_file_path = tempfile.mkstemp(dir=new_temp_dir)
|
||||
with open(tmp_file, "wb") as f:
|
||||
f.write(b"*" * 1024)
|
||||
return tmp_dir
|
||||
return new_temp_dir
|
||||
|
||||
|
||||
# pytest > 2.9 only needs @pytest.fixture
|
||||
@pytest.yield_fixture
|
||||
def temp_dir_1024_delete():
|
||||
def temp_dir_1024_delete(temp_dir):
|
||||
""" 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.
|
||||
"""
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
tmp_file, tmp_file_path = tempfile.mkstemp(dir=tmp_dir)
|
||||
with tempfile.TemporaryDirectory(dir=temp_dir) as new_temp_dir:
|
||||
tmp_file, tmp_file_path = tempfile.mkstemp(dir=new_temp_dir)
|
||||
with open(tmp_file, "wb") as f:
|
||||
f.write(b"*" * 1024)
|
||||
yield tmp_dir
|
||||
yield new_temp_dir
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_file_1024():
|
||||
def temp_file_1024(temp_dir):
|
||||
""" Create a temporary file of a particular size (1024 bytes). """
|
||||
|
||||
with tempfile.NamedTemporaryFile(delete=False) as tmp_file:
|
||||
with tempfile.NamedTemporaryFile(delete=False, dir=temp_dir) as tmp_file:
|
||||
tmp_file.write(b"*" * 1024)
|
||||
return tmp_file.name
|
||||
|
||||
|
||||
# pytest > 2.9 only needs @pytest.fixture
|
||||
@pytest.yield_fixture
|
||||
def temp_file_1024_delete():
|
||||
def temp_file_1024_delete(temp_dir):
|
||||
"""
|
||||
Create a temporary file of a particular size (1024 bytes).
|
||||
The temporary file will be deleted after fixture usage.
|
||||
"""
|
||||
|
||||
with tempfile.NamedTemporaryFile() as tmp_file:
|
||||
with tempfile.NamedTemporaryFile(dir=temp_dir) as tmp_file:
|
||||
tmp_file.write(b"*" * 1024)
|
||||
tmp_file.flush()
|
||||
yield tmp_file.name
|
||||
|
@ -64,13 +64,13 @@ class TestSettings:
|
||||
settings_obj.fill_in_defaults()
|
||||
assert settings_obj._settings["version"] == "DUMMY_VERSION_1.2.3"
|
||||
|
||||
def test_load(self, settings_obj):
|
||||
def test_load(self, temp_dir, settings_obj):
|
||||
custom_settings = {
|
||||
"version": "CUSTOM_VERSION",
|
||||
"socks_port": 9999,
|
||||
"use_stealth": True,
|
||||
}
|
||||
tmp_file, tmp_file_path = tempfile.mkstemp()
|
||||
tmp_file, tmp_file_path = tempfile.mkstemp(dir=temp_dir)
|
||||
with open(tmp_file, "w") as f:
|
||||
json.dump(custom_settings, f)
|
||||
settings_obj.filename = tmp_file_path
|
||||
@ -83,12 +83,12 @@ class TestSettings:
|
||||
os.remove(tmp_file_path)
|
||||
assert os.path.exists(tmp_file_path) is False
|
||||
|
||||
def test_save(self, monkeypatch, settings_obj):
|
||||
def test_save(self, monkeypatch, temp_dir, settings_obj):
|
||||
monkeypatch.setattr(strings, "_", lambda _: "")
|
||||
|
||||
settings_filename = "default_settings.json"
|
||||
tmp_dir = tempfile.gettempdir()
|
||||
settings_path = os.path.join(tmp_dir, settings_filename)
|
||||
new_temp_dir = tempfile.mkdtemp(dir=temp_dir)
|
||||
settings_path = os.path.join(new_temp_dir, settings_filename)
|
||||
settings_obj.filename = settings_path
|
||||
settings_obj.save()
|
||||
with open(settings_path, "r") as f:
|
||||
|
@ -42,7 +42,7 @@ DEFAULT_ZW_FILENAME_REGEX = re.compile(r"^onionshare_[a-z2-7]{6}.zip$")
|
||||
RANDOM_STR_REGEX = re.compile(r"^[a-z2-7]+$")
|
||||
|
||||
|
||||
def web_obj(common_obj, mode, num_files=0):
|
||||
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 """
|
||||
common_obj.settings = Settings(common_obj)
|
||||
strings.load_strings(common_obj)
|
||||
@ -58,7 +58,7 @@ def web_obj(common_obj, mode, num_files=0):
|
||||
# Add files
|
||||
files = []
|
||||
for _ in range(num_files):
|
||||
with tempfile.NamedTemporaryFile(delete=False) as tmp_file:
|
||||
with tempfile.NamedTemporaryFile(delete=False, dir=temp_dir) as tmp_file:
|
||||
tmp_file.write(b"*" * 1024)
|
||||
files.append(tmp_file.name)
|
||||
web.share_mode.set_file_info(files)
|
||||
@ -70,8 +70,8 @@ def web_obj(common_obj, mode, num_files=0):
|
||||
|
||||
|
||||
class TestWeb:
|
||||
def test_share_mode(self, common_obj):
|
||||
web = web_obj(common_obj, "share", 3)
|
||||
def test_share_mode(self, temp_dir, common_obj):
|
||||
web = web_obj(temp_dir, common_obj, "share", 3)
|
||||
assert web.mode == "share"
|
||||
with web.app.test_client() as c:
|
||||
# Load / without auth
|
||||
@ -95,8 +95,8 @@ class TestWeb:
|
||||
assert res.status_code == 200
|
||||
assert res.mimetype == "application/zip"
|
||||
|
||||
def test_share_mode_autostop_sharing_on(self, common_obj, temp_file_1024):
|
||||
web = web_obj(common_obj, "share", 3)
|
||||
def test_share_mode_autostop_sharing_on(self, temp_dir, common_obj, temp_file_1024):
|
||||
web = web_obj(temp_dir, common_obj, "share", 3)
|
||||
web.settings.set("share", "autostop_sharing", True)
|
||||
|
||||
assert web.running == True
|
||||
@ -110,8 +110,8 @@ class TestWeb:
|
||||
|
||||
assert web.running == False
|
||||
|
||||
def test_share_mode_autostop_sharing_off(self, common_obj, temp_file_1024):
|
||||
web = web_obj(common_obj, "share", 3)
|
||||
def test_share_mode_autostop_sharing_off(self, temp_dir, common_obj, temp_file_1024):
|
||||
web = web_obj(temp_dir, common_obj, "share", 3)
|
||||
web.settings.set("share", "autostop_sharing", False)
|
||||
|
||||
assert web.running == True
|
||||
@ -124,8 +124,8 @@ class TestWeb:
|
||||
assert res.mimetype == "application/zip"
|
||||
assert web.running == True
|
||||
|
||||
def test_receive_mode(self, common_obj):
|
||||
web = web_obj(common_obj, "receive")
|
||||
def test_receive_mode(self, temp_dir, common_obj):
|
||||
web = web_obj(temp_dir, common_obj, "receive")
|
||||
assert web.mode == "receive"
|
||||
|
||||
with web.app.test_client() as c:
|
||||
@ -144,8 +144,8 @@ class TestWeb:
|
||||
res.get_data()
|
||||
assert res.status_code == 200
|
||||
|
||||
def test_public_mode_on(self, common_obj):
|
||||
web = web_obj(common_obj, "receive")
|
||||
def test_public_mode_on(self, temp_dir, common_obj):
|
||||
web = web_obj(temp_dir, common_obj, "receive")
|
||||
web.settings.set("general", "public", True)
|
||||
|
||||
with web.app.test_client() as c:
|
||||
@ -154,8 +154,8 @@ class TestWeb:
|
||||
data1 = res.get_data()
|
||||
assert res.status_code == 200
|
||||
|
||||
def test_public_mode_off(self, common_obj):
|
||||
web = web_obj(common_obj, "receive")
|
||||
def test_public_mode_off(self, temp_dir, common_obj):
|
||||
web = web_obj(temp_dir, common_obj, "receive")
|
||||
web.settings.set("general", "public", False)
|
||||
|
||||
with web.app.test_client() as c:
|
||||
|
@ -75,15 +75,6 @@ class TestShare(GuiBaseTest):
|
||||
tab.get_mode().server_status.file_selection.file_list.add_file(self.tmpfiles[1])
|
||||
self.file_selection_widget_has_files(tab, num_files + 2)
|
||||
|
||||
def add_large_file(self, tab):
|
||||
"""Add a large file to the share"""
|
||||
size = 1024 * 1024 * 155
|
||||
with open("/tmp/large_file", "wb") as fout:
|
||||
fout.write(os.urandom(size))
|
||||
tab.get_mode().server_status.file_selection.file_list.add_file(
|
||||
"/tmp/large_file"
|
||||
)
|
||||
|
||||
def add_delete_buttons_hidden(self, tab):
|
||||
"""Test that the add and delete buttons are hidden when the server starts"""
|
||||
self.assertFalse(
|
||||
@ -321,16 +312,6 @@ class TestShare(GuiBaseTest):
|
||||
self.run_all_share_mode_started_tests(tab)
|
||||
self.run_all_share_mode_individual_file_download_tests(tab)
|
||||
|
||||
def run_all_large_file_tests(self, tab):
|
||||
"""Same as above but with a larger file"""
|
||||
self.run_all_share_mode_setup_tests(tab)
|
||||
self.add_large_file(tab)
|
||||
self.run_all_share_mode_started_tests(tab, startup_time=15000)
|
||||
self.assertTrue(tab.filesize_warning.isVisible())
|
||||
self.server_is_stopped(tab)
|
||||
self.web_server_is_stopped(tab)
|
||||
self.server_status_indicator_says_closed(tab)
|
||||
|
||||
def run_all_share_mode_persistent_tests(self, tab):
|
||||
"""Same as end-to-end share tests but also test the password is the same on multiple shared"""
|
||||
self.run_all_share_mode_setup_tests(tab)
|
||||
@ -521,3 +502,24 @@ class TestShare(GuiBaseTest):
|
||||
self.run_all_share_mode_individual_file_tests(tab)
|
||||
|
||||
self.close_all_tabs()
|
||||
|
||||
@pytest.mark.gui
|
||||
def test_large_download(self):
|
||||
"""
|
||||
Test a large download
|
||||
"""
|
||||
tab = self.new_share_tab()
|
||||
|
||||
self.run_all_common_setup_tests()
|
||||
self.run_all_share_mode_setup_tests(tab)
|
||||
tab.get_mode().server_status.file_selection.file_list.add_file(
|
||||
self.tmpfile_large
|
||||
)
|
||||
self.run_all_share_mode_started_tests(tab, startup_time=15000)
|
||||
self.assertTrue(tab.get_mode().filesize_warning.isVisible())
|
||||
self.download_share(tab)
|
||||
self.server_is_stopped(tab)
|
||||
self.web_server_is_stopped(tab)
|
||||
self.server_status_indicator_says_closed(tab)
|
||||
|
||||
self.close_all_tabs()
|
||||
|
Loading…
x
Reference in New Issue
Block a user