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

change(aurweb): add parallel tests and improve aurweb.db

Browse source 
This change utilizes pytest-xdist to perform a multiproc test
run and reworks aurweb.db's code. We no longer use a global
engine, session or Session, but we now use a memo of engines
and sessions as they are requested, based on the PYTEST_CURRENT_TEST
environment variable, which is available during testing.

Additionally, this change strips several SQLite components
out of the Python code-base.

SQLite is still compatible with PHP and sharness tests, but
not with our FastAPI implementation.

More changes:
------------
- Remove use of aurweb.db.session global in other code.
- Use new aurweb.db.name() dynamic db name function in env.py.
- Added 'addopts' to pytest.ini which utilizes multiprocessing.
    - Highly recommended to leave this be or modify `-n auto` to
      `-n {cpu_threads}` where cpu_threads is at least 2.

Signed-off-by: Kevin Morris <kevr@0cost.org>
This commit is contained in:
Kevin Morris 2021-11-17 00:33:41 -08:00
parent 07aac768d6
commit fa43f6bc3e
No known key found for this signature in database
GPG key ID: F7E46DED420788F3
55 changed files with 781 additions and 884 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

251
aurweb/db.py
View file

@ -1,29 +1,34 @@
import functools
import hashlib
import math
import os
import re
from typing import Iterable, NewType
from sqlalchemy import event
from sqlalchemy.orm import Query, scoped_session
import sqlalchemy
from sqlalchemy import create_engine, event
from sqlalchemy.engine.base import Engine
from sqlalchemy.engine.url import URL
from sqlalchemy.orm import Query, Session, SessionTransaction, scoped_session, sessionmaker
import aurweb.config
import aurweb.util
# See get_engine.
engine = None
from aurweb import logging
# ORM Session class.
Session = None
logger = logging.get_logger(__name__)
# Global ORM Session object.
session = None
DRIVERS = {
"mysql": "mysql+mysqldb"
}
# Global introspected object memo.
introspected = dict()
# A mocked up type.
Base = NewType("aurweb.models.declarative_base.Base", "Base")
# Some types we don't get access to in this module.
Base = NewType("Base", "aurweb.models.declarative_base.Base")
def make_random_value(table: str, column: str):
@ -56,14 +61,85 @@ def make_random_value(table: str, column: str):
length = col.type.length
string = aurweb.util.make_random_string(length)
while session.query(table).filter(column == string).first():
while query(table).filter(column == string).first():
string = aurweb.util.make_random_string(length)
return string
def get_session():
def test_name() -> str:
"""
Return the unhashed database name.
The unhashed database name is determined (lower = higher priority) by:
-------------------------------------------
1. {test_suite} portion of PYTEST_CURRENT_TEST
2. aurweb.config.get("database", "name")
During `pytest` runs, the PYTEST_CURRENT_TEST environment variable
is set to the current test in the format `{test_suite}::{test_func}`.
This allows tests to use a suite-specific database for its runs,
which decouples database state from test suites.
:return: Unhashed database name
"""
db = os.environ.get("PYTEST_CURRENT_TEST",
aurweb.config.get("database", "name"))
return db.split(":")[0]
def name() -> str:
"""
Return sanitized database name that can be used for tests or production.
If test_name() starts with "test/", the database name is SHA-1 hashed,
prefixed with 'db', and returned. Otherwise, test_name() is passed
through and not hashed at all.
:return: SHA1-hashed database name prefixed with 'db'
"""
dbname = test_name()
if not dbname.startswith("test/"):
return dbname
sha1 = hashlib.sha1(dbname.encode()).hexdigest()
return "db" + sha1
# Module-private global memo used to store SQLAlchemy sessions.
_sessions = dict()
def get_session(engine: Engine = None) -> Session:
""" Return aurweb.db's global session. """
return session
dbname = name()
global _sessions
if dbname not in _sessions:
if not engine: # pragma: no cover
engine = get_engine()
Session = scoped_session(
sessionmaker(autocommit=True, autoflush=False, bind=engine))
_sessions[dbname] = Session()
# If this is the first grab of this session, log out the
# database name used.
raw_dbname = test_name()
logger.debug(f"DBName({raw_dbname}): {dbname}")
return _sessions.get(dbname)
def pop_session(dbname: str) -> None:
"""
Pop a Session out of the private _sessions memo.
:param dbname: Database name
:raises KeyError: When `dbname` does not exist in the memo
"""
global _sessions
_sessions.pop(dbname)
def refresh(model: Base) -> Base:
@ -121,41 +197,40 @@ def add(model: Base) -> Base:
return model
def begin():
def begin() -> SessionTransaction:
""" Begin an SQLAlchemy SessionTransaction. """
return get_session().begin()
def get_sqlalchemy_url():
def get_sqlalchemy_url() -> URL:
"""
Build an SQLAlchemy for use with create_engine based on the aurweb configuration.
"""
import sqlalchemy
Build an SQLAlchemy URL for use with create_engine.
constructor = sqlalchemy.engine.url.URL
:return: sqlalchemy.engine.url.URL
"""
constructor = URL
parts = sqlalchemy.__version__.split('.')
major = int(parts[0])
minor = int(parts[1])
if major == 1 and minor >= 4: # pragma: no cover
constructor = sqlalchemy.engine.url.URL.create
constructor = URL.create
aur_db_backend = aurweb.config.get('database', 'backend')
if aur_db_backend == 'mysql':
if aurweb.config.get_with_fallback('database', 'port', fallback=None):
port = aurweb.config.get('database', 'port')
param_query = None
else:
port = None
param_query = {
'unix_socket': aurweb.config.get('database', 'socket')
}
param_query = {}
port = aurweb.config.get_with_fallback("database", "port", None)
if not port:
param_query["unix_socket"] = aurweb.config.get(
"database", "socket")
return constructor(
'mysql+mysqldb',
DRIVERS.get(aur_db_backend),
username=aurweb.config.get('database', 'user'),
password=aurweb.config.get('database', 'password'),
password=aurweb.config.get_with_fallback('database', 'password',
fallback=None),
host=aurweb.config.get('database', 'host'),
database=aurweb.config.get('database', 'name'),
database=name(),
port=port,
query=param_query
)
@ -168,58 +243,83 @@ def get_sqlalchemy_url():
raise ValueError('unsupported database backend')
def get_engine(echo: bool = False):
def sqlite_regexp(regex, item) -> bool: # pragma: no cover
""" Method which mimics SQL's REGEXP for SQLite. """
return bool(re.search(regex, str(item)))
def setup_sqlite(engine: Engine) -> None: # pragma: no cover
""" Perform setup for an SQLite engine. """
@event.listens_for(engine, "connect")
def do_begin(conn, record):
create_deterministic_function = functools.partial(
conn.create_function,
deterministic=True
)
create_deterministic_function("REGEXP", 2, sqlite_regexp)
# Module-private global memo used to store SQLAlchemy engines.
_engines = dict()
def get_engine(dbname: str = None, echo: bool = False) -> Engine:
"""
Return the global SQLAlchemy engine.
Return the SQLAlchemy engine for `dbname`.
The engine is created on the first call to get_engine and then stored in the
`engine` global variable for the next calls.
:param dbname: Database name (default: aurweb.db.name())
:param echo: Flag passed through to sqlalchemy.create_engine
:return: SQLAlchemy Engine instance
"""
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
if not dbname:
dbname = name()
global engine, session, Session
if engine is None:
global _engines
if dbname not in _engines:
db_backend = aurweb.config.get("database", "backend")
connect_args = dict()
db_backend = aurweb.config.get("database", "backend")
if db_backend == "sqlite":
# check_same_thread is for a SQLite technicality
# https://fastapi.tiangolo.com/tutorial/sql-databases/#note
is_sqlite = bool(db_backend == "sqlite")
if is_sqlite: # pragma: no cover
connect_args["check_same_thread"] = False
engine = create_engine(get_sqlalchemy_url(),
connect_args=connect_args,
echo=echo)
kwargs = {
"echo": echo,
"connect_args": connect_args
}
_engines[dbname] = create_engine(get_sqlalchemy_url(), **kwargs)
Session = scoped_session(
sessionmaker(autocommit=True, autoflush=False, bind=engine))
session = Session()
if is_sqlite: # pragma: no cover
setup_sqlite(_engines.get(dbname))
if db_backend == "sqlite":
# For SQLite, we need to add some custom functions as
# they are used in the reference graph method.
def regexp(regex, item):
return bool(re.search(regex, str(item)))
@event.listens_for(engine, "connect")
def do_begin(conn, record):
create_deterministic_function = functools.partial(
conn.create_function,
deterministic=True
)
create_deterministic_function("REGEXP", 2, regexp)
return engine
return _engines.get(dbname)
def kill_engine():
global engine, Session, session
if engine:
session.close()
engine.dispose()
engine = Session = session = None
def pop_engine(dbname: str) -> None:
"""
Pop an Engine out of the private _engines memo.
:param dbname: Database name
:raises KeyError: When `dbname` does not exist in the memo
"""
global _engines
_engines.pop(dbname)
def kill_engine() -> None:
""" Close the current session and dispose of the engine. """
dbname = name()
session = get_session()
session.close()
pop_session(dbname)
engine = get_engine()
engine.dispose()
pop_engine(dbname)
def connect():
@ -248,7 +348,9 @@ class ConnectionExecutor:
def paramstyle(self):
return self._paramstyle
def execute(self, query, params=()):
def execute(self, query, params=()): # pragma: no cover
# TODO: SQLite support has been removed in FastAPI. It remains
# here to fund its support for PHP until it is removed.
if self._paramstyle in ('format', 'pyformat'):
query = query.replace('%', '%%').replace('?', '%s')
elif self._paramstyle == 'qmark':
@ -278,16 +380,19 @@ class Connection:
if aur_db_backend == 'mysql':
import MySQLdb
aur_db_host = aurweb.config.get('database', 'host')
aur_db_name = aurweb.config.get('database', 'name')
aur_db_name = name()
aur_db_user = aurweb.config.get('database', 'user')
aur_db_pass = aurweb.config.get('database', 'password')
aur_db_pass = aurweb.config.get_with_fallback(
'database', 'password', str())
aur_db_socket = aurweb.config.get('database', 'socket')
self._conn = MySQLdb.connect(host=aur_db_host,
user=aur_db_user,
passwd=aur_db_pass,
db=aur_db_name,
unix_socket=aur_db_socket)
elif aur_db_backend == 'sqlite':
elif aur_db_backend == 'sqlite': # pragma: no cover
# TODO: SQLite support has been removed in FastAPI. It remains
# here to fund its support for PHP until it is removed.
import sqlite3
aur_db_name = aurweb.config.get('database', 'name')
self._conn = sqlite3.connect(aur_db_name)

6
aurweb/routers/auth.py
View file

@ -6,7 +6,7 @@ from fastapi.responses import HTMLResponse, RedirectResponse
import aurweb.config
from aurweb import cookies
from aurweb import cookies, db
from aurweb.auth import auth_required
from aurweb.l10n import get_translator_for_request
from aurweb.models import User
@ -45,9 +45,7 @@ async def login_post(request: Request,
raise HTTPException(status_code=HTTPStatus.BAD_REQUEST,
detail=_("Bad Referer header."))
from aurweb.db import session
user = session.query(User).filter(User.Username == user).first()
user = db.query(User).filter(User.Username == user).first()
if not user:
return await login_template(request, next,
errors=["Bad username or password."])

8
aurweb/routers/packages.py
View file

@ -1014,12 +1014,12 @@ def pkgbase_disown_instance(request: Request, pkgbase: models.PackageBase):
models.PackageComaintainer.Priority.asc()
).limit(1).first()
if co:
with db.begin():
with db.begin():
if co:
pkgbase.Maintainer = co.User
db.delete(co)
else:
pkgbase.Maintainer = None
else:
pkgbase.Maintainer = None
notif.send()

4
aurweb/schema.py
View file

@ -16,13 +16,13 @@ db_backend = aurweb.config.get("database", "backend")
@compiles(TINYINT, 'sqlite')
def compile_tinyint_sqlite(type_, compiler, **kw):
def compile_tinyint_sqlite(type_, compiler, **kw): # pragma: no cover
"""TINYINT is not supported on SQLite. Substitute it with INTEGER."""
return 'INTEGER'
@compiles(BIGINT, 'sqlite')
def compile_bigint_sqlite(type_, compiler, **kw):
def compile_bigint_sqlite(type_, compiler, **kw): # pragma: no cover
"""
For SQLite's AUTOINCREMENT to work on BIGINT columns, we need to map BIGINT
to INTEGER. Aside from that, BIGINT is the same as INTEGER for SQLite.

68
aurweb/testing/__init__.py
View file

@ -1,26 +1,6 @@
from itertools import chain
import aurweb.db
def references_graph(table):
""" Taken from Django's sqlite3/operations.py. """
query = """
WITH tables AS (
SELECT :table name
UNION
SELECT sqlite_master.name
FROM sqlite_master
JOIN tables ON (sql REGEXP :regexp_1 || tables.name || :regexp_2)
) SELECT name FROM tables;
"""
params = {
"table": table,
"regexp_1": r'(?i)\s+references\s+("|\')?',
"regexp_2": r'("|\')?\s*\(',
}
cursor = aurweb.db.get_session().execute(query, params=params)
return [row[0] for row in cursor.fetchall()]
from aurweb import models
def setup_test_db(*args):
@ -47,22 +27,38 @@ def setup_test_db(*args):
aurweb.db.get_engine()
tables = list(args)
if not tables:
tables = [
models.AcceptedTerm.__tablename__,
models.ApiRateLimit.__tablename__,
models.Ban.__tablename__,
models.Group.__tablename__,
models.License.__tablename__,
models.OfficialProvider.__tablename__,
models.Package.__tablename__,
models.PackageBase.__tablename__,
models.PackageBlacklist.__tablename__,
models.PackageComaintainer.__tablename__,
models.PackageComment.__tablename__,
models.PackageDependency.__tablename__,
models.PackageGroup.__tablename__,
models.PackageKeyword.__tablename__,
models.PackageLicense.__tablename__,
models.PackageNotification.__tablename__,
models.PackageRelation.__tablename__,
models.PackageRequest.__tablename__,
models.PackageSource.__tablename__,
models.PackageVote.__tablename__,
models.Session.__tablename__,
models.SSHPubKey.__tablename__,
models.Term.__tablename__,
models.TUVote.__tablename__,
models.TUVoteInfo.__tablename__,
models.User.__tablename__,
]
db_backend = aurweb.config.get("database", "backend")
if db_backend != "sqlite": # pragma: no cover
aurweb.db.get_session().execute("SET FOREIGN_KEY_CHECKS = 0")
else:
# We're using sqlite, setup tables to be deleted without violating
# foreign key constraints by graphing references.
tables = set(chain.from_iterable(
references_graph(table) for table in tables))
aurweb.db.get_session().execute("SET FOREIGN_KEY_CHECKS = 0")
for table in tables:
aurweb.db.get_session().execute(f"DELETE FROM {table}")
if db_backend != "sqlite": # pragma: no cover
aurweb.db.get_session().execute("SET FOREIGN_KEY_CHECKS = 1")
# Expunge all objects from SQLAlchemy's IdentityMap.
aurweb.db.get_session().execute("SET FOREIGN_KEY_CHECKS = 1")
aurweb.db.get_session().expunge_all()

8
migrations/env.py
View file

@ -41,8 +41,8 @@ def run_migrations_offline():
script output.
"""
db_name = aurweb.config.get("database", "name")
logging.info(f"Performing offline migration on database '{db_name}'.")
dbname = aurweb.db.name()
logging.info(f"Performing offline migration on database '{dbname}'.")
context.configure(
url=aurweb.db.get_sqlalchemy_url(),
target_metadata=target_metadata,
@ -61,8 +61,8 @@ def run_migrations_online():
and associate a connection with the context.
"""
db_name = aurweb.config.get("database", "name")
logging.info(f"Performing online migration on database '{db_name}'.")
dbname = aurweb.db.name()
logging.info(f"Performing online migration on database '{dbname}'.")
connectable = sqlalchemy.create_engine(
aurweb.db.get_sqlalchemy_url(),
poolclass=sqlalchemy.pool.NullPool,

6
pytest.ini
View file

@ -8,3 +8,9 @@
# https://bugs.python.org/issue45097
filterwarnings =
ignore::DeprecationWarning:asyncio.base_events
# Build in coverage and pytest-xdist multiproc testing.
addopts = --cov=aurweb --cov-append --dist load --dist loadfile -n auto
# Our pytest units are located in the ./test/ directory.
testpaths = test

178
test/conftest.py Normal file
View file

@ -0,0 +1,178 @@
"""
pytest configuration.
The conftest.py file is used to define pytest-global fixtures
or actions run before tests.
Module scoped fixtures:
----------------------
- setup_database
- db_session (depends: setup_database)
Function scoped fixtures:
------------------------
- db_test (depends: db_session)
Tests in aurweb which access the database **must** use the `db_test`
function fixture. Most database tests simply require this fixture in
an autouse=True setup fixture, or for fixtures used in DB tests example:
# In scenarios which there are no other database fixtures
# or other database fixtures dependency paths don't always
# hit `db_test`.
@pytest.fixture(autouse=True)
def setup(db_test):
return
# In scenarios where we can embed the `db_test` fixture in
# specific fixtures that already exist.
@pytest.fixture
def user(db_test):
with db.begin():
user = db.create(User, ...)
yield user
The `db_test` fixture triggers our module-level database fixtures,
then clears the database for each test function run in that module.
It is done this way because migration has a large cost; migrating
ahead of each function takes too long when compared to this method.
"""
import pytest
from filelock import FileLock
from sqlalchemy import create_engine
from sqlalchemy.engine import URL
from sqlalchemy.engine.base import Engine
from sqlalchemy.orm import scoped_session
import aurweb.config
import aurweb.db
from aurweb import initdb, logging, testing
logger = logging.get_logger(__name__)
def test_engine() -> Engine:
"""
Return a privileged SQLAlchemy engine with no database.
This method is particularly useful for providing an engine that
can be used to create and drop databases from an SQL server.
:return: SQLAlchemy Engine instance (not connected to a database)
"""
unix_socket = aurweb.config.get_with_fallback("database", "socket", None)
kwargs = {
"username": aurweb.config.get("database", "user"),
"password": aurweb.config.get_with_fallback(
"database", "password", None),
"host": aurweb.config.get("database", "host"),
"port": aurweb.config.get_with_fallback("database", "port", None),
"query": {
"unix_socket": unix_socket
}
}
backend = aurweb.config.get("database", "backend")
driver = aurweb.db.DRIVERS.get(backend)
return create_engine(URL.create(driver, **kwargs))
class AlembicArgs:
"""
Masquerade an ArgumentParser like structure.
This structure is needed to pass conftest-specific arguments
to initdb.run duration database creation.
"""
verbose = False
use_alembic = True
def _create_database(engine: Engine, dbname: str) -> None:
"""
Create a test database.
:param engine: Engine returned by test_engine()
:param dbname: Database name to create
"""
conn = engine.connect()
conn.execute(f"CREATE DATABASE {dbname}")
conn.close()
initdb.run(AlembicArgs)
def _drop_database(engine: Engine, dbname: str) -> None:
"""
Drop a test database.
:param engine: Engine returned by test_engine()
:param dbname: Database name to drop
"""
aurweb.schema.metadata.drop_all(bind=engine)
conn = engine.connect()
conn.execute(f"DROP DATABASE {dbname}")
conn.close()
@pytest.fixture(scope="module")
def setup_database(tmp_path_factory: pytest.fixture,
worker_id: pytest.fixture) -> None:
""" Create and drop a database for the suite this fixture is used in. """
engine = test_engine()
dbname = aurweb.db.name()
if worker_id == "master": # pragma: no cover
# If we're not running tests through multiproc pytest-xdist.
yield _create_database(engine, dbname)
_drop_database(engine, dbname)
return
root_tmp_dir = tmp_path_factory.getbasetemp().parent
fn = root_tmp_dir / dbname
with FileLock(str(fn) + ".lock"):
if fn.is_file():
# If the data file exists, skip database creation.
yield
else:
# Otherwise, create the data file and create the database.
fn.write_text("1")
yield _create_database(engine, dbname)
_drop_database(engine, dbname)
@pytest.fixture(scope="module")
def db_session(setup_database: pytest.fixture) -> scoped_session:
"""
Yield a database session based on aurweb.db.name().
The returned session is popped out of persistence after the test is run.
"""
# After the test runs, aurweb.db.name() ends up returning the
# configured database, because PYTEST_CURRENT_TEST is removed.
dbname = aurweb.db.name()
session = aurweb.db.get_session()
yield session
# Close the session and pop it.
session.close()
aurweb.db.pop_session(dbname)
@pytest.fixture
def db_test(db_session: scoped_session) -> None:
"""
Database test fixture.
This fixture should be included in any tests which access the
database. It ensures that a test database is created and
alembic migrated, takes care of dropping the database when
the module is complete, and runs setup_test_db() to clear out
tables for each test.
Tests using this fixture should access the database
session via aurweb.db.get_session().
"""
testing.setup_test_db()

37
test/test_accepted_term.py
View file

@ -2,38 +2,33 @@ import pytest
from sqlalchemy.exc import IntegrityError
from aurweb.db import create, query
from aurweb import db
from aurweb.models.accepted_term import AcceptedTerm
from aurweb.models.account_type import AccountType
from aurweb.models.account_type import USER_ID
from aurweb.models.term import Term
from aurweb.models.user import User
from aurweb.testing import setup_test_db
user = term = accepted_term = None
@pytest.fixture(autouse=True)
def setup():
global user, term, accepted_term
def setup(db_test):
global user, term
setup_test_db("Users", "AcceptedTerms", "Terms")
with db.begin():
user = db.create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword",
AccountTypeID=USER_ID)
account_type = query(AccountType,
AccountType.AccountType == "User").first()
user = create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword",
AccountType=account_type)
term = create(Term, Description="Test term", URL="https://test.term")
term = db.create(Term, Description="Test term",
URL="https://test.term")
yield term
# Eradicate any terms we created.
setup_test_db("AcceptedTerms", "Terms")
def test_accepted_term():
accepted_term = create(AcceptedTerm, User=user, Term=term)
with db.begin():
accepted_term = db.create(AcceptedTerm, User=user, Term=term)
# Make sure our AcceptedTerm relationships got initialized properly.
assert accepted_term.User == user
@ -42,14 +37,10 @@ def test_accepted_term():
def test_accepted_term_null_user_raises_exception():
from aurweb.db import session
with pytest.raises(IntegrityError):
create(AcceptedTerm, Term=term)
session.rollback()
AcceptedTerm(Term=term)
def test_accepted_term_null_term_raises_exception():
from aurweb.db import session
with pytest.raises(IntegrityError):
create(AcceptedTerm, User=user)
session.rollback()
AcceptedTerm(User=user)

54
test/test_account_type.py
View file

@ -1,31 +1,29 @@
import pytest
from aurweb.db import begin, create, delete, query
from aurweb import db
from aurweb.models.account_type import AccountType
from aurweb.models.user import User
from aurweb.testing import setup_test_db
account_type = None
@pytest.fixture(autouse=True)
def setup():
setup_test_db("Users")
global account_type
with begin():
account_type = create(AccountType, AccountType="TestUser")
yield account_type
with begin():
delete(account_type)
def setup(db_test):
return
def test_account_type():
@pytest.fixture
def account_type() -> AccountType:
with db.begin():
account_type_ = db.create(AccountType, AccountType="TestUser")
yield account_type_
with db.begin():
db.delete(account_type_)
def test_account_type(account_type):
""" Test creating an AccountType, and reading its columns. """
# Make sure it got created and was given an ID.
# Make sure it got db.created and was given an ID.
assert bool(account_type.ID)
# Next, test our string functions.
@ -34,20 +32,20 @@ def test_account_type():
"<AccountType(ID='%s', AccountType='TestUser')>" % (
account_type.ID)
record = query(AccountType,
AccountType.AccountType == "TestUser").first()
record = db.query(AccountType,
AccountType.AccountType == "TestUser").first()
assert account_type == record
def test_user_account_type_relationship():
with begin():
user = create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword",
AccountType=account_type)
def test_user_account_type_relationship(account_type):
with db.begin():
user = db.create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword",
AccountType=account_type)
assert user.AccountType == account_type
# This must be deleted here to avoid foreign key issues when
# This must be db.deleted here to avoid foreign key issues when
# deleting the temporary AccountType in the fixture.
with begin():
delete(user)
with db.begin():
db.delete(user)

9
test/test_accounts_routes.py
View file

@ -20,7 +20,6 @@ from aurweb.models.session import Session
from aurweb.models.ssh_pub_key import SSHPubKey, get_fingerprint
from aurweb.models.term import Term
from aurweb.models.user import User
from aurweb.testing import setup_test_db
from aurweb.testing.html import get_errors
from aurweb.testing.requests import Request
@ -50,11 +49,9 @@ def make_ssh_pubkey():
@pytest.fixture(autouse=True)
def setup():
def setup(db_test):
global user
setup_test_db("Users", "Sessions", "Bans", "Terms", "AcceptedTerms")
account_type = query(AccountType,
AccountType.AccountType == "User").first()
@ -65,10 +62,6 @@ def setup():
yield user
# Remove term records so other tests don't get them
# and falsely redirect.
setup_test_db("Terms", "AcceptedTerms")
@pytest.fixture
def tu_user():

19
test/test_api_rate_limit.py
View file

@ -3,19 +3,18 @@ import pytest
from sqlalchemy.exc import IntegrityError
from aurweb import db
from aurweb.db import create
from aurweb.models.api_rate_limit import ApiRateLimit
from aurweb.testing import setup_test_db
@pytest.fixture(autouse=True)
def setup():
setup_test_db("ApiRateLimit")
def setup(db_test):
return
def test_api_rate_key_creation():
with db.begin():
rate = create(ApiRateLimit, IP="127.0.0.1", Requests=10, WindowStart=1)
rate = db.create(ApiRateLimit, IP="127.0.0.1", Requests=10,
WindowStart=1)
assert rate.IP == "127.0.0.1"
assert rate.Requests == 10
assert rate.WindowStart == 1
@ -23,19 +22,15 @@ def test_api_rate_key_creation():
def test_api_rate_key_ip_default():
with db.begin():
api_rate_limit = create(ApiRateLimit, Requests=10, WindowStart=1)
api_rate_limit = db.create(ApiRateLimit, Requests=10, WindowStart=1)
assert api_rate_limit.IP == str()
def test_api_rate_key_null_requests_raises_exception():
with pytest.raises(IntegrityError):
with db.begin():
create(ApiRateLimit, IP="127.0.0.1", WindowStart=1)
db.rollback()
ApiRateLimit(IP="127.0.0.1", WindowStart=1)
def test_api_rate_key_null_window_start_raises_exception():
with pytest.raises(IntegrityError):
with db.begin():
create(ApiRateLimit, IP="127.0.0.1", Requests=1)
db.rollback()
ApiRateLimit(IP="127.0.0.1", Requests=1)

26
test/test_auth.py
View file

@ -6,28 +6,22 @@ from sqlalchemy.exc import IntegrityError
from aurweb import db
from aurweb.auth import AnonymousUser, BasicAuthBackend, account_type_required, has_credential
from aurweb.db import create, query
from aurweb.models.account_type import USER, USER_ID, AccountType
from aurweb.models.account_type import USER, USER_ID
from aurweb.models.session import Session
from aurweb.models.user import User
from aurweb.testing import setup_test_db
from aurweb.testing.requests import Request
user = backend = request = None
@pytest.fixture(autouse=True)
def setup():
def setup(db_test):
global user, backend, request
setup_test_db("Users", "Sessions")
account_type = query(AccountType,
AccountType.AccountType == "User").first()
with db.begin():
user = create(User, Username="test", Email="test@example.com",
RealName="Test User", Passwd="testPassword",
AccountType=account_type)
user = db.create(User, Username="test", Email="test@example.com",
RealName="Test User", Passwd="testPassword",
AccountTypeID=USER_ID)
backend = BasicAuthBackend()
request = Request()
@ -56,10 +50,8 @@ async def test_auth_backend_invalid_user_id():
# Create a new session with a fake user id.
now_ts = datetime.utcnow().timestamp()
with pytest.raises(IntegrityError):
with db.begin():
create(Session, UsersID=666, SessionID="realSession",
LastUpdateTS=now_ts + 5)
db.rollback()
Session(UsersID=666, SessionID="realSession",
LastUpdateTS=now_ts + 5)
@pytest.mark.asyncio
@ -68,8 +60,8 @@ async def test_basic_auth_backend():
# equal the real_user.
now_ts = datetime.utcnow().timestamp()
with db.begin():
create(Session, UsersID=user.ID, SessionID="realSession",
LastUpdateTS=now_ts + 5)
db.create(Session, UsersID=user.ID, SessionID="realSession",
LastUpdateTS=now_ts + 5)
request.cookies["AURSID"] = "realSession"
_, result = await backend.authenticate(request)
assert result == user

5
test/test_auth_routes.py
View file

@ -13,7 +13,6 @@ from aurweb.db import begin, create, query
from aurweb.models.account_type import AccountType
from aurweb.models.session import Session
from aurweb.models.user import User
from aurweb.testing import setup_test_db
# Some test global constants.
TEST_USERNAME = "test"
@ -27,11 +26,9 @@ user = client = None
@pytest.fixture(autouse=True)
def setup():
def setup(db_test):
global user, client
setup_test_db("Users", "Sessions", "Bans")
account_type = query(AccountType,
AccountType.AccountType == "User").first()

9
test/test_ban.py
View file

@ -9,18 +9,15 @@ from sqlalchemy import exc as sa_exc
from aurweb import db
from aurweb.db import create
from aurweb.models.ban import Ban, is_banned
from aurweb.testing import setup_test_db
from aurweb.testing.requests import Request
ban = request = None
@pytest.fixture(autouse=True)
def setup():
def setup(db_test):
global ban, request
setup_test_db("Bans")
ts = datetime.utcnow() + timedelta(seconds=30)
with db.begin():
ban = create(Ban, IPAddress="127.0.0.1", BanTS=ts)
@ -33,8 +30,6 @@ def test_ban():
def test_invalid_ban():
from aurweb.db import session
with pytest.raises(sa_exc.IntegrityError):
bad_ban = Ban(BanTS=datetime.utcnow())
@ -44,7 +39,7 @@ def test_invalid_ban():
with warnings.catch_warnings():
warnings.simplefilter("ignore", sa_exc.SAWarning)
with db.begin():
session.add(bad_ban)
db.add(bad_ban)
# Since we got a transaction failure, we need to rollback.
db.rollback()

7
test/test_cache.py
View file

@ -3,14 +3,11 @@ import pytest
from aurweb import cache, db
from aurweb.models.account_type import USER_ID
from aurweb.models.user import User
from aurweb.testing import setup_test_db
@pytest.fixture(autouse=True)
def setup():
setup_test_db(
User.__tablename__
)
def setup(db_test):
return
class StubRedis:

7
test/test_captcha.py
View file

@ -1,8 +1,15 @@
import hashlib
import pytest
from aurweb import captcha
@pytest.fixture(autouse=True)
def setup(db_test):
return
def test_captcha_salts():
""" Make sure we can get some captcha salts. """
salts = captcha.get_captcha_salts()

110
test/test_db.py
View file

@ -12,7 +12,6 @@ import aurweb.initdb
from aurweb import db
from aurweb.models.account_type import AccountType
from aurweb.testing import setup_test_db
class Args:
@ -96,16 +95,10 @@ def make_temp_mysql_config():
@pytest.fixture(autouse=True)
def setup_db():
def setup(db_test):
if os.path.exists("/tmp/aurweb.sqlite3"):
os.remove("/tmp/aurweb.sqlite3")
# In various places in this test, we reinitialize the engine.
# Make sure we kill the previous engine before initializing
# it via setup_test_db().
aurweb.db.kill_engine()
setup_test_db()
def test_sqlalchemy_sqlite_url():
tmpctx, tmp = make_temp_sqlite_config()
@ -159,24 +152,6 @@ def test_sqlalchemy_unknown_backend():
def test_db_connects_without_fail():
""" This only tests the actual config supplied to pytest. """
db.connect()
assert db.engine is not None
def test_connection_class_sqlite_without_fail():
tmpctx, tmp = make_temp_sqlite_config()
with tmpctx:
with mock.patch.dict(os.environ, {"AUR_CONFIG": tmp}):
aurweb.config.rehash()
aurweb.db.kill_engine()
aurweb.initdb.run(Args())
conn = db.Connection()
cur = conn.execute(
"SELECT AccountType FROM AccountTypes WHERE ID = ?", (1,))
account_type = cur.fetchone()[0]
assert account_type == "User"
aurweb.config.rehash()
def test_connection_class_unsupported_backend():
@ -200,83 +175,6 @@ def test_connection_mysql():
aurweb.config.rehash()
@mock.patch("sqlite3.connect", mock.MagicMock(return_value=DBConnection()))
@mock.patch.object(sqlite3, "paramstyle", "qmark")
def test_connection_sqlite():
db.Connection()
@mock.patch("sqlite3.connect", mock.MagicMock(return_value=DBConnection()))
@mock.patch.object(sqlite3, "paramstyle", "format")
def test_connection_execute_paramstyle_format():
tmpctx, tmp = make_temp_sqlite_config()
with tmpctx:
with mock.patch.dict(os.environ, {"AUR_CONFIG": tmp}):
aurweb.config.rehash()
aurweb.db.kill_engine()
aurweb.initdb.run(Args())
# Test SQLite route of clearing tables.
setup_test_db("Users", "Bans")
conn = db.Connection()
# First, test ? to %s format replacement.
account_types = conn\
.execute("SELECT * FROM AccountTypes WHERE AccountType = ?",
["User"]).fetchall()
assert account_types == \
["SELECT * FROM AccountTypes WHERE AccountType = %s", ["User"]]
# Test other format replacement.
account_types = conn\
.execute("SELECT * FROM AccountTypes WHERE AccountType = %",
["User"]).fetchall()
assert account_types == \
["SELECT * FROM AccountTypes WHERE AccountType = %%", ["User"]]
aurweb.config.rehash()
@mock.patch("sqlite3.connect", mock.MagicMock(return_value=DBConnection()))
@mock.patch.object(sqlite3, "paramstyle", "qmark")
def test_connection_execute_paramstyle_qmark():
tmpctx, tmp = make_temp_sqlite_config()
with tmpctx:
with mock.patch.dict(os.environ, {"AUR_CONFIG": tmp}):
aurweb.config.rehash()
aurweb.db.kill_engine()
aurweb.initdb.run(Args())
conn = db.Connection()
# We don't modify anything when using qmark, so test equality.
account_types = conn\
.execute("SELECT * FROM AccountTypes WHERE AccountType = ?",
["User"]).fetchall()
assert account_types == \
["SELECT * FROM AccountTypes WHERE AccountType = ?", ["User"]]
aurweb.config.rehash()
@mock.patch("sqlite3.connect", mock.MagicMock(return_value=DBConnection()))
@mock.patch.object(sqlite3, "paramstyle", "unsupported")
def test_connection_execute_paramstyle_unsupported():
tmpctx, tmp = make_temp_sqlite_config()
with tmpctx:
with mock.patch.dict(os.environ, {"AUR_CONFIG": tmp}):
aurweb.config.rehash()
conn = db.Connection()
with pytest.raises(ValueError, match="unsupported paramstyle"):
conn.execute(
"SELECT * FROM AccountTypes WHERE AccountType = ?",
["User"]
).fetchall()
aurweb.config.rehash()
def test_create_delete():
with db.begin():
account_type = db.create(AccountType, AccountType="test")
@ -318,3 +216,9 @@ def test_connection_executor_mysql_paramstyle():
def test_connection_executor_sqlite_paramstyle():
executor = db.ConnectionExecutor(None, backend="sqlite")
assert executor.paramstyle() == sqlite3.paramstyle
def test_name_without_pytest_current_test():
with mock.patch.dict("os.environ", {}, clear=True):
dbname = aurweb.db.name()
assert dbname == aurweb.config.get("database", "name")

5
test/test_dependency_type.py
View file

@ -2,12 +2,11 @@ import pytest
from aurweb.db import begin, create, delete, query
from aurweb.models.dependency_type import DependencyType
from aurweb.testing import setup_test_db
@pytest.fixture(autouse=True)
def setup():
setup_test_db()
def setup(db_test):
return
def test_dependency_types():

9
test/test_group.py
View file

@ -4,12 +4,11 @@ from sqlalchemy.exc import IntegrityError
from aurweb import db
from aurweb.models.group import Group
from aurweb.testing import setup_test_db
@pytest.fixture(autouse=True)
def setup():
setup_test_db("Groups")
def setup(db_test):
return
def test_group_creation():
@ -21,6 +20,4 @@ def test_group_creation():
def test_group_null_name_raises_exception():
with pytest.raises(IntegrityError):
with db.begin():
db.create(Group)
db.rollback()
Group()

11
test/test_homepage.py
View file

@ -18,7 +18,6 @@ from aurweb.models.package_request import PackageRequest
from aurweb.models.request_type import DELETION_ID, RequestType
from aurweb.models.user import User
from aurweb.redis import redis_connection
from aurweb.testing import setup_test_db
from aurweb.testing.html import parse_root
from aurweb.testing.requests import Request
@ -26,14 +25,8 @@ client = TestClient(app)
@pytest.fixture(autouse=True)
def setup():
yield setup_test_db(
User.__tablename__,
Package.__tablename__,
PackageBase.__tablename__,
PackageComaintainer.__tablename__,
PackageRequest.__tablename__
)
def setup(db_test):
return
@pytest.fixture

5
test/test_html.py
View file

@ -8,14 +8,13 @@ from fastapi.testclient import TestClient
from aurweb import asgi, db
from aurweb.models.account_type import TRUSTED_USER_ID, USER_ID, AccountType
from aurweb.models.user import User
from aurweb.testing import setup_test_db
from aurweb.testing.html import get_errors, get_successes, parse_root
from aurweb.testing.requests import Request
@pytest.fixture(autouse=True)
def setup():
setup_test_db(User.__tablename__)
def setup(db_test):
return
@pytest.fixture

9
test/test_initdb.py
View file

@ -1,3 +1,5 @@
import pytest
import aurweb.config
import aurweb.db
import aurweb.initdb
@ -5,6 +7,11 @@ import aurweb.initdb
from aurweb.models.account_type import AccountType
@pytest.fixture(autouse=True)
def setup(db_test):
return
class Args:
use_alembic = True
verbose = True
@ -15,6 +22,8 @@ def test_run():
aurweb.db.kill_engine()
metadata.drop_all(aurweb.db.get_engine())
aurweb.initdb.run(Args())
# Check that constant table rows got added via initdb.
record = aurweb.db.query(AccountType,
AccountType.AccountType == "User").first()
assert record is not None

9
test/test_license.py
View file

@ -4,12 +4,11 @@ from sqlalchemy.exc import IntegrityError
from aurweb import db
from aurweb.models.license import License
from aurweb.testing import setup_test_db
@pytest.fixture(autouse=True)
def setup():
setup_test_db("Licenses")
def setup(db_test):
return
def test_license_creation():
@ -21,6 +20,4 @@ def test_license_creation():
def test_license_null_name_raises_exception():
with pytest.raises(IntegrityError):
with db.begin():
db.create(License)
db.rollback()
License()

23
test/test_official_provider.py
View file

@ -4,12 +4,11 @@ from sqlalchemy.exc import IntegrityError
from aurweb import db
from aurweb.models.official_provider import OfficialProvider
from aurweb.testing import setup_test_db
@pytest.fixture(autouse=True)
def setup():
setup_test_db("OfficialProviders")
def setup(db_test):
return
def test_official_provider_creation():
@ -53,26 +52,14 @@ def test_official_provider_cs():
def test_official_provider_null_name_raises_exception():
with pytest.raises(IntegrityError):
with db.begin():
db.create(OfficialProvider,
Repo="some-repo",
Provides="some-provides")
db.rollback()
OfficialProvider(Repo="some-repo", Provides="some-provides")
def test_official_provider_null_repo_raises_exception():
with pytest.raises(IntegrityError):
with db.begin():
db.create(OfficialProvider,
Name="some-name",
Provides="some-provides")
db.rollback()
OfficialProvider(Name="some-name", Provides="some-provides")
def test_official_provider_null_provides_raises_exception():
with pytest.raises(IntegrityError):
with db.begin():
db.create(OfficialProvider,
Name="some-name",
Repo="some-repo")
db.rollback()
OfficialProvider(Name="some-name", Repo="some-repo")

5
test/test_package.py
View file

@ -8,17 +8,14 @@ from aurweb.models.account_type import AccountType
from aurweb.models.package import Package
from aurweb.models.package_base import PackageBase
from aurweb.models.user import User
from aurweb.testing import setup_test_db
user = pkgbase = package = None
@pytest.fixture(autouse=True)
def setup():
def setup(db_test):
global user, pkgbase, package
setup_test_db("Packages", "PackageBases", "Users")
account_type = db.query(AccountType,
AccountType.AccountType == "User").first()

5
test/test_package_base.py
View file

@ -8,17 +8,14 @@ from aurweb import db
from aurweb.models.account_type import AccountType
from aurweb.models.package_base import PackageBase
from aurweb.models.user import User
from aurweb.testing import setup_test_db
user = None
@pytest.fixture(autouse=True)
def setup():
def setup(db_test):
global user
setup_test_db("Users", "PackageBases")
account_type = db.query(AccountType,
AccountType.AccountType == "User").first()
with db.begin():

16
test/test_package_blacklist.py
View file

@ -6,20 +6,18 @@ from aurweb import db
from aurweb.models.package_base import PackageBase
from aurweb.models.package_blacklist import PackageBlacklist
from aurweb.models.user import User
from aurweb.testing import setup_test_db
user = pkgbase = None
@pytest.fixture(autouse=True)
def setup():
def setup(db_test):
global user, pkgbase
setup_test_db("PackageBlacklist", "PackageBases", "Users")
user = db.create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword")
pkgbase = db.create(PackageBase, Name="test-package", Maintainer=user)
with db.begin():
user = db.create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword")
pkgbase = db.create(PackageBase, Name="test-package", Maintainer=user)
def test_package_blacklist_creation():
@ -31,6 +29,4 @@ def test_package_blacklist_creation():
def test_package_blacklist_null_name_raises_exception():
with pytest.raises(IntegrityError):
with db.begin():
db.create(PackageBlacklist)
db.rollback()
PackageBlacklist()

28
test/test_package_comaintainer.py
View file

@ -2,29 +2,28 @@ import pytest
from sqlalchemy.exc import IntegrityError
from aurweb.db import create, rollback
from aurweb import db
from aurweb.models.package_base import PackageBase
from aurweb.models.package_comaintainer import PackageComaintainer
from aurweb.models.user import User
from aurweb.testing import setup_test_db
user = pkgbase = None
@pytest.fixture(autouse=True)
def setup():
def setup(db_test):
global user, pkgbase
setup_test_db("Users", "PackageBases", "PackageComaintainers")
user = create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword")
pkgbase = create(PackageBase, Name="test-package", Maintainer=user)
with db.begin():
user = db.create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword")
pkgbase = db.create(PackageBase, Name="test-package", Maintainer=user)
def test_package_comaintainer_creation():
package_comaintainer = create(PackageComaintainer, User=user,
PackageBase=pkgbase, Priority=5)
with db.begin():
package_comaintainer = db.create(PackageComaintainer, User=user,
PackageBase=pkgbase, Priority=5)
assert bool(package_comaintainer)
assert package_comaintainer.User == user
assert package_comaintainer.PackageBase == pkgbase
@ -33,17 +32,14 @@ def test_package_comaintainer_creation():
def test_package_comaintainer_null_user_raises_exception():
with pytest.raises(IntegrityError):
create(PackageComaintainer, PackageBase=pkgbase, Priority=1)
rollback()
PackageComaintainer(PackageBase=pkgbase, Priority=1)
def test_package_comaintainer_null_pkgbase_raises_exception():
with pytest.raises(IntegrityError):
create(PackageComaintainer, User=user, Priority=1)
rollback()
PackageComaintainer(User=user, Priority=1)
def test_package_comaintainer_null_priority_raises_exception():
with pytest.raises(IntegrityError):
create(PackageComaintainer, User=user, PackageBase=pkgbase)
rollback()
PackageComaintainer(User=user, PackageBase=pkgbase)

59
test/test_package_comment.py
View file

@ -2,70 +2,55 @@ import pytest
from sqlalchemy.exc import IntegrityError
from aurweb.db import begin, create, query, rollback
from aurweb.models.account_type import AccountType
from aurweb import db
from aurweb.models.account_type import USER_ID
from aurweb.models.package_base import PackageBase
from aurweb.models.package_comment import PackageComment
from aurweb.models.user import User
from aurweb.testing import setup_test_db
user = pkgbase = None
@pytest.fixture(autouse=True)
def setup():
setup_test_db("PackageBases", "PackageComments", "Users")
def setup(db_test):
global user, pkgbase
account_type = query(AccountType,
AccountType.AccountType == "User").first()
with begin():
user = create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword",
AccountType=account_type)
pkgbase = create(PackageBase, Name="test-package", Maintainer=user)
with db.begin():
user = db.create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword",
AccountTypeID=USER_ID)
pkgbase = db.create(PackageBase, Name="test-package", Maintainer=user)
def test_package_comment_creation():
with begin():
package_comment = create(PackageComment,
PackageBase=pkgbase,
User=user,
Comments="Test comment.",
RenderedComment="Test rendered comment.")
with db.begin():
package_comment = db.create(PackageComment, PackageBase=pkgbase,
User=user, Comments="Test comment.",
RenderedComment="Test rendered comment.")
assert bool(package_comment.ID)
def test_package_comment_null_package_base_raises_exception():
with pytest.raises(IntegrityError):
with begin():
create(PackageComment, User=user, Comments="Test comment.",
RenderedComment="Test rendered comment.")
rollback()
PackageComment(User=user, Comments="Test comment.",
RenderedComment="Test rendered comment.")
def test_package_comment_null_user_raises_exception():
with pytest.raises(IntegrityError):
with begin():
create(PackageComment, PackageBase=pkgbase,
Comments="Test comment.",
RenderedComment="Test rendered comment.")
rollback()
PackageComment(PackageBase=pkgbase,
Comments="Test comment.",
RenderedComment="Test rendered comment.")
def test_package_comment_null_comments_raises_exception():
with pytest.raises(IntegrityError):
with begin():
create(PackageComment, PackageBase=pkgbase, User=user,
RenderedComment="Test rendered comment.")
rollback()
PackageComment(PackageBase=pkgbase, User=user,
RenderedComment="Test rendered comment.")
def test_package_comment_null_renderedcomment_defaults():
with begin():
record = create(PackageComment,
PackageBase=pkgbase,
User=user,
Comments="Test comment.")
with db.begin():
record = db.create(PackageComment, PackageBase=pkgbase,
User=user, Comments="Test comment.")
assert record.RenderedComment == str()

95
test/test_package_dependency.py
View file

@ -3,117 +3,70 @@ import pytest
from sqlalchemy.exc import IntegrityError
from aurweb import db
from aurweb.db import create, query
from aurweb.models.account_type import AccountType
from aurweb.models.dependency_type import DependencyType
from aurweb.models.account_type import USER_ID
from aurweb.models.dependency_type import CHECKDEPENDS_ID, DEPENDS_ID, MAKEDEPENDS_ID, OPTDEPENDS_ID
from aurweb.models.package import Package
from aurweb.models.package_base import PackageBase
from aurweb.models.package_dependency import PackageDependency
from aurweb.models.user import User
from aurweb.testing import setup_test_db
user = pkgbase = package = None
@pytest.fixture(autouse=True)
def setup():
def setup(db_test):
global user, pkgbase, package
setup_test_db("Users", "PackageBases", "Packages", "PackageDepends")
account_type = query(AccountType,
AccountType.AccountType == "User").first()
with db.begin():
user = create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword",
AccountType=account_type)
pkgbase = create(PackageBase,
Name="test-package",
Maintainer=user)
package = create(Package,
PackageBase=pkgbase,
Name=pkgbase.Name,
Description="Test description.",
URL="https://test.package")
user = db.create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword",
AccountTypeID=USER_ID)
pkgbase = db.create(PackageBase,
Name="test-package",
Maintainer=user)
package = db.create(Package,
PackageBase=pkgbase,
Name=pkgbase.Name,
Description="Test description.",
URL="https://test.package")
def test_package_dependencies():
depends = query(DependencyType, DependencyType.Name == "depends").first()
with db.begin():
pkgdep = create(PackageDependency, Package=package,
DependencyType=depends,
DepName="test-dep")
pkgdep = db.create(PackageDependency, Package=package,
DepTypeID=DEPENDS_ID, DepName="test-dep")
assert pkgdep.DepName == "test-dep"
assert pkgdep.Package == package
assert pkgdep.DependencyType == depends
assert pkgdep in depends.package_dependencies
assert pkgdep in package.package_dependencies
makedepends = query(DependencyType,
DependencyType.Name == "makedepends").first()
with db.begin():
pkgdep.DependencyType = makedepends
assert pkgdep.DepName == "test-dep"
assert pkgdep.Package == package
assert pkgdep.DependencyType == makedepends
assert pkgdep in makedepends.package_dependencies
assert pkgdep in package.package_dependencies
pkgdep.DepTypeID = MAKEDEPENDS_ID
checkdepends = query(DependencyType,
DependencyType.Name == "checkdepends").first()
with db.begin():
pkgdep.DependencyType = checkdepends
assert pkgdep.DepName == "test-dep"
assert pkgdep.Package == package
assert pkgdep.DependencyType == checkdepends
assert pkgdep in checkdepends.package_dependencies
assert pkgdep in package.package_dependencies
pkgdep.DepTypeID = CHECKDEPENDS_ID
optdepends = query(DependencyType,
DependencyType.Name == "optdepends").first()
with db.begin():
pkgdep.DependencyType = optdepends
assert pkgdep.DepName == "test-dep"
assert pkgdep.Package == package
assert pkgdep.DependencyType == optdepends
assert pkgdep in optdepends.package_dependencies
assert pkgdep in package.package_dependencies
pkgdep.DepTypeID = OPTDEPENDS_ID
assert not pkgdep.is_package()
with db.begin():
base = create(PackageBase, Name=pkgdep.DepName, Maintainer=user)
create(Package, PackageBase=base, Name=pkgdep.DepName)
base = db.create(PackageBase, Name=pkgdep.DepName, Maintainer=user)
db.create(Package, PackageBase=base, Name=pkgdep.DepName)
assert pkgdep.is_package()
def test_package_dependencies_null_package_raises_exception():
depends = query(DependencyType, DependencyType.Name == "depends").first()
with pytest.raises(IntegrityError):
with db.begin():
create(PackageDependency,
DependencyType=depends,
DepName="test-dep")
db.rollback()
PackageDependency(DepTypeID=DEPENDS_ID, DepName="test-dep")
def test_package_dependencies_null_dependency_type_raises_exception():
with pytest.raises(IntegrityError):
with db.begin():
create(PackageDependency,
Package=package,
DepName="test-dep")
db.rollback()
PackageDependency(Package=package, DepName="test-dep")
def test_package_dependencies_null_depname_raises_exception():
depends = query(DependencyType, DependencyType.Name == "depends").first()
with pytest.raises(IntegrityError):
with db.begin():
create(PackageDependency,
Package=package,
DependencyType=depends)
db.rollback()
PackageDependency(DepTypeID=DEPENDS_ID, Package=package)

37
test/test_package_group.py
View file

@ -2,51 +2,44 @@ import pytest
from sqlalchemy.exc import IntegrityError
from aurweb.db import create, query
from aurweb.models.account_type import AccountType
from aurweb import db
from aurweb.models.account_type import USER_ID
from aurweb.models.group import Group
from aurweb.models.package import Package
from aurweb.models.package_base import PackageBase
from aurweb.models.package_group import PackageGroup
from aurweb.models.user import User
from aurweb.testing import setup_test_db
user = group = pkgbase = package = None
@pytest.fixture(autouse=True)
def setup():
def setup(db_test):
global user, group, pkgbase, package
setup_test_db("Users", "PackageBases", "Packages",
"Groups", "PackageGroups")
with db.begin():
user = db.create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword",
AccountTypeID=USER_ID)
group = db.create(Group, Name="Test Group")
account_type = query(AccountType,
AccountType.AccountType == "User").first()
user = create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword",
AccountType=account_type)
group = create(Group, Name="Test Group")
pkgbase = create(PackageBase, Name="test-package", Maintainer=user)
package = create(Package, PackageBase=pkgbase, Name=pkgbase.Name)
with db.begin():
pkgbase = db.create(PackageBase, Name="test-package", Maintainer=user)
package = db.create(Package, PackageBase=pkgbase, Name=pkgbase.Name)
def test_package_group():
package_group = create(PackageGroup, Package=package, Group=group)
with db.begin():
package_group = db.create(PackageGroup, Package=package, Group=group)
assert package_group.Group == group
assert package_group.Package == package
def test_package_group_null_package_raises_exception():
from aurweb.db import session
with pytest.raises(IntegrityError):
create(PackageGroup, Group=group)
session.rollback()
PackageGroup(Group=group)
def test_package_group_null_group_raises_exception():
from aurweb.db import session
with pytest.raises(IntegrityError):
create(PackageGroup, Package=package)
session.rollback()
PackageGroup(Package=package)

37
test/test_package_keyword.py
View file

@ -2,44 +2,37 @@ import pytest
from sqlalchemy.exc import IntegrityError
from aurweb.db import create, query
from aurweb.models.account_type import AccountType
from aurweb import db
from aurweb.models.account_type import USER_ID
from aurweb.models.package_base import PackageBase
from aurweb.models.package_keyword import PackageKeyword
from aurweb.models.user import User
from aurweb.testing import setup_test_db
user = pkgbase = None
@pytest.fixture(autouse=True)
def setup():
def setup(db_test):
global user, pkgbase
setup_test_db("Users", "PackageBases", "PackageKeywords")
account_type = query(AccountType,
AccountType.AccountType == "User").first()
user = create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword",
AccountType=account_type)
pkgbase = create(PackageBase,
Name="beautiful-package",
Maintainer=user)
with db.begin():
user = db.create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword",
AccountTypeID=USER_ID)
pkgbase = db.create(PackageBase,
Name="beautiful-package",
Maintainer=user)
def test_package_keyword():
pkg_keyword = create(PackageKeyword,
PackageBase=pkgbase,
Keyword="test")
with db.begin():
pkg_keyword = db.create(PackageKeyword,
PackageBase=pkgbase,
Keyword="test")
assert pkg_keyword in pkgbase.keywords
assert pkgbase == pkg_keyword.PackageBase
def test_package_keyword_null_pkgbase_raises_exception():
from aurweb.db import session
with pytest.raises(IntegrityError):
create(PackageKeyword,
Keyword="test")
session.rollback()
PackageKeyword(Keyword="test")

38
test/test_package_license.py
View file

@ -2,51 +2,45 @@ import pytest
from sqlalchemy.exc import IntegrityError
from aurweb.db import create, query
from aurweb.models.account_type import AccountType
from aurweb import db
from aurweb.models.account_type import USER_ID
from aurweb.models.license import License
from aurweb.models.package import Package
from aurweb.models.package_base import PackageBase
from aurweb.models.package_license import PackageLicense
from aurweb.models.user import User
from aurweb.testing import setup_test_db
user = license = pkgbase = package = None
@pytest.fixture(autouse=True)
def setup():
def setup(db_test):
global user, license, pkgbase, package
setup_test_db("Users", "PackageBases", "Packages",
"Licenses", "PackageLicenses")
with db.begin():
user = db.create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword",
AccountTypeID=USER_ID)
license = db.create(License, Name="Test License")
account_type = query(AccountType,
AccountType.AccountType == "User").first()
user = create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword",
AccountType=account_type)
license = create(License, Name="Test License")
pkgbase = create(PackageBase, Name="test-package", Maintainer=user)
package = create(Package, PackageBase=pkgbase, Name=pkgbase.Name)
with db.begin():
pkgbase = db.create(PackageBase, Name="test-package", Maintainer=user)
package = db.create(Package, PackageBase=pkgbase, Name=pkgbase.Name)
def test_package_license():
package_license = create(PackageLicense, Package=package, License=license)
with db.begin():
package_license = db.create(PackageLicense, Package=package,
License=license)
assert package_license.License == license
assert package_license.Package == package
def test_package_license_null_package_raises_exception():
from aurweb.db import session
with pytest.raises(IntegrityError):
create(PackageLicense, License=license)
session.rollback()
PackageLicense(License=license)
def test_package_license_null_license_raises_exception():
from aurweb.db import session
with pytest.raises(IntegrityError):
create(PackageLicense, Package=package)
session.rollback()
PackageLicense(Package=package)

25
test/test_package_notification.py
View file

@ -2,29 +2,28 @@ import pytest
from sqlalchemy.exc import IntegrityError
from aurweb.db import create, rollback
from aurweb import db
from aurweb.models.package_base import PackageBase
from aurweb.models.package_notification import PackageNotification
from aurweb.models.user import User
from aurweb.testing import setup_test_db
user = pkgbase = None
@pytest.fixture(autouse=True)
def setup():
def setup(db_test):
global user, pkgbase
setup_test_db("Users", "PackageBases", "PackageNotifications")
user = create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword")
pkgbase = create(PackageBase, Name="test-package", Maintainer=user)
with db.begin():
user = db.create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword")
pkgbase = db.create(PackageBase, Name="test-package", Maintainer=user)
def test_package_notification_creation():
package_notification = create(PackageNotification, User=user,
PackageBase=pkgbase)
with db.begin():
package_notification = db.create(
PackageNotification, User=user, PackageBase=pkgbase)
assert bool(package_notification)
assert package_notification.User == user
assert package_notification.PackageBase == pkgbase
@ -32,11 +31,9 @@ def test_package_notification_creation():
def test_package_notification_null_user_raises_exception():
with pytest.raises(IntegrityError):
create(PackageNotification, PackageBase=pkgbase)
rollback()
PackageNotification(PackageBase=pkgbase)
def test_package_notification_null_pkgbase_raises_exception():
with pytest.raises(IntegrityError):
create(PackageNotification, User=user)
rollback()
PackageNotification(User=user)

90
test/test_package_relation.py
View file

@ -1,103 +1,63 @@
import pytest
from sqlalchemy.exc import IntegrityError, OperationalError
from sqlalchemy.exc import IntegrityError
from aurweb import db
from aurweb.db import create, query
from aurweb.models.account_type import AccountType
from aurweb.models.account_type import USER_ID
from aurweb.models.package import Package
from aurweb.models.package_base import PackageBase
from aurweb.models.package_relation import PackageRelation
from aurweb.models.relation_type import RelationType
from aurweb.models.relation_type import CONFLICTS_ID, PROVIDES_ID, REPLACES_ID
from aurweb.models.user import User
from aurweb.testing import setup_test_db
user = pkgbase = package = None
@pytest.fixture(autouse=True)
def setup():
def setup(db_test):
global user, pkgbase, package
setup_test_db("Users", "PackageBases", "Packages", "PackageRelations")
account_type = query(AccountType,
AccountType.AccountType == "User").first()
with db.begin():
user = create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword",
AccountType=account_type)
pkgbase = create(PackageBase,
Name="test-package",
Maintainer=user)
package = create(Package,
PackageBase=pkgbase,
Name=pkgbase.Name,
Description="Test description.",
URL="https://test.package")
user = db.create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword",
AccountTypeID=USER_ID)
pkgbase = db.create(PackageBase,
Name="test-package",
Maintainer=user)
package = db.create(Package,
PackageBase=pkgbase,
Name=pkgbase.Name,
Description="Test description.",
URL="https://test.package")
def test_package_relation():
conflicts = query(RelationType, RelationType.Name == "conflicts").first()
with db.begin():
pkgrel = create(PackageRelation, Package=package,
RelationType=conflicts,
RelName="test-relation")
pkgrel = db.create(PackageRelation, Package=package,
RelTypeID=CONFLICTS_ID,
RelName="test-relation")
assert pkgrel.RelName == "test-relation"
assert pkgrel.Package == package
assert pkgrel.RelationType == conflicts
assert pkgrel in conflicts.package_relations
assert pkgrel in package.package_relations
provides = query(RelationType, RelationType.Name == "provides").first()
with db.begin():
pkgrel.RelationType = provides
assert pkgrel.RelName == "test-relation"
assert pkgrel.Package == package
assert pkgrel.RelationType == provides
assert pkgrel in provides.package_relations
assert pkgrel in package.package_relations
pkgrel.RelTypeID = PROVIDES_ID
replaces = query(RelationType, RelationType.Name == "replaces").first()
with db.begin():
pkgrel.RelationType = replaces
assert pkgrel.RelName == "test-relation"
assert pkgrel.Package == package
assert pkgrel.RelationType == replaces
assert pkgrel in replaces.package_relations
assert pkgrel in package.package_relations
pkgrel.RelTypeID = REPLACES_ID
def test_package_relation_null_package_raises_exception():
conflicts = query(RelationType, RelationType.Name == "conflicts").first()
assert conflicts is not None
with pytest.raises(IntegrityError):
with db.begin():
create(PackageRelation,
RelationType=conflicts,
RelName="test-relation")
db.rollback()
PackageRelation(RelTypeID=CONFLICTS_ID, RelName="test-relation")
def test_package_relation_null_relation_type_raises_exception():
with pytest.raises(IntegrityError):
with db.begin():
create(PackageRelation,
Package=package,
RelName="test-relation")
db.rollback()
PackageRelation(Package=package, RelName="test-relation")
def test_package_relation_null_relname_raises_exception():
depends = query(RelationType, RelationType.Name == "conflicts").first()
assert depends is not None
with pytest.raises((OperationalError, IntegrityError)):
with db.begin():
create(PackageRelation,
Package=package,
RelationType=depends)
db.rollback()
with pytest.raises(IntegrityError):
PackageRelation(Package=package, RelTypeID=CONFLICTS_ID)

107
test/test_package_request.py
View file

@ -5,41 +5,35 @@ import pytest
from sqlalchemy.exc import IntegrityError
from aurweb import db
from aurweb.db import create, query, rollback
from aurweb.models.account_type import USER_ID
from aurweb.models.package_base import PackageBase
from aurweb.models.package_request import (ACCEPTED, ACCEPTED_ID, CLOSED, CLOSED_ID, PENDING, PENDING_ID, REJECTED,
REJECTED_ID, PackageRequest)
from aurweb.models.request_type import RequestType
from aurweb.models.request_type import MERGE_ID
from aurweb.models.user import User
from aurweb.testing import setup_test_db
user = pkgbase = None
@pytest.fixture(autouse=True)
def setup():
def setup(db_test):
global user, pkgbase
setup_test_db("PackageRequests", "PackageBases", "Users")
with db.begin():
user = create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword")
pkgbase = create(PackageBase, Name="test-package", Maintainer=user)
user = db.create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword",
AccountTypeID=USER_ID)
pkgbase = db.create(PackageBase, Name="test-package", Maintainer=user)
def test_package_request_creation():
request_type = query(RequestType, RequestType.Name == "merge").first()
assert request_type.Name == "merge"
with db.begin():
package_request = create(PackageRequest, RequestType=request_type,
User=user, PackageBase=pkgbase,
PackageBaseName=pkgbase.Name,
Comments=str(), ClosureComment=str())
package_request = db.create(PackageRequest, ReqTypeID=MERGE_ID,
User=user, PackageBase=pkgbase,
PackageBaseName=pkgbase.Name,
Comments=str(), ClosureComment=str())
assert bool(package_request.ID)
assert package_request.RequestType == request_type
assert package_request.User == user
assert package_request.PackageBase == pkgbase
assert package_request.PackageBaseName == pkgbase.Name
@ -47,22 +41,18 @@ def test_package_request_creation():
assert package_request.ClosureComment == str()
# Make sure that everything is cross-referenced with relationships.
assert package_request in request_type.package_requests
assert package_request in user.package_requests
assert package_request in pkgbase.requests
def test_package_request_closed():
request_type = query(RequestType, RequestType.Name == "merge").first()
assert request_type.Name == "merge"
ts = int(datetime.utcnow().timestamp())
with db.begin():
package_request = create(PackageRequest, RequestType=request_type,
User=user, PackageBase=pkgbase,
PackageBaseName=pkgbase.Name,
Closer=user, ClosedTS=ts,
Comments=str(), ClosureComment=str())
package_request = db.create(PackageRequest, ReqTypeID=MERGE_ID,
User=user, PackageBase=pkgbase,
PackageBaseName=pkgbase.Name,
Closer=user, ClosedTS=ts,
Comments=str(), ClosureComment=str())
assert package_request.Closer == user
assert package_request.ClosedTS == ts
@ -73,73 +63,54 @@ def test_package_request_closed():
def test_package_request_null_request_type_raises_exception():
with pytest.raises(IntegrityError):
with db.begin():
create(PackageRequest, User=user, PackageBase=pkgbase,
PackageBaseName=pkgbase.Name,
Comments=str(), ClosureComment=str())
rollback()
PackageRequest(User=user, PackageBase=pkgbase,
PackageBaseName=pkgbase.Name,
Comments=str(), ClosureComment=str())
def test_package_request_null_user_raises_exception():
request_type = query(RequestType, RequestType.Name == "merge").first()
with pytest.raises(IntegrityError):
with db.begin():
create(PackageRequest, RequestType=request_type,
PackageBase=pkgbase, PackageBaseName=pkgbase.Name,
Comments=str(), ClosureComment=str())
rollback()
PackageRequest(ReqTypeID=MERGE_ID,
PackageBase=pkgbase, PackageBaseName=pkgbase.Name,
Comments=str(), ClosureComment=str())
def test_package_request_null_package_base_raises_exception():
request_type = query(RequestType, RequestType.Name == "merge").first()
with pytest.raises(IntegrityError):
with db.begin():
create(PackageRequest, RequestType=request_type,
User=user, PackageBaseName=pkgbase.Name,
Comments=str(), ClosureComment=str())
rollback()
PackageRequest(ReqTypeID=MERGE_ID,
User=user, PackageBaseName=pkgbase.Name,
Comments=str(), ClosureComment=str())
def test_package_request_null_package_base_name_raises_exception():
request_type = query(RequestType, RequestType.Name == "merge").first()
with pytest.raises(IntegrityError):
with db.begin():
create(PackageRequest, RequestType=request_type,
User=user, PackageBase=pkgbase,
Comments=str(), ClosureComment=str())
rollback()
PackageRequest(ReqTypeID=MERGE_ID,
User=user, PackageBase=pkgbase,
Comments=str(), ClosureComment=str())
def test_package_request_null_comments_raises_exception():
request_type = query(RequestType, RequestType.Name == "merge").first()
with pytest.raises(IntegrityError):
with db.begin():
create(PackageRequest, RequestType=request_type, User=user,
PackageBase=pkgbase, PackageBaseName=pkgbase.Name,
ClosureComment=str())
rollback()
PackageRequest(ReqTypeID=MERGE_ID, User=user,
PackageBase=pkgbase, PackageBaseName=pkgbase.Name,
ClosureComment=str())
def test_package_request_null_closure_comment_raises_exception():
request_type = query(RequestType, RequestType.Name == "merge").first()
with pytest.raises(IntegrityError):
with db.begin():
create(PackageRequest, RequestType=request_type, User=user,
PackageBase=pkgbase, PackageBaseName=pkgbase.Name,
Comments=str())
rollback()
PackageRequest(ReqTypeID=MERGE_ID, User=user,
PackageBase=pkgbase, PackageBaseName=pkgbase.Name,
Comments=str())
def test_package_request_status_display():
""" Test status_display() based on the Status column value. """
request_type = query(RequestType, RequestType.Name == "merge").first()
with db.begin():
pkgreq = create(PackageRequest, RequestType=request_type,
User=user, PackageBase=pkgbase,
PackageBaseName=pkgbase.Name,
Comments=str(), ClosureComment=str(),
Status=PENDING_ID)
pkgreq = db.create(PackageRequest, ReqTypeID=MERGE_ID,
User=user, PackageBase=pkgbase,
PackageBaseName=pkgbase.Name,
Comments=str(), ClosureComment=str(),
Status=PENDING_ID)
assert pkgreq.status_display() == PENDING
with db.begin():

2
test/test_package_source.py
View file

@ -14,7 +14,7 @@ user = pkgbase = package = None
@pytest.fixture(autouse=True)
def setup():
def setup(db_test):
global user, pkgbase, package
setup_test_db("PackageSources", "Packages", "PackageBases", "Users")

29
test/test_package_vote.py
View file

@ -4,30 +4,30 @@ import pytest
from sqlalchemy.exc import IntegrityError
from aurweb.db import create, rollback
from aurweb import db
from aurweb.models.package_base import PackageBase
from aurweb.models.package_vote import PackageVote
from aurweb.models.user import User
from aurweb.testing import setup_test_db
user = pkgbase = None
@pytest.fixture(autouse=True)
def setup():
def setup(db_test):
global user, pkgbase
setup_test_db("Users", "PackageBases", "PackageVotes")
user = create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword")
pkgbase = create(PackageBase, Name="test-package", Maintainer=user)
with db.begin():
user = db.create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword")
pkgbase = db.create(PackageBase, Name="test-package", Maintainer=user)
def test_package_vote_creation():
ts = int(datetime.utcnow().timestamp())
package_vote = create(PackageVote, User=user, PackageBase=pkgbase,
VoteTS=ts)
with db.begin():
package_vote = db.create(PackageVote, User=user,
PackageBase=pkgbase, VoteTS=ts)
assert bool(package_vote)
assert package_vote.User == user
assert package_vote.PackageBase == pkgbase
@ -36,17 +36,14 @@ def test_package_vote_creation():
def test_package_vote_null_user_raises_exception():
with pytest.raises(IntegrityError):
create(PackageVote, PackageBase=pkgbase, VoteTS=1)
rollback()
PackageVote(PackageBase=pkgbase, VoteTS=1)
def test_package_vote_null_pkgbase_raises_exception():
with pytest.raises(IntegrityError):
create(PackageVote, User=user, VoteTS=1)
rollback()
PackageVote(User=user, VoteTS=1)
def test_package_vote_null_votets_raises_exception():
with pytest.raises(IntegrityError):
create(PackageVote, User=user, PackageBase=pkgbase)
rollback()
PackageVote(User=user, PackageBase=pkgbase)

23
test/test_packages_routes.py
View file

@ -28,7 +28,6 @@ from aurweb.models.package_vote import PackageVote
from aurweb.models.relation_type import CONFLICTS_ID, PROVIDES_ID, REPLACES_ID, RelationType
from aurweb.models.request_type import DELETION_ID, MERGE_ID, RequestType
from aurweb.models.user import User
from aurweb.testing import setup_test_db
from aurweb.testing.html import get_errors, get_successes, parse_root
from aurweb.testing.requests import Request
@ -65,21 +64,8 @@ def create_package_rel(package: Package,
@pytest.fixture(autouse=True)
def setup():
setup_test_db(
User.__tablename__,
Package.__tablename__,
PackageBase.__tablename__,
PackageDependency.__tablename__,
PackageRelation.__tablename__,
PackageKeyword.__tablename__,
PackageVote.__tablename__,
PackageNotification.__tablename__,
PackageComaintainer.__tablename__,
PackageComment.__tablename__,
PackageRequest.__tablename__,
OfficialProvider.__tablename__
)
def setup(db_test):
return
@pytest.fixture
@ -91,12 +77,11 @@ def client() -> TestClient:
@pytest.fixture
def user() -> User:
""" Yield a user. """
account_type = db.query(AccountType, AccountType.ID == USER_ID).first()
with db.begin():
user = db.create(User, Username="test",
Email="test@example.org",
Passwd="testPassword",
AccountType=account_type)
AccountTypeID=USER_ID)
yield user
@ -1173,7 +1158,7 @@ def test_pkgbase_comments(client: TestClient, maintainer: User, user: User,
PackageNotification.UserID == maintainer.ID
).first()
with db.begin():
db.session.delete(db_notif)
db.delete(db_notif)
# Now, let's edit the comment we just created.
comment_id = int(headers[0].attrib["id"].split("-")[-1])

17
test/test_packages_util.py
View file

@ -6,7 +6,7 @@ from fastapi import HTTPException
from fastapi.testclient import TestClient
from aurweb import asgi, db
from aurweb.models.account_type import USER_ID, AccountType
from aurweb.models.account_type import USER_ID
from aurweb.models.official_provider import OFFICIAL_BASE, OfficialProvider
from aurweb.models.package import Package
from aurweb.models.package_base import PackageBase
@ -15,29 +15,20 @@ from aurweb.models.package_vote import PackageVote
from aurweb.models.user import User
from aurweb.packages import util
from aurweb.redis import kill_redis
from aurweb.testing import setup_test_db
@pytest.fixture(autouse=True)
def setup():
setup_test_db(
User.__tablename__,
Package.__tablename__,
PackageBase.__tablename__,
PackageVote.__tablename__,
PackageNotification.__tablename__,
OfficialProvider.__tablename__
)
def setup(db_test):
return
@pytest.fixture
def maintainer() -> User:
account_type = db.query(AccountType, AccountType.ID == USER_ID).first()
with db.begin():
maintainer = db.create(User, Username="test_maintainer",
Email="test_maintainer@examepl.org",
Passwd="testPassword",
AccountType=account_type)
AccountTypeID=USER_ID)
yield maintainer

7
test/test_popupdate.py
View file

@ -1,5 +1,12 @@
import pytest
from aurweb.scripts import popupdate
@pytest.fixture(autouse=True)
def setup(db_test):
return
def test_popupdate():
popupdate.main()

20
test/test_ratelimit.py
View file

@ -8,15 +8,14 @@ from aurweb import config, db, logging
from aurweb.models import ApiRateLimit
from aurweb.ratelimit import check_ratelimit
from aurweb.redis import redis_connection
from aurweb.testing import setup_test_db
from aurweb.testing.requests import Request
logger = logging.get_logger(__name__)
@pytest.fixture(autouse=True)
def setup():
setup_test_db(ApiRateLimit.__tablename__)
def setup(db_test):
return
@pytest.fixture
@ -31,27 +30,36 @@ def pipeline():
yield pipeline
config_getint = config.getint
def mock_config_getint(section: str, key: str):
if key == "request_limit":
return 4
elif key == "window_length":
return 100
return config.getint(section, key)
return config_getint(section, key)
config_getboolean = config.getboolean
def mock_config_getboolean(return_value: int = 0):
def fn(section: str, key: str):
if section == "ratelimit" and key == "cache":
return return_value
return config.getboolean(section, key)
return config_getboolean(section, key)
return fn
config_get = config.get
def mock_config_get(return_value: str = "none"):
def fn(section: str, key: str):
if section == "options" and key == "cache":
return return_value
return config.get(section, key)
return config_get(section, key)
return fn

5
test/test_relation_type.py
View file

@ -2,12 +2,11 @@ import pytest
from aurweb import db
from aurweb.models.relation_type import RelationType
from aurweb.testing import setup_test_db
@pytest.fixture(autouse=True)
def setup():
setup_test_db()
def setup(db_test):
return
def test_relation_type_creation():

5
test/test_request_type.py
View file

@ -2,12 +2,11 @@ import pytest
from aurweb import db
from aurweb.models.request_type import DELETION_ID, MERGE_ID, ORPHAN_ID, RequestType
from aurweb.testing import setup_test_db
@pytest.fixture(autouse=True)
def setup():
setup_test_db()
def setup(db_test):
return
def test_request_type_creation():

12
test/test_routes.py
View file

@ -10,27 +10,21 @@ from fastapi.testclient import TestClient
from aurweb import db
from aurweb.asgi import app
from aurweb.models.account_type import AccountType
from aurweb.models.account_type import USER_ID
from aurweb.models.user import User
from aurweb.testing import setup_test_db
from aurweb.testing.requests import Request
user = client = None
@pytest.fixture(autouse=True)
def setup():
def setup(db_test):
global user, client
setup_test_db("Users", "Sessions")
account_type = db.query(AccountType,
AccountType.AccountType == "User").first()
with db.begin():
user = db.create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword",
AccountType=account_type)
AccountTypeID=USER_ID)
client = TestClient(app)

8
test/test_rpc.py
View file

@ -26,7 +26,6 @@ from aurweb.models.package_vote import PackageVote
from aurweb.models.relation_type import RelationType
from aurweb.models.user import User
from aurweb.redis import redis_connection
from aurweb.testing import setup_test_db
def make_request(path, headers: Dict[str, str] = {}):
@ -35,11 +34,8 @@ def make_request(path, headers: Dict[str, str] = {}):
@pytest.fixture(autouse=True)
def setup():
# Set up tables.
setup_test_db("Users", "PackageBases", "Packages", "Licenses",
"PackageDepends", "PackageRelations", "PackageLicenses",
"PackageKeywords", "PackageVotes", "ApiRateLimit")
def setup(db_test):
# TODO: Rework this into organized fixtures.
# Create test package details.
with begin():

8
test/test_rss.py
View file

@ -12,17 +12,13 @@ from aurweb.models.account_type import AccountType
from aurweb.models.package import Package
from aurweb.models.package_base import PackageBase
from aurweb.models.user import User
from aurweb.testing import setup_test_db
logger = logging.get_logger(__name__)
@pytest.fixture(autouse=True)
def setup():
setup_test_db(
Package.__tablename__,
PackageBase.__tablename__,
User.__tablename__)
def setup(db_test):
return
@pytest.fixture

5
test/test_session.py
View file

@ -8,17 +8,14 @@ from aurweb import db
from aurweb.models.account_type import AccountType
from aurweb.models.session import Session, generate_unique_sid
from aurweb.models.user import User
from aurweb.testing import setup_test_db
account_type = user = session = None
@pytest.fixture(autouse=True)
def setup():
def setup(db_test):
global account_type, user, session
setup_test_db("Users", "Sessions")
account_type = db.query(AccountType,
AccountType.AccountType == "User").first()
with db.begin():

14
test/test_ssh_pub_key.py
View file

@ -1,10 +1,9 @@
import pytest
from aurweb import db
from aurweb.models.account_type import AccountType
from aurweb.models.account_type import USER_ID
from aurweb.models.ssh_pub_key import SSHPubKey, get_fingerprint
from aurweb.models.user import User
from aurweb.testing import setup_test_db
TEST_SSH_PUBKEY = """
ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABgQCycoCi5yGCvSclH2wmNBUuwsYEzRZZBJaQquRc4ysl+Tg+/jiDkR3Zn9fIznC4KnFoyrIHzkKuePZ3bNDYwkZxkJKoWBCh4hXKDXSm87FMN0+VDC+1QxF/z0XaAGr/P6f4XukabyddypBdnHcZiplbw+YOSqcAE2TCqOlSXwNMOcF9U89UsR/Q9i9I52hlvU0q8+fZVGhou1KCowFSnHYtrr5KYJ04CXkJ13DkVf3+pjQWyrByvBcf1hGEaczlgfobrrv/y96jDhgfXucxliNKLdufDPPkii3LhhsNcDmmI1VZ3v0irKvd9WZuauqloobY84zEFcDTyjn0hxGjVeYFejm4fBnvjga0yZXORuWksdNfXWLDxFk6MDDd1jF0ExRbP+OxDuU4IVyIuDL7S3cnbf2YjGhkms/8voYT2OBE7FwNlfv98Kr0NUp51zpf55Arxn9j0Rz9xTA7FiODQgCn6iQ0SDtzUNL0IKTCw26xJY5gzMxbfpvzPQGeulx/ioM= kevr@volcano
@ -14,21 +13,16 @@ user = ssh_pub_key = None
@pytest.fixture(autouse=True)
def setup():
def setup(db_test):
global user, ssh_pub_key
setup_test_db("Users", "SSHPubKeys")
account_type = db.query(AccountType,
AccountType.AccountType == "User").first()
with db.begin():
user = db.create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword",
AccountType=account_type)
AccountTypeID=USER_ID)
with db.begin():
ssh_pub_key = db.create(SSHPubKey,
UserID=user.ID,
ssh_pub_key = db.create(SSHPubKey, UserID=user.ID,
Fingerprint="testFingerprint",
PubKey="testPubKey")

18
test/test_term.py
View file

@ -4,17 +4,11 @@ from sqlalchemy.exc import IntegrityError
from aurweb import db
from aurweb.models.term import Term
from aurweb.testing import setup_test_db
@pytest.fixture(autouse=True)
def setup():
setup_test_db("Terms")
yield None
# Wipe em out just in case records are leftover.
setup_test_db("Terms")
def setup(db_test):
return
def test_term_creation():
@ -29,13 +23,9 @@ def test_term_creation():
def test_term_null_description_raises_exception():
with pytest.raises(IntegrityError):
with db.begin():
db.create(Term, URL="https://fake_url.io")
db.rollback()
Term(URL="https://fake_url.io")
def test_term_null_url_raises_exception():
with pytest.raises(IntegrityError):
with db.begin():
db.create(Term, Description="Term description")
db.rollback()
Term(Description="Term description")

5
test/test_trusted_user_routes.py
View file

@ -14,7 +14,6 @@ from aurweb.models.account_type import AccountType
from aurweb.models.tu_vote import TUVote
from aurweb.models.tu_voteinfo import TUVoteInfo
from aurweb.models.user import User
from aurweb.testing import setup_test_db
from aurweb.testing.requests import Request
DATETIME_REGEX = r'^[0-9]{4}-[0-9]{2}-[0-9]{2}$'
@ -76,8 +75,8 @@ def assert_past_vote_html(row, expected):
@pytest.fixture(autouse=True)
def setup():
setup_test_db("TU_Votes", "TU_VoteInfo", "Users")
def setup(db_test):
return
@pytest.fixture

43
test/test_tu_vote.py
View file

@ -4,53 +4,48 @@ import pytest
from sqlalchemy.exc import IntegrityError
from aurweb.db import create, query, rollback
from aurweb.models.account_type import AccountType
from aurweb import db
from aurweb.models.account_type import TRUSTED_USER_ID
from aurweb.models.tu_vote import TUVote
from aurweb.models.tu_voteinfo import TUVoteInfo
from aurweb.models.user import User
from aurweb.testing import setup_test_db
user = tu_voteinfo = None
@pytest.fixture(autouse=True)
def setup():
def setup(db_test):
global user, tu_voteinfo
setup_test_db("Users", "TU_VoteInfo", "TU_Votes")
tu_type = query(AccountType,
AccountType.AccountType == "Trusted User").first()
user = create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword",
AccountType=tu_type)
ts = int(datetime.utcnow().timestamp())
tu_voteinfo = create(TUVoteInfo,
Agenda="Blah blah.",
User=user.Username,
Submitted=ts, End=ts + 5,
Quorum=0.5,
Submitter=user)
with db.begin():
user = db.create(User, Username="test", Email="test@example.org",
RealName="Test User", Passwd="testPassword",
AccountTypeID=TRUSTED_USER_ID)
tu_voteinfo = db.create(TUVoteInfo,
Agenda="Blah blah.",
User=user.Username,
Submitted=ts, End=ts + 5,
Quorum=0.5,
Submitter=user)
def test_tu_vote_creation():
tu_vote = create(TUVote, User=user, VoteInfo=tu_voteinfo)
with db.begin():
tu_vote = db.create(TUVote, User=user, VoteInfo=tu_voteinfo)
assert tu_vote.VoteInfo == tu_voteinfo
assert tu_vote.User == user
assert tu_vote in user.tu_votes
assert tu_vote in tu_voteinfo.tu_votes
def test_tu_vote_null_user_raises_exception():
with pytest.raises(IntegrityError):
create(TUVote, VoteInfo=tu_voteinfo)
rollback()
TUVote(VoteInfo=tu_voteinfo)
def test_tu_vote_null_voteinfo_raises_exception():
with pytest.raises(IntegrityError):
create(TUVote, User=user)
rollback()
TUVote(User=user)

5
test/test_tu_voteinfo.py
View file

@ -9,17 +9,14 @@ from aurweb.db import create, query, rollback
from aurweb.models.account_type import AccountType
from aurweb.models.tu_voteinfo import TUVoteInfo
from aurweb.models.user import User
from aurweb.testing import setup_test_db
user = None
@pytest.fixture(autouse=True)
def setup():
def setup(db_test):
global user
setup_test_db("Users", "PackageBases", "TU_VoteInfo")
tu_type = query(AccountType,
AccountType.AccountType == "Trusted User").first()
with db.begin():

14
test/test_user.py
View file

@ -19,27 +19,15 @@ from aurweb.models.package_vote import PackageVote
from aurweb.models.session import Session
from aurweb.models.ssh_pub_key import SSHPubKey
from aurweb.models.user import User
from aurweb.testing import setup_test_db
from aurweb.testing.requests import Request
account_type = user = None
@pytest.fixture(autouse=True)
def setup():
def setup(db_test):
global account_type, user
setup_test_db(
User.__tablename__,
Session.__tablename__,
Ban.__tablename__,
SSHPubKey.__tablename__,
Package.__tablename__,
PackageBase.__tablename__,
PackageVote.__tablename__,
PackageNotification.__tablename__
)
account_type = db.query(AccountType,
AccountType.AccountType == "User").first()