diff options
Diffstat (limited to 'tests')
5 files changed, 96 insertions, 96 deletions
diff --git a/tests/incite/collections/test_df_collection_item_thl_web.py b/tests/incite/collections/test_df_collection_item_thl_web.py index 061c576..b8b955f 100644 --- a/tests/incite/collections/test_df_collection_item_thl_web.py +++ b/tests/incite/collections/test_df_collection_item_thl_web.py @@ -12,7 +12,8 @@ import dask.dataframe as dd import pandas as pd import pytest from dask.distributed import Client as DaskClient -from distributed import Client, Scheduler, Worker +from dask.distributed import Scheduler as DaskScheduler +from dask.distributed import Worker as DaskWorker # noinspection PyUnresolvedReferences from distributed.utils_test import ( @@ -36,7 +37,6 @@ if TYPE_CHECKING: DFCollection, DFCollectionItem, ) - from generalresearch.managers.thl.ledger_manager.thl_ledger import ThlLedgerManager from generalresearch.models.thl.product import Product from generalresearch.models.thl.user import User @@ -57,12 +57,11 @@ unsupported_mock_types = { } -def combo_object() -> Generator[str]: - for x in iter_product( +def combo_object() -> Generator[tuple[DFCollectionType, str]]: + yield from iter_product( df_collections, ["15min", "45min", "1H"], - ): - yield from x + ) class TestDFCollectionItemBase: @@ -170,9 +169,12 @@ class TestDFCollectionItemMethod: def test_has_mysql( self, + df_collection_data_type: DFCollectionType, + offset: str, + duration: timedelta, + delete_df_collection: Callable[..., None], df_collection: DFCollection, thl_web_rr: PostgresConfig, - delete_df_collection: Callable[..., None], ): delete_df_collection(coll=df_collection) @@ -197,6 +199,9 @@ class TestDFCollectionItemMethod: @pytest.mark.skip def test_update_partial_archive( self, + df_collection_data_type: DFCollectionType, + offset: str, + duration: timedelta, ): # for i in collection.items: # assert i.update_partial_archive() @@ -206,8 +211,11 @@ class TestDFCollectionItemMethod: @pytest.mark.skip def test_create_partial_archive( self, + df_collection_data_type: DFCollectionType, + offset: str, + duration: timedelta, ): - assert 1 + 1 == 2 + pass def test_dict( self, @@ -262,14 +270,14 @@ class TestDFCollectionItemMethod: else: assert isinstance(df, pd.DataFrame) assert df.empty - assert set(df.columns) == set(df_collection._schema.columns.keys()) + assert set(df.columns) == set(df_collection.type_schema.columns.keys()) incite_item_factory(user=u1, item=item) df = item.from_db() assert isinstance(df, pd.DataFrame) assert not df.empty - assert set(df.columns) == set(df_collection._schema.columns.keys()) + assert set(df.columns) == set(df_collection.type_schema.columns.keys()) if df_collection.data_type == DFCollectionType.LEDGER: # The number of rows in this dataframe will change depending # on the mocking of data. It's because if the account has @@ -315,32 +323,23 @@ class TestDFCollectionItemMethod: df = item.from_mysql_standard() assert isinstance(df, pd.DataFrame) assert df.empty - assert set(df.columns) == set(df_collection._schema.columns.keys()) + assert set(df.columns) == set(df_collection.type_schema.columns.keys()) incite_item_factory(user=u1, item=item) df = item.from_mysql_standard() assert isinstance(df, pd.DataFrame) assert not df.empty - assert set(df.columns) == set(df_collection._schema.columns.keys()) + assert set(df.columns) == set(df_collection.type_schema.columns.keys()) assert df.shape[0] > 0 def test_from_mysql_ledger( self, df_collection: DFCollection, - user: User, - create_main_accounts: Callable[..., None], - offset: str, - duration: timedelta, - thl_web_rw: PostgresConfig, - thl_ledger_manager: ThlLedgerManager, - df_collection_data_type: DFCollectionType, user_factory: Callable[..., User], product: Product, - client_no_amm: DaskClient, incite_item_factory: Callable[..., None], delete_df_collection: Callable[..., None], - mnt_filepath: GRLDatasets, ): if df_collection.data_type != DFCollectionType.LEDGER: @@ -382,16 +381,10 @@ class TestDFCollectionItemMethod: def test_to_archive( self, df_collection: DFCollection, - user: User, - offset: str, - duration: timedelta, - df_collection_data_type: DFCollectionType, user_factory: Callable[..., User], product: Product, - client_no_amm: DaskClient, incite_item_factory: Callable[..., None], delete_df_collection: Callable[..., None], - mnt_filepath: GRLDatasets, ): if df_collection.data_type in unsupported_mock_types: @@ -418,14 +411,9 @@ class TestDFCollectionItemMethod: def test__to_archive( self, - df_collection_data_type: DFCollectionType, df_collection: DFCollection, user_factory: Callable[..., User], product: Product, - offset: str, - duration: timedelta, - client_no_amm: DaskClient, - user: User, incite_item_factory: Callable[..., None], delete_df_collection: Callable[..., None], mnt_filepath: GRLDatasets, @@ -492,18 +480,27 @@ class TestDFCollectionItemMethod: @pytest.mark.skip def test_to_archive_numbered_partial( self, + df_collection_data_type: DFCollectionType, + offset: str, + duration: timedelta, ): pass @pytest.mark.skip def test_initial_load( self, + df_collection_data_type: DFCollectionType, + offset: str, + duration: timedelta, ): pass @pytest.mark.skip def test_clear_corrupt_archive( self, + df_collection_data_type: DFCollectionType, + offset: str, + duration: timedelta, ): pass @@ -515,18 +512,6 @@ class TestDFCollectionItemMethod: class TestDFCollectionItemMethodBase: @pytest.mark.skip - def test_path_exists( - self, - ): - pass - - @pytest.mark.skip - def test_next_numbered_path( - self, - ): - pass - - @pytest.mark.skip def test_search_highest_numbered_path( self, df_collection_data_type: DFCollectionType, @@ -538,12 +523,18 @@ class TestDFCollectionItemMethodBase: @pytest.mark.skip def test_tmp_filename( self, + df_collection_data_type: DFCollectionType, + offset: str, + duration: timedelta, ): pass @pytest.mark.skip def test_tmp_path( self, + df_collection_data_type: DFCollectionType, + offset: str, + duration: timedelta, ): pass @@ -640,7 +631,8 @@ class TestDFCollectionItemMethodBase: self, df_collection: DFCollection, ): - schema: DataFrameSchema = df_collection._schema + schema: DataFrameSchema = df_collection.type_schema + assert schema.metadata aa = schema.metadata[ARCHIVE_AFTER] # It shouldn't be None, it can be timedelta(seconds=0) @@ -657,6 +649,9 @@ class TestDFCollectionItemMethodBase: @pytest.mark.skip def test_set_empty( self, + df_collection_data_type: DFCollectionType, + offset: str, + duration: timedelta, ): pass @@ -687,17 +682,26 @@ class TestDFCollectionItemMethodBase: @pytest.mark.skip def test_validate_df( self, + df_collection_data_type: DFCollectionType, + offset: str, + duration: timedelta, ): pass @pytest.mark.skip def test_from_archive( self, + df_collection_data_type: DFCollectionType, + offset: str, + duration: timedelta, ): pass def test__to_dict( self, + df_collection_data_type: DFCollectionType, + offset: str, + duration: timedelta, df_collection: DFCollection, ): @@ -718,29 +722,38 @@ class TestDFCollectionItemMethodBase: @pytest.mark.skip def test_delete_partial( self, + df_collection_data_type: DFCollectionType, + offset: str, + duration: timedelta, ): pass @pytest.mark.skip def test_cleanup_partials( self, + df_collection_data_type: DFCollectionType, + offset: str, + duration: timedelta, ): pass @pytest.mark.skip def test_delete_dangling_partials( self, + df_collection_data_type: DFCollectionType, + offset: str, + duration: timedelta, ): pass @gen_cluster(client=True, nthreads=[("127.0.0.1", 1)]) -async def test_client(client, s, worker): +async def test_client(client: DaskClient, s: DaskScheduler, worker: DaskWorker): """c,s,a are all required - the secondary Worker (b) is not required""" - assert isinstance(client, Client) - assert isinstance(s, Scheduler) - assert isinstance(worker, Worker) + assert isinstance(client, DaskClient) + assert isinstance(s, DaskScheduler) + assert isinstance(worker, DaskWorker) @pytest.mark.parametrize( @@ -750,13 +763,17 @@ async def test_client(client, s, worker): @gen_cluster(client=True, nthreads=[("127.0.0.1", 1)]) @pytest.mark.anyio async def test_client_parametrize( - c, s, w, df_collection_data_type: DFCollectionType, offset: str + c: DaskClient, + s: DaskScheduler, + w: DaskWorker, + df_collection_data_type: DFCollectionType, + offset: str, ): """c,s,a are all required - the secondary Worker (b) is not required""" - assert isinstance(c, Client), f"c is not Client, it's {type(c)}" - assert isinstance(s, Scheduler), f"s is not Scheduler, it's {type(s)}" - assert isinstance(w, Worker), f"w is not Worker, it's {type(w)}" + assert isinstance(c, DaskClient), f"c is not Client, it's {type(c)}" + assert isinstance(s, DaskScheduler), f"s is not Scheduler, it's {type(s)}" + assert isinstance(w, DaskWorker), f"w is not Worker, it's {type(w)}" assert df_collection_data_type is not None assert isinstance(offset, str) @@ -923,7 +940,8 @@ class TestDFCollectionItemFunctionalTest: return u1: User = user_factory(product=product) - schema: DataFrameSchema = df_collection._schema + schema: DataFrameSchema = df_collection.type_schema + assert schema.metadata aa = schema.metadata[ARCHIVE_AFTER] assert isinstance(aa, timedelta) diff --git a/tests/incite/collections/test_df_collection_thl_marketplaces.py b/tests/incite/collections/test_df_collection_thl_marketplaces.py index 6ad0cb4..0f79b81 100644 --- a/tests/incite/collections/test_df_collection_thl_marketplaces.py +++ b/tests/incite/collections/test_df_collection_thl_marketplaces.py @@ -1,3 +1,4 @@ +from collections.abc import Generator from datetime import UTC, datetime from itertools import product from typing import TYPE_CHECKING @@ -18,8 +19,8 @@ if TYPE_CHECKING: from generalresearch.pg_helper import PostgresConfig -def combo_object(): - for x in product( +def combo_object() -> Generator[tuple[type, str]]: + yield from product( [ InnovateSurveyHistoryCollection, MorningSurveyTimeseriesCollection, @@ -27,8 +28,7 @@ def combo_object(): SpectrumSurveyTimeseriesCollection, ], ["5min", "6H", "30D"], - ): - yield from x + ) @pytest.mark.parametrize("df_coll, offset", combo_object()) diff --git a/tests/incite/collections/test_df_collection_thl_web.py b/tests/incite/collections/test_df_collection_thl_web.py index 20d7187..3c5e41b 100644 --- a/tests/incite/collections/test_df_collection_thl_web.py +++ b/tests/incite/collections/test_df_collection_thl_web.py @@ -15,8 +15,8 @@ from generalresearch.incite.collections.base import ( ) -def combo_object() -> Generator[tuple]: - for x in product( +def combo_object() -> Generator[tuple[DFCollectionType, str]]: + yield from product( [ DFCollectionType.USER, DFCollectionType.WALL, @@ -26,8 +26,7 @@ def combo_object() -> Generator[tuple]: DFCollectionType.LEDGER, ], ["30min", "1H"], - ): - yield from x + ) @pytest.mark.parametrize( diff --git a/tests/incite/mergers/test_merge_collection.py b/tests/incite/mergers/test_merge_collection.py index 3f478bd..7ed3996 100644 --- a/tests/incite/mergers/test_merge_collection.py +++ b/tests/incite/mergers/test_merge_collection.py @@ -34,8 +34,11 @@ class TestMergeCollection: def test_init( self, - mnt_filepath: GRLDatasets, merge_type: MergeType, + offset: str, + duration: timedelta, + start: datetime, + mnt_filepath: GRLDatasets, ): with pytest.raises(expected_exception=ValueError) as cm: MergeCollection(archive_path=mnt_filepath.data_src) @@ -88,6 +91,9 @@ class TestMergeCollection: def test_schema( self, + offset: str, + duration: timedelta, + start: datetime, mnt_filepath: GRLDatasets, merge_type: MergeType, ): diff --git a/tests/models/test_finance.py b/tests/models/test_finance.py index c579d78..502c596 100644 --- a/tests/models/test_finance.py +++ b/tests/models/test_finance.py @@ -7,6 +7,7 @@ from random import randint from typing import TYPE_CHECKING from uuid import uuid4 +import dask.dataframe as dd import pandas as pd import pytest from dask.distributed import Client as DaskClient @@ -32,11 +33,11 @@ if TYPE_CHECKING: from generalresearch.incite.collections.thl_web import LedgerDFCollection from generalresearch.incite.mergers.pop_ledger import PopLedgerMerge from generalresearch.managers.thl.ledger_manager.thl_ledger import ThlLedgerManager + from generalresearch.managers.thl.product import ProductManager from generalresearch.models.thl.ledger import LedgerAccount from generalresearch.models.thl.product import Product from generalresearch.models.thl.session import Session from generalresearch.models.thl.user import User - from generalresearch.pg_helper import PostgresConfig fake = Faker() @@ -659,10 +660,9 @@ class TestBusinessBalanceInitialize: @pytest.mark.parametrize( - argnames="offset, duration", + argnames="duration", argvalues=list( iter_product( - ["12h", "2D"], [timedelta(days=2), timedelta(days=5)], ) ), @@ -671,17 +671,15 @@ class TestProductFinanceData: def test_base( self, + duration: timedelta, product: Product, user_factory: Callable[..., User], start: datetime, - duration: timedelta, thl_ledger_manager: ThlLedgerManager, session_with_tx_factory: Callable[..., None], ): # -- Build & Setup - # assert ledger_collection.start is None - # assert ledger_collection.offset is None u: User = user_factory(product=product, created=ledger_collection.start) for item in ledger_collection.items: @@ -738,15 +736,6 @@ class TestProductFinanceData: assert len(res) == len({i.time for i in res}) -@pytest.mark.parametrize( - argnames="offset, duration", - argvalues=list( - iter_product( - ["12h", "2D"], - [timedelta(days=2), timedelta(days=5)], - ) - ), -) class TestPOPFinancialData: def test_base( @@ -757,7 +746,6 @@ class TestPOPFinancialData: user_factory: Callable[..., User], product: Product, start: datetime, - duration: timedelta, create_main_accounts: Callable[..., None], session_with_tx_factory: Callable[..., Session], thl_ledger_manager: ThlLedgerManager, @@ -768,8 +756,6 @@ class TestPOPFinancialData: delete_ledger_db() create_main_accounts() delete_df_collection(coll=ledger_collection) - # assert ledger_collection.start is None - # assert ledger_collection.offset is None users = [] for _ in range(5): @@ -833,22 +819,10 @@ class TestPOPFinancialData: # 1 product: Product, multiple Users assert len(users) == len(accounts) - # We group on days, and duration is a parameter to parametrize - assert isinstance(duration, timedelta) - # -- Teardown delete_df_collection(ledger_collection) -@pytest.mark.parametrize( - argnames="offset, duration", - argvalues=list( - iter_product( - ["12h", "1D"], - [timedelta(days=2), timedelta(days=3)], - ) - ), -) class TestBusinessBalanceData: def test_from_pandas( self, @@ -859,7 +833,7 @@ class TestBusinessBalanceData: product: Product, create_main_accounts: Callable[..., None], thl_ledger_manager: ThlLedgerManager, - thl_web_rr: PostgresConfig, + product_manager: ProductManager, delete_df_collection: Callable[..., None], delete_ledger_db: Callable[..., None], session_with_tx_factory: Callable[..., Session], @@ -898,15 +872,18 @@ class TestBusinessBalanceData: columns=numerical_col_names + ["account_id"], filters=[("account_id", "in", [account.uuid])], ) + assert isinstance(ddf, dd.DataFrame) ddf = ddf.groupby("account_id").sum() df: pd.DataFrame = client_no_amm.compute(collections=ddf, sync=True) assert isinstance(df, pd.DataFrame) instance = BusinessBalances.from_pandas( - input_data=df, accounts=[account], thl_pg_config=thl_web_rr + product_manager=product_manager, + input_data=df, + accounts=[account], ) - balance: int = thl_lm.get_account_balance(account=account) + balance: int = thl_ledger_manager.get_account_balance(account=account) assert instance.balance == balance assert instance.net == balance |