external/aurweb
0
0
Fork 0
mirror of https://gitlab.archlinux.org/archlinux/aurweb.git synced 2025-02-03 10:43:03 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'sqlalchemy2' into 'master'

Browse source 
Draft: Upgrade to sqlalchemy 2 & fix db sessions

See merge request archlinux/aurweb!756
This commit is contained in:
Mario Oenning 2023-11-22 21:58:58 +00:00
commit 8db5dc3328
19 changed files with 273 additions and 176 deletions
Download patch file Download diff file Expand all files Collapse all files

26
test/conftest.py
View file

@ -43,7 +43,7 @@ from multiprocessing import Lock
import py
import pytest
from sqlalchemy import create_engine
from sqlalchemy import create_engine, text
from sqlalchemy.engine import URL
from sqlalchemy.engine.base import Engine
from sqlalchemy.exc import ProgrammingError
@ -104,15 +104,16 @@ def _create_database(engine: Engine, dbname: str) -> None:
:param dbname: Database name to create
"""
conn = engine.connect()
try:
conn.execute(f"CREATE DATABASE {dbname}")
except ProgrammingError: # pragma: no cover
# The database most likely already existed if we hit
# a ProgrammingError. Just drop the database and try
# again. If at that point things still fail, any
# exception will be propogated up to the caller.
conn.execute(f"DROP DATABASE {dbname}")
conn.execute(f"CREATE DATABASE {dbname}")
with conn.begin():
try:
conn.execute(text(f"CREATE DATABASE {dbname}"))
except ProgrammingError: # pragma: no cover
# The database most likely already existed if we hit
# a ProgrammingError. Just drop the database and try
# again. If at that point things still fail, any
# exception will be propogated up to the caller.
conn.execute(text(f"DROP DATABASE {dbname}"))
conn.execute(text(f"CREATE DATABASE {dbname}"))
conn.close()
initdb.run(AlembicArgs)
@ -124,9 +125,10 @@ def _drop_database(engine: Engine, dbname: str) -> None:
:param engine: Engine returned by test_engine()
:param dbname: Database name to drop
"""
aurweb.schema.metadata.drop_all(bind=engine)
# aurweb.schema.metadata.drop_all(bind=engine)
conn = engine.connect()
conn.execute(f"DROP DATABASE {dbname}")
with conn.begin():
conn.execute(text(f"DROP DATABASE {dbname}"))
conn.close()

8
test/test_accounts_routes.py
View file

@ -830,6 +830,7 @@ def test_post_account_edit_type_as_dev(client: TestClient, pm_user: User):
request.cookies = cookies
resp = request.post(endpoint, data=data)
assert resp.status_code == int(HTTPStatus.OK)
db.refresh(user2)
assert user2.AccountTypeID == at.DEVELOPER_ID
@ -850,6 +851,7 @@ def test_post_account_edit_invalid_type_as_pm(client: TestClient, pm_user: User)
request.cookies = cookies
resp = request.post(endpoint, data=data)
assert resp.status_code == int(HTTPStatus.BAD_REQUEST)
db.refresh(user2)
assert user2.AccountTypeID == at.USER_ID
errors = get_errors(resp.text)
@ -1020,6 +1022,7 @@ def test_post_account_edit_inactivity(client: TestClient, user: User):
assert resp.status_code == int(HTTPStatus.OK)
# Make sure the user record got updated correctly.
db.refresh(user)
assert user.InactivityTS > 0
post_data.update({"J": False})
@ -1028,6 +1031,7 @@ def test_post_account_edit_inactivity(client: TestClient, user: User):
resp = request.post(f"/account/{user.Username}/edit", data=post_data)
assert resp.status_code == int(HTTPStatus.OK)
db.refresh(user)
assert user.InactivityTS == 0
@ -1050,6 +1054,7 @@ def test_post_account_edit_suspended(client: TestClient, user: User):
assert resp.status_code == int(HTTPStatus.OK)
# Make sure the user record got updated correctly.
db.refresh(user)
assert user.Suspended
# Let's make sure the DB got updated properly.
assert user.session is None
@ -1207,6 +1212,7 @@ def test_post_account_edit_password(client: TestClient, user: User):
assert response.status_code == int(HTTPStatus.OK)
db.refresh(user)
assert user.valid_password("newPassword")
@ -1273,6 +1279,7 @@ def test_post_account_edit_self_type_as_pm(client: TestClient, pm_user: User):
resp = request.post(endpoint, data=data)
assert resp.status_code == int(HTTPStatus.OK)
db.refresh(pm_user)
assert pm_user.AccountTypeID == USER_ID
@ -1308,6 +1315,7 @@ def test_post_account_edit_other_user_type_as_pm(
assert resp.status_code == int(HTTPStatus.OK)
# Let's make sure the DB got updated properly.
db.refresh(user2)
assert user2.AccountTypeID == PACKAGE_MAINTAINER_ID
# and also that this got logged out at DEBUG level.

3
test/test_initdb.py
View file

@ -24,5 +24,6 @@ def test_run():
aurweb.initdb.run(Args())
# Check that constant table rows got added via initdb.
record = aurweb.db.query(AccountType, AccountType.AccountType == "User").first()
with aurweb.db.begin():
record = aurweb.db.query(AccountType, AccountType.AccountType == "User").first()
assert record is not None

1
test/test_package_maintainer_routes.py
View file

@ -768,6 +768,7 @@ def test_pm_proposal_vote(client, proposal):
assert response.status_code == int(HTTPStatus.OK)
# Check that the proposal record got updated.
db.refresh(voteinfo)
assert voteinfo.Yes == yes + 1
# Check that the new PMVote exists.

6
test/test_packages_routes.py
View file

@ -1528,6 +1528,7 @@ def test_packages_post_disown_as_maintainer(
errors = get_errors(resp.text)
expected = "You did not select any packages to disown."
assert errors[0].text.strip() == expected
db.refresh(package)
assert package.PackageBase.Maintainer is not None
# Try to disown `package` without giving the confirm argument.
@ -1552,6 +1553,7 @@ def test_packages_post_disown_as_maintainer(
data={"action": "disown", "IDs": [package.ID], "confirm": True},
)
assert resp.status_code == int(HTTPStatus.BAD_REQUEST)
db.refresh(package)
assert package.PackageBase.Maintainer is not None
errors = get_errors(resp.text)
expected = "You are not allowed to disown one of the packages you selected."
@ -1565,6 +1567,7 @@ def test_packages_post_disown_as_maintainer(
data={"action": "disown", "IDs": [package.ID], "confirm": True},
)
db.get_session().expire_all()
assert package.PackageBase.Maintainer is None
successes = get_successes(resp.text)
expected = "The selected packages have been disowned."
@ -1649,6 +1652,7 @@ def test_packages_post_delete(
# Whoo. Now, let's finally make a valid request as `pm_user`
# to delete `package`.
pkgname = package.PackageBase.Name
with client as request:
request.cookies = pm_cookies
resp = request.post(
@ -1661,7 +1665,7 @@ def test_packages_post_delete(
assert successes[0].text.strip() == expected
# Expect that the package deletion was logged.
pkgbases = [package.PackageBase.Name]
pkgbases = [pkgname]
expected = (
f"Privileged user '{pm_user.Username}' deleted the "
f"following package bases: {str(pkgbases)}."

26
test/test_pkgbase_routes.py
View file

@ -688,6 +688,7 @@ def test_pkgbase_comment_pin_as_co(
assert resp.status_code == int(HTTPStatus.SEE_OTHER)
# Assert that PinnedTS got set.
db.refresh(comment)
assert comment.PinnedTS > 0
# Unpin the comment we just pinned.
@ -698,6 +699,7 @@ def test_pkgbase_comment_pin_as_co(
assert resp.status_code == int(HTTPStatus.SEE_OTHER)
# Let's assert that PinnedTS was unset.
db.refresh(comment)
assert comment.PinnedTS == 0
@ -716,6 +718,7 @@ def test_pkgbase_comment_pin(
assert resp.status_code == int(HTTPStatus.SEE_OTHER)
# Assert that PinnedTS got set.
db.refresh(comment)
assert comment.PinnedTS > 0
# Unpin the comment we just pinned.
@ -726,6 +729,7 @@ def test_pkgbase_comment_pin(
assert resp.status_code == int(HTTPStatus.SEE_OTHER)
# Let's assert that PinnedTS was unset.
db.refresh(comment)
assert comment.PinnedTS == 0
@ -1040,6 +1044,7 @@ def test_pkgbase_flag(
request.cookies = cookies
resp = request.post(endpoint, data={"comments": "Test"})
assert resp.status_code == int(HTTPStatus.SEE_OTHER)
db.refresh(pkgbase)
assert pkgbase.Flagger == user
assert pkgbase.FlaggerComment == "Test"
@ -1077,6 +1082,7 @@ def test_pkgbase_flag(
request.cookies = user2_cookies
resp = request.post(endpoint)
assert resp.status_code == int(HTTPStatus.SEE_OTHER)
db.refresh(pkgbase)
assert pkgbase.Flagger == user
# Now, test that the 'maintainer' user can.
@ -1085,6 +1091,7 @@ def test_pkgbase_flag(
request.cookies = maint_cookies
resp = request.post(endpoint)
assert resp.status_code == int(HTTPStatus.SEE_OTHER)
db.refresh(pkgbase)
assert pkgbase.Flagger is None
# Flag it again.
@ -1098,6 +1105,7 @@ def test_pkgbase_flag(
request.cookies = cookies
resp = request.post(endpoint)
assert resp.status_code == int(HTTPStatus.SEE_OTHER)
db.refresh(pkgbase)
assert pkgbase.Flagger is None
@ -1170,6 +1178,7 @@ def test_pkgbase_vote(client: TestClient, user: User, package: Package):
vote = pkgbase.package_votes.filter(PackageVote.UsersID == user.ID).first()
assert vote is not None
db.refresh(pkgbase)
assert pkgbase.NumVotes == 1
# Remove vote.
@ -1181,6 +1190,7 @@ def test_pkgbase_vote(client: TestClient, user: User, package: Package):
vote = pkgbase.package_votes.filter(PackageVote.UsersID == user.ID).first()
assert vote is None
db.refresh(pkgbase)
assert pkgbase.NumVotes == 0
@ -1592,9 +1602,9 @@ def test_pkgbase_merge_post(
assert resp.status_code == int(HTTPStatus.SEE_OTHER)
# Save these relationships for later comparison.
comments = package.PackageBase.comments.all()
notifs = package.PackageBase.notifications.all()
votes = package.PackageBase.package_votes.all()
comments = [row.__dict__ for row in package.PackageBase.comments.all()]
notifs = [row.__dict__ for row in package.PackageBase.notifications.all()]
votes = [row.__dict__ for row in package.PackageBase.package_votes.all()]
# Merge the package into target.
endpoint = f"/pkgbase/{package.PackageBase.Name}/merge"
@ -1612,9 +1622,13 @@ def test_pkgbase_merge_post(
# Assert that the original comments, notifs and votes we setup
# got migrated to target as intended.
assert comments == target.comments.all()
assert notifs == target.notifications.all()
assert votes == target.package_votes.all()
db.get_session().refresh(target)
assert len(comments) == target.comments.count()
assert comments[0]["PackageBaseID"] != target.ID
assert len(notifs) == target.notifications.count()
assert notifs[0]["PackageBaseID"] != target.ID
assert len(votes) == target.package_votes.count()
assert votes[0]["PackageBaseID"] != target.ID
# ...and that the package got deleted.
package = db.query(Package).filter(Package.Name == pkgname).first()

4
test/test_requests.py
View file

@ -649,6 +649,7 @@ def test_orphan_request(
assert resp.headers.get("location") == f"/pkgbase/{pkgbase.Name}"
# We should have unset the maintainer.
db.refresh(pkgbase)
assert pkgbase.Maintainer is None
# We should have removed the comaintainers.
@ -748,6 +749,7 @@ def test_orphan_as_maintainer(client: TestClient, auser: User, pkgbase: PackageB
# As the pkgbase maintainer, disowning the package just ends up
# either promoting the lowest priority comaintainer or removing
# the associated maintainer relationship altogether.
db.refresh(pkgbase)
assert pkgbase.Maintainer is None
@ -1044,6 +1046,7 @@ def test_requests_close_post(client: TestClient, user: User, pkgreq: PackageRequ
resp = request.post(f"/requests/{pkgreq.ID}/close")
assert resp.status_code == int(HTTPStatus.SEE_OTHER)
db.refresh(pkgreq)
assert pkgreq.Status == REJECTED_ID
assert pkgreq.Closer == user
assert pkgreq.ClosureComment == str()
@ -1060,6 +1063,7 @@ def test_requests_close_post_rejected(
)
assert resp.status_code == int(HTTPStatus.SEE_OTHER)
db.refresh(pkgreq)
assert pkgreq.Status == REJECTED_ID
assert pkgreq.Closer == user
assert pkgreq.ClosureComment == str()

1
test/test_routes.py
View file

@ -102,6 +102,7 @@ def test_user_language(client: TestClient, user: User):
req.cookies = {"AURSID": sid}
response = req.post("/language", data=post_data)
assert response.status_code == int(HTTPStatus.SEE_OTHER)
db.refresh(user)
assert user.LangPreference == "de"

10
test/test_voteinfo.py
View file

@ -1,5 +1,5 @@
import pytest
from sqlalchemy.exc import IntegrityError
from sqlalchemy.exc import IntegrityError, SAWarning
from aurweb import db, time
from aurweb.db import create, rollback
@ -109,7 +109,7 @@ def test_voteinfo_null_submitter_raises(user: User):
def test_voteinfo_null_agenda_raises(user: User):
with pytest.raises(IntegrityError):
with pytest.raises(IntegrityError), pytest.warns(SAWarning):
with db.begin():
create(
VoteInfo,
@ -123,7 +123,7 @@ def test_voteinfo_null_agenda_raises(user: User):
def test_voteinfo_null_user_raises(user: User):
with pytest.raises(IntegrityError):
with pytest.raises(IntegrityError), pytest.warns(SAWarning):
with db.begin():
create(
VoteInfo,
@ -137,7 +137,7 @@ def test_voteinfo_null_user_raises(user: User):
def test_voteinfo_null_submitted_raises(user: User):
with pytest.raises(IntegrityError):
with pytest.raises(IntegrityError), pytest.warns(SAWarning):
with db.begin():
create(
VoteInfo,
@ -151,7 +151,7 @@ def test_voteinfo_null_submitted_raises(user: User):
def test_voteinfo_null_end_raises(user: User):
with pytest.raises(IntegrityError):
with pytest.raises(IntegrityError), pytest.warns(SAWarning):
with db.begin():
create(
VoteInfo,