Add onionshare CLI to cli folder, move GUI to desktop folder, and start refactoring it to work with briefcase

This commit is contained in:
Micah Lee 2020-10-12 22:40:55 -07:00
parent 93e90c89ae
commit a54f99adf6
583 changed files with 14871 additions and 474 deletions

View file

209
desktop/tests/conftest.py Normal file
View file

@ -0,0 +1,209 @@
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
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"
)
parser.addoption(
"--runtor", action="store_true", default=False, help="run tor tests"
)
def pytest_collection_modifyitems(config, items):
if not config.getoption("--runtor"):
# --runtor given in cli: do not skip tor tests
skip_tor = pytest.mark.skip(reason="need --runtor option to run")
for item in items:
if "tor" in item.keywords:
item.add_marker(skip_tor)
if not config.getoption("--rungui"):
# --rungui given in cli: do not skip GUI tests
skip_gui = pytest.mark.skip(reason="need --rungui option to run")
for item in items:
if "gui" in item.keywords:
item.add_marker(skip_gui)
@pytest.fixture
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).
"""
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 new_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
particular size (1024 bytes). The temporary directory (including
the file inside) will be deleted after fixture usage.
"""
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 new_temp_dir
@pytest.fixture
def temp_file_1024(temp_dir):
""" 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)
return tmp_file.name
# pytest > 2.9 only needs @pytest.fixture
@pytest.yield_fixture
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(dir=temp_dir, delete=False) as tmp_file:
tmp_file.write(b"*" * 1024)
tmp_file.flush()
tmp_file.close()
yield tmp_file.name
# pytest > 2.9 only needs @pytest.fixture
@pytest.yield_fixture(scope="session")
def custom_zw():
zw = web.share_mode.ZipWriter(
common.Common(),
zip_filename=common.Common.random_string(4, 6),
processed_size_callback=lambda _: "custom_callback",
)
yield zw
zw.close()
os.remove(zw.zip_filename)
# pytest > 2.9 only needs @pytest.fixture
@pytest.yield_fixture(scope="session")
def default_zw():
zw = web.share_mode.ZipWriter(common.Common())
yield zw
zw.close()
tmp_dir = os.path.dirname(zw.zip_filename)
try:
shutil.rmtree(tmp_dir, ignore_errors=True)
except:
pass
@pytest.fixture
def locale_en(monkeypatch):
monkeypatch.setattr("locale.getdefaultlocale", lambda: ("en_US", "UTF-8"))
@pytest.fixture
def locale_fr(monkeypatch):
monkeypatch.setattr("locale.getdefaultlocale", lambda: ("fr_FR", "UTF-8"))
@pytest.fixture
def locale_invalid(monkeypatch):
monkeypatch.setattr("locale.getdefaultlocale", lambda: ("xx_XX", "UTF-8"))
@pytest.fixture
def locale_ru(monkeypatch):
monkeypatch.setattr("locale.getdefaultlocale", lambda: ("ru_RU", "UTF-8"))
@pytest.fixture
def platform_darwin(monkeypatch):
monkeypatch.setattr("platform.system", lambda: "Darwin")
@pytest.fixture # (scope="session")
def platform_linux(monkeypatch):
monkeypatch.setattr("platform.system", lambda: "Linux")
@pytest.fixture
def platform_windows(monkeypatch):
monkeypatch.setattr("platform.system", lambda: "Windows")
@pytest.fixture
def sys_argv_sys_prefix(monkeypatch):
monkeypatch.setattr("sys.argv", [sys.prefix])
@pytest.fixture
def sys_frozen(monkeypatch):
monkeypatch.setattr("sys.frozen", True, raising=False)
@pytest.fixture
def sys_meipass(monkeypatch):
monkeypatch.setattr("sys._MEIPASS", os.path.expanduser("~"), raising=False)
@pytest.fixture # (scope="session")
def sys_onionshare_dev_mode(monkeypatch):
monkeypatch.setattr("sys.onionshare_dev_mode", True, raising=False)
@pytest.fixture
def time_time_100(monkeypatch):
monkeypatch.setattr("time.time", lambda: 100)
@pytest.fixture
def time_strftime(monkeypatch):
monkeypatch.setattr("time.strftime", lambda _: "Jun 06 2013 11:05:00")
@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"
strings.load_strings(_common)
return settings.Settings(_common)

View file

@ -0,0 +1,467 @@
import pytest
import unittest
import json
import os
import requests
import shutil
import base64
import tempfile
import secrets
import platform
from PyQt5 import QtCore, QtTest, QtWidgets
from onionshare import strings
from onionshare.common import Common
from onionshare.settings import Settings
from onionshare.onion import Onion
from onionshare.web import Web
from onionshare_gui import Application, MainWindow, GuiCommon
from onionshare_gui.tab.mode.share_mode import ShareMode
from onionshare_gui.tab.mode.receive_mode import ReceiveMode
from onionshare_gui.tab.mode.website_mode import WebsiteMode
class GuiBaseTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
common = Common(verbose=True)
# Delete any old test data that might exist
shutil.rmtree(common.build_data_dir(), ignore_errors=True)
qtapp = Application(common)
common.gui = GuiCommon(common, qtapp, local_only=True)
cls.gui = MainWindow(common, filenames=None)
cls.gui.qtapp = qtapp
# Create some random files to test with
cls.tmpdir = tempfile.TemporaryDirectory()
cls.tmpfiles = []
for _ in range(10):
filename = os.path.join(cls.tmpdir.name, f"{secrets.token_hex(4)}.txt")
with open(filename, "w") as file:
file.write(secrets.token_hex(10))
cls.tmpfiles.append(filename)
# A file called "test.txt"
cls.tmpfile_test = os.path.join(cls.tmpdir.name, "test.txt")
with open(cls.tmpfile_test, "w") as file:
file.write("onionshare")
# A file called "test2.txt"
cls.tmpfile_test2 = os.path.join(cls.tmpdir.name, "test2.txt")
with open(cls.tmpfile_test2, "w") as file:
file.write("onionshare2")
# A file called "index.html"
cls.tmpfile_index_html = os.path.join(cls.tmpdir.name, "index.html")
with open(cls.tmpfile_index_html, "w") as file:
file.write(
"<html><body><p>This is a test website hosted by OnionShare</p></body></html>"
)
# A large file
size = 1024 * 1024 * 155
cls.tmpfile_large = os.path.join(cls.tmpdir.name, "large_file")
with open(cls.tmpfile_large, "wb") as fout:
fout.write(os.urandom(size))
@classmethod
def tearDownClass(cls):
# Quit
cls.gui.qtapp.clipboard().clear()
QtCore.QTimer.singleShot(200, cls.gui.close_dialog.accept_button.click)
cls.gui.close()
cls.gui.cleanup()
# Shared test methods
def verify_new_tab(self, tab):
# Make sure the new tab widget is showing, and no mode has been started
QtTest.QTest.qWait(1000)
self.assertTrue(tab.new_tab.isVisible())
self.assertFalse(hasattr(tab, "share_mode"))
self.assertFalse(hasattr(tab, "receive_mode"))
self.assertFalse(hasattr(tab, "website_mode"))
self.assertFalse(hasattr(tab, "chat_mode"))
def new_share_tab(self):
tab = self.gui.tabs.widget(0)
self.verify_new_tab(tab)
# Share files
tab.share_button.click()
self.assertFalse(tab.new_tab.isVisible())
self.assertTrue(tab.share_mode.isVisible())
return tab
def new_share_tab_with_files(self):
tab = self.new_share_tab()
# Add files
for filename in self.tmpfiles:
tab.share_mode.server_status.file_selection.file_list.add_file(filename)
return tab
def new_receive_tab(self):
tab = self.gui.tabs.widget(0)
self.verify_new_tab(tab)
# Receive files
tab.receive_button.click()
self.assertFalse(tab.new_tab.isVisible())
self.assertTrue(tab.receive_mode.isVisible())
return tab
def new_website_tab(self):
tab = self.gui.tabs.widget(0)
self.verify_new_tab(tab)
# Publish website
tab.website_button.click()
self.assertFalse(tab.new_tab.isVisible())
self.assertTrue(tab.website_mode.isVisible())
return tab
def new_website_tab_with_files(self):
tab = self.new_website_tab()
# Add files
for filename in self.tmpfiles:
tab.website_mode.server_status.file_selection.file_list.add_file(filename)
return tab
def close_all_tabs(self):
for _ in range(self.gui.tabs.count()):
tab = self.gui.tabs.widget(0)
QtCore.QTimer.singleShot(200, tab.close_dialog.accept_button.click)
self.gui.tabs.tabBar().tabButton(0, QtWidgets.QTabBar.RightSide).click()
def gui_loaded(self):
"""Test that the GUI actually is shown"""
self.assertTrue(self.gui.show)
def window_title_seen(self):
"""Test that the window title is OnionShare"""
self.assertEqual(self.gui.windowTitle(), "OnionShare")
def server_status_bar_is_visible(self):
"""Test that the status bar is visible"""
self.assertTrue(self.gui.status_bar.isVisible())
def mode_settings_widget_is_visible(self, tab):
"""Test that the mode settings are visible"""
self.assertTrue(tab.get_mode().mode_settings_widget.isVisible())
def mode_settings_widget_is_hidden(self, tab):
"""Test that the mode settings are hidden when the server starts"""
self.assertFalse(tab.get_mode().mode_settings_widget.isVisible())
def click_toggle_history(self, tab):
"""Test that we can toggle Download or Upload history by clicking the toggle button"""
currently_visible = tab.get_mode().history.isVisible()
tab.get_mode().toggle_history.click()
self.assertEqual(tab.get_mode().history.isVisible(), not currently_visible)
def history_indicator(self, tab, indicator_count="1"):
"""Test that we can make sure the history is toggled off, do an action, and the indiciator works"""
# Make sure history is toggled off
if tab.get_mode().history.isVisible():
tab.get_mode().toggle_history.click()
self.assertFalse(tab.get_mode().history.isVisible())
# Indicator should not be visible yet
self.assertFalse(tab.get_mode().toggle_history.indicator_label.isVisible())
if type(tab.get_mode()) == ReceiveMode:
# Upload a file
files = {"file[]": open(self.tmpfiles[0], "rb")}
url = f"http://127.0.0.1:{tab.app.port}/upload"
if tab.settings.get("general", "public"):
requests.post(url, files=files)
else:
requests.post(
url,
files=files,
auth=requests.auth.HTTPBasicAuth(
"onionshare", tab.get_mode().web.password
),
)
QtTest.QTest.qWait(2000)
if type(tab.get_mode()) == ShareMode:
# Download files
url = f"http://127.0.0.1:{tab.app.port}/download"
if tab.settings.get("general", "public"):
requests.get(url)
else:
requests.get(
url,
auth=requests.auth.HTTPBasicAuth(
"onionshare", tab.get_mode().web.password
),
)
QtTest.QTest.qWait(2000)
# Indicator should be visible, have a value of "1"
self.assertTrue(tab.get_mode().toggle_history.indicator_label.isVisible())
self.assertEqual(
tab.get_mode().toggle_history.indicator_label.text(), indicator_count
)
# Toggle history back on, indicator should be hidden again
tab.get_mode().toggle_history.click()
self.assertFalse(tab.get_mode().toggle_history.indicator_label.isVisible())
def history_is_not_visible(self, tab):
"""Test that the History section is not visible"""
self.assertFalse(tab.get_mode().history.isVisible())
def history_is_visible(self, tab):
"""Test that the History section is visible"""
self.assertTrue(tab.get_mode().history.isVisible())
def server_working_on_start_button_pressed(self, tab):
"""Test we can start the service"""
# Should be in SERVER_WORKING state
tab.get_mode().server_status.server_button.click()
self.assertEqual(tab.get_mode().server_status.status, 1)
def toggle_indicator_is_reset(self, tab):
self.assertEqual(tab.get_mode().toggle_history.indicator_count, 0)
self.assertFalse(tab.get_mode().toggle_history.indicator_label.isVisible())
def server_status_indicator_says_starting(self, tab):
"""Test that the Server Status indicator shows we are Starting"""
self.assertEqual(
tab.get_mode().server_status_label.text(),
strings._("gui_status_indicator_share_working"),
)
def server_status_indicator_says_scheduled(self, tab):
"""Test that the Server Status indicator shows we are Scheduled"""
self.assertEqual(
tab.get_mode().server_status_label.text(),
strings._("gui_status_indicator_share_scheduled"),
)
def server_is_started(self, tab, startup_time=2000):
"""Test that the server has started"""
QtTest.QTest.qWait(startup_time)
# Should now be in SERVER_STARTED state
self.assertEqual(tab.get_mode().server_status.status, 2)
def web_server_is_running(self, tab):
"""Test that the web server has started"""
try:
requests.get(f"http://127.0.0.1:{tab.app.port}/")
self.assertTrue(True)
except requests.exceptions.ConnectionError:
self.assertTrue(False)
def have_a_password(self, tab):
"""Test that we have a valid password"""
if not tab.settings.get("general", "public"):
self.assertRegex(tab.get_mode().server_status.web.password, r"(\w+)-(\w+)")
else:
self.assertIsNone(tab.get_mode().server_status.web.password, r"(\w+)-(\w+)")
def add_button_visible(self, tab):
"""Test that the add button should be visible"""
if platform.system() == "Darwin":
self.assertTrue(
tab.get_mode().server_status.file_selection.add_files_button.isVisible()
)
self.assertTrue(
tab.get_mode().server_status.file_selection.add_folder_button.isVisible()
)
else:
self.assertTrue(
tab.get_mode().server_status.file_selection.add_button.isVisible()
)
def url_description_shown(self, tab):
"""Test that the URL label is showing"""
self.assertTrue(tab.get_mode().server_status.url_description.isVisible())
def have_copy_url_button(self, tab):
"""Test that the Copy URL button is shown and that the clipboard is correct"""
self.assertTrue(tab.get_mode().server_status.copy_url_button.isVisible())
tab.get_mode().server_status.copy_url_button.click()
clipboard = tab.common.gui.qtapp.clipboard()
if tab.settings.get("general", "public"):
self.assertEqual(clipboard.text(), f"http://127.0.0.1:{tab.app.port}")
else:
self.assertEqual(
clipboard.text(),
f"http://onionshare:{tab.get_mode().server_status.web.password}@127.0.0.1:{tab.app.port}",
)
def have_show_qr_code_button(self, tab):
"""Test that the Show QR Code URL button is shown and that it loads a QR Code Dialog"""
self.assertTrue(
tab.get_mode().server_status.show_url_qr_code_button.isVisible()
)
def accept_dialog():
window = tab.common.gui.qtapp.activeWindow()
if window:
window.close()
QtCore.QTimer.singleShot(500, accept_dialog)
tab.get_mode().server_status.show_url_qr_code_button.click()
def server_status_indicator_says_started(self, tab):
"""Test that the Server Status indicator shows we are started"""
if type(tab.get_mode()) == ReceiveMode:
self.assertEqual(
tab.get_mode().server_status_label.text(),
strings._("gui_status_indicator_receive_started"),
)
if type(tab.get_mode()) == ShareMode:
self.assertEqual(
tab.get_mode().server_status_label.text(),
strings._("gui_status_indicator_share_started"),
)
def web_page(self, tab, string):
"""Test that the web page contains a string"""
url = f"http://127.0.0.1:{tab.app.port}/"
if tab.settings.get("general", "public"):
r = requests.get(url)
else:
r = requests.get(
url,
auth=requests.auth.HTTPBasicAuth(
"onionshare", tab.get_mode().web.password
),
)
self.assertTrue(string in r.text)
def history_widgets_present(self, tab):
"""Test that the relevant widgets are present in the history view after activity has taken place"""
self.assertFalse(tab.get_mode().history.empty.isVisible())
self.assertTrue(tab.get_mode().history.not_empty.isVisible())
def counter_incremented(self, tab, count):
"""Test that the counter has incremented"""
self.assertEqual(tab.get_mode().history.completed_count, count)
def server_is_stopped(self, tab):
"""Test that the server stops when we click Stop"""
if (
type(tab.get_mode()) == ReceiveMode
or (
type(tab.get_mode()) == ShareMode
and not tab.settings.get("share", "autostop_sharing")
)
or (type(tab.get_mode()) == WebsiteMode)
):
tab.get_mode().server_status.server_button.click()
self.assertEqual(tab.get_mode().server_status.status, 0)
self.assertFalse(
tab.get_mode().server_status.show_url_qr_code_button.isVisible()
)
self.assertFalse(tab.get_mode().server_status.copy_url_button.isVisible())
self.assertFalse(tab.get_mode().server_status.url.isVisible())
self.assertFalse(tab.get_mode().server_status.url_description.isVisible())
self.assertFalse(
tab.get_mode().server_status.copy_hidservauth_button.isVisible()
)
def web_server_is_stopped(self, tab):
"""Test that the web server also stopped"""
QtTest.QTest.qWait(800)
try:
requests.get(f"http://127.0.0.1:{tab.app.port}/")
self.assertTrue(False)
except requests.exceptions.ConnectionError:
self.assertTrue(True)
def server_status_indicator_says_closed(self, tab):
"""Test that the Server Status indicator shows we closed"""
if type(tab.get_mode()) == ReceiveMode:
self.assertEqual(
tab.get_mode().server_status_label.text(),
strings._("gui_status_indicator_receive_stopped"),
)
if type(tab.get_mode()) == ShareMode:
if not tab.settings.get("share", "autostop_sharing"):
self.assertEqual(
tab.get_mode().server_status_label.text(),
strings._("gui_status_indicator_share_stopped"),
)
else:
self.assertEqual(
tab.get_mode().server_status_label.text(),
strings._("closing_automatically"),
)
def clear_all_history_items(self, tab, count):
if count == 0:
tab.get_mode().history.clear_button.click()
self.assertEqual(len(tab.get_mode().history.item_list.items.keys()), count)
def file_selection_widget_has_files(self, tab, num=3):
"""Test that the number of items in the list is as expected"""
self.assertEqual(
tab.get_mode().server_status.file_selection.get_num_files(), num
)
def add_remove_buttons_hidden(self, tab):
"""Test that the add and remove buttons are hidden when the server starts"""
if platform.system() == "Darwin":
self.assertFalse(
tab.get_mode().server_status.file_selection.add_files_button.isVisible()
)
self.assertFalse(
tab.get_mode().server_status.file_selection.add_folder_button.isVisible()
)
else:
self.assertFalse(
tab.get_mode().server_status.file_selection.add_button.isVisible()
)
self.assertFalse(
tab.get_mode().server_status.file_selection.remove_button.isVisible()
)
# Auto-stop timer tests
def set_timeout(self, tab, timeout):
"""Test that the timeout can be set"""
timer = QtCore.QDateTime.currentDateTime().addSecs(timeout)
tab.get_mode().mode_settings_widget.autostop_timer_widget.setDateTime(timer)
self.assertTrue(
tab.get_mode().mode_settings_widget.autostop_timer_widget.dateTime(), timer
)
def autostop_timer_widget_hidden(self, tab):
"""Test that the auto-stop timer widget is hidden when share has started"""
self.assertFalse(
tab.get_mode().mode_settings_widget.autostop_timer_widget.isVisible()
)
def server_timed_out(self, tab, wait):
"""Test that the server has timed out after the timer ran out"""
QtTest.QTest.qWait(wait)
# We should have timed out now
self.assertEqual(tab.get_mode().server_status.status, 0)
# Grouped tests follow from here
def run_all_common_setup_tests(self):
self.gui_loaded()
self.window_title_seen()
self.server_status_bar_is_visible()

4
desktop/tests/pytest.ini Normal file
View file

@ -0,0 +1,4 @@
[pytest]
markers =
gui: marks tests as a GUI test
tor: marks tests as a Tor GUI test

9
desktop/tests/run.bat Normal file
View file

@ -0,0 +1,9 @@
pytest -vvv tests\test_cli.py
pytest -vvv tests\test_cli_common.py
pytest -vvv tests\test_cli_settings.py
pytest -vvv tests\test_cli_strings.py
pytest -vvv tests\test_cli_web.py
pytest -vvv --rungui tests\test_gui_tabs.py
pytest -vvv --rungui tests\test_gui_share.py
pytest -vvv --rungui tests\test_gui_receive.py
pytest -vvv --rungui tests\test_gui_website.py

26
desktop/tests/run.sh Executable file
View file

@ -0,0 +1,26 @@
#!/bin/bash
# The script runs python tests
# Firstly, all CLI tests are run
# Then, all the GUI tests are run individually
# to avoid segmentation fault
PARAMS=""
while [ ! $# -eq 0 ]
do
case "$1" in
--rungui)
PARAMS="$PARAMS --rungui"
;;
--runtor)
PARAMS="$PARAMS --runtor"
;;
esac
shift
done
pytest $PARAMS -vvv ./tests/test_cli*.py || exit 1
for filename in ./tests/test_gui_*.py; do
pytest $PARAMS -vvv --no-qt-log $filename || exit 1
done

56
desktop/tests/test_cli.py Normal file
View file

@ -0,0 +1,56 @@
import os
import pytest
from onionshare import OnionShare
from onionshare.common import Common
from onionshare.mode_settings import ModeSettings
class MyOnion:
def __init__(self):
self.auth_string = "TestHidServAuth"
self.private_key = ""
self.scheduled_key = None
@staticmethod
def start_onion_service(self, mode_settings_obj, await_publication=True, save_scheduled_key=False):
return "test_service_id.onion"
@pytest.fixture
def onionshare_obj():
common = Common()
return OnionShare(common, MyOnion())
@pytest.fixture
def mode_settings_obj():
common = Common()
return ModeSettings(common)
class TestOnionShare:
def test_init(self, onionshare_obj):
assert onionshare_obj.hidserv_dir is None
assert onionshare_obj.onion_host is None
assert onionshare_obj.cleanup_filenames == []
assert onionshare_obj.local_only is False
def test_start_onion_service(self, onionshare_obj, mode_settings_obj):
onionshare_obj.start_onion_service(mode_settings_obj)
assert 17600 <= onionshare_obj.port <= 17650
assert onionshare_obj.onion_host == "test_service_id.onion"
def test_start_onion_service_local_only(self, onionshare_obj, mode_settings_obj):
onionshare_obj.local_only = True
onionshare_obj.start_onion_service(mode_settings_obj)
assert onionshare_obj.onion_host == "127.0.0.1:{}".format(onionshare_obj.port)
def test_cleanup(self, onionshare_obj, temp_dir_1024, temp_file_1024):
onionshare_obj.cleanup_filenames = [temp_dir_1024, temp_file_1024]
onionshare_obj.cleanup()
assert os.path.exists(temp_dir_1024) is False
assert os.path.exists(temp_dir_1024) is False
assert onionshare_obj.cleanup_filenames == []

View file

@ -0,0 +1,275 @@
import contextlib
import inspect
import io
import os
import random
import re
import socket
import sys
import zipfile
import pytest
PASSWORD_REGEX = re.compile(r"^([a-z]+)(-[a-z]+)?-([a-z]+)(-[a-z]+)?$")
# TODO: Improve the Common tests to test it all as a single class
class TestBuildPassword:
@pytest.mark.parametrize(
"test_input,expected",
(
# VALID, two lowercase words, separated by a hyphen
("syrup-enzyme", True),
("caution-friday", True),
# VALID, two lowercase words, with one hyphenated compound word
("drop-down-thimble", True),
("unmixed-yo-yo", True),
# VALID, two lowercase hyphenated compound words, separated by hyphen
("yo-yo-drop-down", True),
("felt-tip-t-shirt", True),
("hello-world", True),
# INVALID
("Upper-Case", False),
("digits-123", False),
("too-many-hyphens-", False),
("symbols-!@#$%", False),
),
)
def test_build_password_regex(self, test_input, expected):
""" Test that `PASSWORD_REGEX` accounts for the following patterns
There are a few hyphenated words in `wordlist.txt`:
* drop-down
* felt-tip
* t-shirt
* yo-yo
These words cause a few extra potential password patterns:
* word-word
* hyphenated-word-word
* word-hyphenated-word
* hyphenated-word-hyphenated-word
"""
assert bool(PASSWORD_REGEX.match(test_input)) == expected
def test_build_password_unique(self, common_obj, sys_onionshare_dev_mode):
assert common_obj.build_password() != common_obj.build_password()
class TestDirSize:
def test_temp_dir_size(self, common_obj, temp_dir_1024_delete):
""" dir_size() should return the total size (in bytes) of all files
in a particular directory.
"""
assert common_obj.dir_size(temp_dir_1024_delete) == 1024
class TestEstimatedTimeRemaining:
@pytest.mark.parametrize(
"test_input,expected",
(
((2, 676, 12), "8h14m16s"),
((14, 1049, 30), "1h26m15s"),
((21, 450, 1), "33m42s"),
((31, 1115, 80), "11m39s"),
((336, 989, 32), "2m12s"),
((603, 949, 38), "36s"),
((971, 1009, 83), "1s"),
),
)
def test_estimated_time_remaining(
self, common_obj, test_input, expected, time_time_100
):
assert common_obj.estimated_time_remaining(*test_input) == expected
@pytest.mark.parametrize(
"test_input",
(
(10, 20, 100), # if `time_elapsed == 0`
(0, 37, 99), # if `download_rate == 0`
),
)
def test_raises_zero_division_error(self, common_obj, test_input, time_time_100):
with pytest.raises(ZeroDivisionError):
common_obj.estimated_time_remaining(*test_input)
class TestFormatSeconds:
@pytest.mark.parametrize(
"test_input,expected",
(
(0, "0s"),
(26, "26s"),
(60, "1m"),
(947.35, "15m47s"),
(1847, "30m47s"),
(2193.94, "36m34s"),
(3600, "1h"),
(13426.83, "3h43m47s"),
(16293, "4h31m33s"),
(18392.14, "5h6m32s"),
(86400, "1d"),
(129674, "1d12h1m14s"),
(56404.12, "15h40m4s"),
),
)
def test_format_seconds(self, common_obj, test_input, expected):
assert common_obj.format_seconds(test_input) == expected
# TODO: test negative numbers?
@pytest.mark.parametrize("test_input", ("string", lambda: None, [], {}, set()))
def test_invalid_input_types(self, common_obj, test_input):
with pytest.raises(TypeError):
common_obj.format_seconds(test_input)
class TestGetAvailablePort:
@pytest.mark.parametrize(
"port_min,port_max",
((random.randint(1024, 1500), random.randint(1800, 2048)) for _ in range(50)),
)
def test_returns_an_open_port(self, common_obj, port_min, port_max):
""" get_available_port() should return an open port within the range """
port = common_obj.get_available_port(port_min, port_max)
assert port_min <= port <= port_max
with socket.socket() as tmpsock:
tmpsock.bind(("127.0.0.1", port))
class TestGetPlatform:
def test_darwin(self, platform_darwin, common_obj):
assert common_obj.platform == "Darwin"
def test_linux(self, platform_linux, common_obj):
assert common_obj.platform == "Linux"
def test_windows(self, platform_windows, common_obj):
assert common_obj.platform == "Windows"
# TODO: double-check these tests
class TestGetResourcePath:
def test_onionshare_dev_mode(self, common_obj, sys_onionshare_dev_mode):
prefix = os.path.join(
os.path.dirname(
os.path.dirname(
os.path.abspath(inspect.getfile(inspect.currentframe()))
)
),
"share",
)
assert common_obj.get_resource_path(
os.path.join(prefix, "test_filename")
) == os.path.join(prefix, "test_filename")
def test_linux(self, common_obj, platform_linux, sys_argv_sys_prefix):
prefix = os.path.join(sys.prefix, "share", "onionshare")
assert common_obj.get_resource_path(
os.path.join(prefix, "test_filename")
) == os.path.join(prefix, "test_filename")
def test_frozen_darwin(self, common_obj, platform_darwin, sys_frozen, sys_meipass):
prefix = os.path.join(sys._MEIPASS, "share")
assert common_obj.get_resource_path(
os.path.join(prefix, "test_filename")
) == os.path.join(prefix, "test_filename")
class TestGetTorPaths:
@pytest.mark.skipif(sys.platform != "Darwin", reason="requires MacOS")
def test_get_tor_paths_darwin(
self, platform_darwin, common_obj, sys_frozen, sys_meipass
):
base_path = os.path.dirname(
os.path.dirname(os.path.dirname(common_obj.get_resource_path("")))
)
tor_path = os.path.join(base_path, "Resources", "Tor", "tor")
tor_geo_ip_file_path = os.path.join(base_path, "Resources", "Tor", "geoip")
tor_geo_ipv6_file_path = os.path.join(base_path, "Resources", "Tor", "geoip6")
obfs4proxy_file_path = os.path.join(base_path, "Resources", "Tor", "obfs4proxy")
assert common_obj.get_tor_paths() == (
tor_path,
tor_geo_ip_file_path,
tor_geo_ipv6_file_path,
obfs4proxy_file_path,
)
@pytest.mark.skipif(sys.platform != "Linux", reason="requires Linux")
def test_get_tor_paths_linux(self, platform_linux, common_obj):
(
tor_path,
tor_geo_ip_file_path,
tor_geo_ipv6_file_path,
_, # obfs4proxy is optional
) = common_obj.get_tor_paths()
assert os.path.basename(tor_path) == "tor"
assert (
tor_geo_ip_file_path == "/usr/share/tor/geoip"
or tor_geo_ip_file_path == "/usr/local/share/tor/geoip"
)
assert (
tor_geo_ipv6_file_path == "/usr/share/tor/geoip6"
or tor_geo_ipv6_file_path == "/usr/local/share/tor/geoip6"
)
@pytest.mark.skipif(sys.platform != "win32", reason="requires Windows")
def test_get_tor_paths_windows(self, platform_windows, common_obj, sys_frozen):
base_path = os.path.join(
os.path.dirname(os.path.dirname(common_obj.get_resource_path(""))), "tor"
)
tor_path = os.path.join(os.path.join(base_path, "Tor"), "tor.exe")
obfs4proxy_file_path = os.path.join(
os.path.join(base_path, "Tor"), "obfs4proxy.exe"
)
tor_geo_ip_file_path = os.path.join(
os.path.join(os.path.join(base_path, "Data"), "Tor"), "geoip"
)
tor_geo_ipv6_file_path = os.path.join(
os.path.join(os.path.join(base_path, "Data"), "Tor"), "geoip6"
)
assert common_obj.get_tor_paths() == (
tor_path,
tor_geo_ip_file_path,
tor_geo_ipv6_file_path,
obfs4proxy_file_path,
)
class TestHumanReadableFilesize:
@pytest.mark.parametrize(
"test_input,expected",
(
(1024 ** 0, "1.0 B"),
(1024 ** 1, "1.0 KiB"),
(1024 ** 2, "1.0 MiB"),
(1024 ** 3, "1.0 GiB"),
(1024 ** 4, "1.0 TiB"),
(1024 ** 5, "1.0 PiB"),
(1024 ** 6, "1.0 EiB"),
(1024 ** 7, "1.0 ZiB"),
(1024 ** 8, "1.0 YiB"),
),
)
def test_human_readable_filesize(self, common_obj, test_input, expected):
assert common_obj.human_readable_filesize(test_input) == expected
class TestLog:
def test_output(self, common_obj, time_strftime):
common_obj.verbose = True
# From: https://stackoverflow.com/questions/1218933
with io.StringIO() as buf, contextlib.redirect_stdout(buf):
common_obj.log("TestModule", "dummy_func")
common_obj.log("TestModule", "dummy_func", "TEST_MSG")
output = buf.getvalue()
line_one, line_two, _ = output.split("\n")
assert line_one == "[Jun 06 2013 11:05:00] TestModule.dummy_func"
assert line_two == "[Jun 06 2013 11:05:00] TestModule.dummy_func: TEST_MSG"

View file

@ -0,0 +1,149 @@
import json
import os
import tempfile
import sys
import pytest
from onionshare import common, settings, strings
@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)
class TestSettings:
def test_init(self, settings_obj):
expected_settings = {
"version": "DUMMY_VERSION_1.2.3",
"connection_type": "bundled",
"control_port_address": "127.0.0.1",
"control_port_port": 9051,
"socks_address": "127.0.0.1",
"socks_port": 9050,
"socket_file_path": "/var/run/tor/control",
"auth_type": "no_auth",
"auth_password": "",
"use_autoupdate": True,
"autoupdate_timestamp": None,
"no_bridges": True,
"tor_bridges_use_obfs4": False,
"tor_bridges_use_meek_lite_azure": False,
"tor_bridges_use_custom_bridges": "",
"persistent_tabs": [],
}
for key in settings_obj._settings:
# Skip locale, it will not always default to the same thing
if key != "locale":
assert settings_obj._settings[key] == settings_obj.default_settings[key]
assert settings_obj._settings[key] == expected_settings[key]
def test_fill_in_defaults(self, settings_obj):
del settings_obj._settings["version"]
settings_obj.fill_in_defaults()
assert settings_obj._settings["version"] == "DUMMY_VERSION_1.2.3"
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(dir=temp_dir)
with open(tmp_file, "w") as f:
json.dump(custom_settings, f)
settings_obj.filename = tmp_file_path
settings_obj.load()
assert settings_obj._settings["version"] == "CUSTOM_VERSION"
assert settings_obj._settings["socks_port"] == 9999
assert settings_obj._settings["use_stealth"] is True
os.remove(tmp_file_path)
assert os.path.exists(tmp_file_path) is False
def test_save(self, monkeypatch, temp_dir, settings_obj):
monkeypatch.setattr(strings, "_", lambda _: "")
settings_filename = "default_settings.json"
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:
settings = json.load(f)
assert settings_obj._settings == settings
os.remove(settings_path)
assert os.path.exists(settings_path) is False
def test_get(self, settings_obj):
assert settings_obj.get("version") == "DUMMY_VERSION_1.2.3"
assert settings_obj.get("connection_type") == "bundled"
assert settings_obj.get("control_port_address") == "127.0.0.1"
assert settings_obj.get("control_port_port") == 9051
assert settings_obj.get("socks_address") == "127.0.0.1"
assert settings_obj.get("socks_port") == 9050
assert settings_obj.get("socket_file_path") == "/var/run/tor/control"
assert settings_obj.get("auth_type") == "no_auth"
assert settings_obj.get("auth_password") == ""
assert settings_obj.get("use_autoupdate") is True
assert settings_obj.get("autoupdate_timestamp") is None
assert settings_obj.get("autoupdate_timestamp") is None
assert settings_obj.get("no_bridges") is True
assert settings_obj.get("tor_bridges_use_obfs4") is False
assert settings_obj.get("tor_bridges_use_meek_lite_azure") is False
assert settings_obj.get("tor_bridges_use_custom_bridges") == ""
def test_set_version(self, settings_obj):
settings_obj.set("version", "CUSTOM_VERSION")
assert settings_obj._settings["version"] == "CUSTOM_VERSION"
def test_set_control_port_port(self, settings_obj):
settings_obj.set("control_port_port", 999)
assert settings_obj._settings["control_port_port"] == 999
settings_obj.set("control_port_port", "NON_INTEGER")
assert settings_obj._settings["control_port_port"] == 9051
def test_set_socks_port(self, settings_obj):
settings_obj.set("socks_port", 888)
assert settings_obj._settings["socks_port"] == 888
settings_obj.set("socks_port", "NON_INTEGER")
assert settings_obj._settings["socks_port"] == 9050
@pytest.mark.skipif(sys.platform != "Darwin", reason="requires Darwin")
def test_filename_darwin(self, monkeypatch, platform_darwin):
obj = settings.Settings(common.Common())
assert obj.filename == os.path.expanduser(
"~/Library/Application Support/OnionShare-testdata/onionshare.json"
)
@pytest.mark.skipif(sys.platform != "Linux", reason="requires Linux")
def test_filename_linux(self, monkeypatch, platform_linux):
obj = settings.Settings(common.Common())
assert obj.filename == os.path.expanduser(
"~/.config/onionshare-testdata/onionshare.json"
)
@pytest.mark.skipif(sys.platform != "win32", reason="requires Windows")
def test_filename_windows(self, monkeypatch, platform_windows):
obj = settings.Settings(common.Common())
assert obj.filename == os.path.expanduser(
"~\\AppData\\Roaming\\OnionShare-testdata\\onionshare.json"
)
def test_set_custom_bridge(self, settings_obj):
settings_obj.set(
"tor_bridges_use_custom_bridges",
"Bridge 45.3.20.65:9050 21300AD88890A49C429A6CB9959CFD44490A8F6E",
)
assert (
settings_obj._settings["tor_bridges_use_custom_bridges"]
== "Bridge 45.3.20.65:9050 21300AD88890A49C429A6CB9959CFD44490A8F6E"
)

View file

@ -0,0 +1,46 @@
# -*- coding: utf-8 -*-
import types
import pytest
from onionshare import strings
from onionshare.settings import Settings
# # Stub get_resource_path so it finds the correct path while running tests
# def get_resource_path(filename):
# resources_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'share')
# path = os.path.join(resources_dir, filename)
# return path
# common.get_resource_path = get_resource_path
def test_underscore_is_function():
assert callable(strings._) and isinstance(strings._, types.FunctionType)
class TestLoadStrings:
def test_load_strings_defaults_to_english(
self, common_obj, locale_en, sys_onionshare_dev_mode
):
""" load_strings() loads English by default """
common_obj.settings = Settings(common_obj)
strings.load_strings(common_obj)
assert strings._("not_a_readable_file") == "{0:s} is not a readable file."
def test_load_strings_loads_other_languages(
self, common_obj, locale_fr, sys_onionshare_dev_mode
):
""" load_strings() loads other languages in different locales """
common_obj.settings = Settings(common_obj)
common_obj.settings.set("locale", "fr")
strings.load_strings(common_obj)
assert strings._("not_a_readable_file") == "{0:s} nest pas un fichier lisible."
def test_load_invalid_locale(
self, common_obj, locale_invalid, sys_onionshare_dev_mode
):
""" load_strings() raises a KeyError for an invalid locale """
with pytest.raises(KeyError):
common_obj.settings = Settings(common_obj)
common_obj.settings.set("locale", "XX")
strings.load_strings(common_obj)

View file

@ -0,0 +1,233 @@
import contextlib
import inspect
import io
import os
import random
import re
import socket
import sys
import zipfile
import tempfile
import base64
import pytest
from werkzeug.datastructures import Headers
from onionshare.common import Common
from onionshare import strings
from onionshare.web import Web
from onionshare.settings import Settings
from onionshare.mode_settings import ModeSettings
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(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)
mode_settings = ModeSettings(common_obj)
web = Web(common_obj, False, mode_settings, mode)
web.generate_password()
web.running = True
web.app.testing = True
# Share mode
if mode == "share":
# Add files
files = []
for _ in range(num_files):
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)
# Receive mode
else:
pass
return web
class TestWeb:
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
res = c.get("/")
res.get_data()
assert res.status_code == 401
# Load / with invalid auth
res = c.get("/", headers=self._make_auth_headers("invalid"))
res.get_data()
assert res.status_code == 401
# Load / with valid auth
res = c.get("/", headers=self._make_auth_headers(web.password))
res.get_data()
assert res.status_code == 200
# Download
res = c.get("/download", headers=self._make_auth_headers(web.password))
res.get_data()
assert res.status_code == 200
assert (
res.mimetype == "application/zip"
or res.mimetype == "application/x-zip-compressed"
)
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
with web.app.test_client() as c:
# Download the first time
res = c.get("/download", headers=self._make_auth_headers(web.password))
res.get_data()
assert res.status_code == 200
assert (
res.mimetype == "application/zip"
or res.mimetype == "application/x-zip-compressed"
)
assert web.running == False
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
with web.app.test_client() as c:
# Download the first time
res = c.get("/download", headers=self._make_auth_headers(web.password))
res.get_data()
assert res.status_code == 200
assert (
res.mimetype == "application/zip"
or res.mimetype == "application/x-zip-compressed"
)
assert web.running == True
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:
# Load / without auth
res = c.get("/")
res.get_data()
assert res.status_code == 401
# Load / with invalid auth
res = c.get("/", headers=self._make_auth_headers("invalid"))
res.get_data()
assert res.status_code == 401
# Load / with valid auth
res = c.get("/", headers=self._make_auth_headers(web.password))
res.get_data()
assert res.status_code == 200
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:
# Loading / should work without auth
res = c.get("/")
data1 = res.get_data()
assert res.status_code == 200
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:
# Load / without auth
res = c.get("/")
res.get_data()
assert res.status_code == 401
# But static resources should work without auth
res = c.get(f"{web.static_url_path}/css/style.css")
res.get_data()
assert res.status_code == 200
# Load / with valid auth
res = c.get("/", headers=self._make_auth_headers(web.password))
res.get_data()
assert res.status_code == 200
def _make_auth_headers(self, password):
auth = base64.b64encode(b"onionshare:" + password.encode()).decode()
h = Headers()
h.add("Authorization", "Basic " + auth)
return h
class TestZipWriterDefault:
@pytest.mark.parametrize(
"test_input",
(
f"onionshare_{''.join(random.choice('abcdefghijklmnopqrstuvwxyz234567') for _ in range(6))}.zip"
for _ in range(50)
),
)
def test_default_zw_filename_regex(self, test_input):
assert bool(DEFAULT_ZW_FILENAME_REGEX.match(test_input))
def test_zw_filename(self, default_zw):
zw_filename = os.path.basename(default_zw.zip_filename)
assert bool(DEFAULT_ZW_FILENAME_REGEX.match(zw_filename))
def test_zipfile_filename_matches_zipwriter_filename(self, default_zw):
assert default_zw.z.filename == default_zw.zip_filename
def test_zipfile_allow_zip64(self, default_zw):
assert default_zw.z._allowZip64 is True
def test_zipfile_mode(self, default_zw):
assert default_zw.z.mode == "w"
def test_callback(self, default_zw):
assert default_zw.processed_size_callback(None) is None
def test_add_file(self, default_zw, temp_file_1024_delete):
default_zw.add_file(temp_file_1024_delete)
zipfile_info = default_zw.z.getinfo(os.path.basename(temp_file_1024_delete))
assert zipfile_info.compress_type == zipfile.ZIP_DEFLATED
assert zipfile_info.file_size == 1024
def test_add_directory(self, temp_dir_1024_delete, default_zw):
previous_size = default_zw._size # size before adding directory
default_zw.add_dir(temp_dir_1024_delete)
assert default_zw._size == previous_size + 1024
class TestZipWriterCustom:
@pytest.mark.parametrize(
"test_input",
(
Common.random_string(
random.randint(2, 50), random.choice((None, random.randint(2, 50)))
)
for _ in range(50)
),
)
def test_random_string_regex(self, test_input):
assert bool(RANDOM_STR_REGEX.match(test_input))
def test_custom_filename(self, custom_zw):
assert bool(RANDOM_STR_REGEX.match(custom_zw.zip_filename))
def test_custom_callback(self, custom_zw):
assert custom_zw.processed_size_callback(None) == "custom_callback"

View file

@ -0,0 +1,252 @@
import pytest
import os
import requests
import shutil
import sys
from datetime import datetime, timedelta
from PyQt5 import QtCore, QtTest
from .gui_base_test import GuiBaseTest
class TestReceive(GuiBaseTest):
# Shared test methods
def upload_file(
self, tab, file_to_upload, expected_basename, identical_files_at_once=False
):
"""Test that we can upload the file"""
# Wait 2 seconds to make sure the filename, based on timestamp, isn't accidentally reused
QtTest.QTest.qWait(2000)
files = {"file[]": open(file_to_upload, "rb")}
url = f"http://127.0.0.1:{tab.app.port}/upload"
if tab.settings.get("general", "public"):
requests.post(url, files=files)
if identical_files_at_once:
# Send a duplicate upload to test for collisions
requests.post(url, files=files)
else:
requests.post(
url,
files=files,
auth=requests.auth.HTTPBasicAuth(
"onionshare", tab.get_mode().web.password
),
)
if identical_files_at_once:
# Send a duplicate upload to test for collisions
requests.post(
url,
files=files,
auth=requests.auth.HTTPBasicAuth(
"onionshare", tab.get_mode().web.password
),
)
QtTest.QTest.qWait(1000)
# Make sure the file is within the last 10 seconds worth of fileames
exists = False
now = datetime.now()
for _ in range(10):
date_dir = now.strftime("%Y-%m-%d")
if identical_files_at_once:
time_dir = now.strftime("%H.%M.%S-1")
else:
time_dir = now.strftime("%H.%M.%S")
receive_mode_dir = os.path.join(
tab.settings.get("receive", "data_dir"), date_dir, time_dir
)
expected_filename = os.path.join(receive_mode_dir, expected_basename)
if os.path.exists(expected_filename):
exists = True
break
now = now - timedelta(seconds=1)
self.assertTrue(exists)
def upload_file_should_fail(self, tab):
"""Test that we can't upload the file when permissions are wrong, and expected content is shown"""
QtTest.QTest.qWait(1000)
files = {"file[]": open(self.tmpfile_test, "rb")}
url = f"http://127.0.0.1:{tab.app.port}/upload"
if tab.settings.get("general", "public"):
r = requests.post(url, files=files)
else:
r = requests.post(
url,
files=files,
auth=requests.auth.HTTPBasicAuth(
"onionshare", tab.get_mode().web.password
),
)
def accept_dialog():
window = tab.common.gui.qtapp.activeWindow()
if window:
window.close()
QtCore.QTimer.singleShot(1000, accept_dialog)
self.assertTrue("Error uploading, please inform the OnionShare user" in r.text)
def try_without_auth_in_non_public_mode(self, tab):
r = requests.post(f"http://127.0.0.1:{tab.app.port}/upload")
self.assertEqual(r.status_code, 401)
r = requests.get(f"http://127.0.0.1:{tab.app.port}/close")
self.assertEqual(r.status_code, 401)
# 'Grouped' tests follow from here
def run_all_receive_mode_setup_tests(self, tab):
"""Set up a share in Receive mode and start it"""
self.history_is_not_visible(tab)
self.click_toggle_history(tab)
self.history_is_visible(tab)
self.server_working_on_start_button_pressed(tab)
self.server_status_indicator_says_starting(tab)
self.server_is_started(tab)
self.web_server_is_running(tab)
self.have_a_password(tab)
self.url_description_shown(tab)
self.have_copy_url_button(tab)
self.have_show_qr_code_button(tab)
self.server_status_indicator_says_started(tab)
self.web_page(tab, "Select the files you want to send, then click")
def run_all_receive_mode_tests(self, tab):
"""Upload files in receive mode and stop the share"""
self.run_all_receive_mode_setup_tests(tab)
if not tab.settings.get("general", "public"):
self.try_without_auth_in_non_public_mode(tab)
self.upload_file(tab, self.tmpfile_test, "test.txt")
self.history_widgets_present(tab)
self.counter_incremented(tab, 1)
self.upload_file(tab, self.tmpfile_test, "test.txt")
self.counter_incremented(tab, 2)
self.upload_file(tab, self.tmpfile_test2, "test2.txt")
self.counter_incremented(tab, 3)
self.upload_file(tab, self.tmpfile_test2, "test2.txt")
self.counter_incremented(tab, 4)
# Test uploading the same file twice at the same time, and make sure no collisions
self.upload_file(tab, self.tmpfile_test, "test.txt", True)
self.counter_incremented(tab, 6)
self.history_indicator(tab, "2")
self.server_is_stopped(tab)
self.web_server_is_stopped(tab)
self.server_status_indicator_says_closed(tab)
self.server_working_on_start_button_pressed(tab)
self.server_is_started(tab)
self.history_indicator(tab, "2")
def run_all_clear_all_button_tests(self, tab):
"""Test the Clear All history button"""
self.run_all_receive_mode_setup_tests(tab)
self.upload_file(tab, self.tmpfile_test, "test.txt")
self.history_widgets_present(tab)
self.clear_all_history_items(tab, 0)
self.upload_file(tab, self.tmpfile_test, "test.txt")
self.clear_all_history_items(tab, 2)
def run_all_upload_non_writable_dir_tests(self, tab):
"""Test uploading a file when the data_dir is non-writable"""
upload_dir = os.path.join(self.tmpdir.name, "OnionShare")
shutil.rmtree(upload_dir, ignore_errors=True)
os.makedirs(upload_dir, 0o700)
# Set the upload dir setting
tab.get_mode().data_dir_lineedit.setText(upload_dir)
tab.settings.set("receive", "data_dir", upload_dir)
self.run_all_receive_mode_setup_tests(tab)
os.chmod(upload_dir, 0o400)
self.upload_file_should_fail(tab)
self.server_is_stopped(tab)
self.web_server_is_stopped(tab)
self.server_status_indicator_says_closed(tab)
os.chmod(upload_dir, 0o700)
# Tests
@pytest.mark.gui
def test_clear_all_button(self):
"""
Clear all history items should work
"""
tab = self.new_receive_tab()
self.run_all_common_setup_tests()
self.run_all_clear_all_button_tests(tab)
self.close_all_tabs()
@pytest.mark.gui
def test_autostop_timer(self):
"""
Test autostop timer
"""
tab = self.new_receive_tab()
tab.get_mode().mode_settings_widget.toggle_advanced_button.click()
tab.get_mode().mode_settings_widget.autostop_timer_checkbox.click()
self.run_all_common_setup_tests()
self.run_all_receive_mode_setup_tests(tab)
self.set_timeout(tab, 5)
self.autostop_timer_widget_hidden(tab)
self.server_timed_out(tab, 15000)
self.web_server_is_stopped(tab)
self.close_all_tabs()
@pytest.mark.gui
def test_upload(self):
"""
Test uploading files
"""
tab = self.new_receive_tab()
self.run_all_common_setup_tests()
self.run_all_receive_mode_tests(tab)
self.close_all_tabs()
@pytest.mark.gui
@pytest.mark.skipif(sys.platform == "win32", reason="Windows doesn't have chmod")
def test_upload_non_writable_dir(self):
"""
Test uploading files to a non-writable directory
"""
tab = self.new_receive_tab()
self.run_all_upload_non_writable_dir_tests(tab)
self.close_all_tabs()
@pytest.mark.gui
def test_public_upload(self):
"""
Test uploading files in public mode
"""
tab = self.new_receive_tab()
tab.get_mode().mode_settings_widget.public_checkbox.click()
self.run_all_common_setup_tests()
self.run_all_receive_mode_tests(tab)
self.close_all_tabs()
@pytest.mark.gui
@pytest.mark.skipif(sys.platform == "win32", reason="Windows doesn't have chmod")
def test_public_upload_non_writable_dir(self):
"""
Test uploading files to a non-writable directory in public mode
"""
tab = self.new_receive_tab()
tab.get_mode().mode_settings_widget.public_checkbox.click()
self.run_all_upload_non_writable_dir_tests(tab)
self.close_all_tabs()

View file

@ -0,0 +1,629 @@
import pytest
import os
import requests
import tempfile
import zipfile
from PyQt5 import QtCore, QtTest
from .gui_base_test import GuiBaseTest
class TestShare(GuiBaseTest):
# Shared test methods
def removing_all_files_hides_remove_button(self, tab):
"""Test that clicking on the file item shows the remove button. Test that removing the only item in the list hides the remove button"""
rect = tab.get_mode().server_status.file_selection.file_list.visualItemRect(
tab.get_mode().server_status.file_selection.file_list.item(0)
)
QtTest.QTest.mouseClick(
tab.get_mode().server_status.file_selection.file_list.viewport(),
QtCore.Qt.LeftButton,
pos=rect.center(),
)
# Remove button should be visible
self.assertTrue(
tab.get_mode().server_status.file_selection.remove_button.isVisible()
)
# Click remove, remove button should still be visible since we have one more file
tab.get_mode().server_status.file_selection.remove_button.click()
rect = tab.get_mode().server_status.file_selection.file_list.visualItemRect(
tab.get_mode().server_status.file_selection.file_list.item(0)
)
QtTest.QTest.mouseClick(
tab.get_mode().server_status.file_selection.file_list.viewport(),
QtCore.Qt.LeftButton,
pos=rect.center(),
)
self.assertTrue(
tab.get_mode().server_status.file_selection.remove_button.isVisible()
)
tab.get_mode().server_status.file_selection.remove_button.click()
# No more files, the remove button should be hidden
self.assertFalse(
tab.get_mode().server_status.file_selection.remove_button.isVisible()
)
def add_a_file_and_remove_using_its_remove_widget(self, tab):
"""Test that we can also remove a file by clicking on its [X] widget"""
num_files = tab.get_mode().server_status.file_selection.get_num_files()
tab.get_mode().server_status.file_selection.file_list.add_file(self.tmpfiles[0])
tab.get_mode().server_status.file_selection.file_list.item(
0
).item_button.click()
self.file_selection_widget_has_files(tab, num_files)
def add_a_file_and_remove_using_remove_all_widget(self, tab):
"""Test that we can also remove all files by clicking on the Remove All widget"""
tab.get_mode().server_status.file_selection.file_list.add_file(self.tmpfiles[0])
tab.get_mode().server_status.file_selection.file_list.add_file(self.tmpfiles[1])
tab.get_mode().remove_all_button.click()
# Should be no files after clearing all
self.file_selection_widget_has_files(tab, 0)
def file_selection_widget_read_files(self, tab):
"""Re-add some files to the list so we can share"""
num_files = tab.get_mode().server_status.file_selection.get_num_files()
tab.get_mode().server_status.file_selection.file_list.add_file(self.tmpfiles[0])
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 download_share(self, tab):
"""Test that we can download the share"""
url = f"http://127.0.0.1:{tab.app.port}/download"
if tab.settings.get("general", "public"):
r = requests.get(url)
else:
r = requests.get(
url,
auth=requests.auth.HTTPBasicAuth(
"onionshare", tab.get_mode().server_status.web.password
),
)
tmp_file = tempfile.NamedTemporaryFile("wb", delete=False)
tmp_file.write(r.content)
tmp_file.close()
z = zipfile.ZipFile(tmp_file.name)
QtTest.QTest.qWait(50)
self.assertEqual("onionshare", z.read("test.txt").decode("utf-8"))
QtTest.QTest.qWait(500)
def individual_file_is_viewable_or_not(self, tab):
"""
Test that an individual file is viewable (when in autostop_sharing is false) or that it
isn't (when not in autostop_sharing is true)
"""
url = f"http://127.0.0.1:{tab.app.port}"
download_file_url = f"http://127.0.0.1:{tab.app.port}/test.txt"
if tab.settings.get("general", "public"):
r = requests.get(url)
else:
r = requests.get(
url,
auth=requests.auth.HTTPBasicAuth(
"onionshare", tab.get_mode().server_status.web.password
),
)
if tab.settings.get("share", "autostop_sharing"):
self.assertFalse('a href="/test.txt"' in r.text)
if tab.settings.get("general", "public"):
r = requests.get(download_file_url)
else:
r = requests.get(
download_file_url,
auth=requests.auth.HTTPBasicAuth(
"onionshare", tab.get_mode().server_status.web.password
),
)
self.assertEqual(r.status_code, 404)
self.download_share(tab)
else:
self.assertTrue('a href="test.txt"' in r.text)
if tab.settings.get("general", "public"):
r = requests.get(download_file_url)
else:
r = requests.get(
download_file_url,
auth=requests.auth.HTTPBasicAuth(
"onionshare", tab.get_mode().server_status.web.password
),
)
tmp_file = tempfile.NamedTemporaryFile("wb", delete=False)
tmp_file.write(r.content)
tmp_file.close()
with open(tmp_file.name, "r") as f:
self.assertEqual("onionshare", f.read())
os.remove(tmp_file.name)
QtTest.QTest.qWait(500)
def hit_401(self, tab):
"""Test that the server stops after too many 401s, or doesn't when in public mode"""
# In non-public mode, get ready to accept the dialog
if not tab.settings.get("general", "public"):
def accept_dialog():
window = tab.common.gui.qtapp.activeWindow()
if window:
window.close()
QtCore.QTimer.singleShot(1000, accept_dialog)
# Make 20 requests with guessed passwords
url = f"http://127.0.0.1:{tab.app.port}/"
for _ in range(20):
password_guess = self.gui.common.build_password()
requests.get(
url, auth=requests.auth.HTTPBasicAuth("onionshare", password_guess)
)
# In public mode, we should still be running (no rate-limiting)
if tab.settings.get("general", "public"):
self.web_server_is_running(tab)
# In non-public mode, we should be shut down (rate-limiting)
else:
self.web_server_is_stopped(tab)
def set_autostart_timer(self, tab, timer):
"""Test that the timer can be set"""
schedule = QtCore.QDateTime.currentDateTime().addSecs(timer)
tab.get_mode().mode_settings_widget.autostart_timer_widget.setDateTime(schedule)
self.assertTrue(
tab.get_mode().mode_settings_widget.autostart_timer_widget.dateTime(),
schedule,
)
def autostart_timer_widget_hidden(self, tab):
"""Test that the auto-start timer widget is hidden when share has started"""
self.assertFalse(
tab.get_mode().mode_settings_widget.autostart_timer_widget.isVisible()
)
def scheduled_service_started(self, tab, wait):
"""Test that the server has timed out after the timer ran out"""
QtTest.QTest.qWait(wait)
# We should have started now
self.assertEqual(tab.get_mode().server_status.status, 2)
def cancel_the_share(self, tab):
"""Test that we can cancel a share before it's started up """
self.server_working_on_start_button_pressed(tab)
self.server_status_indicator_says_scheduled(tab)
self.add_remove_buttons_hidden(tab)
self.mode_settings_widget_is_hidden(tab)
self.set_autostart_timer(tab, 10)
QtTest.QTest.qWait(500)
QtTest.QTest.mousePress(
tab.get_mode().server_status.server_button, QtCore.Qt.LeftButton
)
QtTest.QTest.qWait(100)
QtTest.QTest.mouseRelease(
tab.get_mode().server_status.server_button, QtCore.Qt.LeftButton
)
QtTest.QTest.qWait(500)
self.assertEqual(
tab.get_mode().server_status.status,
tab.get_mode().server_status.STATUS_STOPPED,
)
self.server_is_stopped(tab)
self.web_server_is_stopped(tab)
# Grouped tests follow from here
def run_all_share_mode_setup_tests(self, tab):
"""Tests in share mode prior to starting a share"""
tab.get_mode().server_status.file_selection.file_list.add_file(
self.tmpfile_test
)
tab.get_mode().server_status.file_selection.file_list.add_file(self.tmpfiles[0])
tab.get_mode().server_status.file_selection.file_list.add_file(self.tmpfiles[1])
self.file_selection_widget_has_files(tab, 3)
self.history_is_not_visible(tab)
self.click_toggle_history(tab)
self.history_is_visible(tab)
self.removing_all_files_hides_remove_button(tab)
self.add_a_file_and_remove_using_its_remove_widget(tab)
self.file_selection_widget_read_files(tab)
def run_all_share_mode_started_tests(self, tab, startup_time=2000):
"""Tests in share mode after starting a share"""
self.server_working_on_start_button_pressed(tab)
self.server_status_indicator_says_starting(tab)
self.add_remove_buttons_hidden(tab)
self.mode_settings_widget_is_hidden(tab)
self.server_is_started(tab, startup_time)
self.web_server_is_running(tab)
self.have_a_password(tab)
self.url_description_shown(tab)
self.have_copy_url_button(tab)
self.have_show_qr_code_button(tab)
self.server_status_indicator_says_started(tab)
def run_all_share_mode_download_tests(self, tab):
"""Tests in share mode after downloading a share"""
tab.get_mode().server_status.file_selection.file_list.add_file(
self.tmpfile_test
)
self.web_page(tab, "Total size")
self.download_share(tab)
self.history_widgets_present(tab)
self.server_is_stopped(tab)
self.web_server_is_stopped(tab)
self.server_status_indicator_says_closed(tab)
self.add_button_visible(tab)
self.server_working_on_start_button_pressed(tab)
self.toggle_indicator_is_reset(tab)
self.server_is_started(tab)
self.history_indicator(tab)
def run_all_share_mode_individual_file_download_tests(self, tab):
"""Tests in share mode after downloading a share"""
self.web_page(tab, "Total size")
self.individual_file_is_viewable_or_not(tab)
self.history_widgets_present(tab)
self.server_is_stopped(tab)
self.web_server_is_stopped(tab)
self.server_status_indicator_says_closed(tab)
self.add_button_visible(tab)
self.server_working_on_start_button_pressed(tab)
self.server_is_started(tab)
self.history_indicator(tab)
def run_all_share_mode_tests(self, tab):
"""End-to-end share tests"""
self.run_all_share_mode_setup_tests(tab)
self.run_all_share_mode_started_tests(tab)
self.run_all_share_mode_download_tests(tab)
def run_all_clear_all_history_button_tests(self, tab):
"""Test the Clear All history button"""
self.run_all_share_mode_setup_tests(tab)
self.run_all_share_mode_started_tests(tab)
self.individual_file_is_viewable_or_not(tab)
self.history_widgets_present(tab)
self.clear_all_history_items(tab, 0)
self.individual_file_is_viewable_or_not(tab)
self.clear_all_history_items(tab, 2)
def run_all_remove_all_file_selection_button_tests(self, tab):
"""Test the Remove All File Selection button"""
self.run_all_share_mode_setup_tests(tab)
self.add_a_file_and_remove_using_remove_all_widget(tab)
def run_all_share_mode_individual_file_tests(self, tab):
"""Tests in share mode when viewing an individual file"""
self.run_all_share_mode_setup_tests(tab)
self.run_all_share_mode_started_tests(tab)
self.run_all_share_mode_individual_file_download_tests(tab)
# Tests
@pytest.mark.gui
def test_autostart_and_autostop_timer_mismatch(self):
"""
If autostart timer is after autostop timer, a warning should be thrown
"""
tab = self.new_share_tab()
tab.get_mode().mode_settings_widget.toggle_advanced_button.click()
tab.get_mode().mode_settings_widget.autostart_timer_checkbox.click()
tab.get_mode().mode_settings_widget.autostop_timer_checkbox.click()
def accept_dialog():
window = tab.common.gui.qtapp.activeWindow()
if window:
window.close()
self.run_all_common_setup_tests()
self.run_all_share_mode_setup_tests(tab)
self.set_autostart_timer(tab, 15)
self.set_timeout(tab, 5)
QtCore.QTimer.singleShot(200, accept_dialog)
tab.get_mode().server_status.server_button.click()
self.server_is_stopped(tab)
self.close_all_tabs()
@pytest.mark.gui
def test_autostart_timer(self):
"""
Autostart timer should automatically start
"""
tab = self.new_share_tab()
tab.get_mode().mode_settings_widget.toggle_advanced_button.click()
tab.get_mode().mode_settings_widget.autostart_timer_checkbox.click()
self.run_all_common_setup_tests()
self.run_all_share_mode_setup_tests(tab)
self.set_autostart_timer(tab, 2)
self.server_working_on_start_button_pressed(tab)
self.autostart_timer_widget_hidden(tab)
self.server_status_indicator_says_scheduled(tab)
self.web_server_is_stopped(tab)
self.scheduled_service_started(tab, 2200)
self.web_server_is_running(tab)
self.close_all_tabs()
@pytest.mark.gui
def test_autostart_timer_too_short(self):
"""
Autostart timer should throw a warning if the scheduled time is too soon
"""
tab = self.new_share_tab()
tab.get_mode().mode_settings_widget.toggle_advanced_button.click()
tab.get_mode().mode_settings_widget.autostart_timer_checkbox.click()
def accept_dialog():
window = tab.common.gui.qtapp.activeWindow()
if window:
window.close()
self.run_all_common_setup_tests()
self.run_all_share_mode_setup_tests(tab)
# Set a low timeout
self.set_autostart_timer(tab, 2)
QtTest.QTest.qWait(2200)
QtCore.QTimer.singleShot(200, accept_dialog)
tab.get_mode().server_status.server_button.click()
self.assertEqual(tab.get_mode().server_status.status, 0)
self.close_all_tabs()
@pytest.mark.gui
def test_autostart_timer_cancel(self):
"""
Test canceling a scheduled share
"""
tab = self.new_share_tab()
tab.get_mode().mode_settings_widget.toggle_advanced_button.click()
tab.get_mode().mode_settings_widget.autostart_timer_checkbox.click()
self.run_all_common_setup_tests()
self.run_all_share_mode_setup_tests(tab)
self.cancel_the_share(tab)
self.close_all_tabs()
@pytest.mark.gui
def test_clear_all_history_button(self):
"""
Test clearing all history items
"""
tab = self.new_share_tab()
tab.get_mode().autostop_sharing_checkbox.click()
self.run_all_common_setup_tests()
self.run_all_clear_all_history_button_tests(tab)
self.close_all_tabs()
@pytest.mark.gui
def test_remove_all_file_selection_button(self):
"""
Test remove all file items at once
"""
tab = self.new_share_tab()
self.run_all_common_setup_tests()
self.run_all_remove_all_file_selection_button_tests(tab)
self.close_all_tabs()
@pytest.mark.gui
def test_public_mode(self):
"""
Public mode shouldn't have a password
"""
tab = self.new_share_tab()
tab.get_mode().mode_settings_widget.public_checkbox.click()
self.run_all_common_setup_tests()
self.run_all_share_mode_tests(tab)
self.close_all_tabs()
@pytest.mark.gui
def test_without_autostop_sharing(self):
"""
Disable autostop sharing after first download
"""
tab = self.new_share_tab()
tab.get_mode().autostop_sharing_checkbox.click()
self.run_all_common_setup_tests()
self.run_all_share_mode_tests(tab)
self.close_all_tabs()
@pytest.mark.gui
def test_download(self):
"""
Test downloading in share mode
"""
tab = self.new_share_tab()
self.run_all_common_setup_tests()
self.run_all_share_mode_tests(tab)
self.close_all_tabs()
@pytest.mark.gui
def test_individual_files_without_autostop_sharing(self):
"""
Test downloading individual files with autostop sharing disabled
"""
tab = self.new_share_tab()
tab.get_mode().autostop_sharing_checkbox.click()
self.run_all_common_setup_tests()
self.run_all_share_mode_individual_file_tests(tab)
self.close_all_tabs()
@pytest.mark.gui
def test_individual_files(self):
"""
Test downloading individual files
"""
tab = self.new_share_tab()
self.run_all_common_setup_tests()
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()
@pytest.mark.gui
def test_persistent_password(self):
"""
Test a large download
"""
tab = self.new_share_tab()
tab.get_mode().mode_settings_widget.persistent_checkbox.click()
self.run_all_common_setup_tests()
self.run_all_share_mode_setup_tests(tab)
self.run_all_share_mode_started_tests(tab)
password = tab.get_mode().server_status.web.password
self.run_all_share_mode_download_tests(tab)
self.run_all_share_mode_started_tests(tab)
self.assertEqual(tab.get_mode().server_status.web.password, password)
self.run_all_share_mode_download_tests(tab)
self.close_all_tabs()
@pytest.mark.gui
def test_autostop_timer(self):
"""
Test the autostop timer
"""
tab = self.new_share_tab()
tab.get_mode().mode_settings_widget.toggle_advanced_button.click()
tab.get_mode().mode_settings_widget.autostop_timer_checkbox.click()
self.run_all_common_setup_tests()
self.run_all_share_mode_setup_tests(tab)
self.set_timeout(tab, 5)
self.run_all_share_mode_started_tests(tab)
self.autostop_timer_widget_hidden(tab)
self.server_timed_out(tab, 10000)
self.web_server_is_stopped(tab)
self.close_all_tabs()
@pytest.mark.gui
def test_autostop_timer_too_short(self):
"""
Test the autostop timer when the timeout is too short
"""
tab = self.new_share_tab()
tab.get_mode().mode_settings_widget.toggle_advanced_button.click()
tab.get_mode().mode_settings_widget.autostop_timer_checkbox.click()
def accept_dialog():
window = tab.common.gui.qtapp.activeWindow()
if window:
window.close()
self.run_all_common_setup_tests()
self.run_all_share_mode_setup_tests(tab)
# Set a low timeout
self.set_timeout(tab, 2)
QtTest.QTest.qWait(2100)
QtCore.QTimer.singleShot(2200, accept_dialog)
tab.get_mode().server_status.server_button.click()
self.assertEqual(tab.get_mode().server_status.status, 0)
self.close_all_tabs()
@pytest.mark.gui
def test_unreadable_file(self):
"""
Sharing an unreadable file should throw a warning
"""
tab = self.new_share_tab()
def accept_dialog():
window = tab.common.gui.qtapp.activeWindow()
if window:
window.close()
self.run_all_share_mode_setup_tests(tab)
QtCore.QTimer.singleShot(200, accept_dialog)
tab.get_mode().server_status.file_selection.file_list.add_file(
"/tmp/nonexistent.txt"
)
self.file_selection_widget_has_files(tab, 3)
self.close_all_tabs()
@pytest.mark.gui
def test_401_triggers_ratelimit(self):
"""
Rate limit should be triggered
"""
tab = self.new_share_tab()
def accept_dialog():
window = tab.common.gui.qtapp.activeWindow()
if window:
window.close()
tab.get_mode().autostop_sharing_checkbox.click()
self.run_all_common_setup_tests()
self.run_all_share_mode_tests(tab)
self.hit_401(tab)
self.close_all_tabs()
@pytest.mark.gui
def test_401_public_skips_ratelimit(self):
"""
Public mode should skip the rate limit
"""
tab = self.new_share_tab()
def accept_dialog():
window = tab.common.gui.qtapp.activeWindow()
if window:
window.close()
tab.get_mode().autostop_sharing_checkbox.click()
tab.get_mode().mode_settings_widget.public_checkbox.click()
self.run_all_common_setup_tests()
self.run_all_share_mode_tests(tab)
self.hit_401(tab)
self.close_all_tabs()

View file

@ -0,0 +1,236 @@
import pytest
import os
from PyQt5 import QtCore, QtTest, QtWidgets
from .gui_base_test import GuiBaseTest
class TestTabs(GuiBaseTest):
# Shared test methods
def close_tab_with_active_server(self, tab):
# Start the server
self.assertEqual(
tab.get_mode().server_status.status,
tab.get_mode().server_status.STATUS_STOPPED,
)
tab.get_mode().server_status.server_button.click()
self.assertEqual(
tab.get_mode().server_status.status,
tab.get_mode().server_status.STATUS_WORKING,
)
QtTest.QTest.qWait(1000)
self.assertEqual(
tab.get_mode().server_status.status,
tab.get_mode().server_status.STATUS_STARTED,
)
# Prepare to reject the dialog
QtCore.QTimer.singleShot(0, tab.close_dialog.reject_button.click)
# Close tab
self.gui.tabs.tabBar().tabButton(0, QtWidgets.QTabBar.RightSide).click()
# The tab should still be open
self.assertFalse(tab.new_tab.isVisible())
self.assertTrue(tab.get_mode().isVisible())
# Prepare to accept the dialog
QtCore.QTimer.singleShot(0, tab.close_dialog.accept_button.click)
# Close tab
self.gui.tabs.tabBar().tabButton(0, QtWidgets.QTabBar.RightSide).click()
# The tab should be closed
self.assertTrue(self.gui.tabs.widget(0).new_tab.isVisible())
def close_persistent_tab(self, tab):
# There shouldn't be a persistent settings file
self.assertFalse(os.path.exists(tab.settings.filename))
# Click the persistent checkbox
tab.get_mode().server_status.mode_settings_widget.persistent_checkbox.click()
QtTest.QTest.qWait(100)
# There should be a persistent settings file now
self.assertTrue(os.path.exists(tab.settings.filename))
# Prepare to reject the dialog
QtCore.QTimer.singleShot(0, tab.close_dialog.reject_button.click)
# Close tab
self.gui.tabs.tabBar().tabButton(0, QtWidgets.QTabBar.RightSide).click()
# The tab should still be open
self.assertFalse(tab.new_tab.isVisible())
self.assertTrue(tab.get_mode().isVisible())
# There should be a persistent settings file still
self.assertTrue(os.path.exists(tab.settings.filename))
# Prepare to accept the dialog
QtCore.QTimer.singleShot(0, tab.close_dialog.accept_button.click)
# Close tab
self.gui.tabs.tabBar().tabButton(0, QtWidgets.QTabBar.RightSide).click()
# The tab should be closed
self.assertTrue(self.gui.tabs.widget(0).new_tab.isVisible())
# The persistent settings file should be deleted
self.assertFalse(os.path.exists(tab.settings.filename))
# Tests
@pytest.mark.gui
def test_01_common_tests(self):
"""Run all common tests"""
self.run_all_common_setup_tests()
@pytest.mark.gui
def test_02_starts_with_one_new_tab(self):
"""There should be one "New Tab" tab open"""
self.assertEqual(self.gui.tabs.count(), 1)
self.assertTrue(self.gui.tabs.widget(0).new_tab.isVisible())
@pytest.mark.gui
def test_03_new_tab_button_opens_new_tabs(self):
"""Clicking the "+" button should open new tabs"""
self.assertEqual(self.gui.tabs.count(), 1)
self.gui.tabs.new_tab_button.click()
self.gui.tabs.new_tab_button.click()
self.gui.tabs.new_tab_button.click()
self.assertEqual(self.gui.tabs.count(), 4)
@pytest.mark.gui
def test_04_close_tab_button_closes_tabs(self):
"""Clicking the "x" button should close tabs"""
self.assertEqual(self.gui.tabs.count(), 4)
self.gui.tabs.tabBar().tabButton(0, QtWidgets.QTabBar.RightSide).click()
self.gui.tabs.tabBar().tabButton(0, QtWidgets.QTabBar.RightSide).click()
self.gui.tabs.tabBar().tabButton(0, QtWidgets.QTabBar.RightSide).click()
self.assertEqual(self.gui.tabs.count(), 1)
@pytest.mark.gui
def test_05_closing_last_tab_opens_new_one(self):
"""Closing the last tab should open a new tab"""
self.assertEqual(self.gui.tabs.count(), 1)
# Click share button
self.gui.tabs.widget(0).share_button.click()
self.assertFalse(self.gui.tabs.widget(0).new_tab.isVisible())
self.assertTrue(self.gui.tabs.widget(0).share_mode.isVisible())
# Close the tab
self.gui.tabs.tabBar().tabButton(0, QtWidgets.QTabBar.RightSide).click()
# A new tab should be opened
self.assertEqual(self.gui.tabs.count(), 1)
self.assertTrue(self.gui.tabs.widget(0).new_tab.isVisible())
@pytest.mark.gui
def test_06_new_tab_mode_buttons_show_correct_modes(self):
"""Clicking the mode buttons in a new tab should change the mode of the tab"""
# New tab, share files
self.gui.tabs.new_tab_button.click()
self.gui.tabs.widget(1).share_button.click()
self.assertFalse(self.gui.tabs.widget(1).new_tab.isVisible())
self.assertTrue(self.gui.tabs.widget(1).share_mode.isVisible())
self.assertEqual(
self.gui.status_bar.server_status_label.text(), "Ready to share"
)
# New tab, receive files
self.gui.tabs.new_tab_button.click()
self.gui.tabs.widget(2).receive_button.click()
self.assertFalse(self.gui.tabs.widget(2).new_tab.isVisible())
self.assertTrue(self.gui.tabs.widget(2).receive_mode.isVisible())
self.assertEqual(
self.gui.status_bar.server_status_label.text(), "Ready to receive"
)
# New tab, publish website
self.gui.tabs.new_tab_button.click()
self.gui.tabs.widget(3).website_button.click()
self.assertFalse(self.gui.tabs.widget(3).new_tab.isVisible())
self.assertTrue(self.gui.tabs.widget(3).website_mode.isVisible())
self.assertEqual(
self.gui.status_bar.server_status_label.text(), "Ready to share"
)
# Close tabs
self.gui.tabs.tabBar().tabButton(0, QtWidgets.QTabBar.RightSide).click()
self.gui.tabs.tabBar().tabButton(0, QtWidgets.QTabBar.RightSide).click()
self.gui.tabs.tabBar().tabButton(0, QtWidgets.QTabBar.RightSide).click()
self.gui.tabs.tabBar().tabButton(0, QtWidgets.QTabBar.RightSide).click()
@pytest.mark.gui
def test_07_close_share_tab_while_server_started_should_warn(self):
"""Closing a share mode tab when the server is running should throw a warning"""
tab = self.new_share_tab_with_files()
self.close_tab_with_active_server(tab)
@pytest.mark.gui
def test_08_close_receive_tab_while_server_started_should_warn(self):
"""Closing a recieve mode tab when the server is running should throw a warning"""
tab = self.new_receive_tab()
self.close_tab_with_active_server(tab)
@pytest.mark.gui
def test_09_close_website_tab_while_server_started_should_warn(self):
"""Closing a website mode tab when the server is running should throw a warning"""
tab = self.new_website_tab_with_files()
self.close_tab_with_active_server(tab)
@pytest.mark.gui
def test_10_close_persistent_share_tab_shows_warning(self):
"""Closing a share mode tab that's persistent should show a warning"""
tab = self.new_share_tab_with_files()
self.close_persistent_tab(tab)
@pytest.mark.gui
def test_11_close_persistent_receive_tab_shows_warning(self):
"""Closing a receive mode tab that's persistent should show a warning"""
tab = self.new_receive_tab()
self.close_persistent_tab(tab)
@pytest.mark.gui
def test_12_close_persistent_website_tab_shows_warning(self):
"""Closing a website mode tab that's persistent should show a warning"""
tab = self.new_website_tab_with_files()
self.close_persistent_tab(tab)
@pytest.mark.gui
def test_13_quit_with_server_started_should_warn(self):
"""Quitting OnionShare with any active servers should show a warning"""
tab = self.new_share_tab()
# Start the server
self.assertEqual(
tab.get_mode().server_status.status,
tab.get_mode().server_status.STATUS_STOPPED,
)
tab.get_mode().server_status.server_button.click()
self.assertEqual(
tab.get_mode().server_status.status,
tab.get_mode().server_status.STATUS_WORKING,
)
QtTest.QTest.qWait(500)
self.assertEqual(
tab.get_mode().server_status.status,
tab.get_mode().server_status.STATUS_STARTED,
)
# Prepare to reject the dialog
QtCore.QTimer.singleShot(0, self.gui.close_dialog.reject_button.click)
# Close the window
self.gui.close()
# The window should still be open
self.assertTrue(self.gui.isVisible())
# Stop the server
tab.get_mode().server_status.server_button.click()

View file

@ -0,0 +1,107 @@
import pytest
import os
import requests
import shutil
from datetime import datetime, timedelta
from PyQt5 import QtCore, QtTest
from .gui_base_test import GuiBaseTest
class TestWebsite(GuiBaseTest):
# Shared test methods
def view_website(self, tab):
"""Test that we can download the share"""
url = f"http://127.0.0.1:{tab.app.port}/"
if tab.settings.get("general", "public"):
r = requests.get(url)
else:
r = requests.get(
url,
auth=requests.auth.HTTPBasicAuth(
"onionshare", tab.get_mode().server_status.web.password
),
)
QtTest.QTest.qWait(500)
self.assertTrue("This is a test website hosted by OnionShare" in r.text)
def check_csp_header(self, tab):
"""Test that the CSP header is present when enabled or vice versa"""
url = f"http://127.0.0.1:{tab.app.port}/"
if tab.settings.get("general", "public"):
r = requests.get(url)
else:
r = requests.get(
url,
auth=requests.auth.HTTPBasicAuth(
"onionshare", tab.get_mode().server_status.web.password
),
)
QtTest.QTest.qWait(500)
if tab.settings.get("website", "disable_csp"):
self.assertFalse("Content-Security-Policy" in r.headers)
else:
self.assertTrue("Content-Security-Policy" in r.headers)
def run_all_website_mode_setup_tests(self, tab):
"""Tests in website mode prior to starting a share"""
tab.get_mode().server_status.file_selection.file_list.add_file(
self.tmpfile_index_html
)
for filename in self.tmpfiles:
tab.get_mode().server_status.file_selection.file_list.add_file(filename)
self.file_selection_widget_has_files(tab, 11)
self.history_is_not_visible(tab)
self.click_toggle_history(tab)
self.history_is_visible(tab)
def run_all_website_mode_started_tests(self, tab, startup_time=500):
"""Tests in website mode after starting a share"""
self.server_working_on_start_button_pressed(tab)
self.server_status_indicator_says_starting(tab)
self.add_remove_buttons_hidden(tab)
self.server_is_started(tab, startup_time)
self.web_server_is_running(tab)
self.have_a_password(tab)
self.url_description_shown(tab)
self.have_copy_url_button(tab)
self.have_show_qr_code_button(tab)
self.server_status_indicator_says_started(tab)
def run_all_website_mode_download_tests(self, tab):
"""Tests in website mode after viewing the site"""
self.run_all_website_mode_setup_tests(tab)
self.run_all_website_mode_started_tests(tab, startup_time=500)
self.view_website(tab)
self.check_csp_header(tab)
self.history_widgets_present(tab)
self.server_is_stopped(tab)
self.web_server_is_stopped(tab)
self.server_status_indicator_says_closed(tab)
self.add_button_visible(tab)
# Tests
@pytest.mark.gui
def test_website(self):
"""
Test website mode
"""
tab = self.new_website_tab()
self.run_all_website_mode_download_tests(tab)
self.close_all_tabs()
@pytest.mark.gui
def test_csp_enabled(self):
"""
Test disabling CSP
"""
tab = self.new_website_tab()
tab.get_mode().disable_csp_checkbox.click()
self.run_all_website_mode_download_tests(tab)
self.close_all_tabs()