1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
|
"""Test to ensure that bootstrapping a new instance works with the fietsupdate
script, as described in the documentation.
"""
import configparser
import contextlib
import logging
import os
import shutil
import subprocess
import venv
from pathlib import Path
import sqlalchemy
LOGGER = logging.getLogger(__name__)
REPO_BASE = Path(__file__).parent.parent.parent
@contextlib.contextmanager
def chdir(path: Path):
"""Change path as context manager.
Changes back to the previous path on ``__exit__``.
"""
cwd = Path.cwd()
os.chdir(path)
try:
yield
finally:
os.chdir(cwd)
def install_fietsboek(venv_path: Path) -> Path:
"""Creates a virtual env in the given path and installs fietsboek.
:param venv_path: Path in which the virtual environment should be created.
:return: The path to the ``bin`` directory in which all scripts reside.
"""
venv.create(venv_path, with_pip=True)
subprocess.check_call([venv_path / "bin" / "pip", "install", str(REPO_BASE)])
return venv_path / "bin"
def create_config(config_name: Path):
"""Copies the testing.ini config and fills in the placeholders.
:param config_name: Path to the resulting config.
"""
shutil.copy(REPO_BASE / "testing.ini", config_name)
config = config_name.read_text()
config = config.replace("# %% fietsboek.data_dir %%", "fietsboek.data_dir = %(here)s/data")
config_name.write_text(config)
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):
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"))
|
