|
|
|
@ -52,6 +52,20 @@ def example_users():
@@ -52,6 +52,20 @@ def example_users():
|
|
|
|
|
] |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@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 |
|
|
|
@ -76,6 +90,17 @@ def test_sql_repo_get_order(session, example_orders):
@@ -76,6 +90,17 @@ def test_sql_repo_get_order(session, example_orders):
|
|
|
|
|
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 |
|
|
|
|
|
|
|
|
@ -112,6 +137,17 @@ def test_sql_repo_get_user(session, example_users):
@@ -112,6 +137,17 @@ def test_sql_repo_get_user(session, example_users):
|
|
|
|
|
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 |
|
|
|
|
|
|
|
|
@ -123,6 +159,17 @@ def test_sql_repo_get_user_by_username(session, example_users):
@@ -123,6 +159,17 @@ def test_sql_repo_get_user_by_username(session, example_users):
|
|
|
|
|
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 |
|
|
|
|
|
|
|
|
@ -134,6 +181,17 @@ def test_sql_repo_get_user_by_email(session, example_users):
@@ -134,6 +181,17 @@ def test_sql_repo_get_user_by_email(session, example_users):
|
|
|
|
|
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 |
|
|
|
|
|
|
|
|
@ -157,3 +215,70 @@ def test_sql_search_vendor(session, example_users):
@@ -157,3 +215,70 @@ def test_sql_search_vendor(session, example_users):
|
|
|
|
|
|
|
|
|
|
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]] |
|
|
|
|