Commit a836892c authored by Kevin Morris's avatar Kevin Morris
Browse files

aurweb.db: add query, create, delete helpers



Takes sqlalchemy kwargs or stanzas:

query(Model, Model.Column == value)
query(Model, and_(Model.Column == value, Model.Column != "BAD!"))

Updated tests to reflect the new utility and a comment about upcoming
function deprecation is added to get_account_type().

From here on, phase out the use of get_account_type().

+ aurweb.db: Added create utility function
+ aurweb.db: Added delete utility function

The `delete` function can be used to delete a record by search
kwargs directly.

Example:
    delete(User, User.ID == 6)

All three functions added in this commit are typically useful to
perform these operations without having to import aurweb.db.session.
Removes a bit of redundancy overall.

Signed-off-by: Kevin Morris's avatarKevin Morris <kevr@0cost.org>
parent 5185df62
......@@ -11,6 +11,24 @@ Session = None
session = None
def query(model, *args, **kwargs):
return session.query(model).filter(*args, **kwargs)
def create(model, *args, **kwargs):
instance = model(*args, **kwargs)
session.add(instance)
session.commit()
return instance
def delete(model, *args, **kwargs):
instance = session.query(model).filter(*args, **kwargs)
for record in instance:
session.delete(record)
session.commit()
def get_sqlalchemy_url():
"""
Build an SQLAlchemy for use with create_engine based on the aurweb configuration.
......
import pytest
from aurweb.models.account_type import AccountType
from aurweb.models.user import User
from aurweb.testing import setup_test_db
from aurweb.testing.models import make_user
account_type = None
@pytest.fixture(autouse=True)
def setup():
setup_test_db()
setup_test_db("Users")
def test_account_type():
""" Test creating an AccountType, and reading its columns. """
from aurweb.db import session
global account_type
account_type = AccountType(AccountType="TestUser")
session.add(account_type)
session.commit()
yield account_type
session.delete(account_type)
session.commit()
def test_account_type():
""" Test creating an AccountType, and reading its columns. """
from aurweb.db import session
# Make sure it got created and was given an ID.
assert bool(account_type.ID)
......@@ -25,4 +39,20 @@ def test_account_type():
"<AccountType(ID='%s', AccountType='TestUser')>" % (
account_type.ID)
session.delete(account_type)
record = session.query(AccountType).filter(
AccountType.AccountType == "TestUser").first()
assert account_type == record
def test_user_account_type_relationship():
from aurweb.db import session
user = make_user(Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword",
AccountType=account_type)
assert user.AccountType == account_type
assert account_type.users.filter(User.ID == user.ID).first()
session.delete(user)
session.commit()
......@@ -3,6 +3,7 @@ import re
import sqlite3
import tempfile
from datetime import datetime
from unittest import mock
import mysql.connector
......@@ -11,6 +12,7 @@ import pytest
import aurweb.config
from aurweb import db
from aurweb.models.account_type import AccountType
from aurweb.testing import setup_test_db
......@@ -39,7 +41,7 @@ class DBConnection:
@pytest.fixture(autouse=True)
def setup_db():
setup_test_db()
setup_test_db("Bans")
def test_sqlalchemy_sqlite_url():
......@@ -174,3 +176,12 @@ def test_connection_execute_paramstyle_unsupported():
"SELECT * FROM AccountTypes WHERE AccountType = ?",
["User"]
).fetchall()
def test_create_delete():
db.create(AccountType, AccountType="test")
record = db.query(AccountType, AccountType.AccountType == "test").first()
assert record is not None
db.delete(AccountType, AccountType.AccountType == "test")
record = db.query(AccountType, AccountType.AccountType == "test").first()
assert record is None
......@@ -7,14 +7,26 @@ import pytest
from fastapi.testclient import TestClient
from aurweb.asgi import app
from aurweb.db import query
from aurweb.models.account_type import AccountType
from aurweb.testing import setup_test_db
client = TestClient(app)
user = None
@pytest.fixture
def setup():
setup_test_db("Users", "Session")
global user
setup_test_db("Users", "Sessions")
account_type = query(AccountType,
AccountType.AccountType == "User").first()
user = make_user(Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword",
AccountType=account_type)
def test_index():
......@@ -54,6 +66,7 @@ def test_language_invalid_next():
response = req.post("/language", data=post_data)
assert response.status_code == int(HTTPStatus.BAD_REQUEST)
def test_language_query_params():
""" Test the language post route with query params. """
next = urllib.parse.quote_plus("/")
......@@ -73,4 +86,5 @@ def test_error_messages():
response1 = client.get("/thisroutedoesnotexist")
response2 = client.get("/raisefivethree")
assert response1.status_code == int(HTTPStatus.NOT_FOUND)
assert response2.status_code == int(HTTPStatus.SERVICE_UNAVAILABLE)
\ No newline at end of file
assert response2.status_code == int(HTTPStatus.SERVICE_UNAVAILABLE)
import pytest
import aurweb.config
from aurweb.db import query
from aurweb.models.account_type import AccountType
from aurweb.models.user import User
from aurweb.testing import setup_test_db
......@@ -26,7 +29,6 @@ def test_user():
Salt="efgh", ResetKey="blahblah")
session.add(user)
session.commit()
assert user in account_type.users
# Make sure the user was created and given an ID.
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment