diff options
Diffstat (limited to 'tests/cli')
| -rw-r--r-- | tests/cli/conftest.py | 14 | ||||
| -rw-r--r-- | tests/cli/test_fietsctl.py | 241 |
2 files changed, 255 insertions, 0 deletions
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 |