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

7 files changed, 421 insertions, 3 deletions

diff --git a/tests/assets/picture01.jpg b/tests/assets/picture01.jpg
new 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
new file mode 100644
index 0000000..3e38495
--- /dev/null
+++ b/tests/assets/picture02.jpg
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 add3b3f..c31554d 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -65,6 +65,8 @@ 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)
@@ -85,6 +87,7 @@ def _cleanup_data(app_settings):
         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):
diff --git a/tests/playwright/test_basic.py b/tests/playwright/test_basic.py
index 3ae0f58..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):
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],
+    )