diff options
Diffstat (limited to 'test_utils')
| -rw-r--r-- | test_utils/conftest.py | 205 | ||||
| -rw-r--r-- | test_utils/incite/conftest.py | 3 | ||||
| -rw-r--r-- | test_utils/managers/gr/__init__.py | 0 | ||||
| -rw-r--r-- | test_utils/managers/gr/conftest.py | 151 | ||||
| -rw-r--r-- | test_utils/managers/grliq/__init__.py | 0 | ||||
| -rw-r--r-- | test_utils/managers/grliq/conftest.py | 61 | ||||
| -rw-r--r-- | test_utils/managers/thl/__init__.py | 0 | ||||
| -rw-r--r-- | test_utils/managers/thl/conftest.py | 112 | ||||
| -rw-r--r-- | test_utils/models/conftest.py | 201 |
9 files changed, 528 insertions, 205 deletions
diff --git a/test_utils/conftest.py b/test_utils/conftest.py index 232c1fc..9c80065 100644 --- a/test_utils/conftest.py +++ b/test_utils/conftest.py @@ -1,22 +1,20 @@ import os import shutil -import sys from datetime import datetime, timedelta, timezone from os.path import join as pjoin from pathlib import Path from typing import TYPE_CHECKING, Callable, Generator from uuid import uuid4 -import django import pytest import redis from _pytest.config import Config -from django.conf import settings as django_settings -from django.core.management import call_command from dotenv import load_dotenv -from pydantic import MariaDBDsn, PostgresDsn +from pydantic import MariaDBDsn, PostgresDsn, TypeAdapter +from pydantic_core import MultiHostHost from redis import Redis +from generalresearch.models.custom_types import InternalHostname, PostgresDict from generalresearch.pg_helper import PostgresConfig from generalresearch.redis_helper import RedisConfig from generalresearch.sql_helper import SqlHelper @@ -28,7 +26,7 @@ if TYPE_CHECKING: @pytest.fixture(scope="session") -def env_file_path(pytestconfig: Config) -> str: +def env_file_path(pytestconfig: Config) -> Path: root_path = pytestconfig.rootpath env_file = ".env.test" @@ -40,18 +38,16 @@ def env_file_path(pytestconfig: Config) -> str: for env_path in candidates: if os.path.exists(env_path): load_dotenv(dotenv_path=env_path, override=True) - return os.path.normpath(env_path) + return Path(os.path.normpath(env_path)) raise AssertionError(f"No .env.test file found in: {', '.join(candidates)}") @pytest.fixture(scope="session") -def settings(env_file_path: str) -> "GRLBaseSettings": +def settings(env_file_path: Path) -> "GRLBaseSettings": from generalresearch.config import GRLBaseSettings - print(f"{env_file_path=}") - - s = GRLBaseSettings(_env_file=env_file_path) + s = GRLBaseSettings() if s.thl_mkpl_rr_db is not None: if s.spectrum_rw_db is None: @@ -71,14 +67,28 @@ def settings(env_file_path: str) -> "GRLBaseSettings": def postgres_instance(settings: "GRLBaseSettings") -> Generator[PostgresDsn]: """Create a ephemeral postgresql instance for us to use during pytest. - This is simplified, and only based off a single host. We don't want to - create multiple migrated tmp databases for each rw/rr/ro connection + This does not create any tables, or schema definitions within the instance. + What this does is simply: + + 1. Create a database on a known, consistent, staging or unittest + defined Postgres server. + + 2. Return the PostgresDsn of that table + + 3. On shutdown, go ahead and delete that database after the + tests have finished. """ - assert settings.thl_web_rw_db - # assert settings.thl_web_rw_db.host + msg = "Must define Postgres test settings" + assert settings.testing_postgres, msg + assert settings.testing_postgres_user, msg + assert settings.testing_postgres_pass, msg - dsn: PostgresDsn = settings.thl_web_rw_db + db_uri, db_user, db_pass = ( + settings.testing_postgres, + settings.testing_postgres_user, + settings.testing_postgres_pass, + ) # Connect to default DB to create the new one from psycopg import connect @@ -86,40 +96,74 @@ def postgres_instance(settings: "GRLBaseSettings") -> Generator[PostgresDsn]: now = datetime.now(timezone.utc) ts: str = now.strftime("%Y-%m-%d") - db_name = f"unittest-{ts}-{uuid4().hex[:6]}" - print("XXX", str(dsn)) - conn = connect(str(dsn)) + + db_path_connect = f"postgres://{db_user}:{db_pass}@{db_uri}" + db_path = f"{db_path_connect}/{db_name}" + + # The DATABASE does NOT yet exist on the Postgres SERVER, thus + # we first must connect only to the SERVER (eg: default postgres path used) + conn = connect(f"{db_path_connect}/postgres") conn.autocommit = True cur = conn.cursor() cur.execute(SQL("CREATE DATABASE {}").format(Identifier(db_name))) cur.close() conn.close() - host = dsn.hosts()[0] - db_url = ( - f"postgres://{host['username']}:{host['password']}@{host['host']}/{db_name}" - ) - - yield PostgresDsn(db_url) + yield PostgresDsn(db_path) # Teardown: drop the DB after the session - conn = connect(str(dsn)) + conn = connect(f"{db_path_connect}/postgres") conn.autocommit = True cur = conn.cursor() - # cur.execute(SQL("DROP DATABASE {}").format(Identifier(db_name))) + cur.execute(SQL("DROP DATABASE {} WITH (FORCE)").format(Identifier(db_name))) cur.close() conn.close() @pytest.fixture(scope="session") -def django_db_setup(settings: "GRLBaseSettings") -> Callable[..., None]: +def postgres_instance_dict( + postgres_instance: PostgresDsn, +) -> Generator[PostgresDict]: + host = postgres_instance.hosts()[0] + assert host is not None + + msg = "Must have full Postgres details" + assert host["host"], msg + assert host["username"], msg + assert host["password"], msg + + assert postgres_instance.path + + yield PostgresDict( + username=host["username"], + password=host["password"], + host=host["host"], + name=postgres_instance.path.lstrip("/"), + port=5432, + ) + + +@pytest.fixture(scope="session") +def postgres_instance_host( + postgres_instance_dict: PostgresDict, +) -> Generator[InternalHostname]: + adapter = TypeAdapter(InternalHostname) + value = adapter.validate_python(postgres_instance_dict["host"]) + yield value + + +@pytest.fixture(scope="session") +def django_db_factory( + postgres_instance: PostgresDsn, postgres_instance_dict: PostgresDict +) -> Callable[..., PostgresDsn]: - def _inner(): + import django + from django.apps import apps + from django.conf import settings as django_settings + from django.core.management import call_command - assert settings.thl_web_rw_db - dsn: PostgresDsn = settings.thl_web_rw_db - host = dsn.hosts()[0] + def _inner(django_project: str = "generalresearch.thl_django"): # 1. Bootstrapping Django settings if not django_settings.configured: @@ -127,93 +171,70 @@ def django_db_setup(settings: "GRLBaseSettings") -> Callable[..., None]: DATABASES={ "default": { "ENGINE": "django.db.backends.postgresql", - # PostgresDsn stores path as "/dbname" - "NAME": str(dsn.path).lstrip("/"), - "USER": host["username"], - "PASSWORD": host["password"], - "HOST": host["host"], - "PORT": "5432", + "NAME": postgres_instance_dict["name"], + "USER": postgres_instance_dict["username"], + "PASSWORD": postgres_instance_dict["password"], + "HOST": postgres_instance_dict["host"], + "PORT": postgres_instance_dict["port"], } }, INSTALLED_APPS=[ "django.contrib.postgres", "django.contrib.contenttypes", - "generalresearch.thl_django", + django_project, ], ) django.setup() - from django.apps import apps - - for model in apps.get_models(): - print(f"Discovered model: {model._meta.label}") + # for model in apps.get_models(): + # print(f"Discovered model: {model._meta.label}") # 2. Run migrations directly during fixture activation call_command("migrate") + # 3. Return the Dsn so the factory gives a way to connect + return postgres_instance + return _inner @pytest.fixture(scope="session") -def thl_web_rr( - settings: "GRLBaseSettings", postgres_instance: PostgresDsn, django_db_setup -) -> PostgresConfig: - dsn = settings.thl_web_rr_db - assert dsn - assert dsn.path - - if dsn.path not in ["/", "/postgres"]: - assert "/unittest-" in dsn.path - - db_path = postgres_instance.path - host = dsn.hosts()[0] - db_url = f"postgres://{host['username']}:{host['password']}@{host['host']}{db_path}" - - # Run Migrations now. - django_db_setup() +def thl_web_rr(django_db_factory: Callable[..., PostgresDsn]) -> PostgresConfig: return PostgresConfig( - dsn=PostgresDsn(db_url), + dsn=django_db_factory("generalresearch.thl_django"), connect_timeout=1, statement_timeout=5, ) @pytest.fixture(scope="session") -def thl_web_rw( - settings: "GRLBaseSettings", postgres_instance: PostgresDsn, django_db_setup -) -> PostgresConfig: - dsn = settings.thl_web_rw_db - assert dsn - assert dsn.path +def thl_web_rw(thl_web_rr: PostgresConfig) -> PostgresConfig: + return thl_web_rr - if dsn.path not in ["/", "/postgres"]: - assert "/unittest-" in dsn.path - db_path = postgres_instance.path - host = dsn.hosts()[0] - db_url = f"postgres://{host['username']}:{host['password']}@{host['host']}{db_path}" - - # Run Migrations now. - django_db_setup() +@pytest.fixture(scope="session") +def gr_db(django_db_factory: Callable[..., PostgresDsn]) -> PostgresConfig: return PostgresConfig( - dsn=PostgresDsn(db_url), + dsn=django_db_factory("gr_carer"), connect_timeout=1, statement_timeout=5, ) @pytest.fixture(scope="session") -def gr_db(settings: "GRLBaseSettings") -> PostgresConfig: - dsn = settings.gr_db - assert dsn - assert dsn.path +def grliq_db(postgres_instance: PostgresDsn) -> PostgresConfig: - if dsn.path not in ["/", "/postgres"]: - assert "/unittest-" in dsn.path + # test_words = {"localhost", "127.0.0.1", "unittest", "grliq-test"} + # assert any(w in str(postgres_config.dsn) for w in test_words), "check grliq postgres_config" + # assert "grliqdeceezpocymo" not in str(postgres_config.dsn), "check grliq postgres_config" - return PostgresConfig(dsn=settings.gr_db, connect_timeout=5, statement_timeout=2) + return PostgresConfig( + dsn=postgres_instance, + connect_timeout=1, + statement_timeout=5, + ) @pytest.fixture(scope="session") @@ -234,26 +255,6 @@ def spectrum_rw(settings: "GRLBaseSettings") -> SqlHelper: @pytest.fixture(scope="session") -def grliq_db(settings: "GRLBaseSettings") -> PostgresConfig: - dsn = settings.grliq_db - assert dsn - assert dsn.path - - if dsn.path not in ["/", "/postgres"]: - assert "/unittest-" in dsn.path - - # test_words = {"localhost", "127.0.0.1", "unittest", "grliq-test"} - # assert any(w in str(postgres_config.dsn) for w in test_words), "check grliq postgres_config" - # assert "grliqdeceezpocymo" not in str(postgres_config.dsn), "check grliq postgres_config" - - return PostgresConfig( - dsn=settings.grliq_db, - connect_timeout=2, - statement_timeout=2, - ) - - -@pytest.fixture(scope="session") def thl_redis(settings: "GRLBaseSettings") -> "Redis": # todo: this should get replaced with redisconfig (in most places) # I'm not sure where this would be? in the domain name? diff --git a/test_utils/incite/conftest.py b/test_utils/incite/conftest.py index 058093e..0e2f7bd 100644 --- a/test_utils/incite/conftest.py +++ b/test_utils/incite/conftest.py @@ -10,9 +10,6 @@ import pytest from _pytest.fixtures import SubRequest from faker import Faker -# from test_utils.managers.ledger.conftest import session_with_tx_factory -# from test_utils.models.conftest import session_factory - if TYPE_CHECKING: from generalresearch.config import GRLBaseSettings from generalresearch.incite.base import GRLDatasets diff --git a/test_utils/managers/gr/__init__.py b/test_utils/managers/gr/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/test_utils/managers/gr/__init__.py diff --git a/test_utils/managers/gr/conftest.py b/test_utils/managers/gr/conftest.py new file mode 100644 index 0000000..3e3a4ad --- /dev/null +++ b/test_utils/managers/gr/conftest.py @@ -0,0 +1,151 @@ +from __future__ import annotations + +from typing import Callable +from uuid import uuid4 + +import pytest +from pydantic import PositiveInt +from pydantic_extra_types.phone_numbers import PhoneNumber + +from generalresearch.managers.gr.authentication import GRUserManager +from generalresearch.managers.gr.business import ( + BusinessAddressManager, + BusinessBankAccountManager, + BusinessManager, +) +from generalresearch.managers.gr.team import TeamManager +from generalresearch.models.custom_types import UUIDStr +from generalresearch.models.gr.authentication import GRUser +from generalresearch.models.gr.business import ( + Business, + BusinessAddress, + BusinessBankAccount, + BusinessType, + TransferMethod, +) +from generalresearch.models.gr.team import Team + + +@pytest.fixture +def gr_user_factory(gr_um: GRUserManager) -> Callable[..., GRUser]: + + def _inner( + sub: str | None = None, + is_superuser: bool = False, + ) -> GRUser: + sub = sub or f"{uuid4().hex}-{uuid4().hex}" + + return gr_um.create( + sub=sub, + is_superuser=is_superuser, + ) + + return _inner + + +@pytest.fixture +def gr_business_bank_account_factory( + gr_bbam: BusinessBankAccountManager, +) -> Callable[..., BusinessBankAccount]: + + def _inner( + business_id: PositiveInt, + uuid: UUIDStr | None = None, + transfer_method: TransferMethod | None = None, + account_number: str | None = None, + routing_number: str | None = None, + iban: str | None = None, + swift: str | None = None, + ): + from generalresearch.models.gr.business import TransferMethod + + return gr_bbam.create( + business_id=business_id, + uuid=uuid or uuid4().hex, + transfer_method=transfer_method or TransferMethod.ACH, + account_number=account_number or uuid4().hex[:6], + routing_number=routing_number or uuid4().hex[:6], + iban=iban or uuid4().hex[:6], + swift=swift or uuid4().hex[:6], + ) + + return _inner + + +@pytest.fixture +def gr_business_address_factory( + gr_bam: BusinessAddressManager, +) -> Callable[..., BusinessAddress]: + + def _inner( + business_id: PositiveInt, + uuid: UUIDStr | None = None, + line_1: str | None = None, + line_2: str | None = None, + city: str | None = None, + state: str | None = None, + postal_code: str | None = None, + phone_number: PhoneNumber | None = None, + country: str | None = None, + ): + uuid = uuid or uuid4().hex + line_1 = line_1 or "abc" + line_2 = line_2 or "bczx" + city = city or "Downingtown" + state = state or "CA" + postal_code = postal_code or "94041" + phone_number = None + country = country or "US" + + return gr_bam.create( + business_id=business_id, + uuid=uuid, + line_1=line_1, + line_2=line_2, + city=city, + state=state, + postal_code=postal_code, + phone_number=phone_number, + country=country, + ) + + return _inner + + +@pytest.fixture +def gr_business_factory( + gr_bm: BusinessManager, +) -> Callable[..., Business]: + + def _inner( + uuid: UUIDStr | None = None, + name: str | None = None, + team: Team | None = None, + kind: BusinessType | None = None, + tax_number: str | None = None, + ) -> Business: + from random import randint + + uuid = uuid or uuid4().hex + name = name or "< Unknown >" + tax_number = tax_number or str(randint(1, 999_999_999)) + + return gr_bm.create( + uuid=uuid, name=name, team=team, kind=kind, tax_number=tax_number + ) + + return _inner + + +@pytest.fixture +def gr_team( + gr_tm: TeamManager, +) -> Callable[..., Team]: + + def _inner(uuid: UUIDStr | None = None, name: str | None = None) -> Team: + uuid = uuid or uuid4().hex + name = name or f"name-{uuid4().hex[:12]}" + + return gr_tm.create(uuid=uuid, name=name) + + return _inner diff --git a/test_utils/managers/grliq/__init__.py b/test_utils/managers/grliq/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/test_utils/managers/grliq/__init__.py diff --git a/test_utils/managers/grliq/conftest.py b/test_utils/managers/grliq/conftest.py new file mode 100644 index 0000000..525d8c8 --- /dev/null +++ b/test_utils/managers/grliq/conftest.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Callable +from uuid import uuid4 + +import pytest + +from generalresearch.grliq.managers import DUMMY_GRLIQ_DATA +from generalresearch.grliq.managers.forensic_data import GrlIqDataManager +from generalresearch.grliq.models.forensic_data import GrlIqData + + +@pytest.fixture +def grliq_data_factory(grliq_dm: GrlIqDataManager) -> Callable[..., GrlIqData]: + + def _inner( + is_attempt_allowed: bool = True, + product_id: str | None = None, + product_user_id: str | None = None, + uuid: str | None = None, + mid: str | None = None, + created_at: datetime | None = None, + ) -> GrlIqData: + """ + Creates a dummy record in the db with a GrlIqData (data), GrlIqCheckerResults (result_data), + and GrlIqForensicCategoryResult (category_results) + :param is_attempt_allowed: Whether the attempt is allowed. + :param product_id: product_id of user + :param product_user_id: product_user_id of user + :param uuid: uuid for the grliq data record + :param mid: the thl_session:uuid / mid for the attempt. + :return: + """ + import copy + + res: GrlIqData = copy.deepcopy(DUMMY_GRLIQ_DATA[int(is_attempt_allowed)]) + + product_id = product_id or uuid4().hex + product_user_id = product_user_id or uuid4().hex + uuid = uuid or uuid4().hex + mid = mid or uuid4().hex + created_at = created_at or datetime.now(tz=timezone.utc) + + res["data"].product_id = product_id + res["data"].product_user_id = product_user_id + res["data"].uuid = uuid + res["data"].mid = mid + res["data"].created_at = created_at + res["result_data"].uuid = uuid + res["category_result"].uuid = uuid + + return grliq_dm.create( + iq_data=res["data"], + result_data=res["result_data"], + category_result=res["category_result"], + fraud_score=res["category_result"].fraud_score, + is_attempt_allowed=res["category_result"].is_attempt_allowed(), + ) + + return _inner diff --git a/test_utils/managers/thl/__init__.py b/test_utils/managers/thl/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/test_utils/managers/thl/__init__.py diff --git a/test_utils/managers/thl/conftest.py b/test_utils/managers/thl/conftest.py new file mode 100644 index 0000000..76e4226 --- /dev/null +++ b/test_utils/managers/thl/conftest.py @@ -0,0 +1,112 @@ +from __future__ import annotations + +from decimal import Decimal +from random import randint +from typing import Callable + +import faker +from pydantic import PositiveInt + +from generalresearch.managers.thl.ipinfo import IPGeonameManager, IPInformationManager +from generalresearch.models.custom_types import IPvAnyAddressStr +from generalresearch.models.thl.ipinfo import IPGeoname, IPInformation, UserType + +fake = faker.Faker() + + +def ipgeoname_factory(ipgeoname_manager: IPGeonameManager) -> Callable[..., IPGeoname]: + + def _inner( + geoname_id: PositiveInt | None = None, + continent_code: str | None = None, + continent_name: str | None = None, + country_iso: str | None = None, + country_name: str | None = None, + subdivision_1_iso: str | None = None, + subdivision_1_name: str | None = None, + subdivision_2_iso: str | None = None, + subdivision_2_name: str | None = None, + city_name: str | None = None, + metro_code: int | None = None, + time_zone: str | None = None, + is_in_european_union: bool | None = None, + ) -> IPGeoname: + + return ipgeoname_manager.create( + geoname_id=geoname_id or randint(1, 999_999_999), + continent_code=continent_code or "na", + continent_name=continent_name or "North America", + country_iso=country_iso or "us", + country_name=country_name or "United States", + subdivision_1_iso=subdivision_1_iso or "fl", + subdivision_1_name=subdivision_1_name or "Florida", + subdivision_2_iso=subdivision_2_iso, + subdivision_2_name=subdivision_2_name, + city_name=city_name, + metro_code=metro_code, + time_zone=time_zone, + is_in_european_union=is_in_european_union, + ) + + return _inner + + +def ipinformation_factory( + ipinformation_manager: IPInformationManager, +) -> Callable[..., IPInformation]: + + def _inner( + ip: IPvAnyAddressStr | None = None, + geoname_id: PositiveInt | None = None, + country_iso: str | None = None, + registered_country_iso: str | None = None, + is_anonymous: bool | None = None, + is_anonymous_vpn: bool | None = None, + is_hosting_provider: bool | None = None, + is_public_proxy: bool | None = None, + is_tor_exit_node: bool | None = None, + is_residential_proxy: bool | None = None, + autonomous_system_number: PositiveInt | None = None, + autonomous_system_organization: str | None = None, + domain: str | None = None, + isp: str | None = None, + mobile_country_code: str | None = None, + mobile_network_code: str | None = None, + network: str | None = None, + organization: str | None = None, + static_ip_score: float | None = None, + user_type: UserType | None = None, + postal_code: str | None = None, + latitude: Decimal | None = None, + longitude: Decimal | None = None, + accuracy_radius: int | None = None, + ) -> IPInformation: + + return ipinformation_manager.create( + ip=ip or fake.ipv4_public(), + geoname_id=geoname_id, + country_iso=country_iso or fake.country_code(), + registered_country_iso=registered_country_iso, + is_anonymous=is_anonymous, + is_anonymous_vpn=is_anonymous_vpn, + is_hosting_provider=is_hosting_provider, + is_public_proxy=is_public_proxy, + is_tor_exit_node=is_tor_exit_node, + is_residential_proxy=is_residential_proxy, + autonomous_system_number=autonomous_system_number, + autonomous_system_organization=autonomous_system_organization, + domain=domain, + isp=isp, + mobile_country_code=mobile_country_code, + mobile_network_code=mobile_network_code, + network=network, + organization=organization, + static_ip_score=static_ip_score, + user_type=user_type, + postal_code=postal_code, + latitude=latitude, + longitude=longitude, + accuracy_radius=accuracy_radius, + ) + + return _inner diff --git a/test_utils/models/conftest.py b/test_utils/models/conftest.py index 64bdec6..1133d32 100644 --- a/test_utils/models/conftest.py +++ b/test_utils/models/conftest.py @@ -1,11 +1,14 @@ +from __future__ import annotations + from datetime import datetime, timedelta, timezone from decimal import Decimal from random import choice as randchoice from random import randint -from typing import TYPE_CHECKING, Callable, Dict, List, Optional +from typing import TYPE_CHECKING, Callable from uuid import uuid4 import pytest +from fastapi import Request from pydantic import AwareDatetime, PositiveInt from generalresearch.models import Source @@ -84,26 +87,27 @@ def user( @pytest.fixture def user_with_wallet( - request, user_factory: Callable[..., "User"], product_user_wallet_yes: "Product" -) -> "User": + user_factory: Callable[..., User], + product_user_wallet_yes: Product, +) -> User: # A user on a product with user wallet enabled, but they have no money return user_factory(product=product_user_wallet_yes) @pytest.fixture def user_with_wallet_amt( - request, user_factory: Callable[..., "User"], product_amt_true: "Product" -) -> "User": + user_factory: Callable[..., User], product_amt_true: Product +) -> User: # A user on a product with user wallet enabled, on AMT, but they have no money return user_factory(product=product_amt_true) @pytest.fixture(scope="function") def user_factory( - user_manager: "UserManager", thl_web_rr: PostgresConfig -) -> Callable[..., "User"]: + user_manager: UserManager, thl_web_rr: PostgresConfig +) -> Callable[..., User]: - def _inner(product: "Product", created: Optional[datetime] = None) -> "User": + def _inner(product: Product, created: datetime | None = None) -> User: u = user_manager.create_dummy(product=product, created=created) u.prefetch_product(pg_config=thl_web_rr) @@ -113,11 +117,11 @@ def user_factory( @pytest.fixture -def wall_factory(wall_manager: "WallManager") -> Callable[..., "Wall"]: +def wall_factory(wall_manager: WallManager) -> Callable[..., Wall]: def _inner( - session: "Session", wall_status: "Status", req_cpi: Optional[Decimal] = None - ) -> "Wall": + session: Session, wall_status: Status, req_cpi: Decimal | None = None + ) -> Wall: assert session.started <= datetime.now( tz=timezone.utc @@ -153,9 +157,7 @@ def wall_factory(wall_manager: "WallManager") -> Callable[..., "Wall"]: @pytest.fixture -def wall( - session: "Session", user: "User", wall_manager: "WallManager" -) -> Optional["Wall"]: +def wall(session: Session, user: User, wall_manager: WallManager) -> Wall | None: from generalresearch.models.thl.task_status import StatusCode1 wall = wall_manager.create_dummy(session_id=session.id, user_id=user.user_id) @@ -170,20 +172,20 @@ def wall( @pytest.fixture def session_factory( - wall_factory: Callable[..., "Wall"], - session_manager: "SessionManager", - wall_manager: "WallManager", + wall_factory: Callable[..., Wall], + session_manager: SessionManager, + wall_manager: WallManager, utc_hour_ago: datetime, -) -> Callable[..., "Session"]: +) -> Callable[..., Session]: from generalresearch.models.thl.session import Source def _inner( - user: "User", + user: User, # Wall details wall_count: int = 5, wall_req_cpi: Decimal = Decimal(".50"), - wall_req_cpis: Optional[List[Decimal]] = None, - wall_statuses: Optional[List[Status]] = None, + wall_req_cpis: list[Decimal] | None = None, + wall_statuses: list[Status] | None = None, wall_source: Source = Source.TESTING, # Session details final_status: Status = Status.COMPLETE, @@ -236,24 +238,24 @@ def session_factory( @pytest.fixture(scope="function") def finished_session_factory( - session_factory: Callable[..., "Session"], - session_manager: "SessionManager", + session_factory: Callable[..., Session], + session_manager: SessionManager, utc_hour_ago: datetime, -) -> Callable[..., "Session"]: +) -> Callable[..., Session]: from generalresearch.models.thl.session import Source def _inner( - user: "User", + user: User, # Wall details wall_count: int = 5, wall_req_cpi: Decimal = Decimal(".50"), - wall_req_cpis: Optional[List[Decimal]] = None, - wall_statuses: Optional[List[Status]] = None, + wall_req_cpis: list[Decimal] | None = None, + wall_statuses: list[Status] | None = None, wall_source: Source = Source.TESTING, # Session details final_status: Status = Status.COMPLETE, started: datetime = utc_hour_ago, - ) -> "Session": + ) -> Session: s: Session = session_factory( user=user, wall_count=wall_count, @@ -281,9 +283,8 @@ def finished_session_factory( @pytest.fixture def session( - user: "User", session_manager: "SessionManager", wall_manager: "WallManager" -) -> "Session": - from generalresearch.models.thl.session import Session, Wall + user: User, session_manager: SessionManager, wall_manager: WallManager +) -> Session: session: Session = session_manager.create_dummy(user=user, country_iso="us") wall: Wall = wall_manager.create_dummy( @@ -297,7 +298,7 @@ def session( @pytest.fixture -def product(request, product_manager: "ProductManager") -> "Product": +def product(request: Request, product_manager: ProductManager) -> Product: team = getattr(request, "team", None) business = getattr(request, "business", None) @@ -309,13 +310,13 @@ def product(request, product_manager: "ProductManager") -> "Product": @pytest.fixture -def product_factory(product_manager: "ProductManager") -> Callable[..., "Product"]: +def product_factory(product_manager: ProductManager) -> Callable[..., Product]: def _inner( - team: Optional["Team"] = None, - business: Optional["Business"] = None, + team: Team | None = None, + business: Business | None = None, commission_pct: Decimal = Decimal("0.05"), - ) -> "Product": + ) -> Product: return product_manager.create_dummy( team_id=team.uuid if team else None, business_id=business.uuid if business else None, @@ -326,7 +327,7 @@ def product_factory(product_manager: "ProductManager") -> Callable[..., "Product @pytest.fixture -def payout_config(request) -> "PayoutConfig": +def payout_config(request: Request) -> PayoutConfig: from generalresearch.models.thl.product import ( PayoutConfig, PayoutTransformation, @@ -348,8 +349,8 @@ def payout_config(request) -> "PayoutConfig": @pytest.fixture def product_user_wallet_yes( - payout_config: "PayoutConfig", product_manager: "ProductManager" -) -> "Product": + payout_config: PayoutConfig, product_manager: ProductManager +) -> Product: from generalresearch.models.thl.product import UserWalletConfig return product_manager.create_dummy( @@ -358,7 +359,7 @@ def product_user_wallet_yes( @pytest.fixture -def product_user_wallet_no(product_manager: "ProductManager") -> "Product": +def product_user_wallet_no(product_manager: ProductManager) -> Product: from generalresearch.models.thl.product import UserWalletConfig return product_manager.create_dummy( @@ -368,8 +369,8 @@ def product_user_wallet_no(product_manager: "ProductManager") -> "Product": @pytest.fixture def product_amt_true( - product_manager: "ProductManager", payout_config: "PayoutConfig" -) -> "Product": + product_manager: ProductManager, payout_config: PayoutConfig +) -> Product: from generalresearch.models.thl.product import UserWalletConfig return product_manager.create_dummy( @@ -380,19 +381,19 @@ def product_amt_true( @pytest.fixture def bp_payout_factory( - thl_lm: "ThlLedgerManager", - product_manager: "ProductManager", - business_payout_event_manager: "BusinessPayoutEventManager", -) -> Callable[..., "BrokerageProductPayoutEvent"]: + thl_lm: ThlLedgerManager, + product_manager: ProductManager, + business_payout_event_manager: BusinessPayoutEventManager, +) -> Callable[..., BrokerageProductPayoutEvent]: def _inner( - product: Optional["Product"] = None, - amount: Optional["USDCent"] = None, - ext_ref_id: Optional[str] = None, - created: Optional[AwareDatetime] = None, + product: Product | None = None, + amount: USDCent | None = None, + ext_ref_id: str | None = None, + created: AwareDatetime | None = None, skip_wallet_balance_check: bool = False, skip_one_per_day_check: bool = False, - ) -> "BrokerageProductPayoutEvent": + ) -> BrokerageProductPayoutEvent: from generalresearch.currency import USDCent product = product or product_manager.create_dummy() @@ -415,43 +416,43 @@ def bp_payout_factory( @pytest.fixture -def business(request, business_manager: "BusinessManager") -> "Business": +def business(request, business_manager: BusinessManager) -> Business: return business_manager.create_dummy() @pytest.fixture def business_address( - request, business: "Business", business_address_manager: "BusinessAddressManager" -) -> "BusinessAddress": + request, business: "Business", business_address_manager: BusinessAddressManager +) -> BusinessAddress: return business_address_manager.create_dummy(business_id=business.id) @pytest.fixture def business_bank_account( request, - business: "Business", - business_bank_account_manager: "BusinessBankAccountManager", -) -> "BusinessBankAccount": + business: Business, + business_bank_account_manager: BusinessBankAccountManager, +) -> BusinessBankAccount: return business_bank_account_manager.create_dummy(business_id=business.id) @pytest.fixture -def team(request, team_manager: "TeamManager") -> "Team": +def team(request, team_manager: TeamManager) -> Team: return team_manager.create_dummy() @pytest.fixture -def gr_user(gr_um: "GRUserManager") -> "GRUser": +def gr_user(gr_um: GRUserManager) -> GRUser: return gr_um.create_dummy() @pytest.fixture def gr_user_cache( - gr_user: "GRUser", + gr_user: GRUser, gr_db: PostgresConfig, thl_web_rr: PostgresConfig, gr_redis_config: RedisConfig, -) -> "GRUser": +) -> GRUser: gr_user.set_cache( pg_config=gr_db, thl_web_rr=thl_web_rr, redis_config=gr_redis_config ) @@ -459,7 +460,7 @@ def gr_user_cache( @pytest.fixture -def gr_user_factory(gr_um: "GRUserManager") -> Callable[..., "GRUser"]: +def gr_user_factory(gr_um: GRUserManager) -> Callable[..., GRUser]: def _inner(): return gr_um.create_dummy() @@ -469,8 +470,8 @@ def gr_user_factory(gr_um: "GRUserManager") -> Callable[..., "GRUser"]: @pytest.fixture() def gr_user_token( - gr_user: "GRUser", gr_tm: "GRTokenManager", gr_db: PostgresConfig -) -> "GRToken": + gr_user: GRUser, gr_tm: GRTokenManager, gr_db: PostgresConfig +) -> GRToken: gr_tm.create(user_id=gr_user.id) gr_user.prefetch_token(pg_config=gr_db) @@ -480,14 +481,14 @@ def gr_user_token( @pytest.fixture() -def gr_user_token_header(gr_user_token: "GRToken") -> Dict[str, str]: +def gr_user_token_header(gr_user_token: GRToken) -> dict[str, str]: return gr_user_token.auth_header @pytest.fixture(scope="function") def membership( - request, team: "Team", gr_user: "GRUser", team_manager: "TeamManager" -) -> "Membership": + request, team: Team, gr_user: GRUser, team_manager: TeamManager +) -> Membership: assert team.id, "Team must be saved" assert gr_user.id, "GRUser must be saved" return team_manager.add_user(team=team, gr_user=gr_user) @@ -495,14 +496,14 @@ def membership( @pytest.fixture(scope="function") def membership_factory( - team: "Team", - gr_user: "GRUser", - membership_manager: "MembershipManager", - team_manager: "TeamManager", - gr_um: "GRUserManager", -) -> Callable[..., "Membership"]: - - def _inner(**kwargs) -> "Membership": + team: Team, + gr_user: GRUser, + membership_manager: MembershipManager, + team_manager: TeamManager, + gr_um: GRUserManager, +) -> Callable[..., Membership]: + + def _inner(**kwargs) -> Membership: _team = kwargs.get("team", team_manager.create_dummy()) _gr_user = kwargs.get("gr_user", gr_um.create_dummy()) @@ -512,23 +513,23 @@ def membership_factory( @pytest.fixture -def audit_log(audit_log_manager: "AuditLogManager", user: "User") -> "AuditLog": +def audit_log(audit_log_manager: AuditLogManager, user: User) -> AuditLog: return audit_log_manager.create_dummy(user_id=user.user_id) @pytest.fixture def audit_log_factory( - audit_log_manager: "AuditLogManager", -) -> Callable[..., "AuditLog"]: + audit_log_manager: AuditLogManager, +) -> Callable[..., AuditLog]: def _inner( user_id: PositiveInt, - level: Optional["AuditLogLevel"] = None, - event_type: Optional[str] = None, - event_msg: Optional[str] = None, - event_value: Optional[float] = None, - ) -> "AuditLog": + level: AuditLogLevel | None = None, + event_type: str | None = None, + event_msg: str | None = None, + event_value: float | None = None, + ) -> AuditLog: return audit_log_manager.create_dummy( user_id=user_id, level=level, @@ -541,14 +542,14 @@ def audit_log_factory( @pytest.fixture -def ip_geoname(ip_geoname_manager: "IPGeonameManager") -> "IPGeoname": +def ip_geoname(ip_geoname_manager: IPGeonameManager) -> IPGeoname: return ip_geoname_manager.create_dummy() @pytest.fixture def ip_information( - ip_information_manager: "IPInformationManager", ip_geoname: "IPGeoname" -) -> "IPInformation": + ip_information_manager: IPInformationManager, ip_geoname: IPGeoname +) -> IPInformation: return ip_information_manager.create_dummy( geoname_id=ip_geoname.geoname_id, country_iso=ip_geoname.country_iso ) @@ -556,10 +557,10 @@ def ip_information( @pytest.fixture def ip_information_factory( - ip_information_manager: "IPInformationManager", -) -> Callable[..., "IPInformation"]: + ip_information_manager: IPInformationManager, +) -> Callable[..., IPInformation]: - def _inner(ip: str, geoname: "IPGeoname", **kwargs) -> "IPInformation": + def _inner(ip: str, geoname: IPGeoname, **kwargs) -> IPInformation: return ip_information_manager.create_dummy( ip=ip, geoname_id=geoname.geoname_id, @@ -572,25 +573,25 @@ def ip_information_factory( @pytest.fixture def ip_record( - ip_record_manager: "IPRecordManager", ip_geoname: "IPGeoname", user: "User" -) -> "IPRecord": + ip_record_manager: IPRecordManager, ip_geoname: IPGeoname, user: User +) -> IPRecord: return ip_record_manager.create_dummy(user_id=user.user_id) @pytest.fixture def ip_record_factory( - ip_record_manager: "IPRecordManager", user: "User" -) -> Callable[..., "IPRecord"]: + ip_record_manager: IPRecordManager, user: User +) -> Callable[..., IPRecord]: - def _inner(user_id: PositiveInt, ip: Optional[str] = None) -> "IPRecord": + def _inner(user_id: PositiveInt, ip: str | None = None) -> IPRecord: return ip_record_manager.create_dummy(user_id=user_id, ip=ip) return _inner @pytest.fixture(scope="session") -def buyer(buyer_manager: "BuyerManager") -> "Buyer": +def buyer(buyer_manager: BuyerManager) -> Buyer: buyer_code = uuid4().hex buyer_manager.bulk_get_or_create(source=Source.TESTING, codes=[buyer_code]) b = Buyer( @@ -601,7 +602,7 @@ def buyer(buyer_manager: "BuyerManager") -> "Buyer": @pytest.fixture(scope="session") -def buyer_factory(buyer_manager: "BuyerManager") -> Callable[..., "Buyer"]: +def buyer_factory(buyer_manager: BuyerManager) -> Callable[..., Buyer]: def _inner() -> Buyer: return buyer_manager.bulk_get_or_create( @@ -612,7 +613,7 @@ def buyer_factory(buyer_manager: "BuyerManager") -> Callable[..., "Buyer"]: @pytest.fixture(scope="session") -def survey(survey_manager: "SurveyManager", buyer: "Buyer") -> "Survey": +def survey(survey_manager: SurveyManager, buyer: Buyer) -> "Survey": s = Survey(source=Source.TESTING, survey_id=uuid4().hex, buyer_code=buyer.code) survey_manager.create_bulk([s]) return s @@ -620,10 +621,10 @@ def survey(survey_manager: "SurveyManager", buyer: "Buyer") -> "Survey": @pytest.fixture(scope="session") def survey_factory( - survey_manager: "SurveyManager", buyer_factory: Callable[..., "Buyer"] -) -> Callable[..., "Survey"]: + survey_manager: SurveyManager, buyer_factory: Callable[..., Buyer] +) -> Callable[..., Survey]: - def _inner(buyer: Optional[Buyer] = None) -> "Survey": + def _inner(buyer: Buyer | None = None) -> Survey: buyer = buyer or buyer_factory() s = Survey( source=Source.TESTING, |