diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/assets/picture01.jpg | bin | 0 -> 58545 bytes | |||
| -rw-r--r-- | tests/assets/picture02.jpg | bin | 0 -> 52997 bytes | |||
| -rw-r--r-- | tests/bootstrap/test_new_instance.py | 90 | ||||
| -rw-r--r-- | tests/cli/conftest.py | 14 | ||||
| -rw-r--r-- | tests/cli/test_fietsctl.py | 241 | ||||
| -rw-r--r-- | tests/conftest.py | 26 | ||||
| -rw-r--r-- | tests/integration/test_browse.py | 118 | ||||
| -rw-r--r-- | tests/integration/test_pdf.py | 59 | ||||
| -rw-r--r-- | tests/playwright/conftest.py | 18 | ||||
| -rw-r--r-- | tests/playwright/test_basic.py | 34 | ||||
| -rw-r--r-- | tests/playwright/test_journeys.py | 170 | ||||
| -rw-r--r-- | tests/playwright/test_profiles.py | 2 | ||||
| -rw-r--r-- | tests/playwright/test_share.py | 3 | ||||
| -rw-r--r-- | tests/playwright/test_tileproxy.py | 4 | ||||
| -rw-r--r-- | tests/playwright/test_transformers.py | 19 | ||||
| -rw-r--r-- | tests/testutils.py | 159 | ||||
| -rw-r--r-- | tests/unit/test_pdf.py | 58 | ||||
| -rw-r--r-- | tests/unit/test_util.py | 25 |
18 files changed, 965 insertions, 75 deletions
diff --git a/tests/assets/picture01.jpg b/tests/assets/picture01.jpg Binary files differnew file mode 100644 index 0000000..cdd6d1b --- /dev/null +++ b/tests/assets/picture01.jpg diff --git a/tests/assets/picture02.jpg b/tests/assets/picture02.jpg Binary files differnew file mode 100644 index 0000000..3e38495 --- /dev/null +++ b/tests/assets/picture02.jpg diff --git a/tests/bootstrap/test_new_instance.py b/tests/bootstrap/test_new_instance.py index dc3076e..05076f4 100644 --- a/tests/bootstrap/test_new_instance.py +++ b/tests/bootstrap/test_new_instance.py @@ -2,6 +2,7 @@ script, as described in the documentation. """ +import configparser import contextlib import logging import os @@ -10,6 +11,8 @@ import subprocess import venv from pathlib import Path +import sqlalchemy + LOGGER = logging.getLogger(__name__) REPO_BASE = Path(__file__).parent.parent.parent @@ -51,31 +54,66 @@ def create_config(config_name: Path): Path("data").mkdir() +def cleanup_database(config_name: Path): + """Connects to the database and ensures everything is reset. + + :param config_name: Path to the config file. + """ + if not config_name.exists(): + return + + parser = configparser.ConfigParser() + parser["DEFAULT"]["here"] = str(config_name.parent) + parser.read(config_name) + + db_url = parser["app:main"]["sqlalchemy.url"] + engine = sqlalchemy.create_engine(db_url) + + match engine.name: + case "sqlite": + pass + case "postgresql": + with engine.connect() as connection: + connection.execute(sqlalchemy.text("DROP SCHEMA public CASCADE;")) + connection.execute(sqlalchemy.text("CREATE SCHEMA public;")) + connection.commit() + + def test_setup_via_fietsupdate(tmpdir): with chdir(tmpdir): - # We create a new temporary virtual environment with a fresh install, just - # to be sure there's as little interference as possible. - LOGGER.info("Installing Fietsboek into clean env") - binaries_path = install_fietsboek(tmpdir / "venv") - - LOGGER.info("Creating a test configuration") - create_config(Path("testing.ini")) - - # Try to run the migrations - subprocess.check_call( - [binaries_path / "fietsupdate", "update", "-c", "testing.ini", "-f"] - ) - - # Also try to add an administrator - subprocess.check_call([ - binaries_path / "fietsctl", - "user", - "add", - "-c", "testing.ini", - "--email", "foobar@example.com", - "--name", "Foo Bar", - "--password", "raboof", - "--admin", - ]) - - assert True + try: + # We create a new temporary virtual environment with a fresh install, just + # to be sure there's as little interference as possible. + LOGGER.info("Installing Fietsboek into clean env") + binaries_path = install_fietsboek(tmpdir / "venv") + + LOGGER.info("Installing additional SQL engines") + subprocess.check_call( + [binaries_path / "pip", "install", "psycopg2"] + ) + + LOGGER.info("Creating a test configuration") + create_config(Path("testing.ini")) + + # Try to run the migrations + subprocess.check_call( + [binaries_path / "fietsupdate", "update", "-c", "testing.ini", "-f"] + ) + + # Also try to add an administrator + subprocess.check_call([ + binaries_path / "fietsctl", + "user", + "add", + "-c", "testing.ini", + "--email", "foobar@example.com", + "--name", "Foo Bar", + "--password", "raboof", + "--admin", + ]) + + assert True + finally: + # Clean up the database. This is important with anything but SQLite, as + # the tables would otherwise persist and interfere with the other tests. + cleanup_database(Path("testing.ini")) diff --git a/tests/cli/conftest.py b/tests/cli/conftest.py new file mode 100644 index 0000000..97193c9 --- /dev/null +++ b/tests/cli/conftest.py @@ -0,0 +1,14 @@ +import pytest + +fietsctl_results = [] + + +@pytest.hookimpl(hookwrapper=True) +def pytest_runtest_call(item): + fietsctl_results.clear() + + yield + + item.add_report_section( + "call", "fietsctl", "\n\n".join(result.report() for result in fietsctl_results) + ) diff --git a/tests/cli/test_fietsctl.py b/tests/cli/test_fietsctl.py new file mode 100644 index 0000000..c033b23 --- /dev/null +++ b/tests/cli/test_fietsctl.py @@ -0,0 +1,241 @@ +import os +import re +import secrets +import subprocess +from pathlib import Path + +import pytest +from sqlalchemy import select +from sqlalchemy.orm import Session + +from fietsboek import models +from .conftest import fietsctl_results # pylint: disable=relative-beyond-top-level +from ..testutils import populate # pylint: disable=relative-beyond-top-level + + +class CliChecker: + """Result of a CLI tool invocation, with useful methods for tests.""" + + def __init__(self, result): + self.result = result + + @property + def args(self): + """Arguments to the CLI invocation.""" + return self.result.args + + @property + def returncode(self) -> int: + """Return code.""" + return self.result.returncode + + def successful(self) -> bool: + """Whether the invocation was successful (based on the return code).""" + return self.returncode == os.EX_OK + + @property + def stdout(self) -> bytes: + """Standard output.""" + return self.result.stdout or b"" + + def has_line(self, regex: str) -> bool: + """Checks whether the output has a line matching the given regex.""" + pattern = re.compile(regex) + for line in self.stdout.split(b"\n"): + ascii_line = line.decode("ascii", "ignore") + if pattern.search(ascii_line): + return True + return False + + def report(self): + """Formats the output for reporting in pytest.""" + return f"cmd: {self.args} -> {self.returncode}\n" + self.stdout.decode("ascii", "ignore") + + +@pytest.fixture +def fietsctl(app, dbengine, ini_file, data_manager): + """A fietsctl invocation helper.""" + # Ideally, we want to use the ini file. However, the ini doesn't have a data + # dir defined (as we use a temporary path for that), so we need to create a + # new one. + # Ideally², we'd create that ini file in the temporary folder, but that + # gives us the wrong sqlite database (as that uses %(here)). So we could + # either figure out the actual SQLite path and patch that as well, or "give + # up" and create the ini in the correct working dir. + # To do that, we add some random bytes to avoid collisions, and we remove + # the file at the end. + data_dir = data_manager.data_dir + entropy = secrets.token_hex(4) + config_path = Path(ini_file).with_name(f"testing.{entropy}.ini") + assert not config_path.exists(), "file collision - rerun tests!" + config = Path(ini_file).read_text(encoding="ascii") + config = config.replace("# %% fietsboek.data_dir %%", f"fietsboek.data_dir = {data_dir}") + config_path.write_text(config, encoding="ascii") + + def run_inner(args): + cmd = ["fietsctl"] + args + ["-c", str(config_path)] + result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=False) + checker = CliChecker(result) + fietsctl_results.append(checker) + return checker + + yield run_inner + + config_path.unlink() + + +def test_user_add(fietsctl, dbsession): + res = fietsctl( + ["user", "add", "--email", "foo@bar.com", "--name", "Remy", "--password", "hadley"] + ) + assert res.successful() + + res = fietsctl([ + "user", "add", + "--email", "bar@foo.com", + "--name", "James", + "--password", "wilson", + "--admin", + ]) + assert res.successful() + + user: models.User = ( + dbsession.execute(select(models.User).filter_by(email="foo@bar.com")).scalar_one() + ) + + assert user.name == "Remy" + user.check_password("hadley") + assert user.session_secret + assert not user.is_admin + assert user.is_verified + + user: models.User = ( + dbsession.execute(select(models.User).filter_by(email="bar@foo.com")).scalar_one() + ) + + assert user.name == "James" + user.check_password("wilson") + assert user.session_secret + assert user.is_admin + assert user.is_verified + + +def test_user_list(fietsctl, dbengine, data_manager): + populate(dbengine, data_manager) + + res = fietsctl(["user", "list"]) + assert res.successful() + assert res.has_line("av.*jon\\.snow@nw\\.org.*Jon") + assert res.has_line("v.*davos@seaworth\\.com.*Davos") + + +def test_user_del(fietsctl, dbengine, data_manager): + populate(dbengine, data_manager) + + res = fietsctl(["user", "del", "-f", "--email", "jon.snow@nw.org"]) + assert res.successful() + + with dbengine.connect() as conn: + qry = select(models.User).filter_by(email="davos@seaworth.com") + row = conn.execute(qry).scalar_one_or_none() + assert row + qry = select(models.User).filter_by(email="jon.snow@nw.org") + row = conn.execute(qry).scalar_one_or_none() + assert row is None + + res = fietsctl(["user", "del", "-f", "--email", "davos@seaworth.com"]) + assert res.successful() + + with dbengine.connect() as conn: + qry = select(models.User).filter_by(email="davos@seaworth.com") + row = conn.execute(qry).scalar_one_or_none() + assert row is None + + +def test_user_passwd(fietsctl, dbengine, data_manager): + ids = populate(dbengine, data_manager) + + res = fietsctl(["user", "passwd", "--email", "jon.snow@nw.org", "--password", "Ghost"]) + assert res.successful() + + with Session(dbengine) as session: + user = session.get(models.User, ids.jon) + user.check_password("Ghost") + with pytest.raises(models.user.PasswordMismatch): + user.check_password("ygritte") + + +def test_user_modify(fietsctl, dbengine, data_manager): + ids = populate(dbengine, data_manager) + + res = fietsctl([ + "user", "modify", + "--email", "davos@seaworth.com", + "--admin", + "--no-verified", + "--set-email", "ser.davos@seaworth.com", + ]) + assert res.successful() + + with Session(dbengine) as session: + user: models.User = session.get(models.User, ids.davos) + assert not user.is_verified + assert user.is_admin + assert user.email == "ser.davos@seaworth.com" + + +def test_track_list(fietsctl, dbengine, data_manager): + populate(dbengine, data_manager) + + res = fietsctl(["track", "list"]) + assert res.successful() + assert res.has_line("Jon.*Trip around Winterfell") + assert res.has_line("Jon.*Road to Riverrun") + + +def test_track_del(fietsctl, dbengine, data_manager): + ids = populate(dbengine, data_manager) + + res = fietsctl(["track", "del", "-i", str(ids.winterfell)]) + assert not res.successful() + + res = fietsctl(["track", "del", "-i", str(ids.winterfell), "-f"]) + assert res.successful() + + with Session(dbengine) as session: + track = session.get(models.Track, ids.winterfell) + assert track is None + + with pytest.raises(FileNotFoundError): + data_manager.open(ids.winterfell) + + +def test_maintenance_mode_get(fietsctl, data_manager): + res = fietsctl(["maintenance-mode"]) + assert res.successful() + assert res.has_line("Maintenance mode is disabled") + + (data_manager.data_dir / "MAINTENANCE").write_bytes(b"Updating") + res = fietsctl(["maintenance-mode"]) + assert res.successful() + assert res.has_line("enabled: Updating") + + +def test_maintenance_mode_set(fietsctl, data_manager): + assert data_manager.maintenance_mode() is None + + res = fietsctl(["maintenance-mode", "Discombobulating"]) + assert res.successful() + assert data_manager.maintenance_mode() == "Discombobulating" + + res = fietsctl(["maintenance-mode", "Frobnicating"]) + assert res.successful() + assert data_manager.maintenance_mode() == "Frobnicating" + + +def test_maintenance_mode_disable(fietsctl, data_manager): + (data_manager.data_dir / "MAINTENANCE").write_bytes(b"Updating") + + res = fietsctl(["maintenance-mode", "--disable"]) + assert res.successful() + assert data_manager.maintenance_mode() is None diff --git a/tests/conftest.py b/tests/conftest.py index cd74b0b..c31554d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -52,6 +52,7 @@ def dbengine(app_settings, ini_file): yield engine + engine.dispose() Base.metadata.drop_all(bind=engine) alembic.command.stamp(alembic_cfg, None, purge=True) @@ -59,29 +60,40 @@ def dbengine(app_settings, ini_file): def data_manager(app_settings): return DataManager(Path(app_settings["fietsboek.data_dir"])) + +def clean_directory_content(path: Path): + if path.is_dir(): + shutil.rmtree(path) + path.mkdir() + elif path.is_file(): + path.unlink() + + @pytest.fixture(autouse=True) def _cleanup_data(app_settings): yield engine = models.get_engine(app_settings) - db_meta = inspect(engine) + # Load all table names beforehand, as has_table() would cause lock conflicts + tables = inspect(engine).get_table_names() with engine.begin() as connection: for table in reversed(Base.metadata.sorted_tables): # The unit tests don't always set up the tables, so be gentle when # tearing them down - if db_meta.has_table(table.name): + if table.name in tables: connection.execute(table.delete()) # The unit tests also often don't have a data directory, so be gentle here as well if "fietsboek.data_dir" in app_settings: data_dir = Path(app_settings["fietsboek.data_dir"]) - if (data_dir / "tracks").is_dir(): - shutil.rmtree(data_dir / "tracks") - if (data_dir / "users").is_dir(): - shutil.rmtree(data_dir / "users") + clean_directory_content(data_dir / "tracks") + clean_directory_content(data_dir / "users") + clean_directory_content(data_dir / "journeys") + clean_directory_content(data_dir / "MAINTENANCE") @pytest.fixture(scope='module') def app(app_settings, dbengine, tmp_path_factory): app_settings["fietsboek.data_dir"] = str(tmp_path_factory.mktemp("data")) logging.getLogger().setLevel(logging.DEBUG) + logging.getLogger("matplotlib").setLevel(logging.INFO) return main({}, dbengine=dbengine, **app_settings) @pytest.fixture @@ -189,7 +201,7 @@ def logged_in(testapp, route_path, dbsession, tm): tm.abort() with tm: - user = models.User(email='foo@barre.com', is_verified=True) + user = models.User(name="Feu Barre", email='foo@barre.com', is_verified=True) user.set_password("foobar") dbsession.add(user) dbsession.flush() diff --git a/tests/integration/test_browse.py b/tests/integration/test_browse.py index 875821d..1b96e2e 100644 --- a/tests/integration/test_browse.py +++ b/tests/integration/test_browse.py @@ -3,8 +3,10 @@ import zipfile from contextlib import contextmanager from datetime import datetime +from sqlalchemy import delete, inspect + from testutils import load_gpx_asset -from fietsboek import models +from fietsboek import convert, models from fietsboek.models.track import Visibility @@ -21,7 +23,10 @@ def added_tracks(tm, dbsession, owner, data_manager): # objects to the database. tm.abort() + path = convert.smart_convert(load_gpx_asset("Teasi_1.gpx.gz")).path() + tracks = [] + track_ids = [] with tm: track = models.Track( owner=owner, @@ -35,8 +40,10 @@ def added_tracks(tm, dbsession, owner, data_manager): track.date = datetime(2022, 3, 14, 9, 26, 54) dbsession.add(track) dbsession.flush() - data_manager.initialize(track.id).compress_gpx(load_gpx_asset("MyTourbook_1.gpx.gz")) + track.fast_set_path(path) + data_manager.initialize(track.id) tracks.append(track) + track_ids.append(track.id) track = models.Track( owner=owner, @@ -50,14 +57,18 @@ def added_tracks(tm, dbsession, owner, data_manager): track.date = datetime(2022, 10, 29, 13, 37, 11) dbsession.add(track) dbsession.flush() - data_manager.initialize(track.id).compress_gpx(load_gpx_asset("Teasi_1.gpx.gz")) + track.fast_set_path(path) + track.ensure_cache(path) + dbsession.add(track.cache) + data_manager.initialize(track.id) tracks.append(track) + track_ids.append(track.id) tm.begin() tm.doom() try: - yield tracks + yield track_ids finally: tm.abort() with tm: @@ -67,26 +78,113 @@ def added_tracks(tm, dbsession, owner, data_manager): tm.doom() +@contextmanager +def a_lot_of_tracks(tm, dbsession, owner, data_manager): + """Adds some tracks to the database session. + + This function should be used as a context manager and it ensures that the + added tracks are deleted again after the test, to make a clean slate for + the next test. + """ + # The normal transaction is "doomed", so we need to abort it, start a fresh + # one, and then explicitely commit it, otherwise we will not persist the + # objects to the database. + tm.abort() + + gpx_data = load_gpx_asset("MyTourbook_1.gpx.gz") + skel = convert.smart_convert(gpx_data) + path = skel.path() + + tracks = [] + track_ids = [] + with tm: + for index in range(50): + track = models.Track( + owner=owner, + title=f"Traxi {index}", + visibility=Visibility.PUBLIC, + description="One of many", + badges=[], + link_secret="foobar", + tagged_people=[], + ) + track.date = datetime(2022 - index, 3, 14, 9, 26, 59) + dbsession.add(track) + dbsession.flush() + track.fast_set_path(path) + track.ensure_cache(path) + dbsession.add(track.cache) + tracks.append(track) + track_ids.append(track.id) + data_manager.initialize(track.id) + + tm.begin() + tm.doom() + + try: + yield track_ids + finally: + tm.abort() + table = inspect(models.track.TrackPoint).tables[0] + with tm: + for track_id in track_ids: + dbsession.execute(delete(table).where(table.c.track_id == track_id)) + dbsession.execute( + delete(models.TrackCache).where(models.TrackCache.track_id == track_id) + ) + dbsession.execute(delete(models.Track).where(models.Track.id == track_id)) + tm.begin() + tm.doom() + + def test_browse(testapp, dbsession, route_path, logged_in, tm, data_manager): # pylint: disable=too-many-positional-arguments # Ensure there are some tracks in the database with added_tracks(tm, dbsession, logged_in, data_manager): # Now go to the browse page - browse = testapp.get(route_path('browse')) + browse = testapp.get(route_path("browse")) assert "Foobar" in browse.text assert "Barfoo" in browse.text +def test_browse_paged(testapp, dbsession, route_path, logged_in, tm, data_manager): + # pylint: disable=too-many-positional-arguments + with a_lot_of_tracks(tm, dbsession, logged_in, data_manager): + page_1 = testapp.get(route_path("browse", _query=[("page", 1)])) + assert "Traxi 0" in page_1.text + assert "Traxi 10" in page_1.text + assert "Traxi 20" not in page_1.text + assert "Traxi 30" not in page_1.text + assert "Traxi 40" not in page_1.text + + page_2 = testapp.get(route_path("browse", _query=[("page", 2)])) + assert "Traxi 0" not in page_2.text + assert "Traxi 10" not in page_2.text + assert "Traxi 20" in page_2.text + assert "Traxi 30" in page_2.text + assert "Traxi 40" not in page_2.text + + page_3 = testapp.get(route_path("browse", _query=[("page", 3)])) + assert "Traxi 0" not in page_3.text + assert "Traxi 10" not in page_3.text + assert "Traxi 20" not in page_3.text + assert "Traxi 30" not in page_3.text + assert "Traxi 40" in page_3.text + + def test_archive(testapp, dbsession, route_path, logged_in, tm, data_manager): # pylint: disable=too-many-positional-arguments - with added_tracks(tm, dbsession, logged_in, data_manager): + with added_tracks(tm, dbsession, logged_in, data_manager) as tracks: archive = testapp.get( - route_path('track-archive', _query=[("track_id[]", "1"), ("track_id[]", "2")]) + route_path( + "track-archive", + _query=[("track_id[]", tracks[0]), ("track_id[]", tracks[1])], + ) ) result = io.BytesIO(archive.body) - with zipfile.ZipFile(result, 'r') as zipped: + with zipfile.ZipFile(result, "r") as zipped: assert len(zipped.namelist()) == 2 - assert "track_1.gpx" in zipped.namelist() - assert "track_2.gpx" in zipped.namelist() + assert f"track_{tracks[0]}.gpx" in zipped.namelist() + assert f"track_{tracks[1]}.gpx" in zipped.namelist() diff --git a/tests/integration/test_pdf.py b/tests/integration/test_pdf.py new file mode 100644 index 0000000..29cda02 --- /dev/null +++ b/tests/integration/test_pdf.py @@ -0,0 +1,59 @@ +from contextlib import contextmanager +from datetime import datetime + +from testutils import load_gpx_asset +from fietsboek import convert, models +from fietsboek.models.track import Visibility + + +@contextmanager +def a_track(tm, dbsession, owner, data_manager): + """Adds some tracks to the database session. + + This function should be used as a context manager and it ensures that the + added tracks are deleted again after the test, to make a clean slate for + the next test. + """ + # The normal transaction is "doomed", so we need to abort it, start a fresh + # one, and then explicitely commit it, otherwise we will not persist the + # objects to the database. + tm.abort() + + with tm: + track = models.Track( + owner=owner, + title="Goober", + visibility=Visibility.PUBLIC, + description="A bar'd track", + badges=[], + link_secret="raboof", + tagged_people=[], + ) + track.date = datetime(2027, 3, 14, 9, 26, 54) + track.set_path(convert.smart_convert(load_gpx_asset("MyTourbook_1.gpx.gz")).path()) + dbsession.add(track) + dbsession.flush() + data_manager.initialize(track.id) + track_id = track.id + + tm.begin() + tm.doom() + + try: + yield track_id + finally: + tm.abort() + with tm: + dbsession.delete(track) + data_manager.purge(track_id) + tm.begin() + tm.doom() + + +def test_pdf(testapp, dbsession, route_path, logged_in, tm, data_manager): + # pylint: disable=too-many-positional-arguments + # Ensure there are some tracks in the database + with a_track(tm, dbsession, logged_in, data_manager) as track_id: + pdf = testapp.get(route_path("track-pdf", track_id=track_id)) + + assert pdf diff --git a/tests/playwright/conftest.py b/tests/playwright/conftest.py index 4e7f5a4..435daa6 100644 --- a/tests/playwright/conftest.py +++ b/tests/playwright/conftest.py @@ -7,10 +7,12 @@ from wsgiref import simple_server from pyramid.authentication import AuthTktCookieHelper from pyramid.testing import DummyRequest +import fietsboek.config from testutils import load_gpx_asset from fietsboek import models, util, actions from fietsboek.models.track import Visibility, TrackType from fietsboek.config import Config +from fietsboek.views.tileproxy import TileRequester import pytest @@ -55,7 +57,11 @@ def dbaccess(app): through and the running WSGI app cannot read them. """ session_factory = app.registry["dbsession_factory"] - return session_factory() + factory = session_factory() + + yield factory + + factory.close() class Helper: @@ -104,7 +110,10 @@ class Helper: ) def add_track( - self, user: Optional[models.User] = None, track_name: str = "Teasi_1.gpx.gz" + self, + user: Optional[models.User] = None, + track_name: str = "Teasi_1.gpx.gz", + title: str = "Another awesome track", ) -> models.Track: """Add a track to the given user. @@ -112,13 +121,16 @@ class Helper: """ if user is None: user = self.john_doe() + config = fietsboek.config.parse(self.app_settings) with self.dbaccess: user = self.dbaccess.merge(user) track = actions.add_track( self.dbaccess, self.data_manager, + TileRequester(None), + config.public_tile_layers()[0], owner=user, - title="Another awesome track", + title=title, visibility=Visibility.PRIVATE, description="Another description", track_type=TrackType.ORGANIC, diff --git a/tests/playwright/test_basic.py b/tests/playwright/test_basic.py index 231962e..19ef8ff 100644 --- a/tests/playwright/test_basic.py +++ b/tests/playwright/test_basic.py @@ -4,7 +4,7 @@ import pytest from playwright.sync_api import Page, expect from sqlalchemy import select -from testutils import load_gpx_asset +from testutils import asset_path, load_gpx_asset from fietsboek import models from fietsboek.models.track import Visibility @@ -60,6 +60,10 @@ def test_upload(page: Page, playwright_helper, tmp_path, dbaccess): page.get_by_role("button", name="Add Tag").click() page.get_by_label("Description").fill("Beschreibung der tollen Tour") + page.locator("#imageSelector").set_input_files( + [asset_path("picture01.jpg"), asset_path("picture02.jpg")], + ) + page.locator(".btn", has_text="Upload").click() # Once we have finished the upload, extract the ID of the track and check @@ -72,6 +76,7 @@ def test_upload(page: Page, playwright_helper, tmp_path, dbaccess): assert track.visibility == Visibility.PUBLIC assert track.text_tags() == {"Tolle Tour"} assert track.description == "Beschreibung der tollen Tour" + assert len(track.images) == 2 def test_edit(page: Page, playwright_helper, dbaccess): @@ -101,6 +106,33 @@ def test_edit(page: Page, playwright_helper, dbaccess): assert track.description == "Not so descriptive anymore" +def test_edit_change_gpx(page: Page, playwright_helper, tmp_path, dbaccess): + playwright_helper.login() + track_id = playwright_helper.add_track().id + + track = dbaccess.execute(select(models.Track).filter_by(id=track_id)).scalar_one() + old_cache = track.cache.length, track.cache.uphill, track.cache.downhill + + page.goto(f"/track/{track_id}") + page.locator(".btn", has_text="Edit").click() + + gpx_data = load_gpx_asset("Synthetic_BRouter_1.gpx.gz") + gpx_path = tmp_path / "NewGPX.gpx" + with open(gpx_path, "wb") as gpx_fobj: + gpx_fobj.write(gpx_data) + + page.get_by_label("New file for this track").set_input_files(gpx_path) + page.locator(".btn", has_text="Save").click() + + track = dbaccess.execute(select(models.Track).filter_by(id=track_id)).scalar_one() + new_cache = track.cache + dbaccess.refresh(new_cache) + + assert old_cache[0] != new_cache.length + assert old_cache[1] != new_cache.uphill + assert old_cache[2] != new_cache.downhill + + def test_browse(page: Page, playwright_helper, dbaccess): playwright_helper.login() track = playwright_helper.add_track() diff --git a/tests/playwright/test_journeys.py b/tests/playwright/test_journeys.py new file mode 100644 index 0000000..2bfd271 --- /dev/null +++ b/tests/playwright/test_journeys.py @@ -0,0 +1,170 @@ +from playwright.sync_api import Page, expect +from sqlalchemy import select + +from fietsboek import models + + +def add_journey(playwright_helper, dbaccess, title): + """Adds a journey for testing purposes. Returns the journey ID.""" + t_1 = playwright_helper.add_track(None, "Teasi_1.gpx.gz", "trayectoria uno") + t_2 = playwright_helper.add_track(None, "MyTourbook_1.gpx.gz", "trayectoria dos") + + with dbaccess: + journey = models.Journey( + owner=playwright_helper.john_doe(), + title=title, + description="You saw sirens?", + visibility=models.journey.Visibility.PUBLIC, + ) + + dbaccess.add(journey) + dbaccess.flush() + + journey.set_track_ids([t_1.id, t_2.id]) + dbaccess.commit() + dbaccess.refresh(journey, ["id"]) + dbaccess.expunge(journey) + + playwright_helper.data_manager.initialize_journey(journey.id) + + return journey.id + + +def test_journey_list(page: Page, playwright_helper, dbaccess): + playwright_helper.login() + + add_journey(playwright_helper, dbaccess, title="Our Journey") + + page.goto("/journey/") + expect(page.locator("h5", has_text="Our Journey")).to_be_visible() + expect(page.locator("li", has_text="trayectoria uno")).to_be_visible() + expect(page.locator("li", has_text="trayectoria dos")).to_be_visible() + + +def test_journey_new(page: Page, playwright_helper, dbaccess): + playwright_helper.login() + + playwright_helper.add_track(None, "Teasi_1.gpx.gz", "trayectoria uno") + playwright_helper.add_track(None, "MyTourbook_1.gpx.gz", "trayectoria dos") + playwright_helper.add_track(None, "MyTourbook_1.gpx.gz", "trayectoria tres") + + page.goto("/journey/") + page.get_by_text("New journey").click() + + page.get_by_label("Title").fill("My Odyssey") + page.get_by_label("Description").fill("I saw sirens!") + + page.locator("#trackSearch").fill("uno") + page.locator("#trackSearchButton").click() + page.locator(".track-query-response button").click() + + page.locator("#trackSearch").fill("dos") + page.locator("#trackSearchButton").click() + page.locator(".track-query-response button").click() + + page.locator("#trackSearch").fill("tres") + page.locator("#trackSearchButton").click() + page.locator(".track-query-response button").click() + page.locator(".journey-track", has_text="tres").locator(".btn").click() + + page.locator(".btn", has_text="Save").click() + + expect(page.locator("h1", has_text="My Odyssey")).to_be_visible() + + expect(page.locator("h5", has_text="trayectoria uno")).to_be_visible() + expect(page.locator("h5", has_text="trayectoria dos")).to_be_visible() + + journey = dbaccess.execute(select(models.Journey).filter_by(title="My Odyssey")).scalar_one() + + assert journey.title == "My Odyssey" + assert journey.description == "I saw sirens!" + assert len(journey.tracks) == 2 + assert journey.tracks[0].title == "trayectoria uno" + assert journey.tracks[1].title == "trayectoria dos" + + +def test_journey_new_empty_title(page: Page, playwright_helper): + playwright_helper.login() + + playwright_helper.add_track(None, "Teasi_1.gpx.gz", "trayectoria uno") + + page.goto("/journey/") + page.get_by_text("New journey").click() + + page.locator("#trackSearch").fill("uno") + page.locator("#trackSearchButton").click() + page.locator(".track-query-response button").click() + page.locator(".btn", has_text="Save").click() + + expect(page.locator(".invalid-feedback", has_text="A title is required")).to_be_visible() + + +def test_journey_new_no_tracks(page: Page, playwright_helper): + playwright_helper.login() + + page.goto("/journey/") + page.get_by_text("New journey").click() + + page.get_by_label("Title").fill("A title is there!") + + page.locator(".btn", has_text="Save").click() + + expect(page.locator(".invalid-feedback", has_text="A journey must have at least one track"))\ + .to_be_visible() + + +def test_journey_edit(page: Page, playwright_helper, dbaccess): + playwright_helper.login() + + journey_id = add_journey(playwright_helper, dbaccess, title="Your Odyssey") + + page.goto(f"/journey/{journey_id}/") + + expect(page.locator("h1", has_text="Your Odyssey")).to_be_visible() + + page.locator("a", has_text="Edit").click() + + page.get_by_label("Title").fill("Their Odyssey") + page.get_by_label("Description").fill("Where is Homer?") + + expect(page.locator(".track-title", has_text="trayectoria uno")).to_be_visible() + page.locator(".journey-track", has_text="uno").locator(".btn").click() + expect(page.locator(".track-title", has_text="trayectoria uno")).not_to_be_visible() + + page.locator(".btn", has_text="Save").click() + + expect(page.locator("h1", has_text="Their Odyssey")).to_be_visible() + expect(page.locator("h5", has_text="trayectoria uno")).not_to_be_visible() + expect(page.locator("h5", has_text="trayectoria dos")).to_be_visible() + + journey = dbaccess.execute(select(models.Journey).filter_by(title="Their Odyssey")).scalar_one() + + assert journey.title == "Their Odyssey" + assert journey.description == "Where is Homer?" + assert len(journey.tracks) == 1 + assert journey.tracks[0].title == "trayectoria dos" + + +def test_journey_reorder(page: Page, playwright_helper, dbaccess): + playwright_helper.login() + + journey_id = add_journey(playwright_helper, dbaccess, title="Her Journey") + + page.goto(f"/journey/{journey_id}/edit") + + expect(page.locator("h1", has_text="Her Journey")).to_be_visible() + + page.locator(".track-title", has_text="trayectoria uno").drag_to( + page.locator(".track-title", has_text="trayectoria dos"), + target_position={"x": 10, "y": 20}, + ) + + page.locator(".btn", has_text="Save").click() + + expect(page.locator("h1", has_text="Her Journey")).to_be_visible() + + journey = dbaccess.execute(select(models.Journey).filter_by(id=journey_id)).scalar_one() + + assert len(journey.tracks) == 2 + assert journey.tracks[0].title == "trayectoria dos" + assert journey.tracks[1].title == "trayectoria uno" diff --git a/tests/playwright/test_profiles.py b/tests/playwright/test_profiles.py index 7e5fb3c..ffbaab0 100644 --- a/tests/playwright/test_profiles.py +++ b/tests/playwright/test_profiles.py @@ -5,7 +5,7 @@ def test_forbidden(page: Page, playwright_helper): john = playwright_helper.john_doe() with page.expect_response(lambda resp: resp.status == 403): - page.goto(f"/user/{john.id}") + page.goto(f"/user/{john.id}/") def test_profile(page: Page, playwright_helper): diff --git a/tests/playwright/test_share.py b/tests/playwright/test_share.py index de288a0..dcba899 100644 --- a/tests/playwright/test_share.py +++ b/tests/playwright/test_share.py @@ -29,7 +29,8 @@ def test_view_wrong_link(page: Page, playwright_helper, dbaccess): with page.expect_response(lambda resp: resp.status == 403): page.goto(f"/track/{track.id}?secret=foobar") - assert "Forbidden" in page.content() + assert "No entry" in page.content() + assert "not allowed to access" in page.content() def test_change_link(page: Page, playwright_helper, dbaccess): diff --git a/tests/playwright/test_tileproxy.py b/tests/playwright/test_tileproxy.py index d4d3389..2a2bdc0 100644 --- a/tests/playwright/test_tileproxy.py +++ b/tests/playwright/test_tileproxy.py @@ -14,7 +14,7 @@ def test_tileproxy(page: Page, playwright_helper, caplog): # If we're too fast, the log entry might not be there yet, wait 2 more # seconds - if "Skipping tile proxy request for testing URL" not in caplog.messages: + if "Skipping tile request for testing URL" not in caplog.messages: time.sleep(2) - assert "Skipping tile proxy request for testing URL" in caplog.messages + assert "Skipping tile request for testing URL" in caplog.messages diff --git a/tests/playwright/test_transformers.py b/tests/playwright/test_transformers.py index fc89afb..d4e3456 100644 --- a/tests/playwright/test_transformers.py +++ b/tests/playwright/test_transformers.py @@ -26,7 +26,7 @@ def test_transformer_zero_elevation_disabled(page: Page, playwright_helper, tmp_ # Expect early (here and in the other tests) to ensure that the backend has # caught up with executing the transformer. Otherwise it might happen that # we read the database while the request is not finished yet. - expect(page.locator("#detailsUphill")).to_contain_text("167.7 m") + expect(page.locator("#detailsUphill")).to_contain_text("167.79 m") new_track_id = int(page.url.rsplit("/", 1)[1]) track = dbaccess.execute(select(models.Track).filter_by(id=new_track_id)).scalar_one() @@ -90,7 +90,7 @@ def test_transformer_steep_slope_disabled(page: Page, playwright_helper, tmp_pat page.locator(".btn", has_text="Upload").click() - expect(page.locator("#detailsUphill")).to_contain_text("61.54 m") + expect(page.locator("#detailsUphill")).to_contain_text("64.4 m") new_track_id = int(page.url.rsplit("/", 1)[1]) track = dbaccess.execute(select(models.Track).filter_by(id=new_track_id)).scalar_one() @@ -111,11 +111,11 @@ def test_transformer_steep_slope_enabled(page: Page, playwright_helper, tmp_path page.locator(".btn", has_text="Upload").click() - expect(page.locator("#detailsUphill")).to_contain_text("1.2 m") + expect(page.locator("#detailsUphill")).to_contain_text("2.4 m") new_track_id = int(page.url.rsplit("/", 1)[1]) track = dbaccess.execute(select(models.Track).filter_by(id=new_track_id)).scalar_one() - assert track.cache.uphill < 2 + assert track.cache.uphill < 3 def test_transformer_steep_slope_edited(page: Page, playwright_helper, tmp_path, dbaccess): @@ -137,14 +137,14 @@ def test_transformer_steep_slope_edited(page: Page, playwright_helper, tmp_path, page.locator(".btn", has_text="Save").click() - expect(page.locator("#detailsUphill")).to_contain_text("1.2 m") + expect(page.locator("#detailsUphill")).to_contain_text("2.4 m") track_id = int(page.url.rsplit("/", 1)[1]) track = dbaccess.execute(select(models.Track).filter_by(id=track_id)).scalar_one() - assert track.cache.uphill < 2 + assert track.cache.uphill < 3 -def test_transformer_elevation_jump_enabled(page: Page, playwright_helper, tmp_path, data_manager): +def test_transformer_elevation_jump_enabled(page: Page, playwright_helper, tmp_path, dbaccess): playwright_helper.login() page.goto("/") @@ -161,9 +161,10 @@ def test_transformer_elevation_jump_enabled(page: Page, playwright_helper, tmp_p page.locator(".alert", has_text="Upload successful").wait_for() new_track_id = int(page.url.rsplit("/", 1)[1]) - data = data_manager.open(new_track_id) - gpx = gpxpy.parse(data.decompress_gpx()) + gpx = gpxpy.parse( + dbaccess.execute(select(models.Track).filter_by(id=new_track_id)).scalar_one().gpx_xml() + ) points = iter(gpx.walk(only_points=True)) next(points) for prev_point, point in zip(gpx.walk(only_points=True), points): diff --git a/tests/testutils.py b/tests/testutils.py index 810bdf7..e49a0b5 100644 --- a/tests/testutils.py +++ b/tests/testutils.py @@ -1,8 +1,45 @@ """Various utility functions for testing.""" +import datetime import gzip +import io from pathlib import Path +from typing import NamedTuple from playwright.sync_api import Page +from sqlalchemy import Engine +from sqlalchemy.orm import Session + +from fietsboek import convert, models, util +from fietsboek.data import DataManager + + +class PopulationIds(NamedTuple): + """Collection of database IDs that :func:`populate` returns.""" + jon: int + davos: int + winterfell: int + riverrun: int + + +def asset_path(filename: str) -> Path: + """Returns the path to the test asset given by the filename. + + :param filename: The filename of the asset. + :return: The complete path. + """ + asset_dir = Path(__file__).parent / "assets" + return asset_dir / filename + + +def load_test_asset(filename: str) -> bytes: + """Load a test asset. + + Unlike :func:`load_gpx_asset`, this function does not do gzip decompression. + + :param filkename: Name of the asset to load. + :return: The content of the file as bytes. + """ + return asset_path(filename).read_bytes() def load_gpx_asset(filename: str) -> bytes: @@ -14,8 +51,7 @@ def load_gpx_asset(filename: str) -> bytes: :param filename: Name of the asset to load. :return: The content of the asset as bytes. """ - asset_dir = Path(__file__).parent / 'assets' - test_file = asset_dir / filename + test_file = asset_path(filename) with gzip.open(test_file, 'rb') as fobj: return fobj.read() @@ -36,3 +72,122 @@ def extract_and_upload(page: Page, filename: str, tmp_path: Path): page.get_by_label("GPX file").set_input_files(gpx_path) page.locator(".bi-upload").click() + + +def populate(dbengine: Engine, data_manager: DataManager) -> PopulationIds: + """Populates the database and data directory with some test data. + + This adds: + + Jon (jon.snow@nw.org, admin, password: ygritte) + Trip around Winterfell (2 images) + Road to Riverrun + + Davos (davos@seaworth.com, password: 123456) + + :return: An object carrying the database IDs of the added objects. + """ + user_ids = [] + with Session(dbengine) as session: + user = models.User(name="Jon", email="jon.snow@nw.org", is_verified=True, is_admin=True) + user.set_password("ygritte") + user.roll_session_secret() + session.add(user) + session.flush() + user_ids.append(user.id) + + user = models.User(name="Davos", email="davos@seaworth.com", is_verified=True) + user.set_password("123456") + user.roll_session_secret() + session.add(user) + session.flush() + user_ids.append(user.id) + + session.commit() + + for user_id in user_ids: + data_manager.initialize_user(user_id) + + track_ids = [] + # First track for Jon: + + gpx_data = load_gpx_asset("Teasi_1.gpx.gz") + track = convert.smart_convert(gpx_data) + path = track.path() + track.points = [] + track.owner_id = user_ids[0] + track.title = "Trip around Winterfell" + track.visibility = models.track.Visibility.PUBLIC + track.type = models.track.TrackType.ORGANIC + track.description = "I took my sister for a quick trip around Winterfell" + track.badges = [] + track.link_secret = util.random_link_secret() + track.tagged_people = [] + track.date = datetime.datetime(1984, 1, 2, 10, 11, tzinfo=datetime.UTC) + track.transformers = [] + track.sync_tags({"westeros"}) + + with Session(dbengine) as session: + session.add(track) + session.flush() + + assert track.id is not None + track_ids.append(track.id) + manager = data_manager.initialize(track.id) + manager.compress_backup(gpx_data) + + track.fast_set_path(path) + track.ensure_cache(path) + session.add(track.cache) + + image_name = manager.add_image(io.BytesIO(load_test_asset("picture01.jpg")), "PIC001.jpg") + image_meta = models.ImageMetadata(track=track, image_name=image_name) + image_meta.description = "Beautiful sight out of the Hunter's Gate" + session.add(image_meta) + + image_name = manager.add_image(io.BytesIO(load_test_asset("picture02.jpg")), "PIC002.jpg") + image_meta = models.ImageMetadata(track=track, image_name=image_name) + image_meta.description = "Our steel wire horses ready to gallop" + session.add(image_meta) + + session.commit() + + # Second track for Jon: + + gpx_data = load_gpx_asset("MyTourbook_1.gpx.gz") + track = convert.smart_convert(gpx_data) + path = track.path() + track.points = [] + track.owner_id = user_ids[0] + track.title = "Road to Riverrun" + track.visibility = models.track.Visibility.PUBLIC + track.type = models.track.TrackType.ORGANIC + track.description = "Got a wedding to attend!" + track.badges = [] + track.link_secret = util.random_link_secret() + track.tagged_people = [] + track.date = datetime.datetime(1985, 8, 5, 16, 41, tzinfo=datetime.UTC) + track.transformers = [] + track.sync_tags({"westeros"}) + + with Session(dbengine) as session: + session.add(track) + session.flush() + + assert track.id is not None + track_ids.append(track.id) + manager = data_manager.initialize(track.id) + manager.compress_backup(gpx_data) + + track.fast_set_path(path) + track.ensure_cache(path) + session.add(track.cache) + + session.commit() + + return PopulationIds( + jon=user_ids[0], + davos=user_ids[1], + winterfell=track_ids[0], + riverrun=track_ids[1], + ) diff --git a/tests/unit/test_pdf.py b/tests/unit/test_pdf.py new file mode 100644 index 0000000..aafa717 --- /dev/null +++ b/tests/unit/test_pdf.py @@ -0,0 +1,58 @@ +import pytest + +from fietsboek import pdf + + +@pytest.mark.parametrize("value, expected", [ + ('', '""'), + ('a', '"\\u{61}"'), + ('FOO', '"\\u{46}\\u{4f}\\u{4f}"'), + ('äß', '"\\u{e4}\\u{df}"'), + ('"', '"\\u{22}"'), + ("'", '"\\u{27}"'), +]) +def test_typst_string(value, expected): + assert pdf.typst_string(value) == expected + + +@pytest.mark.parametrize("value, expected", [ + ("foo", "foo"), + ("*foo*", "\\*foo\\*"), + ("#strong[foo]", "\\#strong\\[foo\\]"), + ("= foo", "\\= foo"), + ("par 1\n\npar 2", "par 1\n\npar 2"), +]) +def test_typst_escape(value, expected): + assert pdf.typst_escape(value) == expected + + +@pytest.mark.parametrize("md_source, typst_source", [ + ("*foo*", "#emph[foo]\n\n"), + ("**foo**", "#strong[foo]\n\n"), + ("***foo***", "#strong[#emph[foo]]\n\n"), + ("[Teksd](https://link)", + '#link("\\u{68}\\u{74}\\u{74}\\u{70}\\u{73}\\u{3a}' + '\\u{2f}\\u{2f}\\u{6c}\\u{69}\\u{6e}\\u{6b}")[Teksd]\n\n'), + ("""\ +# Uperschrift + +Teksd""", """\ +#heading(level: 1)[Uperschrift] +Teksd\n\n"""), + ("""\ +* Eitem 1 +* Eitem 2""", """\ +#list( +[Eitem 1], +[Eitem 2], +)"""), + ("""\ +1. Eitem 1 +1. Eitem 2""", """\ +#enum( +[Eitem 1], +[Eitem 2], +)"""), +]) +def test_md_to_typst(md_source, typst_source): + assert pdf.md_to_typst(md_source) == typst_source diff --git a/tests/unit/test_util.py b/tests/unit/test_util.py index 6dc8e7d..cc92058 100644 --- a/tests/unit/test_util.py +++ b/tests/unit/test_util.py @@ -71,19 +71,6 @@ def test_guess_gpx_timezone(gpx_file, offset): assert timezone.utcoffset(None) == offset -@pytest.mark.parametrize('gpx_file', [ - 'Teasi_1.gpx.gz', - 'MyTourbook_1.gpx.gz', - 'Synthetic_WT2.gpx.gz', - 'Synthetic_BRouter_1.gpx.gz', -]) -def test_tour_metadata(gpx_file): - # Here we simply make sure that we do not crash the metadata extraction - # function. - gpx_data = load_gpx_asset(gpx_file) - assert util.tour_metadata(gpx_data) is not None - - @pytest.mark.parametrize('mps, kph', [(1, 3.6), (10, 36)]) def test_mps_to_kph(mps, kph): assert util.mps_to_kph(mps) == pytest.approx(kph, 0.1) @@ -112,3 +99,15 @@ def test_tile_url(app_request): assert "{y}" in route_url assert "{z}" in route_url assert "bobby" in route_url + + +@pytest.mark.parametrize("value, expected", [ + ("", b""), + ("foo", b"foo"), + ("<foo>", b"<foo>"), + ("foo bar", b"foo bar"), + ("</gpx>", b"</gpx>"), + ("äÖß", b"äÖß"), +]) +def test_xml_escape(value, expected): + assert util.xml_escape(value) == expected |
