Merge remote-tracking branch 'origin/cli-tests'

1 files changed, 157 insertions, 2 deletions

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],
+    )