diff options
Diffstat (limited to 'tests/incite')
25 files changed, 3602 insertions, 0 deletions
diff --git a/tests/incite/__init__.py b/tests/incite/__init__.py new file mode 100644 index 0000000..2f736e8 --- /dev/null +++ b/tests/incite/__init__.py @@ -0,0 +1,137 @@ +# class TestParquetBehaviors(CleanTempDirectoryTestCls): +# wall_coll = WallDFCollection( +# start=GLOBAL_VARS["wall"].start, +# offset="49h", +# archive_path=f"{settings.incite_mount_dir}/raw/df-collections/{DFCollectionType.WALL.value}", +# ) +# +# def test_filters(self): +# # Using REAL data here +# start = datetime(year=2024, month=1, day=15, hour=12, tzinfo=timezone.utc) +# end = datetime(year=2024, month=1, day=15, hour=20, tzinfo=timezone.utc) +# end_max = datetime( +# year=2024, month=1, day=15, hour=20, tzinfo=timezone.utc +# ) + timedelta(hours=2) +# +# ir = pd.Interval(left=pd.Timestamp(start), right=pd.Timestamp(end)) +# wall_items = [w for w in self.wall_coll.items if w.interval.overlaps(ir)] +# ddf = self.wall_coll.ddf( +# items=wall_items, +# include_partial=True, +# force_rr_latest=False, +# columns=["started", "finished"], +# filters=[ +# ("started", ">=", start), +# ("started", "<", end), +# ], +# ) +# +# df = ddf.compute() +# self.assertIsInstance(df, pd.DataFrame) +# +# # No started=None, and they're all between the started and the end +# self.assertFalse(df.started.isna().any()) +# self.assertFalse((df.started < start).any()) +# self.assertFalse((df.started > end).any()) +# +# # Has finished=None and finished=time, so +# # the finished is all between the started and +# # the end_max +# self.assertTrue(df.finished.isna().any()) +# self.assertTrue((df.finished.dt.year == 2024).any()) +# +# self.assertFalse((df.finished > end_max).any()) +# self.assertFalse((df.finished < start).any()) +# +# # def test_user_id_list(self): +# # # Calling compute turns it into a np.ndarray +# # user_ids = self.instance.ddf( +# # columns=["user_id"] +# # ).user_id.unique().values.compute() +# # self.assertIsInstance(user_ids, np.ndarray) +# # +# # # If ddf filters work with ndarray +# # user_product_merge = <todo: assign> +# # +# # with self.assertRaises(TypeError) as cm: +# # user_product_merge.ddf( +# # filters=[("id", "in", user_ids)]) +# # self.assertIn("Value of 'in' filter must be a list, set or tuple.", str(cm.exception)) +# # +# # # No compute == dask array +# # user_ids = self.instance.ddf( +# # columns=["user_id"] +# # ).user_id.unique().values +# # self.assertIsInstance(user_ids, da.Array) +# # +# # with self.assertRaises(TypeError) as cm: +# # user_product_merge.ddf( +# # filters=[("id", "in", user_ids)]) +# # self.assertIn("Value of 'in' filter must be a list, set or tuple.", str(cm.exception)) +# # +# # # pick a product_id (most active one) +# # self.product_id = instance.df.product_id.value_counts().index[0] +# # self.expected_columns: int = len(instance._schema.columns) +# # self.instance = instance +# +# # def test_basic(self): +# # # now try to load up the data! +# # self.instance.grouped_key = self.product_id +# # +# # # Confirm any of the items are archived +# # self.assertTrue(self.instance.progress.has_archive.eq(True).any()) +# # +# # # Confirm it returns a df +# # df = self.instance.dd().compute() +# # +# # self.assertFalse(df.empty) +# # self.assertEqual(df.shape[1], self.expected_columns) +# # self.assertGreater(df.shape[0], 1) +# # +# # # Confirm that DF only contains this product_id +# # self.assertEqual(df[df.product_id == self.product_id].shape, df.shape) +# +# # def test_god_vs_product_id(self): +# # self.instance.grouped_key = self.product_id +# # df_product_origin = self.instance.dd(columns=None, filters=None).compute() +# # +# # self.instance.grouped_key = None +# # df_god_origin = self.instance.dd(columns=None, +# # filters=[("product_id", "==", self.product_id)]).compute() +# # +# # self.assertTrue(df_god_origin.equals(df_product_origin)) +# +# # +# # instance = POPSessionMerge( +# # start=START, +# # archive_path=self.PATH, +# # group_by="product_id" +# # ) +# # instance.build(U=GLOBAL_VARS["user"], S=GLOBAL_VARS["session"], W=GLOBAL_VARS["wall"]) +# # instance.save(god_only=False) +# # +# # # pick a product_id (most active one) +# # self.product_id = instance.df.product_id.value_counts().index[0] +# # self.expected_columns: int = len(instance._schema.columns) +# # self.instance = instance +# +# +# class TestValidItem(CleanTempDirectoryTestCls): +# +# def test_interval(self): +# for k in GLOBAL_VARS.keys(): +# coll = GLOBAL_VARS[k] +# item = coll.items[0] +# ir = item.interval +# +# self.assertIsInstance(ir, pd.Interval) +# self.assertLess(a=ir.left, b=ir.right) +# +# def test_str(self): +# for k in GLOBAL_VARS.keys(): +# coll = GLOBAL_VARS[k] +# item = coll.items[0] +# +# offset = coll.offset or "–" +# +# self.assertIn(offset, str(item)) diff --git a/tests/incite/collections/__init__.py b/tests/incite/collections/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/tests/incite/collections/__init__.py diff --git a/tests/incite/collections/test_df_collection_base.py b/tests/incite/collections/test_df_collection_base.py new file mode 100644 index 0000000..5aaa729 --- /dev/null +++ b/tests/incite/collections/test_df_collection_base.py @@ -0,0 +1,113 @@ +from datetime import datetime, timezone + +import pandas as pd +import pytest +from pandera import DataFrameSchema + +from generalresearch.incite.collections import ( + DFCollectionType, + DFCollection, +) +from test_utils.incite.conftest import mnt_filepath + +df_collection_types = [e for e in DFCollectionType if e is not DFCollectionType.TEST] + + +@pytest.mark.parametrize("df_coll_type", df_collection_types) +class TestDFCollectionBase: + """None of these tests are about the DFCollection with any specific + data_type... that will be handled in other parameterized tests + + """ + + def test_init(self, mnt_filepath, df_coll_type): + """Try to initialize the DFCollection with various invalid parameters""" + with pytest.raises(expected_exception=ValueError) as cm: + DFCollection(archive_path=mnt_filepath.data_src) + assert "Must explicitly provide a data_type" in str(cm.value) + + # with pytest.raises(expected_exception=ValueError) as cm: + # DFCollection( + # data_type=DFCollectionType.TEST, archive_path=mnt_filepath.data_src + # ) + # assert "Must provide a supported data_type" in str(cm.value) + + instance = DFCollection( + data_type=DFCollectionType.WALL, archive_path=mnt_filepath.data_src + ) + assert instance.data_type == DFCollectionType.WALL + + +@pytest.mark.parametrize("df_coll_type", df_collection_types) +class TestDFCollectionBaseProperties: + + @pytest.mark.skip + def test_df_collection_items(self, mnt_filepath, df_coll_type): + instance = DFCollection( + data_type=df_coll_type, + start=datetime(year=1800, month=1, day=1, tzinfo=timezone.utc), + finished=datetime(year=1900, month=1, day=1, tzinfo=timezone.utc), + offset="100d", + archive_path=mnt_filepath.archive_path(enum_type=df_coll_type), + ) + + assert len(instance.interval_range) == len(instance.items) + assert len(instance.items) == 366 + + def test_df_collection_progress(self, mnt_filepath, df_coll_type): + instance = DFCollection( + data_type=df_coll_type, + start=datetime(year=1800, month=1, day=1, tzinfo=timezone.utc), + finished=datetime(year=1900, month=1, day=1, tzinfo=timezone.utc), + offset="100d", + archive_path=mnt_filepath.archive_path(enum_type=df_coll_type), + ) + + # Progress returns a dataframe with a row each Item + assert isinstance(instance.progress, pd.DataFrame) + assert instance.progress.shape == (366, 6) + + def test_df_collection_schema(self, mnt_filepath, df_coll_type): + instance1 = DFCollection( + data_type=DFCollectionType.WALL, archive_path=mnt_filepath.data_src + ) + + instance2 = DFCollection( + data_type=DFCollectionType.SESSION, archive_path=mnt_filepath.data_src + ) + + assert instance1._schema != instance2._schema + assert isinstance(instance1._schema, DataFrameSchema) + assert isinstance(instance2._schema, DataFrameSchema) + + +class TestDFCollectionBaseMethods: + + @pytest.mark.skip + def test_initial_load(self, mnt_filepath, thl_web_rr): + instance = DFCollection( + pg_config=thl_web_rr, + data_type=DFCollectionType.USER, + start=datetime(year=2022, month=1, day=1, minute=0, tzinfo=timezone.utc), + finished=datetime(year=2022, month=1, day=1, minute=5, tzinfo=timezone.utc), + offset="2min", + archive_path=mnt_filepath.data_src, + ) + + # Confirm that there are no archives available yet + assert instance.progress.has_archive.eq(False).all() + + instance.initial_load() + assert 47 == len(instance.ddf().index) + assert instance.progress.should_archive.eq(True).all() + + # A few archives should have been made + assert not instance.progress.has_archive.eq(False).all() + + @pytest.mark.skip + def test_fetch_force_rr_latest(self): + pass + + @pytest.mark.skip + def test_force_rr_latest(self): + pass diff --git a/tests/incite/collections/test_df_collection_item_base.py b/tests/incite/collections/test_df_collection_item_base.py new file mode 100644 index 0000000..a0c0b0b --- /dev/null +++ b/tests/incite/collections/test_df_collection_item_base.py @@ -0,0 +1,72 @@ +from datetime import datetime, timezone + +import pytest + +from generalresearch.incite.collections import ( + DFCollectionType, + DFCollectionItem, + DFCollection, +) +from test_utils.incite.conftest import mnt_filepath + +df_collection_types = [e for e in DFCollectionType if e is not DFCollectionType.TEST] + + +@pytest.mark.parametrize("df_coll_type", df_collection_types) +class TestDFCollectionItemBase: + + def test_init(self, mnt_filepath, df_coll_type): + collection = DFCollection( + data_type=df_coll_type, + offset="100d", + start=datetime(year=1800, month=1, day=1, tzinfo=timezone.utc), + finished=datetime(year=1900, month=1, day=1, tzinfo=timezone.utc), + archive_path=mnt_filepath.archive_path(enum_type=df_coll_type), + ) + + item = DFCollectionItem() + item._collection = collection + + assert isinstance(item, DFCollectionItem) + + +@pytest.mark.parametrize("df_coll_type", df_collection_types) +class TestDFCollectionItemProperties: + + @pytest.mark.skip + def test_filename(self, df_coll_type): + pass + + +@pytest.mark.parametrize("df_coll_type", df_collection_types) +class TestDFCollectionItemMethods: + + def test_has_mysql_false(self, mnt_filepath, df_coll_type): + collection = DFCollection( + data_type=df_coll_type, + offset="100d", + start=datetime(year=1800, month=1, day=1, tzinfo=timezone.utc), + finished=datetime(year=1900, month=1, day=1, tzinfo=timezone.utc), + archive_path=mnt_filepath.archive_path(enum_type=df_coll_type), + ) + + instance1: DFCollectionItem = collection.items[0] + assert not instance1.has_mysql() + + def test_has_mysql_true(self, thl_web_rr, mnt_filepath, df_coll_type): + collection = DFCollection( + data_type=df_coll_type, + offset="100d", + start=datetime(year=1800, month=1, day=1, tzinfo=timezone.utc), + finished=datetime(year=1900, month=1, day=1, tzinfo=timezone.utc), + archive_path=mnt_filepath.archive_path(enum_type=df_coll_type), + pg_config=thl_web_rr, + ) + + # Has RR, assume unittest server is online + instance2: DFCollectionItem = collection.items[0] + assert instance2.has_mysql() + + @pytest.mark.skip + def test_update_partial_archive(self, df_coll_type): + pass diff --git a/tests/incite/collections/test_df_collection_item_thl_web.py b/tests/incite/collections/test_df_collection_item_thl_web.py new file mode 100644 index 0000000..9c3d67a --- /dev/null +++ b/tests/incite/collections/test_df_collection_item_thl_web.py @@ -0,0 +1,994 @@ +from datetime import datetime, timezone, timedelta +from itertools import product as iter_product +from os.path import join as pjoin +from pathlib import PurePath, Path +from uuid import uuid4 + +import dask.dataframe as dd +import pandas as pd +import pytest +from distributed import Client, Scheduler, Worker + +# noinspection PyUnresolvedReferences +from distributed.utils_test import ( + gen_cluster, + client_no_amm, + loop, + loop_in_thread, + cleanup, + cluster_fixture, + client, +) +from faker import Faker +from pandera import DataFrameSchema +from pydantic import FilePath + +from generalresearch.incite.base import CollectionItemBase +from generalresearch.incite.collections import ( + DFCollectionItem, + DFCollectionType, +) +from generalresearch.incite.schemas import ARCHIVE_AFTER +from generalresearch.models.thl.user import User +from generalresearch.pg_helper import PostgresConfig +from generalresearch.sql_helper import PostgresDsn +from test_utils.incite.conftest import mnt_filepath, incite_item_factory + +fake = Faker() + +df_collections = [ + DFCollectionType.WALL, + DFCollectionType.SESSION, + DFCollectionType.LEDGER, + DFCollectionType.TASK_ADJUSTMENT, +] + +unsupported_mock_types = { + DFCollectionType.IP_INFO, + DFCollectionType.IP_HISTORY, + DFCollectionType.IP_HISTORY_WS, + DFCollectionType.TASK_ADJUSTMENT, +} + + +def combo_object(): + for x in iter_product( + df_collections, + ["15min", "45min", "1H"], + ): + yield x + + +class TestDFCollectionItemBase: + def test_init(self): + instance = CollectionItemBase() + assert isinstance(instance, CollectionItemBase) + assert isinstance(instance.start, datetime) + + +@pytest.mark.parametrize( + argnames="df_collection_data_type, offset", argvalues=combo_object() +) +class TestDFCollectionItemProperties: + + def test_filename(self, df_collection_data_type, df_collection, offset): + for i in df_collection.items: + assert isinstance(i.filename, str) + + assert isinstance(i.path, PurePath) + assert i.path.name == i.filename + + assert i._collection.data_type.name.lower() in i.filename + assert i._collection.offset in i.filename + assert i.start.strftime("%Y-%m-%d-%H-%M-%S") in i.filename + + +@pytest.mark.parametrize( + argnames="df_collection_data_type, offset", argvalues=combo_object() +) +class TestDFCollectionItemPropertiesBase: + + def test_name(self, df_collection_data_type, offset, df_collection): + for i in df_collection.items: + assert isinstance(i.name, str) + + def test_finish(self, df_collection_data_type, offset, df_collection): + for i in df_collection.items: + assert isinstance(i.finish, datetime) + + def test_interval(self, df_collection_data_type, offset, df_collection): + for i in df_collection.items: + assert isinstance(i.interval, pd.Interval) + + def test_partial_filename(self, df_collection_data_type, offset, df_collection): + for i in df_collection.items: + assert isinstance(i.partial_filename, str) + + def test_empty_filename(self, df_collection_data_type, offset, df_collection): + for i in df_collection.items: + assert isinstance(i.empty_filename, str) + + def test_path(self, df_collection_data_type, offset, df_collection): + for i in df_collection.items: + assert isinstance(i.path, FilePath) + + def test_partial_path(self, df_collection_data_type, offset, df_collection): + for i in df_collection.items: + assert isinstance(i.partial_path, FilePath) + + def test_empty_path(self, df_collection_data_type, offset, df_collection): + for i in df_collection.items: + assert isinstance(i.empty_path, FilePath) + + +@pytest.mark.parametrize( + argnames="df_collection_data_type, offset, duration", + argvalues=list( + iter_product( + df_collections, + ["12h", "10D"], + [timedelta(days=10), timedelta(days=45)], + ) + ), +) +class TestDFCollectionItemMethod: + + def test_has_mysql( + self, + df_collection, + thl_web_rr, + offset, + duration, + df_collection_data_type, + delete_df_collection, + ): + delete_df_collection(coll=df_collection) + + df_collection.pg_config = None + for i in df_collection.items: + assert not i.has_mysql() + + # Confirm that the regular connection should work as expected + df_collection.pg_config = thl_web_rr + for i in df_collection.items: + assert i.has_mysql() + + # Make a fake connection and confirm it does NOT work + df_collection.pg_config = PostgresConfig( + dsn=PostgresDsn(f"postgres://root:@127.0.0.1/{uuid4().hex}"), + connect_timeout=5, + statement_timeout=1, + ) + for i in df_collection.items: + assert not i.has_mysql() + + @pytest.mark.skip + def test_update_partial_archive( + self, + df_collection, + offset, + duration, + thl_web_rw, + df_collection_data_type, + delete_df_collection, + ): + # for i in collection.items: + # assert i.update_partial_archive() + # assert df.created.max() < _last_time_block[1] + pass + + @pytest.mark.skip + def test_create_partial_archive( + self, + df_collection, + offset, + duration, + create_main_accounts, + thl_web_rw, + thl_lm, + df_collection_data_type, + user_factory, + product, + client_no_amm, + incite_item_factory, + delete_df_collection, + mnt_filepath, + ): + assert 1 + 1 == 2 + + def test_dict( + self, + df_collection_data_type, + offset, + duration, + df_collection, + delete_df_collection, + ): + delete_df_collection(coll=df_collection) + + for item in df_collection.items: + res = item.to_dict() + assert isinstance(res, dict) + assert len(res.keys()) == 6 + + assert isinstance(res["should_archive"], bool) + assert isinstance(res["has_archive"], bool) + assert isinstance(res["path"], Path) + assert isinstance(res["filename"], str) + + assert isinstance(res["start"], datetime) + assert isinstance(res["finish"], datetime) + assert res["start"] < res["finish"] + + def test_from_mysql( + self, + df_collection_data_type, + df_collection, + offset, + duration, + create_main_accounts, + thl_web_rw, + user_factory, + product, + incite_item_factory, + delete_df_collection, + ): + from generalresearch.models.thl.user import User + + if df_collection.data_type in unsupported_mock_types: + return + + delete_df_collection(coll=df_collection) + u1: User = user_factory(product=product) + + # No data has been loaded, but we can confirm the from_mysql returns + # back an empty data with the correct columns + for item in df_collection.items: + # Unlike .from_mysql_ledger(), .from_mysql_standard() will return + # back and empty df with the correct columns in place + delete_df_collection(coll=df_collection) + df = item.from_mysql() + if df_collection.data_type == DFCollectionType.LEDGER: + assert df is None + else: + assert df.empty + assert set(df.columns) == set(df_collection._schema.columns.keys()) + + incite_item_factory(user=u1, item=item) + + df = item.from_mysql() + assert not df.empty + assert set(df.columns) == set(df_collection._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 + # user wallet on, then there will be more transactions for + # example. + assert df.shape[0] > 0 + + def test_from_mysql_standard( + self, + df_collection_data_type, + df_collection, + offset, + duration, + user_factory, + product, + incite_item_factory, + delete_df_collection, + ): + from generalresearch.models.thl.user import User + + if df_collection.data_type in unsupported_mock_types: + return + u1: User = user_factory(product=product) + + delete_df_collection(coll=df_collection) + + for item in df_collection.items: + item: DFCollectionItem + + if df_collection.data_type == DFCollectionType.LEDGER: + # We're using parametrize, so this If statement is just to + # confirm other Item Types will always raise an assertion + with pytest.raises(expected_exception=AssertionError) as cm: + res = item.from_mysql_standard() + assert ( + "Can't call from_mysql_standard for Ledger DFCollectionItem" + in str(cm.value) + ) + + continue + + # Unlike .from_mysql_ledger(), .from_mysql_standard() will return + # back and empty df with the correct columns in place + df = item.from_mysql_standard() + assert df.empty + assert set(df.columns) == set(df_collection._schema.columns.keys()) + + incite_item_factory(user=u1, item=item) + + df = item.from_mysql_standard() + assert not df.empty + assert set(df.columns) == set(df_collection._schema.columns.keys()) + assert df.shape[0] > 0 + + def test_from_mysql_ledger( + self, + df_collection, + user, + create_main_accounts, + offset, + duration, + thl_web_rw, + thl_lm, + df_collection_data_type, + user_factory, + product, + client_no_amm, + incite_item_factory, + delete_df_collection, + mnt_filepath, + ): + from generalresearch.models.thl.user import User + + if df_collection.data_type != DFCollectionType.LEDGER: + return + u1: User = user_factory(product=product) + + delete_df_collection(coll=df_collection) + + for item in df_collection.items: + item: DFCollectionItem + delete_df_collection(coll=df_collection) + + # Okay, now continue with the actual Ledger Item tests... we need + # to ensure that this item.start - item.finish range hasn't had + # any prior transactions created within that range. + assert item.from_mysql_ledger() is None + + # Create main accounts doesn't matter because it doesn't + # add any transactions to the db + assert item.from_mysql_ledger() is None + + incite_item_factory(user=u1, item=item) + df = item.from_mysql_ledger() + assert isinstance(df, pd.DataFrame) + + # Not only is this a np.int64 to int comparison, but I also know it + # isn't actually measuring anything meaningful. However, it's useful + # as it tells us if the DF contains all the correct TX Entries. I + # figured it's better to count the amount rather than just the + # number of rows. DF == transactions * 2 because there are two + # entries per transactions + # assert df.amount.sum() == total_amt + # assert total_entries == df.shape[0] + + assert not df.tx_id.is_unique + df["net"] = df.direction * df.amount + assert df.groupby("tx_id").net.sum().sum() == 0 + + def test_to_archive( + self, + df_collection, + user, + offset, + duration, + df_collection_data_type, + user_factory, + product, + client_no_amm, + incite_item_factory, + delete_df_collection, + mnt_filepath, + ): + from generalresearch.models.thl.user import User + + if df_collection.data_type in unsupported_mock_types: + return + u1: User = user_factory(product=product) + + delete_df_collection(coll=df_collection) + + for item in df_collection.items: + item: DFCollectionItem + + incite_item_factory(user=u1, item=item) + + # Load up the data that we'll be using for various to_archive + # methods. + df = item.from_mysql() + ddf = dd.from_pandas(df, npartitions=1) + + # (1) Write the basic archive, the issue is that because it's + # an empty pd.DataFrame, it never makes an actual parquet file + assert item.to_archive(ddf=ddf, is_partial=False, overwrite=False) + assert item.has_archive() + assert item.has_archive(include_empty=False) + + def test__to_archive( + self, + df_collection_data_type, + df_collection, + user_factory, + product, + offset, + duration, + client_no_amm, + user, + incite_item_factory, + delete_df_collection, + mnt_filepath, + ): + """We already have a test for the "non-private" version of this, + which primarily just uses the respective Client to determine if + the ddf is_empty or not. + + Therefore, use the private test to check the manual behavior of + passing in the is_empty or overwrite. + """ + if df_collection.data_type in unsupported_mock_types: + return + + delete_df_collection(coll=df_collection) + u1: User = user_factory(product=product) + + for item in df_collection.items: + item: DFCollectionItem + + incite_item_factory(user=u1, item=item) + + # Load up the data that we'll be using for various to_archive + # methods. Will always be empty pd.DataFrames for now... + df = item.from_mysql() + ddf = dd.from_pandas(df, npartitions=1) + + # (1) Confirm a missing ddf (shouldn't bc of type hint) should + # immediately return back False + assert not item._to_archive(ddf=None, is_empty=True) + assert not item._to_archive(ddf=None, is_empty=False) + + # (2) Setting empty overrides any possible state of the ddf + for rand_val in [df, ddf, True, 1_000]: + assert not item.empty_path.exists() + item._to_archive(ddf=rand_val, is_empty=True) + assert item.empty_path.exists() + item.empty_path.unlink() + + # (3) Trigger a warning with overwrite. First write an empty, + # then write it again with override default to confirm it worked, + # then write it again with override=False to confirm it does + # not work. + assert item._to_archive(ddf=ddf, is_empty=True) + res1 = item.empty_path.stat() + + # Returns none because it knows the file (regular, empty, or + # partial) already exists + assert not item._to_archive(ddf=ddf, is_empty=True, overwrite=False) + + # Currently override=True doesn't actually work on empty files + # because it's checked again in .set_empty() and isn't + # aware of the override flag that may be passed in to + # item._to_archive() + with pytest.raises(expected_exception=AssertionError) as cm: + item._to_archive(ddf=rand_val, is_empty=True, overwrite=True) + assert "set_empty is already set; why are you doing this?" in str(cm.value) + + # We can assert the file stats are the same because we were never + # able to go ahead and rewrite or update it in anyway + res2 = item.empty_path.stat() + assert res1 == res2 + + @pytest.mark.skip + def test_to_archive_numbered_partial( + self, df_collection_data_type, df_collection, offset, duration + ): + pass + + @pytest.mark.skip + def test_initial_load( + self, df_collection_data_type, df_collection, offset, duration + ): + pass + + @pytest.mark.skip + def test_clear_corrupt_archive( + self, df_collection_data_type, df_collection, offset, duration + ): + pass + + +@pytest.mark.parametrize( + argnames="df_collection_data_type, offset, duration", + argvalues=list(iter_product(df_collections, ["12h", "10D"], [timedelta(days=15)])), +) +class TestDFCollectionItemMethodBase: + + @pytest.mark.skip + def test_path_exists(self, df_collection_data_type, offset, duration): + pass + + @pytest.mark.skip + def test_next_numbered_path(self, df_collection_data_type, offset, duration): + pass + + @pytest.mark.skip + def test_search_highest_numbered_path( + self, df_collection_data_type, offset, duration + ): + pass + + @pytest.mark.skip + def test_tmp_filename(self, df_collection_data_type, offset, duration): + pass + + @pytest.mark.skip + def test_tmp_path(self, df_collection_data_type, offset, duration): + pass + + def test_is_empty(self, df_collection_data_type, df_collection, offset, duration): + """ + test_has_empty was merged into this because item.has_empty is + an alias for is_empty.. or vis-versa + """ + + for item in df_collection.items: + assert not item.is_empty() + assert not item.has_empty() + + item.empty_path.touch() + + assert item.is_empty() + assert item.has_empty() + + def test_has_partial_archive( + self, df_collection_data_type, df_collection, offset, duration + ): + for item in df_collection.items: + assert not item.has_partial_archive() + item.partial_path.touch() + assert item.has_partial_archive() + + def test_has_archive( + self, df_collection_data_type, df_collection, offset, duration + ): + for item in df_collection.items: + # (1) Originally, nothing exists... so let's just make a file and + # confirm that it works if just touch that path (no validation + # occurs at all). + assert not item.has_archive(include_empty=False) + assert not item.has_archive(include_empty=True) + item.path.touch() + assert item.has_archive(include_empty=False) + assert item.has_archive(include_empty=True) + + item.path.unlink() + assert not item.has_archive(include_empty=False) + assert not item.has_archive(include_empty=True) + + # (2) Same as the above, except make an empty directory + # instead of a file + assert not item.has_archive(include_empty=False) + assert not item.has_archive(include_empty=True) + item.path.mkdir() + assert item.has_archive(include_empty=False) + assert item.has_archive(include_empty=True) + + item.path.rmdir() + assert not item.has_archive(include_empty=False) + assert not item.has_archive(include_empty=True) + + # (3) Rather than make a empty file or dir at the path, let's + # touch the empty_path and confirm the include_empty option + # works + + item.empty_path.touch() + assert not item.has_archive(include_empty=False) + assert item.has_archive(include_empty=True) + + def test_delete_archive( + self, df_collection_data_type, df_collection, offset, duration + ): + for item in df_collection.items: + item: DFCollectionItem + # (1) Confirm that it doesn't raise an error or anything if we + # try to delete files or folders that do not exist + CollectionItemBase.delete_archive(generic_path=item.path) + CollectionItemBase.delete_archive(generic_path=item.empty_path) + CollectionItemBase.delete_archive(generic_path=item.partial_path) + + item.path.touch() + item.empty_path.touch() + item.partial_path.touch() + + CollectionItemBase.delete_archive(generic_path=item.path) + CollectionItemBase.delete_archive(generic_path=item.empty_path) + CollectionItemBase.delete_archive(generic_path=item.partial_path) + + assert not item.path.exists() + assert not item.empty_path.exists() + assert not item.partial_path.exists() + + def test_should_archive( + self, df_collection_data_type, df_collection, offset, duration + ): + schema: DataFrameSchema = df_collection._schema + aa = schema.metadata[ARCHIVE_AFTER] + + # It shouldn't be None, it can be timedelta(seconds=0) + assert isinstance(aa, timedelta) + + for item in df_collection.items: + item: DFCollectionItem + + if datetime.now(tz=timezone.utc) > item.finish + aa: + assert item.should_archive() + else: + assert not item.should_archive() + + @pytest.mark.skip + def test_set_empty(self, df_collection_data_type, df_collection, offset, duration): + pass + + def test_valid_archive( + self, df_collection_data_type, df_collection, offset, duration + ): + # Originally, nothing has been saved or anything.. so confirm it + # always comes back as None + for item in df_collection.items: + assert not item.valid_archive(generic_path=None, sample=None) + + _path = Path(pjoin(df_collection.archive_path, uuid4().hex)) + + # (1) Fail if isfile, but doesn't exist and if we can't read + # it as valid ParquetFile + assert not item.valid_archive(generic_path=_path, sample=None) + _path.touch() + assert not item.valid_archive(generic_path=_path, sample=None) + _path.unlink() + + # (2) Fail if isdir and we can't read it as a valid ParquetFile + _path.mkdir() + assert _path.is_dir() + assert not item.valid_archive(generic_path=_path, sample=None) + _path.rmdir() + + @pytest.mark.skip + def test_validate_df( + self, df_collection_data_type, df_collection, offset, duration + ): + pass + + @pytest.mark.skip + def test_from_archive( + self, df_collection_data_type, df_collection, offset, duration + ): + pass + + def test__to_dict(self, df_collection_data_type, df_collection, offset, duration): + + for item in df_collection.items: + res = item._to_dict() + assert isinstance(res, dict) + assert len(res.keys()) == 6 + + assert isinstance(res["should_archive"], bool) + assert isinstance(res["has_archive"], bool) + assert isinstance(res["path"], Path) + assert isinstance(res["filename"], str) + + assert isinstance(res["start"], datetime) + assert isinstance(res["finish"], datetime) + assert res["start"] < res["finish"] + + @pytest.mark.skip + def test_delete_partial( + self, df_collection_data_type, df_collection, offset, duration + ): + pass + + @pytest.mark.skip + def test_cleanup_partials( + self, df_collection_data_type, df_collection, offset, duration + ): + pass + + @pytest.mark.skip + def test_delete_dangling_partials( + self, df_collection_data_type, df_collection, offset, duration + ): + pass + + +@gen_cluster(client=True, nthreads=[("127.0.0.1", 1)]) +async def test_client(client, s, worker): + """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) + + +@pytest.mark.parametrize( + argnames="df_collection_data_type, offset", + argvalues=combo_object(), +) +@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, offset): + """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 df_collection_data_type is not None + assert isinstance(offset, str) + + +# I cannot figure out how to define the parametrize on the Test, but then have +# sync or async methods within it, with some having or not having the +# gen_cluster decorator set. + + +@pytest.mark.parametrize( + argnames="df_collection_data_type, offset, duration", + argvalues=list(iter_product(df_collections, ["12h", "10D"], [timedelta(days=15)])), +) +class TestDFCollectionItemFunctionalTest: + + def test_to_archive_and_ddf( + self, + df_collection_data_type, + offset, + duration, + client_no_amm, + df_collection, + user, + user_factory, + product, + incite_item_factory, + delete_df_collection, + mnt_filepath, + ): + from generalresearch.models.thl.user import User + + if df_collection.data_type in unsupported_mock_types: + return + u1: User = user_factory(product=product) + + delete_df_collection(coll=df_collection) + df_collection._client = client_no_amm + + # Assert that there are no pre-existing archives + assert df_collection.progress.has_archive.eq(False).all() + res = df_collection.ddf() + assert res is None + + delete_df_collection(coll=df_collection) + for item in df_collection.items: + item: DFCollectionItem + + incite_item_factory(user=u1, item=item) + item.initial_load() + + # I know it seems weird to delete items from the database before we + # proceed with the test. However, the content should have already + # been saved out into an parquet at this point, and I am too lazy + # to write a separate teardown for a collection (and not a + # single Item) + + # Now that we went ahead with the initial_load, Assert that all + # items have archives files saved + assert isinstance(df_collection.progress, pd.DataFrame) + assert df_collection.progress.has_archive.eq(True).all() + + ddf = df_collection.ddf() + shape = df_collection._client.compute(collections=ddf.shape, sync=True) + assert shape[0] > 5 + + def test_filesize_estimate( + self, + df_collection, + user, + offset, + duration, + client_no_amm, + user_factory, + product, + df_collection_data_type, + incite_item_factory, + delete_df_collection, + mnt_filepath, + ): + """A functional test to write some Parquet files for the + DFCollection and then confirm that the files get written + correctly. + + Confirm the files are written correctly by: + (1) Validating their passing the pandera schema + (2) The file or dir has an expected size on disk + """ + import pyarrow.parquet as pq + from generalresearch.models.thl.user import User + import os + + if df_collection.data_type in unsupported_mock_types: + return + delete_df_collection(coll=df_collection) + u1: User = user_factory(product=product) + + # Pick 3 random items to sample for correct filesize + for item in df_collection.items: + item: DFCollectionItem + + incite_item_factory(user=u1, item=item) + item.initial_load(overwrite=True) + + total_bytes = 0 + for fp in pq.ParquetDataset(item.path).files: + total_bytes += os.stat(fp).st_size + + total_mb = total_bytes / 1_048_576 + + assert total_bytes > 1_000 + assert total_mb < 1 + + def test_to_archive_client( + self, + client_no_amm, + df_collection, + user_factory, + product, + offset, + duration, + df_collection_data_type, + incite_item_factory, + delete_df_collection, + mnt_filepath, + ): + from generalresearch.models.thl.user import User + + delete_df_collection(coll=df_collection) + df_collection._client = client_no_amm + u1: User = user_factory(product=product) + + for item in df_collection.items: + item: DFCollectionItem + + if df_collection.data_type in unsupported_mock_types: + continue + + incite_item_factory(user=u1, item=item) + + # Load up the data that we'll be using for various to_archive + # methods. Will always be empty pd.DataFrames for now... + df = item.from_mysql() + ddf = dd.from_pandas(df, npartitions=1) + assert isinstance(ddf, dd.DataFrame) + + # (1) Write the basic archive, the issue is that because it's + # an empty pd.DataFrame, it never makes an actual parquet file + assert not item.has_archive() + saved = item.to_archive(ddf=ddf, is_partial=False, overwrite=False) + assert saved + assert item.has_archive(include_empty=True) + + @pytest.mark.skip + def test_get_items(self, df_collection, product, offset, duration): + with pytest.warns(expected_warning=ResourceWarning) as cm: + df_collection.get_items_last365() + assert "DFCollectionItem has missing archives" in str( + [w.message for w in cm.list] + ) + + res = df_collection.get_items_last365() + assert len(res) == len(df_collection.items) + + def test_saving_protections( + self, + client_no_amm, + df_collection_data_type, + df_collection, + incite_item_factory, + delete_df_collection, + user_factory, + product, + offset, + duration, + mnt_filepath, + ): + """Don't allow creating an archive for data that will likely be + overwritten or updated + """ + from generalresearch.models.thl.user import User + + if df_collection.data_type in unsupported_mock_types: + return + u1: User = user_factory(product=product) + + schema: DataFrameSchema = df_collection._schema + aa = schema.metadata[ARCHIVE_AFTER] + assert isinstance(aa, timedelta) + + delete_df_collection(df_collection) + for item in df_collection.items: + item: DFCollectionItem + + incite_item_factory(user=u1, item=item) + + should_archive = item.should_archive() + res = item.initial_load() + + # self.assertIn("Cannot create archive for such new data", str(cm.records)) + + # .to_archive() will return back True or False depending on if it + # was successful. We want to compare that result to the + # .should_archive() method result + assert should_archive == res + + def test_empty_item( + self, + client_no_amm, + df_collection_data_type, + df_collection, + incite_item_factory, + delete_df_collection, + user, + offset, + duration, + mnt_filepath, + ): + delete_df_collection(coll=df_collection) + + for item in df_collection.items: + assert not item.has_empty() + df: pd.DataFrame = item.from_mysql() + + # We do this check b/c the Ledger returns back None and + # I don't want it to fail when we go to make a ddf + if df is None: + item.set_empty() + else: + ddf = dd.from_pandas(df, npartitions=1) + item.to_archive(ddf=ddf) + + assert item.has_empty() + + def test_file_touching( + self, + client_no_amm, + df_collection_data_type, + df_collection, + incite_item_factory, + delete_df_collection, + user_factory, + product, + offset, + duration, + mnt_filepath, + ): + from generalresearch.models.thl.user import User + + delete_df_collection(coll=df_collection) + df_collection._client = client_no_amm + u1: User = user_factory(product=product) + + for item in df_collection.items: + # Confirm none of the paths exist yet + assert not item.has_archive() + assert not item.path_exists(generic_path=item.path) + assert not item.has_empty() + assert not item.path_exists(generic_path=item.empty_path) + + if df_collection.data_type in unsupported_mock_types: + assert not item.has_archive(include_empty=False) + assert not item.has_empty() + assert not item.path_exists(generic_path=item.empty_path) + else: + incite_item_factory(user=u1, item=item) + item.initial_load() + + assert item.has_archive(include_empty=False) + assert item.path_exists(generic_path=item.path) + assert not item.has_empty() diff --git a/tests/incite/collections/test_df_collection_thl_marketplaces.py b/tests/incite/collections/test_df_collection_thl_marketplaces.py new file mode 100644 index 0000000..0a77938 --- /dev/null +++ b/tests/incite/collections/test_df_collection_thl_marketplaces.py @@ -0,0 +1,75 @@ +from datetime import datetime, timezone +from itertools import product + +import pytest +from pandera import Column, Index, DataFrameSchema + +from generalresearch.incite.collections import DFCollection +from generalresearch.incite.collections import DFCollectionType +from generalresearch.incite.collections.thl_marketplaces import ( + InnovateSurveyHistoryCollection, + MorningSurveyTimeseriesCollection, + SagoSurveyHistoryCollection, + SpectrumSurveyTimeseriesCollection, +) +from test_utils.incite.conftest import mnt_filepath + + +def combo_object(): + for x in product( + [ + InnovateSurveyHistoryCollection, + MorningSurveyTimeseriesCollection, + SagoSurveyHistoryCollection, + SpectrumSurveyTimeseriesCollection, + ], + ["5min", "6H", "30D"], + ): + yield x + + +@pytest.mark.parametrize("df_coll, offset", combo_object()) +class TestDFCollection_thl_marketplaces: + + def test_init(self, mnt_filepath, df_coll, offset, spectrum_rw): + assert issubclass(df_coll, DFCollection) + + # This is stupid, but we need to pull the default from the + # Pydantic field + data_type = df_coll.model_fields["data_type"].default + assert isinstance(data_type, DFCollectionType) + + # (1) Can't be totally empty, needs a path... + with pytest.raises(expected_exception=Exception) as cm: + instance = df_coll() + + # (2) Confirm it only needs the archive_path + instance = df_coll( + archive_path=mnt_filepath.archive_path(enum_type=data_type), + ) + assert isinstance(instance, DFCollection) + + # (3) Confirm it loads with all + instance = df_coll( + archive_path=mnt_filepath.archive_path(enum_type=data_type), + sql_helper=spectrum_rw, + offset=offset, + start=datetime(year=2023, month=6, day=1, minute=0, tzinfo=timezone.utc), + finished=datetime(year=2023, month=6, day=1, minute=5, tzinfo=timezone.utc), + ) + assert isinstance(instance, DFCollection) + + # (4) Now that we initialize the Class, we can access the _schema + assert isinstance(instance._schema, DataFrameSchema) + assert isinstance(instance._schema.index, Index) + + for c in instance._schema.columns.keys(): + assert isinstance(c, str) + col = instance._schema.columns[c] + assert isinstance(col, Column) + + assert instance._schema.coerce, "coerce on all Schemas" + assert isinstance(instance._schema.checks, list) + assert len(instance._schema.checks) == 0 + assert isinstance(instance._schema.metadata, dict) + assert len(instance._schema.metadata.keys()) == 2 diff --git a/tests/incite/collections/test_df_collection_thl_web.py b/tests/incite/collections/test_df_collection_thl_web.py new file mode 100644 index 0000000..e6f464b --- /dev/null +++ b/tests/incite/collections/test_df_collection_thl_web.py @@ -0,0 +1,160 @@ +from datetime import datetime +from itertools import product + +import dask.dataframe as dd +import pandas as pd +import pytest +from pandera import DataFrameSchema + +from generalresearch.incite.collections import DFCollection, DFCollectionType + + +def combo_object(): + for x in product( + [ + DFCollectionType.USER, + DFCollectionType.WALL, + DFCollectionType.SESSION, + DFCollectionType.TASK_ADJUSTMENT, + DFCollectionType.AUDIT_LOG, + DFCollectionType.LEDGER, + ], + ["30min", "1H"], + ): + yield x + + +@pytest.mark.parametrize( + argnames="df_collection_data_type, offset", argvalues=combo_object() +) +class TestDFCollection_thl_web: + + def test_init(self, df_collection_data_type, offset, df_collection): + assert isinstance(df_collection_data_type, DFCollectionType) + assert isinstance(df_collection, DFCollection) + + +@pytest.mark.parametrize( + argnames="df_collection_data_type, offset", argvalues=combo_object() +) +class TestDFCollection_thl_web_Properties: + + def test_items(self, df_collection_data_type, offset, df_collection): + assert isinstance(df_collection.items, list) + for i in df_collection.items: + assert i._collection == df_collection + + def test__schema(self, df_collection_data_type, offset, df_collection): + assert isinstance(df_collection._schema, DataFrameSchema) + + +@pytest.mark.parametrize( + argnames="df_collection_data_type, offset", argvalues=combo_object() +) +class TestDFCollection_thl_web_BaseProperties: + + @pytest.mark.skip + def test__interval_range(self, df_collection_data_type, offset, df_collection): + pass + + def test_interval_start(self, df_collection_data_type, offset, df_collection): + assert isinstance(df_collection.interval_start, datetime) + + def test_interval_range(self, df_collection_data_type, offset, df_collection): + assert isinstance(df_collection.interval_range, list) + + def test_progress(self, df_collection_data_type, offset, df_collection): + assert isinstance(df_collection.progress, pd.DataFrame) + + +@pytest.mark.parametrize( + argnames="df_collection_data_type, offset", argvalues=combo_object() +) +class TestDFCollection_thl_web_Methods: + + @pytest.mark.skip + def test_initial_loads(self, df_collection_data_type, df_collection, offset): + pass + + @pytest.mark.skip + def test_fetch_force_rr_latest( + self, df_collection_data_type, df_collection, offset + ): + pass + + @pytest.mark.skip + def test_force_rr_latest(self, df_collection_data_type, df_collection, offset): + pass + + +@pytest.mark.parametrize( + argnames="df_collection_data_type, offset", argvalues=combo_object() +) +class TestDFCollection_thl_web_BaseMethods: + + def test_fetch_all_paths(self, df_collection_data_type, offset, df_collection): + res = df_collection.fetch_all_paths( + items=None, force_rr_latest=False, include_partial=False + ) + assert isinstance(res, list) + + @pytest.mark.skip + def test_ddf(self, df_collection_data_type, offset, df_collection): + res = df_collection.ddf() + assert isinstance(res, dd.DataFrame) + + # -- cleanup -- + @pytest.mark.skip + def test_schedule_cleanup(self, df_collection_data_type, offset, df_collection): + pass + + @pytest.mark.skip + def test_cleanup(self, df_collection_data_type, offset, df_collection): + pass + + @pytest.mark.skip + def test_cleanup_partials(self, df_collection_data_type, offset, df_collection): + pass + + @pytest.mark.skip + def test_clear_tmp_archives(self, df_collection_data_type, offset, df_collection): + pass + + @pytest.mark.skip + def test_clear_corrupt_archives( + self, df_collection_data_type, offset, df_collection + ): + pass + + @pytest.mark.skip + def test_rebuild_symlinks(self, df_collection_data_type, offset, df_collection): + pass + + # -- Source timing -- + @pytest.mark.skip + def test_get_item(self, df_collection_data_type, offset, df_collection): + pass + + @pytest.mark.skip + def test_get_item_start(self, df_collection_data_type, offset, df_collection): + pass + + @pytest.mark.skip + def test_get_items(self, df_collection_data_type, offset, df_collection): + # If we get all the items from the start of the collection, it + # should include all the items! + res1 = df_collection.items + res2 = df_collection.get_items(since=df_collection.start) + assert len(res1) == len(res2) + + @pytest.mark.skip + def test_get_items_from_year(self, df_collection_data_type, offset, df_collection): + pass + + @pytest.mark.skip + def test_get_items_last90(self, df_collection_data_type, offset, df_collection): + pass + + @pytest.mark.skip + def test_get_items_last365(self, df_collection_data_type, offset, df_collection): + pass diff --git a/tests/incite/collections/test_df_collection_thl_web_ledger.py b/tests/incite/collections/test_df_collection_thl_web_ledger.py new file mode 100644 index 0000000..599d979 --- /dev/null +++ b/tests/incite/collections/test_df_collection_thl_web_ledger.py @@ -0,0 +1,32 @@ +# def test_loaded(self, client_no_amm, collection, new_user_fixture, pop_ledger_merge): +# collection._client = client_no_amm +# +# teardown_events(collection) +# THL_LM.create_main_accounts() +# +# for item in collection.items: +# populate_events(item, user=new_user_fixture) +# item.initial_load() +# +# ddf = collection.ddf( +# force_rr_latest=False, +# include_partial=True, +# filters=[ +# ("created", ">=", collection.start), +# ("created", "<", collection.finished), +# ], +# ) +# +# assert isinstance(ddf, dd.DataFrame) +# df = client_no_amm.compute(collections=ddf, sync=True) +# assert isinstance(df, pd.DataFrame) +# +# # Simple validation check(s) +# assert not df.tx_id.is_unique +# df["net"] = df.direction * df.amount +# assert df.groupby("tx_id").net.sum().sum() == 0 +# +# teardown_events(collection) +# +# +# diff --git a/tests/incite/mergers/__init__.py b/tests/incite/mergers/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/tests/incite/mergers/__init__.py diff --git a/tests/incite/mergers/foundations/__init__.py b/tests/incite/mergers/foundations/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/tests/incite/mergers/foundations/__init__.py diff --git a/tests/incite/mergers/foundations/test_enriched_session.py b/tests/incite/mergers/foundations/test_enriched_session.py new file mode 100644 index 0000000..ec15d38 --- /dev/null +++ b/tests/incite/mergers/foundations/test_enriched_session.py @@ -0,0 +1,138 @@ +from datetime import timedelta, timezone, datetime +from decimal import Decimal +from itertools import product +from typing import Optional + +from generalresearch.incite.schemas.admin_responses import ( + AdminPOPSessionSchema, +) + +import dask.dataframe as dd +import pandas as pd +import pytest + +from test_utils.incite.collections.conftest import ( + session_collection, + wall_collection, +) + + +@pytest.mark.parametrize( + argnames="offset, duration", + argvalues=list( + product( + ["12h", "3D"], + [timedelta(days=5)], + ) + ), +) +class TestEnrichedSession: + + def test_base( + self, + client_no_amm, + product, + user_factory, + wall_collection, + session_collection, + enriched_session_merge, + thl_web_rr, + delete_df_collection, + incite_item_factory, + ): + from generalresearch.models.thl.user import User + + delete_df_collection(coll=session_collection) + + u1: User = user_factory(product=product, created=session_collection.start) + + for item in session_collection.items: + incite_item_factory(item=item, user=u1) + item.initial_load() + + for item in wall_collection.items: + item.initial_load() + + enriched_session_merge.build( + client=client_no_amm, + wall_coll=wall_collection, + session_coll=session_collection, + pg_config=thl_web_rr, + ) + + # -- + + ddf = enriched_session_merge.ddf() + assert isinstance(ddf, dd.DataFrame) + + df = client_no_amm.compute(collections=ddf, sync=True) + assert isinstance(df, pd.DataFrame) + + assert not df.empty + + # -- Teardown + delete_df_collection(session_collection) + + +class TestEnrichedSessionAdmin: + + @pytest.fixture + def start(self) -> "datetime": + return datetime(year=2020, month=3, day=14, tzinfo=timezone.utc) + + @pytest.fixture + def offset(self) -> str: + return "1d" + + @pytest.fixture + def duration(self) -> Optional["timedelta"]: + return timedelta(days=5) + + def test_to_admin_response( + self, + event_report_request, + enriched_session_merge, + client_no_amm, + wall_collection, + session_collection, + thl_web_rr, + session_report_request, + user_factory, + start, + session_factory, + product_factory, + delete_df_collection, + ): + delete_df_collection(coll=wall_collection) + delete_df_collection(coll=session_collection) + + p1 = product_factory() + p2 = product_factory() + + for p in [p1, p2]: + u = user_factory(product=p) + for i in range(50): + s = session_factory( + user=u, + wall_count=1, + wall_req_cpi=Decimal("1.00"), + started=start + timedelta(minutes=i, seconds=1), + ) + wall_collection.initial_load(client=None, sync=True) + session_collection.initial_load(client=None, sync=True) + + enriched_session_merge.build( + client=client_no_amm, + session_coll=session_collection, + wall_coll=wall_collection, + pg_config=thl_web_rr, + ) + + df = enriched_session_merge.to_admin_response( + rr=session_report_request, client=client_no_amm + ) + + assert isinstance(df, pd.DataFrame) + assert not df.empty + assert isinstance(AdminPOPSessionSchema.validate(df), pd.DataFrame) + assert df.index.get_level_values(1).nunique() == 2 diff --git a/tests/incite/mergers/foundations/test_enriched_task_adjust.py b/tests/incite/mergers/foundations/test_enriched_task_adjust.py new file mode 100644 index 0000000..96c214f --- /dev/null +++ b/tests/incite/mergers/foundations/test_enriched_task_adjust.py @@ -0,0 +1,76 @@ +from datetime import timedelta +from itertools import product as iter_product + +import dask.dataframe as dd +import pandas as pd +import pytest + +from test_utils.incite.collections.conftest import ( + wall_collection, + task_adj_collection, + session_collection, +) +from test_utils.incite.mergers.conftest import enriched_wall_merge + + +@pytest.mark.parametrize( + argnames="offset, duration,", + argvalues=list( + iter_product( + ["12h", "3D"], + [timedelta(days=5)], + ) + ), +) +class TestEnrichedTaskAdjust: + + @pytest.mark.skip + def test_base( + self, + client_no_amm, + user_factory, + product, + task_adj_collection, + wall_collection, + session_collection, + enriched_wall_merge, + enriched_task_adjust_merge, + incite_item_factory, + delete_df_collection, + thl_web_rr, + ): + from generalresearch.models.thl.user import User + + # -- Build & Setup + delete_df_collection(coll=session_collection) + u1: User = user_factory(product=product) + + for item in session_collection.items: + incite_item_factory(user=u1, item=item) + item.initial_load() + for item in wall_collection.items: + item.initial_load() + + enriched_wall_merge.build( + client=client_no_amm, + session_coll=session_collection, + wall_coll=wall_collection, + pg_config=thl_web_rr, + ) + + enriched_task_adjust_merge.build( + client=client_no_amm, + task_adjust_coll=task_adj_collection, + enriched_wall=enriched_wall_merge, + pg_config=thl_web_rr, + ) + + # -- + + ddf = enriched_task_adjust_merge.ddf() + assert isinstance(ddf, dd.DataFrame) + + df = client_no_amm.compute(collections=ddf, sync=True) + assert isinstance(df, pd.DataFrame) + + assert not df.empty diff --git a/tests/incite/mergers/foundations/test_enriched_wall.py b/tests/incite/mergers/foundations/test_enriched_wall.py new file mode 100644 index 0000000..8f4995b --- /dev/null +++ b/tests/incite/mergers/foundations/test_enriched_wall.py @@ -0,0 +1,236 @@ +from datetime import timedelta, timezone, datetime +from decimal import Decimal +from itertools import product as iter_product +from typing import Optional + +import dask.dataframe as dd +import pandas as pd +import pytest + +# noinspection PyUnresolvedReferences +from distributed.utils_test import ( + gen_cluster, + client_no_amm, + loop, + loop_in_thread, + cleanup, + cluster_fixture, + client, +) + +from generalresearch.incite.mergers.foundations.enriched_wall import ( + EnrichedWallMergeItem, +) +from test_utils.incite.collections.conftest import ( + session_collection, + wall_collection, +) +from test_utils.incite.conftest import incite_item_factory +from test_utils.incite.mergers.conftest import ( + enriched_wall_merge, +) + + +@pytest.mark.parametrize( + argnames="offset, duration", + argvalues=list(iter_product(["48h", "3D"], [timedelta(days=5)])), +) +class TestEnrichedWall: + + def test_base( + self, + client_no_amm, + product, + user_factory, + wall_collection, + thl_web_rr, + session_collection, + enriched_wall_merge, + delete_df_collection, + incite_item_factory, + ): + from generalresearch.models.thl.user import User + + # -- Build & Setup + delete_df_collection(coll=session_collection) + delete_df_collection(coll=wall_collection) + u1: User = user_factory(product=product, created=session_collection.start) + + for item in session_collection.items: + incite_item_factory(item=item, user=u1) + item.initial_load() + + for item in wall_collection.items: + item.initial_load() + + enriched_wall_merge.build( + client=client_no_amm, + wall_coll=wall_collection, + session_coll=session_collection, + pg_config=thl_web_rr, + ) + + # -- + + ddf = enriched_wall_merge.ddf() + assert isinstance(ddf, dd.DataFrame) + + df = client_no_amm.compute(collections=ddf, sync=True) + assert isinstance(df, pd.DataFrame) + + assert not df.empty + + def test_base_item( + self, + client_no_amm, + product, + user_factory, + wall_collection, + session_collection, + enriched_wall_merge, + delete_df_collection, + thl_web_rr, + incite_item_factory, + ): + # -- Build & Setup + delete_df_collection(coll=session_collection) + u = user_factory(product=product, created=session_collection.start) + + for item in session_collection.items: + incite_item_factory(item=item, user=u) + item.initial_load() + for item in wall_collection.items: + item.initial_load() + + enriched_wall_merge.build( + client=client_no_amm, + wall_coll=wall_collection, + session_coll=session_collection, + pg_config=thl_web_rr, + ) + + # -- + + for item in enriched_wall_merge.items: + assert isinstance(item, EnrichedWallMergeItem) + + path = item.path + + try: + modified_time1 = path.stat().st_mtime + except (Exception,): + modified_time1 = 0 + + item.build( + client=client_no_amm, + wall_coll=wall_collection, + session_coll=session_collection, + pg_config=thl_web_rr, + ) + modified_time2 = path.stat().st_mtime + + # Merger Items can't be updated unless it's a partial, confirm + # that even after attempting to rebuild, it doesn't re-touch + # the file + assert modified_time2 == modified_time1 + + # def test_admin_pop_session_device_type(ew_merge_setup): + # self.build() + # + # rr = ReportRequest( + # report_type=ReportType.POP_EVENT, + # index0="started", + # index1="device_type", + # freq="min", + # start=start, + # ) + # + # df, categories, updated = self.instance.to_admin_response( + # rr=rr, product_ids=[self.product.id], client=client + # ) + # + # assert isinstance(df, pd.DataFrame) + # device_types_str = [str(e.value) for e in DeviceType] + # device_types = df.index.get_level_values(1).values + # assert all([dt in device_types_str for dt in device_types]) + + +class TestEnrichedWallToAdmin: + + @pytest.fixture + def start(self) -> "datetime": + return datetime(year=2020, month=3, day=14, tzinfo=timezone.utc) + + @pytest.fixture + def offset(self) -> str: + return "1d" + + @pytest.fixture + def duration(self) -> Optional["timedelta"]: + return timedelta(days=5) + + def test_empty(self, enriched_wall_merge, client_no_amm, start): + from generalresearch.models.admin.request import ReportRequest + + rr = ReportRequest.model_validate({"interval": "5min", "start": start}) + + res = enriched_wall_merge.to_admin_response( + rr=rr, + client=client_no_amm, + ) + + assert isinstance(res, pd.DataFrame) + + assert res.empty + assert len(res.columns) > 5 + + def test_to_admin_response( + self, + event_report_request, + enriched_wall_merge, + client_no_amm, + wall_collection, + session_collection, + thl_web_rr, + user, + session_factory, + delete_df_collection, + product_factory, + user_factory, + start, + ): + delete_df_collection(coll=wall_collection) + delete_df_collection(coll=session_collection) + + p1 = product_factory() + p2 = product_factory() + + for p in [p1, p2]: + u = user_factory(product=p) + for i in range(50): + s = session_factory( + user=u, + wall_count=2, + wall_req_cpi=Decimal("1.00"), + started=start + timedelta(minutes=i, seconds=1), + ) + + wall_collection.initial_load(client=None, sync=True) + session_collection.initial_load(client=None, sync=True) + + enriched_wall_merge.build( + client=client_no_amm, + wall_coll=wall_collection, + session_coll=session_collection, + pg_config=thl_web_rr, + ) + + df = enriched_wall_merge.to_admin_response( + rr=event_report_request, client=client_no_amm + ) + + assert isinstance(df, pd.DataFrame) + assert not df.empty + # assert len(df) == 1 + # assert user.product_id == df.reset_index().loc[0, "index1"] + assert df.index.get_level_values(1).nunique() == 2 diff --git a/tests/incite/mergers/foundations/test_user_id_product.py b/tests/incite/mergers/foundations/test_user_id_product.py new file mode 100644 index 0000000..f96bfb4 --- /dev/null +++ b/tests/incite/mergers/foundations/test_user_id_product.py @@ -0,0 +1,73 @@ +from datetime import timedelta, datetime, timezone +from itertools import product + +import pandas as pd +import pytest + +# noinspection PyUnresolvedReferences +from distributed.utils_test import ( + gen_cluster, + client_no_amm, + loop, + loop_in_thread, + cleanup, + cluster_fixture, + client, +) + +from generalresearch.incite.mergers.foundations.user_id_product import ( + UserIdProductMergeItem, +) +from test_utils.incite.mergers.conftest import user_id_product_merge + + +@pytest.mark.parametrize( + argnames="offset, duration, start", + argvalues=list( + product( + ["12h", "3D"], + [timedelta(days=5)], + [ + (datetime.now(tz=timezone.utc) - timedelta(days=35)).replace( + microsecond=0 + ) + ], + ) + ), +) +class TestUserIDProduct: + + @pytest.mark.skip + def test_base(self, client_no_amm, user_id_product_merge): + ddf = user_id_product_merge.ddf() + df = client_no_amm.compute(collections=ddf, sync=True) + assert isinstance(df, pd.DataFrame) + assert not df.empty + + @pytest.mark.skip + def test_base_item(self, client_no_amm, user_id_product_merge, user_collection): + assert len(user_id_product_merge.items) == 1 + + for item in user_id_product_merge.items: + assert isinstance(item, UserIdProductMergeItem) + + path = item.path + + try: + modified_time1 = path.stat().st_mtime + except (Exception,): + modified_time1 = 0 + + user_id_product_merge.build(client=client_no_amm, user_coll=user_collection) + modified_time2 = path.stat().st_mtime + + assert modified_time2 > modified_time1 + + @pytest.mark.skip + def test_read(self, client_no_amm, user_id_product_merge): + users_ddf = user_id_product_merge.ddf() + df = client_no_amm.compute(collections=users_ddf, sync=True) + + assert isinstance(df, pd.DataFrame) + assert len(df.columns) == 1 + assert str(df.product_id.dtype) == "category" diff --git a/tests/incite/mergers/test_merge_collection.py b/tests/incite/mergers/test_merge_collection.py new file mode 100644 index 0000000..692cac3 --- /dev/null +++ b/tests/incite/mergers/test_merge_collection.py @@ -0,0 +1,102 @@ +from datetime import datetime, timezone, timedelta +from itertools import product + +import pandas as pd +import pytest +from pandera import DataFrameSchema + +from generalresearch.incite.mergers import ( + MergeCollection, + MergeType, +) +from test_utils.incite.conftest import mnt_filepath + +merge_types = list(e for e in MergeType if e != MergeType.TEST) + + +@pytest.mark.parametrize( + argnames="merge_type, offset, duration, start", + argvalues=list( + product( + merge_types, + ["5min", "6h", "14D"], + [timedelta(days=30)], + [ + (datetime.now(tz=timezone.utc) - timedelta(days=35)).replace( + microsecond=0 + ) + ], + ) + ), +) +class TestMergeCollection: + + def test_init(self, mnt_filepath, merge_type, offset, duration, start): + with pytest.raises(expected_exception=ValueError) as cm: + MergeCollection(archive_path=mnt_filepath.data_src) + assert "Must explicitly provide a merge_type" in str(cm.value) + + instance = MergeCollection( + merge_type=merge_type, + archive_path=mnt_filepath.archive_path(enum_type=merge_type), + ) + assert instance.merge_type == merge_type + + def test_items(self, mnt_filepath, merge_type, offset, duration, start): + instance = MergeCollection( + merge_type=merge_type, + offset=offset, + start=start, + finished=start + duration, + archive_path=mnt_filepath.archive_path(enum_type=merge_type), + ) + + assert len(instance.interval_range) == len(instance.items) + + def test_progress(self, mnt_filepath, merge_type, offset, duration, start): + instance = MergeCollection( + merge_type=merge_type, + offset=offset, + start=start, + finished=start + duration, + archive_path=mnt_filepath.archive_path(enum_type=merge_type), + ) + + assert isinstance(instance.progress, pd.DataFrame) + assert instance.progress.shape[0] > 0 + assert instance.progress.shape[1] == 7 + assert instance.progress["group_by"].isnull().all() + + def test_schema(self, mnt_filepath, merge_type, offset, duration, start): + instance = MergeCollection( + merge_type=merge_type, + archive_path=mnt_filepath.archive_path(enum_type=merge_type), + ) + + assert isinstance(instance._schema, DataFrameSchema) + + def test_load(self, mnt_filepath, merge_type, offset, duration, start): + instance = MergeCollection( + merge_type=merge_type, + start=start, + finished=start + duration, + offset=offset, + archive_path=mnt_filepath.archive_path(enum_type=merge_type), + ) + + # Confirm that there are no archives available yet + assert instance.progress.has_archive.eq(False).all() + + def test_get_items(self, mnt_filepath, merge_type, offset, duration, start): + instance = MergeCollection( + start=start, + finished=start + duration, + offset=offset, + merge_type=merge_type, + archive_path=mnt_filepath.archive_path(enum_type=merge_type), + ) + + # with pytest.raises(expected_exception=ResourceWarning) as cm: + res = instance.get_items_last365() + # assert "has missing archives", str(cm.value) + assert len(res) == len(instance.items) diff --git a/tests/incite/mergers/test_merge_collection_item.py b/tests/incite/mergers/test_merge_collection_item.py new file mode 100644 index 0000000..96f8789 --- /dev/null +++ b/tests/incite/mergers/test_merge_collection_item.py @@ -0,0 +1,66 @@ +from datetime import datetime, timezone, timedelta +from itertools import product +from pathlib import PurePath + +import pytest + +from generalresearch.incite.mergers import MergeCollectionItem, MergeType +from generalresearch.incite.mergers.foundations.enriched_session import ( + EnrichedSessionMerge, +) +from generalresearch.incite.mergers.foundations.enriched_wall import ( + EnrichedWallMerge, +) +from test_utils.incite.mergers.conftest import merge_collection + + +@pytest.mark.parametrize( + argnames="merge_type, offset, duration", + argvalues=list( + product( + [MergeType.ENRICHED_SESSION, MergeType.ENRICHED_WALL], + ["1h"], + [timedelta(days=1)], + ) + ), +) +class TestMergeCollectionItem: + + def test_file_naming(self, merge_collection, offset, duration, start): + assert len(merge_collection.items) == 25 + + items: list[MergeCollectionItem] = merge_collection.items + + for i in items: + i: MergeCollectionItem + + assert isinstance(i.path, PurePath) + assert i.path.name == i.filename + + assert i._collection.merge_type.name.lower() in i.filename + assert i._collection.offset in i.filename + assert i.start.strftime("%Y-%m-%d-%H-%M-%S") in i.filename + + def test_archives(self, merge_collection, offset, duration, start): + assert len(merge_collection.items) == 25 + + for i in merge_collection.items: + assert not i.has_archive() + assert not i.has_empty() + assert not i.is_empty() + assert not i.has_partial_archive() + assert i.has_archive() == i.path_exists(generic_path=i.path) + + res = set([i.should_archive() for i in merge_collection.items]) + assert len(res) == 1 + + def test_item_to_archive(self, merge_collection, offset, duration, start): + for item in merge_collection.items: + item: MergeCollectionItem + assert not item.has_archive() + + # TODO: setup build methods + # ddf = self.build + # saved = instance.to_archive(ddf=ddf) + # self.assertTrue(saved) + # self.assertTrue(instance.has_archive()) diff --git a/tests/incite/mergers/test_pop_ledger.py b/tests/incite/mergers/test_pop_ledger.py new file mode 100644 index 0000000..6f96108 --- /dev/null +++ b/tests/incite/mergers/test_pop_ledger.py @@ -0,0 +1,307 @@ +from datetime import timedelta, datetime, timezone +from itertools import product as iter_product +from typing import Optional + +import pandas as pd +import pytest +from distributed.utils_test import client_no_amm + +from generalresearch.incite.schemas.mergers.pop_ledger import ( + numerical_col_names, +) +from test_utils.incite.collections.conftest import ledger_collection +from test_utils.incite.conftest import mnt_filepath, incite_item_factory +from test_utils.incite.mergers.conftest import pop_ledger_merge +from test_utils.managers.ledger.conftest import create_main_accounts + + +@pytest.mark.parametrize( + argnames="offset, duration", + argvalues=list( + iter_product( + ["12h", "3D"], + [timedelta(days=4)], + ) + ), +) +class TestMergePOPLedger: + + @pytest.fixture + def start(self) -> "datetime": + return datetime(year=2020, month=3, day=14, tzinfo=timezone.utc) + + @pytest.fixture + def duration(self) -> Optional["timedelta"]: + return timedelta(days=5) + + def test_base( + self, + client_no_amm, + ledger_collection, + pop_ledger_merge, + product, + user_factory, + create_main_accounts, + thl_lm, + delete_df_collection, + incite_item_factory, + delete_ledger_db, + ): + from generalresearch.models.thl.ledger import LedgerAccount + + u = user_factory(product=product, created=ledger_collection.start) + + # -- Build & Setup + delete_ledger_db() + create_main_accounts() + delete_df_collection(coll=ledger_collection) + # assert ledger_collection.start is None + # assert ledger_collection.offset is None + + for item in ledger_collection.items: + incite_item_factory(item=item, user=u) + item.initial_load() + + # Confirm any of the items are archived + assert ledger_collection.progress.has_archive.eq(True).all() + + pop_ledger_merge.build( + client=client_no_amm, + ledger_coll=ledger_collection, + ) + # assert pop_ledger_merge.progress.has_archive.eq(True).all() + + ddf = pop_ledger_merge.ddf() + df: pd.DataFrame = client_no_amm.compute(collections=ddf, sync=True) + + assert isinstance(df, pd.DataFrame) + assert not df.empty + + # -- + + user_wallet_account: LedgerAccount = thl_lm.get_account_or_create_user_wallet( + user=u + ) + cash_account: LedgerAccount = thl_lm.get_account_cash() + rev_account: LedgerAccount = thl_lm.get_account_task_complete_revenue() + + item_finishes = [i.finish for i in ledger_collection.items] + item_finishes.sort(reverse=True) + last_item_finish = item_finishes[0] + + # Pure SQL based lookups + cash_balance: int = thl_lm.get_account_balance(account=cash_account) + rev_balance: int = thl_lm.get_account_balance(account=rev_account) + assert cash_balance > rev_balance + + # (1) Test Cash Account + ddf = pop_ledger_merge.ddf( + columns=numerical_col_names, + filters=[ + ("account_id", "==", cash_account.uuid), + ("time_idx", ">=", ledger_collection.start), + ("time_idx", "<", last_item_finish), + ], + ) + df: pd.DataFrame = client_no_amm.compute(collections=ddf, sync=True) + assert df["mp_payment.CREDIT"].sum() == 0 + assert cash_balance > 0 + assert df["mp_payment.DEBIT"].sum() == cash_balance + + # (2) Test Revenue Account + ddf = pop_ledger_merge.ddf( + columns=numerical_col_names, + filters=[ + ("account_id", "==", rev_account.uuid), + ("time_idx", ">=", ledger_collection.start), + ("time_idx", "<", last_item_finish), + ], + ) + df: pd.DataFrame = client_no_amm.compute(collections=ddf, sync=True) + + assert rev_balance == 0 + assert df["bp_payment.CREDIT"].sum() == 0 + assert df["mp_payment.DEBIT"].sum() == 0 + assert df["mp_payment.CREDIT"].sum() > 0 + + # -- Cleanup + delete_ledger_db() + + def test_pydantic_init( + self, + client_no_amm, + ledger_collection, + pop_ledger_merge, + mnt_filepath, + product, + user_factory, + create_main_accounts, + offset, + duration, + start, + thl_lm, + incite_item_factory, + delete_df_collection, + delete_ledger_db, + session_collection, + ): + from generalresearch.models.thl.ledger import LedgerAccount + from generalresearch.models.thl.product import Product + from generalresearch.models.thl.finance import ProductBalances + + u = user_factory(product=product, created=session_collection.start) + + assert ledger_collection.finished is not None + assert isinstance(u.product, Product) + delete_ledger_db() + create_main_accounts(), + delete_df_collection(coll=ledger_collection) + + bp_account: LedgerAccount = thl_lm.get_account_or_create_bp_wallet( + product=u.product + ) + cash_account: LedgerAccount = thl_lm.get_account_cash() + rev_account: LedgerAccount = thl_lm.get_account_task_complete_revenue() + + for item in ledger_collection.items: + incite_item_factory(item=item, user=u) + item.initial_load(overwrite=True) + + pop_ledger_merge.build(client=client_no_amm, ledger_coll=ledger_collection) + + item_finishes = [i.finish for i in ledger_collection.items] + item_finishes.sort(reverse=True) + last_item_finish = item_finishes[0] + + # (1) Filter by the Product Account, this means no cash_account, or + # rev_account transactions will be present in here... + ddf = pop_ledger_merge.ddf( + columns=numerical_col_names + ["time_idx"], + filters=[ + ("account_id", "==", bp_account.uuid), + ("time_idx", ">=", ledger_collection.start), + ("time_idx", "<", last_item_finish), + ], + ) + df: pd.DataFrame = client_no_amm.compute(collections=ddf, sync=True) + df = df.set_index("time_idx") + assert not df.empty + + instance = ProductBalances.from_pandas(input_data=df.sum()) + assert instance.payout == instance.net == instance.bp_payment_credit + assert instance.available_balance < instance.net + assert instance.available_balance + instance.retainer == instance.net + assert instance.balance == thl_lm.get_account_balance(bp_account) + assert df["bp_payment.CREDIT"].sum() == thl_lm.get_account_balance(bp_account) + + # (2) Filter by the Cash Account + ddf = pop_ledger_merge.ddf( + columns=numerical_col_names + ["time_idx"], + filters=[ + ("account_id", "==", cash_account.uuid), + ("time_idx", ">=", ledger_collection.start), + ("time_idx", "<", last_item_finish), + ], + ) + df: pd.DataFrame = client_no_amm.compute(collections=ddf, sync=True) + + cash_balance: int = thl_lm.get_account_balance(account=cash_account) + assert df["bp_payment.CREDIT"].sum() == 0 + assert cash_balance > 0 + assert df["mp_payment.CREDIT"].sum() == 0 + assert df["mp_payment.DEBIT"].sum() == cash_balance + + # (2) Filter by the Revenue Account + ddf = pop_ledger_merge.ddf( + columns=numerical_col_names + ["time_idx"], + filters=[ + ("account_id", "==", rev_account.uuid), + ("time_idx", ">=", ledger_collection.start), + ("time_idx", "<", last_item_finish), + ], + ) + df: pd.DataFrame = client_no_amm.compute(collections=ddf, sync=True) + + rev_balance: int = thl_lm.get_account_balance(account=rev_account) + assert rev_balance == 0 + assert df["bp_payment.CREDIT"].sum() == 0 + assert df["mp_payment.DEBIT"].sum() == 0 + assert df["mp_payment.CREDIT"].sum() > 0 + + def test_resample( + self, + client_no_amm, + ledger_collection, + pop_ledger_merge, + mnt_filepath, + user_factory, + product, + create_main_accounts, + offset, + duration, + start, + thl_lm, + delete_df_collection, + incite_item_factory, + ): + from generalresearch.models.thl.user import User + + assert ledger_collection.finished is not None + delete_df_collection(coll=ledger_collection) + u1: User = user_factory(product=product) + + bp_account = thl_lm.get_account_or_create_bp_wallet(product=u1.product) + + for item in ledger_collection.items: + incite_item_factory(user=u1, item=item) + item.initial_load(overwrite=True) + + pop_ledger_merge.build(client=client_no_amm, ledger_coll=ledger_collection) + + item_finishes = [i.finish for i in ledger_collection.items] + item_finishes.sort(reverse=True) + last_item_finish = item_finishes[0] + + ddf = pop_ledger_merge.ddf( + columns=numerical_col_names + ["time_idx"], + filters=[ + ("account_id", "==", bp_account.uuid), + ("time_idx", ">=", ledger_collection.start), + ("time_idx", "<", last_item_finish), + ], + ) + df = client_no_amm.compute(collections=ddf, sync=True) + assert isinstance(df, pd.DataFrame) + assert isinstance(df.index, pd.Index) + assert not isinstance(df.index, pd.RangeIndex) + + # Now change the index so we can easily resample it + df = df.set_index("time_idx") + assert isinstance(df.index, pd.Index) + assert isinstance(df.index, pd.DatetimeIndex) + + bp_account_balance = thl_lm.get_account_balance(account=bp_account) + + # Initial sum + initial_sum = df.sum().sum() + # assert len(df) == 48 # msg="Original df should be 48 rows" + + # Original (1min) to 5min + df_5min = df.resample("5min").sum() + # assert len(df_5min) == 12 + assert initial_sum == df_5min.sum().sum() + + # 30min + df_30min = df.resample("30min").sum() + # assert len(df_30min) == 2 + assert initial_sum == df_30min.sum().sum() + + # 1hr + df_1hr = df.resample("1h").sum() + # assert len(df_1hr) == 1 + assert initial_sum == df_1hr.sum().sum() + + # 1 day + df_1day = df.resample("1d").sum() + # assert len(df_1day) == 1 + assert initial_sum == df_1day.sum().sum() diff --git a/tests/incite/mergers/test_ym_survey_merge.py b/tests/incite/mergers/test_ym_survey_merge.py new file mode 100644 index 0000000..4c2df6b --- /dev/null +++ b/tests/incite/mergers/test_ym_survey_merge.py @@ -0,0 +1,125 @@ +from datetime import timedelta, timezone, datetime +from itertools import product + +import pandas as pd +import pytest + +# noinspection PyUnresolvedReferences +from distributed.utils_test import ( + gen_cluster, + client_no_amm, + loop, + loop_in_thread, + cleanup, + cluster_fixture, + client, +) + +from test_utils.incite.collections.conftest import wall_collection, session_collection +from test_utils.incite.mergers.conftest import ( + enriched_session_merge, + ym_survey_wall_merge, +) + + +@pytest.mark.parametrize( + argnames="offset, duration, start", + argvalues=list( + product( + ["12h", "3D"], + [timedelta(days=30)], + [ + (datetime.now(tz=timezone.utc) - timedelta(days=35)).replace( + microsecond=0 + ) + ], + ) + ), +) +class TestYMSurveyMerge: + """We override start, not because it's needed on the YMSurveyWall merge, + which operates on a rolling 10-day window, but because we don't want + to mock data in the wall collection and enriched_session_merge from + the 1800s and then wonder why there is no data available in the past + 10 days in the database. + """ + + def test_base( + self, + client_no_amm, + user_factory, + product, + ym_survey_wall_merge, + wall_collection, + session_collection, + enriched_session_merge, + delete_df_collection, + incite_item_factory, + thl_web_rr, + ): + from generalresearch.models.thl.user import User + + delete_df_collection(coll=session_collection) + user: User = user_factory(product=product, created=session_collection.start) + + # -- Build & Setup + assert ym_survey_wall_merge.start is None + assert ym_survey_wall_merge.offset == "10D" + + for item in session_collection.items: + incite_item_factory(item=item, user=user) + item.initial_load() + for item in wall_collection.items: + item.initial_load() + + # Confirm any of the items are archived + assert session_collection.progress.has_archive.eq(True).all() + assert wall_collection.progress.has_archive.eq(True).all() + + enriched_session_merge.build( + client=client_no_amm, + session_coll=session_collection, + wall_coll=wall_collection, + pg_config=thl_web_rr, + ) + assert enriched_session_merge.progress.has_archive.eq(True).all() + + ddf = enriched_session_merge.ddf() + df: pd.DataFrame = client_no_amm.compute(collections=ddf, sync=True) + + assert isinstance(df, pd.DataFrame) + assert not df.empty + + # -- + + ym_survey_wall_merge.build( + client=client_no_amm, + wall_coll=wall_collection, + enriched_session=enriched_session_merge, + ) + assert ym_survey_wall_merge.progress.has_archive.eq(True).all() + + # -- + + ddf = ym_survey_wall_merge.ddf() + df: pd.DataFrame = client_no_amm.compute(collections=ddf, sync=True) + + assert isinstance(df, pd.DataFrame) + assert not df.empty + + # -- + assert df.product_id.nunique() == 1 + assert df.team_id.nunique() == 1 + assert df.source.nunique() > 1 + + started_min_ts = df.started.min() + started_max_ts = df.started.max() + + assert type(started_min_ts) is pd.Timestamp + assert type(started_max_ts) is pd.Timestamp + + started_min: datetime = datetime.fromisoformat(str(started_min_ts)) + started_max: datetime = datetime.fromisoformat(str(started_max_ts)) + + started_delta = started_max - started_min + assert started_delta >= timedelta(days=3) diff --git a/tests/incite/schemas/__init__.py b/tests/incite/schemas/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/tests/incite/schemas/__init__.py diff --git a/tests/incite/schemas/test_admin_responses.py b/tests/incite/schemas/test_admin_responses.py new file mode 100644 index 0000000..43aa399 --- /dev/null +++ b/tests/incite/schemas/test_admin_responses.py @@ -0,0 +1,239 @@ +from datetime import datetime, timezone, timedelta +from random import sample +from typing import List + +import numpy as np +import pandas as pd +import pytest + +from generalresearch.incite.schemas import empty_dataframe_from_schema +from generalresearch.incite.schemas.admin_responses import ( + AdminPOPSchema, + SIX_HOUR_SECONDS, +) +from generalresearch.locales import Localelator + + +class TestAdminPOPSchema: + schema_df = empty_dataframe_from_schema(AdminPOPSchema) + countries = list(Localelator().get_all_countries())[:5] + dates = [datetime(year=2024, month=1, day=i, tzinfo=None) for i in range(1, 10)] + + @classmethod + def assign_valid_vals(cls, df: pd.DataFrame) -> pd.DataFrame: + for c in df.columns: + check_attrs: dict = AdminPOPSchema.columns[c].checks[0].statistics + df[c] = np.random.randint( + check_attrs["min_value"], check_attrs["max_value"], df.shape[0] + ) + + return df + + def test_empty(self): + with pytest.raises(Exception): + AdminPOPSchema.validate(pd.DataFrame()) + + def test_new_empty_df(self): + df = empty_dataframe_from_schema(AdminPOPSchema) + + assert isinstance(df, pd.DataFrame) + assert isinstance(df.index, pd.MultiIndex) + assert df.columns.size == len(AdminPOPSchema.columns) + + def test_valid(self): + # (1) Works with raw naive datetime + dates = [ + datetime(year=2024, month=1, day=i, tzinfo=None).isoformat() + for i in range(1, 10) + ] + df = pd.DataFrame( + index=pd.MultiIndex.from_product( + iterables=[dates, self.countries], names=["index0", "index1"] + ), + columns=self.schema_df.columns, + ) + df = self.assign_valid_vals(df) + + df = AdminPOPSchema.validate(df) + assert isinstance(df, pd.DataFrame) + + # (2) Works with isoformat naive datetime + dates = [datetime(year=2024, month=1, day=i, tzinfo=None) for i in range(1, 10)] + df = pd.DataFrame( + index=pd.MultiIndex.from_product( + iterables=[dates, self.countries], names=["index0", "index1"] + ), + columns=self.schema_df.columns, + ) + df = self.assign_valid_vals(df) + + df = AdminPOPSchema.validate(df) + assert isinstance(df, pd.DataFrame) + + def test_index_tz_parser(self): + tz_dates = [ + datetime(year=2024, month=1, day=i, tzinfo=timezone.utc) + for i in range(1, 10) + ] + + df = pd.DataFrame( + index=pd.MultiIndex.from_product( + iterables=[tz_dates, self.countries], names=["index0", "index1"] + ), + columns=self.schema_df.columns, + ) + df = self.assign_valid_vals(df) + + # Initially, they're all set with a timezone + timestmaps: List[pd.Timestamp] = [i for i in df.index.get_level_values(0)] + assert all([ts.tz == timezone.utc for ts in timestmaps]) + + # After validation, the timezone is removed + df = AdminPOPSchema.validate(df) + timestmaps: List[pd.Timestamp] = [i for i in df.index.get_level_values(0)] + assert all([ts.tz is None for ts in timestmaps]) + + def test_index_tz_no_future_beyond_one_year(self): + now = datetime.now(tz=timezone.utc) + tz_dates = [now + timedelta(days=i * 365) for i in range(1, 10)] + + df = pd.DataFrame( + index=pd.MultiIndex.from_product( + iterables=[tz_dates, self.countries], names=["index0", "index1"] + ), + columns=self.schema_df.columns, + ) + df = self.assign_valid_vals(df) + + with pytest.raises(Exception) as cm: + AdminPOPSchema.validate(df) + + assert ( + "Index 'index0' failed element-wise validator " + "number 0: less_than(" in str(cm.value) + ) + + def test_index_only_str(self): + # --- float64 to str! --- + df = pd.DataFrame( + index=pd.MultiIndex.from_product( + iterables=[self.dates, np.random.rand(1, 10)[0]], + names=["index0", "index1"], + ), + columns=self.schema_df.columns, + ) + df = self.assign_valid_vals(df) + + vals = [i for i in df.index.get_level_values(1)] + assert all([isinstance(v, float) for v in vals]) + + df = AdminPOPSchema.validate(df, lazy=True) + + vals = [i for i in df.index.get_level_values(1)] + assert all([isinstance(v, str) for v in vals]) + + # --- int to str --- + + df = pd.DataFrame( + index=pd.MultiIndex.from_product( + iterables=[self.dates, sample(range(100), 20)], + names=["index0", "index1"], + ), + columns=self.schema_df.columns, + ) + df = self.assign_valid_vals(df) + + vals = [i for i in df.index.get_level_values(1)] + assert all([isinstance(v, int) for v in vals]) + + df = AdminPOPSchema.validate(df, lazy=True) + + vals = [i for i in df.index.get_level_values(1)] + assert all([isinstance(v, str) for v in vals]) + + # a = 1 + assert isinstance(df, pd.DataFrame) + + def test_invalid_parsing(self): + # (1) Timezones AND as strings will still parse correctly + tz_str_dates = [ + datetime( + year=2024, month=1, day=1, minute=i, tzinfo=timezone.utc + ).isoformat() + for i in range(1, 10) + ] + df = pd.DataFrame( + index=pd.MultiIndex.from_product( + iterables=[tz_str_dates, self.countries], + names=["index0", "index1"], + ), + columns=self.schema_df.columns, + ) + df = self.assign_valid_vals(df) + df = AdminPOPSchema.validate(df, lazy=True) + + assert isinstance(df, pd.DataFrame) + timestmaps: List[pd.Timestamp] = [i for i in df.index.get_level_values(0)] + assert all([ts.tz is None for ts in timestmaps]) + + # (2) Timezones are removed + dates = [ + datetime(year=2024, month=1, day=1, minute=i, tzinfo=timezone.utc) + for i in range(1, 10) + ] + df = pd.DataFrame( + index=pd.MultiIndex.from_product( + iterables=[dates, self.countries], names=["index0", "index1"] + ), + columns=self.schema_df.columns, + ) + df = self.assign_valid_vals(df) + + # Has tz before validation, and none after + timestmaps: List[pd.Timestamp] = [i for i in df.index.get_level_values(0)] + assert all([ts.tz is timezone.utc for ts in timestmaps]) + + df = AdminPOPSchema.validate(df, lazy=True) + + timestmaps: List[pd.Timestamp] = [i for i in df.index.get_level_values(0)] + assert all([ts.tz is None for ts in timestmaps]) + + def test_clipping(self): + df = pd.DataFrame( + index=pd.MultiIndex.from_product( + iterables=[self.dates, self.countries], + names=["index0", "index1"], + ), + columns=self.schema_df.columns, + ) + df = self.assign_valid_vals(df) + df = AdminPOPSchema.validate(df) + assert df.elapsed_avg.max() < SIX_HOUR_SECONDS + + # Now that we know it's valid, break the elapsed avg + df["elapsed_avg"] = np.random.randint( + SIX_HOUR_SECONDS, SIX_HOUR_SECONDS + 10_000, df.shape[0] + ) + assert df.elapsed_avg.max() > SIX_HOUR_SECONDS + + # Confirm it doesn't fail if the values are greater, and that + # all the values are clipped to the max + df = AdminPOPSchema.validate(df) + assert df.elapsed_avg.eq(SIX_HOUR_SECONDS).all() + + def test_rounding(self): + df = pd.DataFrame( + index=pd.MultiIndex.from_product( + iterables=[self.dates, self.countries], + names=["index0", "index1"], + ), + columns=self.schema_df.columns, + ) + df = self.assign_valid_vals(df) + + df["payout_avg"] = 2.123456789900002 + + assert df.payout_avg.sum() == 95.5555555455001 + + df = AdminPOPSchema.validate(df) + assert df.payout_avg.sum() == 95.40000000000003 diff --git a/tests/incite/schemas/test_thl_web.py b/tests/incite/schemas/test_thl_web.py new file mode 100644 index 0000000..7f4434b --- /dev/null +++ b/tests/incite/schemas/test_thl_web.py @@ -0,0 +1,70 @@ +import pandas as pd +import pytest +from pandera.errors import SchemaError + + +class TestWallSchema: + + def test_empty(self): + from generalresearch.incite.schemas.thl_web import THLWallSchema + + with pytest.raises(SchemaError): + THLWallSchema.validate(pd.DataFrame()) + + def test_index_missing(self): + from generalresearch.incite.schemas.thl_web import THLWallSchema + + df = pd.DataFrame(columns=THLWallSchema.columns.keys()) + + with pytest.raises(SchemaError) as cm: + THLWallSchema.validate(df) + + def test_no_rows(self): + from generalresearch.incite.schemas.thl_web import THLWallSchema + + df = pd.DataFrame(index=["uuid"], columns=THLWallSchema.columns.keys()) + + with pytest.raises(SchemaError) as cm: + THLWallSchema.validate(df) + + def test_new_empty_df(self): + from generalresearch.incite.schemas import empty_dataframe_from_schema + from generalresearch.incite.schemas.thl_web import THLWallSchema + + df = empty_dataframe_from_schema(THLWallSchema) + assert isinstance(df, pd.DataFrame) + assert df.columns.size == 20 + + +class TestSessionSchema: + + def test_empty(self): + from generalresearch.incite.schemas.thl_web import THLSessionSchema + + with pytest.raises(SchemaError): + THLSessionSchema.validate(pd.DataFrame()) + + def test_index_missing(self): + from generalresearch.incite.schemas.thl_web import THLSessionSchema + + df = pd.DataFrame(columns=THLSessionSchema.columns.keys()) + df.set_index("uuid", inplace=True) + + with pytest.raises(SchemaError) as cm: + THLSessionSchema.validate(df) + + def test_no_rows(self): + from generalresearch.incite.schemas.thl_web import THLSessionSchema + + df = pd.DataFrame(index=["id"], columns=THLSessionSchema.columns.keys()) + + with pytest.raises(SchemaError) as cm: + THLSessionSchema.validate(df) + + def test_new_empty_df(self): + from generalresearch.incite.schemas import empty_dataframe_from_schema + from generalresearch.incite.schemas.thl_web import THLSessionSchema + + df = empty_dataframe_from_schema(THLSessionSchema) + assert isinstance(df, pd.DataFrame) + assert df.columns.size == 21 diff --git a/tests/incite/test_collection_base.py b/tests/incite/test_collection_base.py new file mode 100644 index 0000000..497e5ab --- /dev/null +++ b/tests/incite/test_collection_base.py @@ -0,0 +1,318 @@ +from datetime import datetime, timezone, timedelta +from os.path import exists as pexists, join as pjoin +from pathlib import Path +from uuid import uuid4 + +import numpy as np +import pandas as pd +import pytest +from _pytest._code.code import ExceptionInfo + +from generalresearch.incite.base import CollectionBase +from test_utils.incite.conftest import mnt_filepath + +AGO_15min = (datetime.now(tz=timezone.utc) - timedelta(minutes=15)).replace( + microsecond=0 +) +AGO_1HR = (datetime.now(tz=timezone.utc) - timedelta(hours=1)).replace(microsecond=0) +AGO_2HR = (datetime.now(tz=timezone.utc) - timedelta(hours=2)).replace(microsecond=0) + + +class TestCollectionBase: + def test_init(self, mnt_filepath): + instance = CollectionBase(archive_path=mnt_filepath.data_src) + assert instance.df.empty is True + + def test_init_df(self, mnt_filepath): + # Only an empty pd.DataFrame can ever be provided + instance = CollectionBase( + df=pd.DataFrame({}), archive_path=mnt_filepath.data_src + ) + assert isinstance(instance.df, pd.DataFrame) + + with pytest.raises(expected_exception=ValueError) as cm: + cm: ExceptionInfo + CollectionBase( + df=pd.DataFrame(columns=[0, 1, 2]), archive_path=mnt_filepath.data_src + ) + assert "Do not provide a pd.DataFrame" in str(cm.value) + + with pytest.raises(expected_exception=ValueError) as cm: + cm: ExceptionInfo + CollectionBase( + df=pd.DataFrame(np.random.randint(100, size=(1000, 1)), columns=["A"]), + archive_path=mnt_filepath.data_src, + ) + assert "Do not provide a pd.DataFrame" in str(cm.value) + + def test_init_start(self, mnt_filepath): + with pytest.raises(expected_exception=ValueError) as cm: + cm: ExceptionInfo + CollectionBase( + start=datetime.now(tz=timezone.utc) - timedelta(days=10), + archive_path=mnt_filepath.data_src, + ) + assert "Collection.start must not have microseconds" in str(cm.value) + + with pytest.raises(expected_exception=ValueError) as cm: + cm: ExceptionInfo + tz = timezone(timedelta(hours=-5), "EST") + + CollectionBase( + start=datetime(year=2000, month=1, day=1, tzinfo=tz), + archive_path=mnt_filepath.data_src, + ) + assert "Timezone is not UTC" in str(cm.value) + + instance = CollectionBase(archive_path=mnt_filepath.data_src) + assert instance.start == datetime( + year=2018, month=1, day=1, tzinfo=timezone.utc + ) + + with pytest.raises(expected_exception=ValueError) as cm: + cm: ExceptionInfo + CollectionBase( + start=AGO_2HR, offset="3h", archive_path=mnt_filepath.data_src + ) + assert "Offset must be equal to, or smaller the start timestamp" in str( + cm.value + ) + + def test_init_archive_path(self, mnt_filepath): + """DirectoryPath is apparently smart enough to confirm that the + directory path exists. + """ + + # (1) Basic, confirm an existing path works + instance = CollectionBase(archive_path=mnt_filepath.data_src) + assert instance.archive_path == mnt_filepath.data_src + + # (2) It can't point to a file + file_path = Path(pjoin(mnt_filepath.data_src, f"{uuid4().hex}.zip")) + assert not pexists(file_path) + with pytest.raises(expected_exception=ValueError) as cm: + cm: ExceptionInfo + CollectionBase(archive_path=file_path) + assert "Path does not point to a directory" in str(cm.value) + + # (3) It doesn't create the directory if it doesn't exist + new_path = Path(pjoin(mnt_filepath.data_src, f"{uuid4().hex}/")) + assert not pexists(new_path) + with pytest.raises(expected_exception=ValueError) as cm: + cm: ExceptionInfo + CollectionBase(archive_path=new_path) + assert "Path does not point to a directory" in str(cm.value) + + def test_init_offset(self, mnt_filepath): + with pytest.raises(expected_exception=ValueError) as cm: + cm: ExceptionInfo + CollectionBase(offset="1:X", archive_path=mnt_filepath.data_src) + assert "Invalid offset alias provided. Please review:" in str(cm.value) + + with pytest.raises(expected_exception=ValueError) as cm: + cm: ExceptionInfo + CollectionBase(offset=f"59sec", archive_path=mnt_filepath.data_src) + assert "Must be equal to, or longer than 1 min" in str(cm.value) + + with pytest.raises(expected_exception=ValueError) as cm: + cm: ExceptionInfo + CollectionBase(offset=f"{365 * 101}d", archive_path=mnt_filepath.data_src) + assert "String should have at most 5 characters" in str(cm.value) + + +class TestCollectionBaseProperties: + + def test_items(self, mnt_filepath): + with pytest.raises(expected_exception=NotImplementedError) as cm: + cm: ExceptionInfo + instance = CollectionBase(archive_path=mnt_filepath.data_src) + x = instance.items + assert "Must override" in str(cm.value) + + def test_interval_range(self, mnt_filepath): + instance = CollectionBase(archive_path=mnt_filepath.data_src) + # Private method requires the end parameter + with pytest.raises(expected_exception=AssertionError) as cm: + cm: ExceptionInfo + instance._interval_range(end=None) + assert "an end value must be provided" in str(cm.value) + + # End param must be same as started (which forces utc) + tz = timezone(timedelta(hours=-5), "EST") + with pytest.raises(expected_exception=AssertionError) as cm: + cm: ExceptionInfo + instance._interval_range(end=datetime.now(tz=tz)) + assert "Timezones must match" in str(cm.value) + + res = instance._interval_range(end=datetime.now(tz=timezone.utc)) + assert isinstance(res, pd.IntervalIndex) + assert res.closed_left + assert res.is_non_overlapping_monotonic + assert res.is_monotonic_increasing + assert res.is_unique + + def test_interval_range2(self, mnt_filepath): + instance = CollectionBase(archive_path=mnt_filepath.data_src) + assert isinstance(instance.interval_range, list) + + # 1 hrs ago has 2 x 30min + the future 30min + OFFSET = "30min" + instance = CollectionBase( + start=AGO_1HR, offset=OFFSET, archive_path=mnt_filepath.data_src + ) + assert len(instance.interval_range) == 3 + assert instance.interval_range[0][0] == AGO_1HR + + # 1 hrs ago has 1 x 60min + the future 60min + OFFSET = "60min" + instance = CollectionBase( + start=AGO_1HR, offset=OFFSET, archive_path=mnt_filepath.data_src + ) + assert len(instance.interval_range) == 2 + + def test_progress(self, mnt_filepath): + with pytest.raises(expected_exception=NotImplementedError) as cm: + cm: ExceptionInfo + instance = CollectionBase( + start=AGO_15min, offset="3min", archive_path=mnt_filepath.data_src + ) + x = instance.progress + assert "Must override" in str(cm.value) + + def test_progress2(self, mnt_filepath): + instance = CollectionBase( + start=AGO_2HR, + offset="15min", + archive_path=mnt_filepath.data_src, + ) + assert instance.df.empty + + with pytest.raises(expected_exception=NotImplementedError) as cm: + df = instance.progress + assert "Must override" in str(cm.value) + + def test_items2(self, mnt_filepath): + """There can't be a test for this because the Items need a path whic + isn't possible in the generic form + """ + instance = CollectionBase( + start=AGO_1HR, offset="5min", archive_path=mnt_filepath.data_src + ) + + with pytest.raises(expected_exception=NotImplementedError) as cm: + cm: ExceptionInfo + items = instance.items + assert "Must override" in str(cm.value) + + # item = items[-3] + # ddf = instance.ddf(items=[item], include_partial=True, force_rr_latest=False) + # df = item.validate_ddf(ddf=ddf) + # assert isinstance(df, pd.DataFrame) + # assert len(df.columns) == 16 + # assert str(df.product_id.dtype) == "object" + # assert str(ddf.product_id.dtype) == "string" + + def test_items3(self, mnt_filepath): + instance = CollectionBase( + start=AGO_2HR, + offset="15min", + archive_path=mnt_filepath.data_src, + ) + with pytest.raises(expected_exception=NotImplementedError) as cm: + item = instance.items[0] + assert "Must override" in str(cm.value) + + +class TestCollectionBaseMethodsCleanup: + def test_fetch_force_rr_latest(self, mnt_filepath): + coll = CollectionBase(archive_path=mnt_filepath.data_src) + + with pytest.raises(expected_exception=Exception) as cm: + cm: ExceptionInfo + coll.fetch_force_rr_latest(sources=[]) + assert "Must override" in str(cm.value) + + def test_fetch_all_paths(self, mnt_filepath): + coll = CollectionBase(archive_path=mnt_filepath.data_src) + + with pytest.raises(expected_exception=NotImplementedError) as cm: + cm: ExceptionInfo + coll.fetch_all_paths( + items=None, force_rr_latest=False, include_partial=False + ) + assert "Must override" in str(cm.value) + + +class TestCollectionBaseMethodsCleanup: + @pytest.mark.skip + def test_cleanup_partials(self, mnt_filepath): + instance = CollectionBase(archive_path=mnt_filepath.data_src) + assert instance.cleanup_partials() is None # it doesn't return anything + + def test_clear_tmp_archives(self, mnt_filepath): + instance = CollectionBase(archive_path=mnt_filepath.data_src) + assert instance.clear_tmp_archives() is None # it doesn't return anything + + @pytest.mark.skip + def test_clear_corrupt_archives(self, mnt_filepath): + """TODO: expand this so it actually has corrupt archives that we + check to see if they're removed + """ + instance = CollectionBase(archive_path=mnt_filepath.data_src) + assert instance.clear_corrupt_archives() is None # it doesn't return anything + + @pytest.mark.skip + def test_rebuild_symlinks(self, mnt_filepath): + instance = CollectionBase(archive_path=mnt_filepath.data_src) + assert instance.rebuild_symlinks() is None + + +class TestCollectionBaseMethodsSourceTiming: + + def test_get_item(self, mnt_filepath): + instance = CollectionBase(archive_path=mnt_filepath.data_src) + i = pd.Interval(left=1, right=2, closed="left") + + with pytest.raises(expected_exception=NotImplementedError) as cm: + instance.get_item(interval=i) + assert "Must override" in str(cm.value) + + def test_get_item_start(self, mnt_filepath): + instance = CollectionBase(archive_path=mnt_filepath.data_src) + + dt = datetime.now(tz=timezone.utc) + start = pd.Timestamp(dt) + + with pytest.raises(expected_exception=NotImplementedError) as cm: + instance.get_item_start(start=start) + assert "Must override" in str(cm.value) + + def test_get_items(self, mnt_filepath): + instance = CollectionBase(archive_path=mnt_filepath.data_src) + + dt = datetime.now(tz=timezone.utc) + + with pytest.raises(expected_exception=NotImplementedError) as cm: + instance.get_items(since=dt) + assert "Must override" in str(cm.value) + + def test_get_items_from_year(self, mnt_filepath): + instance = CollectionBase(archive_path=mnt_filepath.data_src) + + with pytest.raises(expected_exception=NotImplementedError) as cm: + instance.get_items_from_year(year=2020) + assert "Must override" in str(cm.value) + + def test_get_items_last90(self, mnt_filepath): + instance = CollectionBase(archive_path=mnt_filepath.data_src) + + with pytest.raises(expected_exception=NotImplementedError) as cm: + instance.get_items_last90() + assert "Must override" in str(cm.value) + + def test_get_items_last365(self, mnt_filepath): + instance = CollectionBase(archive_path=mnt_filepath.data_src) + + with pytest.raises(expected_exception=NotImplementedError) as cm: + instance.get_items_last365() + assert "Must override" in str(cm.value) diff --git a/tests/incite/test_collection_base_item.py b/tests/incite/test_collection_base_item.py new file mode 100644 index 0000000..e5d1d02 --- /dev/null +++ b/tests/incite/test_collection_base_item.py @@ -0,0 +1,223 @@ +from datetime import datetime, timezone +from os.path import join as pjoin +from pathlib import Path +from uuid import uuid4 + +import dask.dataframe as dd +import pandas as pd +import pytest +from pydantic import ValidationError + +from generalresearch.incite.base import CollectionItemBase + + +class TestCollectionItemBase: + def test_init(self): + dt = datetime.now(tz=timezone.utc).replace(microsecond=0) + + instance = CollectionItemBase() + instance2 = CollectionItemBase(start=dt) + + assert isinstance(instance, CollectionItemBase) + assert isinstance(instance2, CollectionItemBase) + + assert instance.start.second == instance2.start.second + assert 0 == instance.start.microsecond == instance2.start.microsecond + + def test_init_start(self): + dt = datetime.now(tz=timezone.utc) + + with pytest.raises(expected_exception=ValidationError) as cm: + CollectionItemBase(start=dt) + + assert "CollectionItem.start must not have microsecond precision" in str( + cm.value + ) + + +class TestCollectionItemBaseProperties: + + def test_finish(self): + instance = CollectionItemBase() + + with pytest.raises(expected_exception=AttributeError) as cm: + res = instance.finish + + def test_interval(self): + instance = CollectionItemBase() + + with pytest.raises(expected_exception=AttributeError) as cm: + res = instance.interval + + def test_filename(self): + instance = CollectionItemBase() + + with pytest.raises(expected_exception=NotImplementedError) as cm: + res = instance.filename + + assert "Do not use CollectionItemBase directly" in str(cm.value) + + def test_partial_filename(self): + instance = CollectionItemBase() + + with pytest.raises(expected_exception=NotImplementedError) as cm: + res = instance.filename + + assert "Do not use CollectionItemBase directly" in str(cm.value) + + def test_empty_filename(self): + instance = CollectionItemBase() + + with pytest.raises(expected_exception=NotImplementedError) as cm: + res = instance.filename + + assert "Do not use CollectionItemBase directly" in str(cm.value) + + def test_path(self): + instance = CollectionItemBase() + + with pytest.raises(expected_exception=AttributeError) as cm: + res = instance.path + + def test_partial_path(self): + instance = CollectionItemBase() + + with pytest.raises(expected_exception=AttributeError) as cm: + res = instance.partial_path + + def test_empty_path(self): + instance = CollectionItemBase() + + with pytest.raises(expected_exception=AttributeError) as cm: + res = instance.empty_path + + +class TestCollectionItemBaseMethods: + + @pytest.mark.skip + def test_next_numbered_path(self): + pass + + @pytest.mark.skip + def test_search_highest_numbered_path(self): + pass + + def test_tmp_filename(self): + instance = CollectionItemBase() + + with pytest.raises(expected_exception=NotImplementedError) as cm: + res = instance.tmp_filename() + assert "Do not use CollectionItemBase directly" in str(cm.value) + + def test_tmp_path(self): + instance = CollectionItemBase() + + with pytest.raises(expected_exception=AttributeError) as cm: + res = instance.tmp_path() + + def test_is_empty(self): + instance = CollectionItemBase() + + with pytest.raises(expected_exception=AttributeError) as cm: + res = instance.is_empty() + + def test_has_empty(self): + instance = CollectionItemBase() + + with pytest.raises(expected_exception=AttributeError) as cm: + res = instance.has_empty() + + def test_has_partial_archive(self): + instance = CollectionItemBase() + + with pytest.raises(expected_exception=AttributeError) as cm: + res = instance.has_partial_archive() + + @pytest.mark.parametrize("include_empty", [True, False]) + def test_has_archive(self, include_empty): + instance = CollectionItemBase() + + with pytest.raises(expected_exception=AttributeError) as cm: + res = instance.has_archive(include_empty=include_empty) + + def test_delete_archive_file(self, mnt_filepath): + path1 = Path(pjoin(mnt_filepath.data_src, f"{uuid4().hex}.zip")) + + # Confirm it doesn't exist, and that delete_archive() doesn't throw + # an error when trying to delete a non-existent file or folder + assert not path1.exists() + CollectionItemBase.delete_archive(generic_path=path1) + # TODO: LOG.warning(f"tried removing non-existent file: {generic_path}") + + # Create it, confirm it exists, delete it, and confirm it doesn't exist + path1.touch() + assert path1.exists() + CollectionItemBase.delete_archive(generic_path=path1) + assert not path1.exists() + + def test_delete_archive_dir(self, mnt_filepath): + path1 = Path(pjoin(mnt_filepath.data_src, f"{uuid4().hex}")) + + # Confirm it doesn't exist, and that delete_archive() doesn't throw + # an error when trying to delete a non-existent file or folder + assert not path1.exists() + CollectionItemBase.delete_archive(generic_path=path1) + # TODO: LOG.warning(f"tried removing non-existent file: {generic_path}") + + # Create it, confirm it exists, delete it, and confirm it doesn't exist + path1.mkdir() + assert path1.exists() + assert path1.is_dir() + CollectionItemBase.delete_archive(generic_path=path1) + assert not path1.exists() + + def test_should_archive(self): + instance = CollectionItemBase() + + with pytest.raises(expected_exception=AttributeError) as cm: + res = instance.should_archive() + + def test_set_empty(self): + instance = CollectionItemBase() + + with pytest.raises(expected_exception=AttributeError) as cm: + res = instance.set_empty() + + def test_valid_archive(self): + instance = CollectionItemBase() + + with pytest.raises(expected_exception=AttributeError) as cm: + res = instance.valid_archive(generic_path=None, sample=None) + + +class TestCollectionItemBaseMethodsORM: + + @pytest.mark.skip + def test_from_archive(self): + pass + + @pytest.mark.parametrize("is_partial", [True, False]) + def test_to_archive(self, is_partial): + instance = CollectionItemBase() + + with pytest.raises(expected_exception=NotImplementedError) as cm: + res = instance.to_archive( + ddf=dd.from_pandas(data=pd.DataFrame()), is_partial=is_partial + ) + assert "Must override" in str(cm.value) + + @pytest.mark.skip + def test__to_dict(self): + pass + + @pytest.mark.skip + def test_delete_partial(self): + pass + + @pytest.mark.skip + def test_cleanup_partials(self): + pass + + @pytest.mark.skip + def test_delete_dangling_partials(self): + pass diff --git a/tests/incite/test_grl_flow.py b/tests/incite/test_grl_flow.py new file mode 100644 index 0000000..c632f9a --- /dev/null +++ b/tests/incite/test_grl_flow.py @@ -0,0 +1,23 @@ +class TestGRLFlow: + + def test_init(self, mnt_filepath, thl_web_rr): + from generalresearch.incite.defaults import ( + ledger_df_collection, + task_df_collection, + pop_ledger as plm, + ) + + from generalresearch.incite.collections.thl_web import ( + LedgerDFCollection, + TaskAdjustmentDFCollection, + ) + from generalresearch.incite.mergers.pop_ledger import PopLedgerMerge + + ledger_df = ledger_df_collection(ds=mnt_filepath, pg_config=thl_web_rr) + assert isinstance(ledger_df, LedgerDFCollection) + + task_df = task_df_collection(ds=mnt_filepath, pg_config=thl_web_rr) + assert isinstance(task_df, TaskAdjustmentDFCollection) + + pop_ledger = plm(ds=mnt_filepath) + assert isinstance(pop_ledger, PopLedgerMerge) diff --git a/tests/incite/test_interval_idx.py b/tests/incite/test_interval_idx.py new file mode 100644 index 0000000..ea2bced --- /dev/null +++ b/tests/incite/test_interval_idx.py @@ -0,0 +1,23 @@ +import pandas as pd +from datetime import datetime, timezone, timedelta + + +class TestIntervalIndex: + + def test_init(self): + start = datetime(year=2000, month=1, day=1) + end = datetime(year=2000, month=1, day=10) + + iv_r: pd.IntervalIndex = pd.interval_range( + start=start, end=end, freq="1d", closed="left" + ) + assert isinstance(iv_r, pd.IntervalIndex) + assert len(iv_r.to_list()) == 9 + + # If the offset is longer than the end - start it will not + # error. It will simply have 0 rows. + iv_r: pd.IntervalIndex = pd.interval_range( + start=start, end=end, freq="30d", closed="left" + ) + assert isinstance(iv_r, pd.IntervalIndex) + assert len(iv_r.to_list()) == 0 |