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
|
import json
import os
import subprocess
import tempfile
from datetime import datetime, timezone
from importlib import resources
from typing import Any, Generator
import pytest
from bson.json_util import object_hook
from flask import Flask
from flask.testing import FlaskClient
from pymongo import MongoClient
from sec_certs_page import app as sec_certs_app
from sec_certs_page import mongo
from sec_certs_page.cc.mongo import create as cc_create
from sec_certs_page.common.mongo import init_collections
from sec_certs_page.fips.mongo import create as fips_create
from sec_certs_page.pp.mongo import create as pp_create
from sec_certs_page.user.models import User, hash_password
from .client import RemoteTestClient
@pytest.fixture(scope="session")
def app():
with sec_certs_app.app_context():
yield sec_certs_app
@pytest.fixture(scope="session")
def mongodb(app):
# Spin-up a temporary MongoDB instance
# Requires `mongod` to be installed and in PATH
# This is used instead of mongomock because mongomock does not support all features
# required by the application (e.g., $text search)
tmpdir = tempfile.TemporaryDirectory()
proc = subprocess.Popen(
[
"mongod",
"--dbpath",
tmpdir.name,
"--replSet",
"rs0",
"--port",
"27666",
],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
try:
cx = MongoClient("mongodb://localhost:27666", directConnection=True)
cx.admin.command("replSetInitiate")
yield cx
finally:
proc.kill()
proc.wait()
tmpdir.cleanup()
@pytest.fixture(autouse=True, scope="session")
def mongo_data(app, mongodb):
# Create the DB structure
cc_create()
fips_create()
pp_create()
# Initialize other collections
init_collections()
for file in resources.files("tests.functional.data.mongo").iterdir():
if not file.name.endswith(".json"):
continue
with file.open("r") as f:
data = json.load(f, object_hook=object_hook)
collection = file.name.removesuffix(".json")
mongo.db[collection].insert_many(data)
yield
@pytest.fixture()
def clean_mongo(app):
queue = []
with mongo.db.watch(full_document_before_change="whenAvailable") as stream:
yield
# Clean up any changes made to the DB during the test
while stream.alive:
change = stream.try_next()
if not change:
break
operation = change["operationType"]
pre_image = change.get("fullDocumentBeforeChange")
doc_id = change["documentKey"]["_id"]
coll = mongo.db[change["ns"]["coll"]]
if operation == "update" or operation == "replace":
if pre_image:
# Restore the previous version of the document
queue.append((coll.replace_one, {"_id": doc_id}, pre_image))
# print(f"Rolled back update/replace for _id {doc_id}")
# else:
# print(f"Pre-image missing for update/replace operation on _id {doc_id}")
elif operation == "insert":
# Remove the inserted document
queue.append((coll.delete_one, {"_id": doc_id}))
# print(f"Rolled back insert for _id {doc_id}")
# else:
# print(f"Unknown operationType: {operation} for _id {doc_id}")
for item in queue:
func = item[0]
args = item[1:]
func(*args)
@pytest.fixture(scope="function")
def raw_client(app: Flask) -> Generator[FlaskClient | RemoteTestClient, Any, None]:
if os.getenv("TEST_REMOTE"):
yield RemoteTestClient("https://sec-certs.org")
else:
with app.app_context():
yield app.test_client()
@pytest.fixture(scope="function")
def client(app: Flask) -> Generator[FlaskClient | RemoteTestClient, Any, None]:
if os.getenv("TEST_REMOTE"):
yield RemoteTestClient("https://sec-certs.org")
else:
with app.app_context(), app.test_client() as testing_client:
yield testing_client
@pytest.fixture()
def user(app) -> Generator[tuple[User, str], Any, None]:
username = "user"
password = "password"
email = "example@example.com"
roles: list[str] = ["chat"]
pwhash = hash_password(password)
user = User(
username, pwhash, email, roles, email_confirmed=True, created_at=datetime.now(timezone.utc), github_id=None
)
res = mongo.db.users.insert_one(user.dict)
yield user, password
mongo.db.users.delete_one({"_id": res.inserted_id})
@pytest.fixture()
def username(user):
user, _ = user
return user.username
@pytest.fixture()
def email(user):
user, _ = user
return user.email
@pytest.fixture()
def password(user):
_, password = user
return password
@pytest.fixture()
def logged_in(raw_client: FlaskClient, username, password, mocker) -> Generator[FlaskClient, Any, None]:
mocker.patch("flask_wtf.csrf.validate_csrf")
with raw_client:
raw_client.post(
"/user/login",
data={"username": username, "password": password, "remember_me": True},
follow_redirects=True,
)
yield raw_client
|
