diff options
Diffstat (limited to 'tests/managers/thl/test_user_manager')
5 files changed, 159 insertions, 60 deletions
diff --git a/tests/managers/thl/test_user_manager/test_base.py b/tests/managers/thl/test_user_manager/test_base.py index 0d7ffef..5d12052 100644 --- a/tests/managers/thl/test_user_manager/test_base.py +++ b/tests/managers/thl/test_user_manager/test_base.py @@ -1,23 +1,37 @@ import logging -from datetime import datetime, timezone +from collections.abc import Callable +from datetime import UTC, datetime from random import randint +from typing import TYPE_CHECKING from uuid import uuid4 import pytest from generalresearch.managers.thl.user_manager import ( - UserCreateNotAllowedError, get_bp_user_create_limit_hourly, ) +from generalresearch.managers.thl.user_manager.exceptions import ( + UserCreateNotAllowedError, +) +from generalresearch.managers.thl.user_manager.mysql_user_manager import ( + MysqlUserManager, +) from generalresearch.managers.thl.user_manager.rate_limit import ( RateLimitItemPerHourConstantKey, + UserManagerLimiter, ) -from generalresearch.managers.thl.user_manager.user_manager import ( - UserManager, -) -from generalresearch.models.thl.product import Product, UserCreateConfig +from generalresearch.models.thl.product import UserCreateConfig from generalresearch.models.thl.user import User +if TYPE_CHECKING: + from generalresearch.managers.thl.product import ProductManager + from generalresearch.managers.thl.user_manager.user_manager import ( + UserManager, + ) + from generalresearch.managers.thl.userhealth import AuditLogManager + from generalresearch.models.thl.product import Product + from generalresearch.pg_helper import PostgresConfig + logger = logging.getLogger() @@ -83,10 +97,11 @@ class TestUserManager: class TestBlockUserManager: - def test_block_user(self, product, user_manager: UserManager): + def test_block_user(self, product: Product, user_manager: UserManager): product_user_id = f"user-{uuid4().hex[:10]}" # mysql_user_manager to skip user creation limit check + assert isinstance(user_manager.mysql_user_manager, MysqlUserManager) user: User = user_manager.mysql_user_manager.create_user( product_id=product.id, product_user_id=product_user_id ) @@ -109,16 +124,19 @@ class TestBlockUserManager: user = user_manager.get_user(user_id=user.user_id) assert user.blocked - def test_block_user_whitelist(self, product, user_manager, thl_web_rw): + def test_block_user_whitelist( + self, product: Product, user_manager: UserManager, thl_web_rw: PostgresConfig + ): product_user_id = f"user-{uuid4().hex[:10]}" # mysql_user_manager to skip user creation limit check + assert isinstance(user_manager.mysql_user_manager, MysqlUserManager) user: User = user_manager.mysql_user_manager.create_user( product_id=product.id, product_user_id=product_user_id ) assert not user.blocked - now = datetime.now(tz=timezone.utc) + now = datetime.now(tz=UTC) # Adds user to whitelist thl_web_rw.execute_write( """ @@ -135,8 +153,13 @@ class TestBlockUserManager: class TestCreateUserManager: - def test_create_user(self, product_manager, thl_web_rw, user_manager): - product: Product = product_manager.create_dummy( + def test_create_user( + self, + product_factory: Callable[..., Product], + thl_web_rw: PostgresConfig, + user_manager: UserManager, + ): + product: Product = product_factory( user_create_config=UserCreateConfig( min_hourly_create_limit=10, max_hourly_create_limit=69 ), @@ -144,6 +167,7 @@ class TestCreateUserManager: product_user_id = f"user-{uuid4().hex[:10]}" + assert isinstance(user_manager.mysql_user_manager, MysqlUserManager) user: User = user_manager.mysql_user_manager.create_user( product_id=product.id, product_user_id=product_user_id ) @@ -156,7 +180,7 @@ class TestCreateUserManager: # make sure thl_user row is created res_thl_user = thl_web_rw.execute_sql_query( - query=f""" + query=""" SELECT * FROM thl_user AS u WHERE u.id = %s @@ -172,8 +196,13 @@ class TestCreateUserManager: assert u2.user_id == user.user_id assert u2.uuid == user.uuid - def test_create_user_integrity_error(self, product_manager, user_manager, caplog): - product: Product = product_manager.create_dummy( + def test_create_user_integrity_error( + self, + user_manager: UserManager, + product_factory: Callable[..., Product], + caplog, + ): + product: Product = product_factory( product_id=uuid4().hex, team_id=uuid4().hex, name=f"Test Product ID #{uuid4().hex[:6]}", @@ -185,6 +214,7 @@ class TestCreateUserManager: product_user_id = f"user-{uuid4().hex[:10]}" rand_msg = f"log-{uuid4().hex}" + assert isinstance(user_manager.mysql_user_manager, MysqlUserManager) with caplog.at_level(logging.INFO): logger.info(rand_msg) user1 = user_manager.mysql_user_manager.create_user( @@ -213,9 +243,14 @@ class TestCreateUserManager: assert user1 == user2 - def test_raise_allow_user_create(self, product_manager, user_manager): + def test_raise_allow_user_create( + self, + product_manager: ProductManager, + user_manager: UserManager, + product_factory: Callable[..., Product], + ): rand_num = randint(25, 200) - product: Product = product_manager.create_dummy( + product: Product = product_factory( product_id=uuid4().hex, team_id=uuid4().hex, name=f"Test Product ID #{uuid4().hex[:6]}", @@ -247,10 +282,11 @@ class TestCreateUserManager: assert key == f"LIMITER/thl-grpc/allow_user_create/{instance.id}" # make sure we clear the key or subsequent tests will fail + assert isinstance(user_manager.user_manager_limiter, UserManagerLimiter) user_manager.user_manager_limiter.storage.clear(key=key) n = 0 - with pytest.raises(expected_exception=UserCreateNotAllowedError) as cm: + with pytest.raises(expected_exception=UserCreateNotAllowedError): for n, _ in enumerate(range(rl_value + 5)): user_manager.user_manager_limiter.raise_allow_user_create( product=product @@ -260,14 +296,16 @@ class TestCreateUserManager: class TestUserManagerMethods: - def test_audit_log(self, user_manager, user, audit_log_manager): + def test_audit_log( + self, user_manager: UserManager, user: User, audit_log_manager: AuditLogManager + ): from generalresearch.models.thl.userhealth import AuditLog res = audit_log_manager.filter_by_user_id(user_id=user.user_id) assert len(res) == 0 msg = uuid4().hex - user_manager.audit_log(user=user, level=30, event_type=msg) + user_manager.audit_log(audit_log_manager, user=user, level=30, event_type=msg) res = audit_log_manager.filter_by_user_id(user_id=user.user_id) assert len(res) == 1 diff --git a/tests/managers/thl/test_user_manager/test_mysql.py b/tests/managers/thl/test_user_manager/test_mysql.py index 0313bbf..ed7d458 100644 --- a/tests/managers/thl/test_user_manager/test_mysql.py +++ b/tests/managers/thl/test_user_manager/test_mysql.py @@ -1,25 +1,28 @@ -from test_utils.models.conftest import user, user_manager +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from generalresearch.managers.thl.user_manager.mysql_user_manager import ( + MysqlUserManager, + ) + from generalresearch.models.thl.user import User class TestUserManagerMysqlNew: - def test_get_notset(self, user_manager): - assert ( - user_manager.mysql_user_manager.get_user_from_mysql(user_id=-3105) is None - ) + def test_get_notset(self, mysql_user_manager: MysqlUserManager): + assert mysql_user_manager.get_user_from_mysql(user_id=-3105) is None - def test_get_user_id(self, user, user_manager): - assert ( - user_manager.mysql_user_manager.get_user_from_mysql(user_id=user.user_id) - == user - ) + def test_get_user_id(self, user: User, mysql_user_manager: MysqlUserManager): + assert mysql_user_manager.get_user_from_mysql(user_id=user.user_id) == user - def test_get_uuid(self, user, user_manager): - u = user_manager.mysql_user_manager.get_user_from_mysql(user_uuid=user.uuid) + def test_get_uuid(self, user: User, mysql_user_manager: MysqlUserManager): + u = mysql_user_manager.get_user_from_mysql(user_uuid=user.uuid) assert u == user - def test_get_ubp(self, user, user_manager): - u = user_manager.mysql_user_manager.get_user_from_mysql( + def test_get_ubp(self, user: User, mysql_user_manager: MysqlUserManager): + u = mysql_user_manager.get_user_from_mysql( product_id=user.product_id, product_user_id=user.product_user_id ) assert u == user diff --git a/tests/managers/thl/test_user_manager/test_redis.py b/tests/managers/thl/test_user_manager/test_redis.py index a69519e..f6b59c9 100644 --- a/tests/managers/thl/test_user_manager/test_redis.py +++ b/tests/managers/thl/test_user_manager/test_redis.py @@ -1,29 +1,41 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + import pytest from generalresearch.managers.base import Permission +from generalresearch.managers.thl.user_manager.redis_user_manager import ( + RedisUserManager, +) + +if TYPE_CHECKING: + from generalresearch.config import GRLBaseSettings + from generalresearch.models.thl.user import User + from generalresearch.pg_helper import PostgresConfig class TestUserManagerRedis: - def test_get_notset(self, user_manager, user): - user_manager.clear_user_inmemory_cache(user=user) - assert user_manager.redis_user_manager.get_user(user_id=user.user_id) is None + def test_get_notset(self, redis_user_manager: RedisUserManager, user: User): + redis_user_manager.clear_user(user=user) + assert redis_user_manager.get_user(user_id=user.user_id) is None - def test_get_user_id(self, user_manager, user): - user_manager.redis_user_manager.set_user(user=user) + def test_get_user_id(self, redis_user_manager: RedisUserManager, user: User): + redis_user_manager.set_user(user=user) - assert user_manager.redis_user_manager.get_user(user_id=user.user_id) == user + assert redis_user_manager.get_user(user_id=user.user_id) == user - def test_get_uuid(self, user_manager, user): - user_manager.redis_user_manager.set_user(user=user) + def test_get_uuid(self, redis_user_manager: RedisUserManager, user: User): + redis_user_manager.set_user(user=user) - assert user_manager.redis_user_manager.get_user(user_uuid=user.uuid) == user + assert redis_user_manager.get_user(user_uuid=user.uuid) == user - def test_get_ubp(self, user_manager, user): - user_manager.redis_user_manager.set_user(user=user) + def test_get_ubp(self, redis_user_manager: RedisUserManager, user: User): + redis_user_manager.set_user(user=user) assert ( - user_manager.redis_user_manager.get_user( + redis_user_manager.get_user( product_id=user.product_id, product_user_id=user.product_user_id ) == user @@ -34,7 +46,13 @@ class TestUserManagerRedis: # I mean, the sets are implicitly tested by the get tests above. no point pass - def test_get_with_cache_prefix(self, settings, user, thl_web_rw, thl_web_rr): + def test_get_with_cache_prefix( + self, + settings: GRLBaseSettings, + user: User, + thl_web_rw: PostgresConfig, + thl_web_rr: PostgresConfig, + ): """ Confirm the prefix functionality is working; we do this so it is easier to migrate between any potentially breaking versions @@ -69,9 +87,11 @@ class TestUserManagerRedis: product_id=user.product_id, product_user_id=user.product_user_id ) + assert isinstance(um1.redis_user_manager, RedisUserManager) res1 = um1.redis_user_manager.client.get(f"user-lookup:user_id:{user.user_id}") assert res1 is not None + assert isinstance(um2.redis_user_manager, RedisUserManager) res2 = um2.redis_user_manager.client.get( f"user-lookup-v2:user_id:{user.user_id}" ) diff --git a/tests/managers/thl/test_user_manager/test_user_fetch.py b/tests/managers/thl/test_user_manager/test_user_fetch.py index a4b3d57..9a279ed 100644 --- a/tests/managers/thl/test_user_manager/test_user_fetch.py +++ b/tests/managers/thl/test_user_manager/test_user_fetch.py @@ -1,14 +1,25 @@ +from __future__ import annotations + +from collections.abc import Callable +from typing import TYPE_CHECKING from uuid import uuid4 import pytest -from generalresearch.models.thl.user import User -from test_utils.models.conftest import product, user_manager, user_factory +if TYPE_CHECKING: + from generalresearch.managers.thl.user_manager.user_manager import UserManager + from generalresearch.models.thl.product import Product + from generalresearch.models.thl.user import User class TestUserManagerFetch: - def test_fetch(self, user_factory, product, user_manager): + def test_fetch( + self, + user_factory: Callable[..., User], + product: Product, + user_manager: UserManager, + ): user1: User = user_factory(product=product) user2: User = user_factory(product=product) res = user_manager.fetch_by_bpuids( @@ -30,7 +41,7 @@ class TestUserManagerFetch: res = user_manager.fetch(user_uuids=[uuid4().hex]) assert len(res) == 0 - def test_fetch_invalid(self, user_manager): + def test_fetch_invalid(self, user_manager: UserManager): with pytest.raises(AssertionError) as e: user_manager.fetch(user_uuids=[], user_ids=None) assert "Must pass ONE of user_ids, user_uuids" in str(e.value) diff --git a/tests/managers/thl/test_user_manager/test_user_metadata.py b/tests/managers/thl/test_user_manager/test_user_metadata.py index 91dc16a..eb6a272 100644 --- a/tests/managers/thl/test_user_manager/test_user_metadata.py +++ b/tests/managers/thl/test_user_manager/test_user_metadata.py @@ -1,20 +1,38 @@ +from __future__ import annotations + +from collections.abc import Callable +from typing import TYPE_CHECKING from uuid import uuid4 import pytest from generalresearch.models.thl.user_profile import UserMetadata -from test_utils.models.conftest import user, user_manager, user_factory + +if TYPE_CHECKING: + from generalresearch.managers.thl.user_manager.user_metadata_manager import ( + UserMetadataManager, + ) + from generalresearch.models.thl.product import Product + from generalresearch.models.thl.user import User class TestUserMetadataManager: - def test_get_notset(self, user, user_manager, user_metadata_manager): + def test_get_notset( + self, + user: User, + user_metadata_manager: UserMetadataManager, + ): # The row in the db won't exist. It just returns the default obj with everything None (except for the user_id) um1 = user_metadata_manager.get(user_id=user.user_id) assert um1 == UserMetadata(user_id=user.user_id) - def test_create(self, user_factory, product, user_metadata_manager): - from generalresearch.models.thl.user import User + def test_create( + self, + user_factory: Callable[..., User], + product: Product, + user_metadata_manager: UserMetadataManager, + ): u1: User = user_factory(product=product) @@ -27,8 +45,12 @@ class TestUserMetadataManager: um2 = user_metadata_manager.get(email_address=email_address) assert um == um2 - def test_create_no_email(self, product, user_factory, user_metadata_manager): - from generalresearch.models.thl.user import User + def test_create_no_email( + self, + product: Product, + user_factory: Callable[..., User], + user_metadata_manager: UserMetadataManager, + ): u1: User = user_factory(product=product) um = UserMetadata(user_id=u1.user_id) @@ -38,8 +60,12 @@ class TestUserMetadataManager: um2 = user_metadata_manager.get(user_id=u1.user_id) assert um == um2 - def test_update(self, product, user_factory, user_metadata_manager): - from generalresearch.models.thl.user import User + def test_update( + self, + product: Product, + user_factory: Callable[..., User], + user_metadata_manager: UserMetadataManager, + ): u: User = user_factory(product=product) @@ -58,8 +84,9 @@ class TestUserMetadataManager: email_address=email_address.replace("example1", "example2"), ) - def test_filter(self, user_factory, product, user_metadata_manager): - from generalresearch.models.thl.user import User + def test_filter( + self, user_factory: Callable[..., User], product: Product, user_metadata_manager + ): user1: User = user_factory(product=product) user2: User = user_factory(product=product) |