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
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
|
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
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):
with Session(dbengine) as session:
user = models.User(name="Shauna", email="vayne@lge.com")
user.set_password("vayne")
user.roll_session_secret()
session.add(user)
user = models.User(name="Berri", email="tx@ar.rak", is_admin=True, is_verified=True)
user.set_password("txarrak")
user.roll_session_secret()
session.add(user)
session.commit()
res = fietsctl(["user", "list"])
assert res.successful()
assert res.has_line("vayne@lge\\.com .* Shauna")
assert res.has_line("av.*tx@ar\\.rak .* Berri")
def test_user_del(fietsctl, dbengine):
with Session(dbengine) as session:
user = models.User(name="Shaun", email="murphy@tg.doc", is_verified=True)
user.set_password("murphy")
user.roll_session_secret()
session.add(user)
user = models.User(name="Aaron", email="gl@ss.man", is_verified=True)
user.set_password("glassman")
user.roll_session_secret()
session.add(user)
session.commit()
res = fietsctl(["user", "del", "-f", "--email", "murphy@tg.doc"])
assert res.successful()
with dbengine.connect() as conn:
qry = select(models.User).filter_by(email="gl@ss.man")
row = conn.execute(qry).scalar_one_or_none()
assert row
qry = select(models.User).filter_by(email="murphy@tg.doc")
row = conn.execute(qry).scalar_one_or_none()
assert row is None
res = fietsctl(["user", "del", "-f", "--email", "gl@ss.man"])
assert res.successful()
with dbengine.connect() as conn:
qry = select(models.User).filter_by(email="gl@ss.man")
row = conn.execute(qry).scalar_one_or_none()
assert row is None
def test_user_passwd(fietsctl, dbengine):
with Session(dbengine) as session:
user = models.User(name="john", email="dori@n", is_verified=True)
user.set_password("dorian")
user.roll_session_secret()
session.add(user)
session.commit()
user_id = user.id
res = fietsctl(["user", "passwd", "--email", "dori@n", "--password", "DORIAN"])
assert res.successful()
with Session(dbengine) as session:
user = session.get(models.User, user_id)
user.check_password("DORIAN")
with pytest.raises(models.user.PasswordMismatch):
user.check_password("dorian")
|
