You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
330 lines
8.7 KiB
330 lines
8.7 KiB
import pytest |
|
|
|
|
|
@pytest.fixture() |
|
def example_orders(): |
|
from datetime import datetime |
|
from ordr3.models import OrderItem, OrderCategory, OrderStatus |
|
|
|
return [ |
|
OrderItem( |
|
1, |
|
"Ethanol", |
|
"1-23", |
|
"VWR", |
|
OrderCategory.SOLVENT, |
|
"5 l", |
|
20.0, |
|
2, |
|
"EUR", |
|
"DFG", |
|
"urgent", |
|
datetime(2020, 2, 3, 15, 14, 13), |
|
"me", |
|
OrderStatus.OPEN, |
|
), |
|
OrderItem( |
|
2, |
|
"Gloves", |
|
"12-3", |
|
"Carl Roth", |
|
OrderCategory.DISPOSABLE, |
|
"100 St.", |
|
40.0, |
|
5, |
|
"USD", |
|
"BMBF", |
|
"no comment", |
|
datetime(2020, 2, 4, 15, 14, 13), |
|
"you", |
|
OrderStatus.APPROVAL, |
|
), |
|
] |
|
|
|
|
|
@pytest.fixture() |
|
def example_users(): |
|
from ordr3.models import User, UserRole |
|
|
|
return [ |
|
User(1, "You", "Jim", "Smith", "jim.smith", "abcd", UserRole.ADMIN,), |
|
User(2, "Me", "Jane", "Doe", "jane.doe", "1234", UserRole.USER), |
|
] |
|
|
|
|
|
@pytest.fixture() |
|
def example_tokens(): |
|
from ordr3.models import PasswordResetToken |
|
from datetime import datetime, timedelta |
|
|
|
valid = datetime.utcnow() + timedelta(days=2) |
|
invalid = datetime.utcnow() - timedelta(days=2) |
|
|
|
return [ |
|
PasswordResetToken("valid_token", 1, valid), |
|
PasswordResetToken("invalid_token", 2, invalid), |
|
] |
|
|
|
|
|
def test_sql_repo_add_order(session, example_orders): |
|
from ordr3.repo import SqlAlchemyRepository |
|
from ordr3.models import OrderItem |
|
|
|
repo = SqlAlchemyRepository(session) |
|
repo.add_order(example_orders[0]) |
|
session.flush() |
|
|
|
order = session.query(OrderItem).first() |
|
|
|
assert order == example_orders[0] |
|
|
|
|
|
def test_sql_repo_delete_order(session, example_orders): |
|
from ordr3.repo import SqlAlchemyRepository |
|
from ordr3.services import create_log_entry |
|
from ordr3.models import LogEntry, OrderStatus, User |
|
|
|
repo = SqlAlchemyRepository(session) |
|
repo.add_order(example_orders[0]) |
|
repo.add_order(example_orders[1]) |
|
user = User(*list("ABCDEFG")) |
|
create_log_entry(example_orders[0], OrderStatus.APPROVAL, user) |
|
create_log_entry(example_orders[1], OrderStatus.APPROVAL, user) |
|
session.flush() |
|
|
|
repo.delete_order(example_orders[0]) |
|
session.flush() |
|
|
|
assert repo.list_orders() == [example_orders[1]] |
|
assert session.query(LogEntry).all() == example_orders[1].log |
|
|
|
|
|
def test_sql_repo_get_order(session, example_orders): |
|
from ordr3.repo import SqlAlchemyRepository |
|
|
|
repo = SqlAlchemyRepository(session) |
|
repo.add_order(example_orders[0]) |
|
repo.add_order(example_orders[1]) |
|
session.flush() |
|
|
|
assert example_orders[1] == repo.get_order(2) |
|
|
|
|
|
def test_sql_repo_get_order_raises_exception(session, example_orders): |
|
from ordr3.repo import SqlAlchemyRepository, RepoItemNotFound |
|
|
|
repo = SqlAlchemyRepository(session) |
|
repo.add_order(example_orders[0]) |
|
session.flush() |
|
|
|
with pytest.raises(RepoItemNotFound): |
|
repo.get_order(2) |
|
|
|
|
|
def test_sql_repo_list_orders(session, example_orders): |
|
from ordr3.repo import SqlAlchemyRepository |
|
|
|
earlier, later = example_orders |
|
repo = SqlAlchemyRepository(session) |
|
repo.add_order(earlier) |
|
repo.add_order(later) |
|
session.flush() |
|
|
|
assert repo.list_orders() == [later, earlier] |
|
|
|
|
|
def test_sql_repo_add_user(session, example_users): |
|
from ordr3.repo import SqlAlchemyRepository |
|
from ordr3.models import User |
|
|
|
repo = SqlAlchemyRepository(session) |
|
repo.add_user(example_users[0]) |
|
session.flush() |
|
|
|
user = session.query(User).first() |
|
|
|
assert user == example_users[0] |
|
|
|
|
|
def test_sql_repo_delte_user(session, example_users): |
|
from ordr3.repo import SqlAlchemyRepository |
|
|
|
repo = SqlAlchemyRepository(session) |
|
repo.add_user(example_users[0]) |
|
repo.add_user(example_users[1]) |
|
session.flush() |
|
|
|
repo.delete_user(example_users[0]) |
|
|
|
assert repo.list_users() == [example_users[1]] |
|
|
|
|
|
def test_sql_repo_get_user(session, example_users): |
|
from ordr3.repo import SqlAlchemyRepository |
|
|
|
repo = SqlAlchemyRepository(session) |
|
repo.add_user(example_users[0]) |
|
repo.add_user(example_users[1]) |
|
session.flush() |
|
|
|
assert example_users[1] == repo.get_user(2) |
|
|
|
|
|
def test_sql_repo_get_user_raises_exception(session, example_users): |
|
from ordr3.repo import SqlAlchemyRepository, RepoItemNotFound |
|
|
|
repo = SqlAlchemyRepository(session) |
|
repo.add_user(example_users[0]) |
|
session.flush() |
|
|
|
with pytest.raises(RepoItemNotFound): |
|
repo.get_user(2) |
|
|
|
|
|
def test_sql_repo_get_user_by_username(session, example_users): |
|
from ordr3.repo import SqlAlchemyRepository |
|
|
|
repo = SqlAlchemyRepository(session) |
|
repo.add_user(example_users[0]) |
|
repo.add_user(example_users[1]) |
|
session.flush() |
|
|
|
assert example_users[1] == repo.get_user_by_username("Me") |
|
|
|
|
|
def test_sql_repo_get_user_by_username_exception(session, example_users): |
|
from ordr3.repo import SqlAlchemyRepository, RepoItemNotFound |
|
|
|
repo = SqlAlchemyRepository(session) |
|
repo.add_user(example_users[0]) |
|
session.flush() |
|
|
|
with pytest.raises(RepoItemNotFound): |
|
repo.get_user_by_username("unknown user name") |
|
|
|
|
|
def test_sql_repo_get_user_by_email(session, example_users): |
|
from ordr3.repo import SqlAlchemyRepository |
|
|
|
repo = SqlAlchemyRepository(session) |
|
repo.add_user(example_users[0]) |
|
repo.add_user(example_users[1]) |
|
session.flush() |
|
|
|
assert example_users[1] == repo.get_user_by_email("jane.doe") |
|
|
|
|
|
def test_sql_repo_get_user_by_email_exception(session, example_users): |
|
from ordr3.repo import SqlAlchemyRepository, RepoItemNotFound |
|
|
|
repo = SqlAlchemyRepository(session) |
|
repo.add_user(example_users[0]) |
|
session.flush() |
|
|
|
with pytest.raises(RepoItemNotFound): |
|
repo.get_user_by_email("unknown email") |
|
|
|
|
|
def test_sql_repo_list_users(session, example_users): |
|
from ordr3.repo import SqlAlchemyRepository |
|
|
|
later, earlier = example_users |
|
repo = SqlAlchemyRepository(session) |
|
repo.add_user(later) |
|
repo.add_user(earlier) |
|
session.flush() |
|
|
|
assert repo.list_users() == [earlier, later] |
|
|
|
|
|
def test_sql_repo_count_new_users(session, example_users): |
|
from ordr3.repo import SqlAlchemyRepository |
|
from ordr3.models import UserRole |
|
|
|
repo = SqlAlchemyRepository(session) |
|
example_users[0].role = UserRole.NEW |
|
repo.add_user(example_users[0]) |
|
repo.add_user(example_users[1]) |
|
session.flush() |
|
|
|
assert repo.count_new_users() == 1 |
|
|
|
|
|
def test_sql_search_vendor(session, example_users): |
|
from ordr3.repo import SqlAlchemyRepository |
|
from ordr3.models import Vendor |
|
|
|
repo = SqlAlchemyRepository(session) |
|
rep = Vendor("sa", "Sigma Aldrich") |
|
session.add(rep) |
|
session.flush() |
|
|
|
assert repo.search_vendor("sa") == "Sigma Aldrich" |
|
assert repo.search_vendor("unknown") is None |
|
|
|
|
|
def test_sql_repo_add_reset_token(session, example_tokens): |
|
from ordr3.repo import SqlAlchemyRepository |
|
from ordr3.models import PasswordResetToken |
|
|
|
repo = SqlAlchemyRepository(session) |
|
repo.add_reset_token(example_tokens[0]) |
|
session.flush() |
|
|
|
token = session.query(PasswordResetToken).first() |
|
|
|
assert token == example_tokens[0] |
|
|
|
|
|
def test_sql_repo_delete_reset_token(session, example_tokens): |
|
from ordr3.repo import SqlAlchemyRepository |
|
from ordr3.models import PasswordResetToken |
|
|
|
repo = SqlAlchemyRepository(session) |
|
repo.add_reset_token(example_tokens[0]) |
|
repo.add_reset_token(example_tokens[1]) |
|
session.flush() |
|
|
|
repo.delete_reset_token(example_tokens[0]) |
|
|
|
tokens = session.query(PasswordResetToken).all() |
|
assert tokens == [example_tokens[1]] |
|
|
|
|
|
def test_sql_repo_get_reset_token(session, example_tokens): |
|
from ordr3.repo import SqlAlchemyRepository |
|
|
|
repo = SqlAlchemyRepository(session) |
|
repo.add_reset_token(example_tokens[0]) |
|
repo.add_reset_token(example_tokens[1]) |
|
session.flush() |
|
|
|
token = repo.get_reset_token("valid_token") |
|
|
|
assert token == example_tokens[0] |
|
|
|
|
|
def test_sql_repo_get_reset_token_raises_exception(session, example_tokens): |
|
from ordr3.repo import SqlAlchemyRepository, RepoItemNotFound |
|
|
|
repo = SqlAlchemyRepository(session) |
|
repo.add_reset_token(example_tokens[0]) |
|
session.flush() |
|
|
|
with pytest.raises(RepoItemNotFound): |
|
repo.get_reset_token("unknown token") |
|
|
|
|
|
def test_clear_stale_reset_tokens(session, example_tokens): |
|
from ordr3.repo import SqlAlchemyRepository |
|
from ordr3.models import PasswordResetToken |
|
|
|
repo = SqlAlchemyRepository(session) |
|
repo.add_reset_token(example_tokens[0]) |
|
repo.add_reset_token(example_tokens[1]) |
|
session.flush() |
|
|
|
repo.clear_stale_reset_tokens() |
|
|
|
tokens = session.query(PasswordResetToken).all() |
|
assert tokens == [example_tokens[0]]
|
|
|