Merge branch 'master' into dev-greg

8 files changed, 367 insertions, 266 deletions

diff --git a/tests/incite/mergers/foundations/test_enriched_session.py b/tests/incite/mergers/foundations/test_enriched_session.py
index 47f243e..71b2442 100644
--- a/tests/incite/mergers/foundations/test_enriched_session.py
+++ b/tests/incite/mergers/foundations/test_enriched_session.py
@@ -1,20 +1,35 @@
-from datetime import datetime, timedelta, timezone
+from __future__ import annotations
+
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
 from decimal import Decimal
 from itertools import product
-from typing import Optional
+from typing import TYPE_CHECKING
 
 import dask.dataframe as dd
 import pandas as pd
 import pytest
+from dask.distributed import Client as DaskClient
 
 from generalresearch.incite.schemas.admin_responses import (
     AdminPOPSessionSchema,
 )
-from generalresearch.pg_helper import PostgresConfig
-from test_utils.incite.collections.conftest import (
-    session_collection,
-    wall_collection,
-)
+
+if TYPE_CHECKING:
+    from generalresearch.incite.collections.thl_web import (
+        SessionDFCollection,
+        WallDFCollection,
+    )
+    from generalresearch.incite.mergers.foundations.enriched_session import (
+        EnrichedSessionMerge,
+    )
+    from generalresearch.models.admin.request import (
+        ReportRequest,
+    )
+    from generalresearch.models.thl.product import Product
+    from generalresearch.models.thl.session import Session
+    from generalresearch.models.thl.user import User
+    from generalresearch.pg_helper import PostgresConfig
 
 
 @pytest.mark.parametrize(
@@ -30,17 +45,16 @@ class TestEnrichedSession:
 
     def test_base(
         self,
-        client_no_amm,
-        product,
-        user_factory,
-        wall_collection,
-        session_collection,
-        enriched_session_merge,
+        client_no_amm: DaskClient,
+        product: Product,
+        user_factory: Callable[..., User],
+        wall_collection: WallDFCollection,
+        session_collection: SessionDFCollection,
+        enriched_session_merge: EnrichedSessionMerge,
         thl_web_rr: PostgresConfig,
-        delete_df_collection,
-        incite_item_factory,
+        delete_df_collection: Callable[..., None],
+        incite_item_factory: Callable[..., None],
     ):
-        from generalresearch.models.thl.user import User
 
         delete_df_collection(coll=session_collection)
 
@@ -77,31 +91,31 @@ class TestEnrichedSession:
 class TestEnrichedSessionAdmin:
 
     @pytest.fixture
-    def start(self) -> "datetime":
-        return datetime(year=2020, month=3, day=14, tzinfo=timezone.utc)
+    def start(self) -> datetime:
+        return datetime(year=2020, month=3, day=14, tzinfo=UTC)
 
     @pytest.fixture
     def offset(self) -> str:
         return "1d"
 
     @pytest.fixture
-    def duration(self) -> Optional["timedelta"]:
+    def duration(self) -> timedelta | None:
         return timedelta(days=5)
 
     def test_to_admin_response(
         self,
-        event_report_request,
-        enriched_session_merge,
-        client_no_amm,
-        wall_collection,
-        session_collection,
+        event_report_request: ReportRequest,
+        enriched_session_merge: EnrichedSessionMerge,
+        client_no_amm: DaskClient,
+        wall_collection: WallDFCollection,
+        session_collection: SessionDFCollection,
         thl_web_rr: PostgresConfig,
-        session_report_request,
-        user_factory,
-        start,
-        session_factory,
-        product_factory,
-        delete_df_collection,
+        session_report_request: ReportRequest,
+        user_factory: Callable[..., User],
+        start: datetime,
+        session_factory: Callable[..., Session],
+        product_factory: Callable[..., Product],
+        delete_df_collection: Callable[..., None],
     ):
         delete_df_collection(coll=wall_collection)
         delete_df_collection(coll=session_collection)
@@ -112,7 +126,7 @@ class TestEnrichedSessionAdmin:
         for p in [p1, p2]:
             u = user_factory(product=p)
             for i in range(50):
-                s = session_factory(
+                _ = session_factory(
                     user=u,
                     wall_count=1,
                     wall_req_cpi=Decimal("1.00"),
diff --git a/tests/incite/mergers/foundations/test_enriched_task_adjust.py b/tests/incite/mergers/foundations/test_enriched_task_adjust.py
index 96c214f..877d22f 100644
--- a/tests/incite/mergers/foundations/test_enriched_task_adjust.py
+++ b/tests/incite/mergers/foundations/test_enriched_task_adjust.py
@@ -1,16 +1,30 @@
+from __future__ import annotations
+
+from collections.abc import Callable
 from datetime import timedelta
 from itertools import product as iter_product
+from typing import TYPE_CHECKING
 
 import dask.dataframe as dd
 import pandas as pd
 import pytest
+from dask.distributed import Client as DaskClient
 
-from test_utils.incite.collections.conftest import (
-    wall_collection,
-    task_adj_collection,
-    session_collection,
-)
-from test_utils.incite.mergers.conftest import enriched_wall_merge
+if TYPE_CHECKING:
+    from generalresearch.incite.collections.thl_web import (
+        SessionDFCollection,
+        TaskAdjustmentDFCollection,
+        WallDFCollection,
+    )
+    from generalresearch.incite.mergers.foundations.enriched_task_adjust import (
+        EnrichedTaskAdjustMerge,
+    )
+    from generalresearch.incite.mergers.foundations.enriched_wall import (
+        EnrichedWallMerge,
+    )
+    from generalresearch.models.thl.product import Product
+    from generalresearch.models.thl.user import User
+    from generalresearch.pg_helper import PostgresConfig
 
 
 @pytest.mark.parametrize(
@@ -27,19 +41,18 @@ 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,
+        client_no_amm: DaskClient,
+        user_factory: Callable[..., User],
+        product: Product,
+        task_adj_collection: TaskAdjustmentDFCollection,
+        wall_collection: WallDFCollection,
+        session_collection: SessionDFCollection,
+        enriched_wall_merge: EnrichedWallMerge,
+        enriched_task_adjust_merge: EnrichedTaskAdjustMerge,
+        incite_item_factory: Callable[..., None],
+        delete_df_collection: Callable[..., None],
+        thl_web_rr: PostgresConfig,
     ):
-        from generalresearch.models.thl.user import User
 
         # -- Build & Setup
         delete_df_collection(coll=session_collection)
diff --git a/tests/incite/mergers/foundations/test_enriched_wall.py b/tests/incite/mergers/foundations/test_enriched_wall.py
index 8f4995b..2b9afb8 100644
--- a/tests/incite/mergers/foundations/test_enriched_wall.py
+++ b/tests/incite/mergers/foundations/test_enriched_wall.py
@@ -1,34 +1,33 @@
-from datetime import timedelta, timezone, datetime
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
 from decimal import Decimal
 from itertools import product as iter_product
-from typing import Optional
+from typing import TYPE_CHECKING
 
 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 dask.distributed import Client as DaskClient
 
 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,
-)
+
+if TYPE_CHECKING:
+    from generalresearch.incite.collections.thl_web import (
+        SessionDFCollection,
+        WallDFCollection,
+    )
+
+    # noinspection PyUnresolvedReferences
+    from generalresearch.incite.mergers.foundations.enriched_wall import (
+        EnrichedWallMerge,
+    )
+    from generalresearch.models.admin.request import ReportRequest
+    from generalresearch.models.thl.product import Product
+    from generalresearch.models.thl.session import Session
+    from generalresearch.models.thl.user import User
+    from generalresearch.pg_helper import PostgresConfig
 
 
 @pytest.mark.parametrize(
@@ -39,17 +38,16 @@ 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,
+        client_no_amm: DaskClient,
+        product: Product,
+        user_factory: Callable[..., User],
+        wall_collection: WallDFCollection,
+        thl_web_rr: PostgresConfig,
+        session_collection: SessionDFCollection,
+        enriched_wall_merge: EnrichedWallMerge,
+        delete_df_collection: Callable[..., None],
+        incite_item_factory: Callable[..., None],
     ):
-        from generalresearch.models.thl.user import User
 
         # -- Build & Setup
         delete_df_collection(coll=session_collection)
@@ -82,15 +80,15 @@ class TestEnrichedWall:
 
     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,
+        client_no_amm: DaskClient,
+        product: Product,
+        user_factory: Callable[..., User],
+        wall_collection: WallDFCollection,
+        session_collection: SessionDFCollection,
+        enriched_wall_merge: EnrichedWallMerge,
+        delete_df_collection: Callable[..., None],
+        thl_web_rr: PostgresConfig,
+        incite_item_factory: Callable[..., None],
     ):
         # -- Build & Setup
         delete_df_collection(coll=session_collection)
@@ -118,7 +116,7 @@ class TestEnrichedWall:
 
             try:
                 modified_time1 = path.stat().st_mtime
-            except (Exception,):
+            except OSError:
                 modified_time1 = 0
 
             item.build(
@@ -158,18 +156,23 @@ class TestEnrichedWall:
 class TestEnrichedWallToAdmin:
 
     @pytest.fixture
-    def start(self) -> "datetime":
-        return datetime(year=2020, month=3, day=14, tzinfo=timezone.utc)
+    def start(self) -> datetime:
+        return datetime(year=2020, month=3, day=14, tzinfo=UTC)
 
     @pytest.fixture
     def offset(self) -> str:
         return "1d"
 
     @pytest.fixture
-    def duration(self) -> Optional["timedelta"]:
+    def duration(self) -> timedelta | None:
         return timedelta(days=5)
 
-    def test_empty(self, enriched_wall_merge, client_no_amm, start):
+    def test_empty(
+        self,
+        enriched_wall_merge: EnrichedWallMerge,
+        client_no_amm: DaskClient,
+        start: datetime,
+    ):
         from generalresearch.models.admin.request import ReportRequest
 
         rr = ReportRequest.model_validate({"interval": "5min", "start": start})
@@ -186,18 +189,18 @@ class TestEnrichedWallToAdmin:
 
     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,
+        event_report_request: ReportRequest,
+        enriched_wall_merge: EnrichedWallMerge,
+        client_no_amm: DaskClient,
+        wall_collection: WallDFCollection,
+        session_collection: SessionDFCollection,
+        thl_web_rr: PostgresConfig,
+        user: User,
+        session_factory: Callable[..., Session],
+        delete_df_collection: Callable[..., None],
+        product_factory: Callable[..., Product],
+        user_factory: Callable[..., User],
+        start: datetime,
     ):
         delete_df_collection(coll=wall_collection)
         delete_df_collection(coll=session_collection)
@@ -208,7 +211,7 @@ class TestEnrichedWallToAdmin:
         for p in [p1, p2]:
             u = user_factory(product=p)
             for i in range(50):
-                s = session_factory(
+                _ = session_factory(
                     user=u,
                     wall_count=2,
                     wall_req_cpi=Decimal("1.00"),
diff --git a/tests/incite/mergers/foundations/test_user_id_product.py b/tests/incite/mergers/foundations/test_user_id_product.py
index f96bfb4..8c4b2f7 100644
--- a/tests/incite/mergers/foundations/test_user_id_product.py
+++ b/tests/incite/mergers/foundations/test_user_id_product.py
@@ -1,24 +1,22 @@
-from datetime import timedelta, datetime, timezone
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
 from itertools import product
+from typing import TYPE_CHECKING
 
 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 dask.distributed import Client as DaskClient
 
 from generalresearch.incite.mergers.foundations.user_id_product import (
     UserIdProductMergeItem,
 )
-from test_utils.incite.mergers.conftest import user_id_product_merge
+
+if TYPE_CHECKING:
+    # noinspection PyUnresolvedReferences
+    from generalresearch.incite.mergers.foundations.user_id_product import (
+        UserIdProductMerge,
+    )
 
 
 @pytest.mark.parametrize(
@@ -27,25 +25,28 @@ from test_utils.incite.mergers.conftest import user_id_product_merge
         product(
             ["12h", "3D"],
             [timedelta(days=5)],
-            [
-                (datetime.now(tz=timezone.utc) - timedelta(days=35)).replace(
-                    microsecond=0
-                )
-            ],
+            [(datetime.now(tz=UTC) - timedelta(days=35)).replace(microsecond=0)],
         )
     ),
 )
 class TestUserIDProduct:
 
     @pytest.mark.skip
-    def test_base(self, client_no_amm, user_id_product_merge):
+    def test_base(
+        self, client_no_amm: DaskClient, user_id_product_merge: UserIdProductMerge
+    ):
         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):
+    def test_base_item(
+        self,
+        client_no_amm: DaskClient,
+        user_id_product_merge: UserIdProductMerge,
+        user_collection,
+    ):
         assert len(user_id_product_merge.items) == 1
 
         for item in user_id_product_merge.items:
@@ -55,7 +56,7 @@ class TestUserIDProduct:
 
             try:
                 modified_time1 = path.stat().st_mtime
-            except (Exception,):
+            except OSError:
                 modified_time1 = 0
 
             user_id_product_merge.build(client=client_no_amm, user_coll=user_collection)
@@ -64,7 +65,9 @@ class TestUserIDProduct:
             assert modified_time2 > modified_time1
 
     @pytest.mark.skip
-    def test_read(self, client_no_amm, user_id_product_merge):
+    def test_read(
+        self, client_no_amm: DaskClient, user_id_product_merge: UserIdProductMerge
+    ):
         users_ddf = user_id_product_merge.ddf()
         df = client_no_amm.compute(collections=users_ddf, sync=True)
 
diff --git a/tests/incite/mergers/test_merge_collection.py b/tests/incite/mergers/test_merge_collection.py
index ec507bc..7ed3996 100644
--- a/tests/incite/mergers/test_merge_collection.py
+++ b/tests/incite/mergers/test_merge_collection.py
@@ -1,17 +1,22 @@
-from datetime import datetime, timezone, timedelta
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
 from itertools import product
+from typing import TYPE_CHECKING
 
 import pandas as pd
 import pytest
 from pandera.pandas import DataFrameSchema
 
-from generalresearch.incite.mergers import (
+from generalresearch.incite.mergers.base import (
     MergeCollection,
     MergeType,
 )
-from test_utils.incite.conftest import mnt_filepath
 
-merge_types = list(e for e in MergeType if e != MergeType.TEST)
+if TYPE_CHECKING:
+    from generalresearch.incite.base import GRLDatasets
+
+merge_types = [e for e in MergeType if e != MergeType.TEST]
 
 
 @pytest.mark.parametrize(
@@ -21,17 +26,20 @@ merge_types = list(e for e in MergeType if e != MergeType.TEST)
             merge_types,
             ["5min", "6h", "14D"],
             [timedelta(days=30)],
-            [
-                (datetime.now(tz=timezone.utc) - timedelta(days=35)).replace(
-                    microsecond=0
-                )
-            ],
+            [(datetime.now(tz=UTC) - timedelta(days=35)).replace(microsecond=0)],
         )
     ),
 )
 class TestMergeCollection:
 
-    def test_init(self, mnt_filepath, merge_type, offset, duration, start):
+    def test_init(
+        self,
+        merge_type: MergeType,
+        offset: str,
+        duration: timedelta,
+        start: datetime,
+        mnt_filepath: GRLDatasets,
+    ):
         with pytest.raises(expected_exception=ValueError) as cm:
             MergeCollection(archive_path=mnt_filepath.data_src)
         assert "Must explicitly provide a merge_type" in str(cm.value)
@@ -42,7 +50,14 @@ class TestMergeCollection:
         )
         assert instance.merge_type == merge_type
 
-    def test_items(self, mnt_filepath, merge_type, offset, duration, start):
+    def test_items(
+        self,
+        mnt_filepath: GRLDatasets,
+        merge_type: MergeType,
+        offset: str,
+        duration: timedelta,
+        start: datetime,
+    ):
         instance = MergeCollection(
             merge_type=merge_type,
             offset=offset,
@@ -53,7 +68,14 @@ class TestMergeCollection:
 
         assert len(instance.interval_range) == len(instance.items)
 
-    def test_progress(self, mnt_filepath, merge_type, offset, duration, start):
+    def test_progress(
+        self,
+        mnt_filepath: GRLDatasets,
+        merge_type: MergeType,
+        offset: str,
+        duration: timedelta,
+        start: datetime,
+    ):
         instance = MergeCollection(
             merge_type=merge_type,
             offset=offset,
@@ -67,7 +89,14 @@ class TestMergeCollection:
         assert instance.progress.shape[1] == 7
         assert instance.progress["group_by"].isnull().all()
 
-    def test_schema(self, mnt_filepath, merge_type, offset, duration, start):
+    def test_schema(
+        self,
+        offset: str,
+        duration: timedelta,
+        start: datetime,
+        mnt_filepath: GRLDatasets,
+        merge_type: MergeType,
+    ):
         instance = MergeCollection(
             merge_type=merge_type,
             archive_path=mnt_filepath.archive_path(enum_type=merge_type),
@@ -75,7 +104,14 @@ class TestMergeCollection:
 
         assert isinstance(instance._schema, DataFrameSchema)
 
-    def test_load(self, mnt_filepath, merge_type, offset, duration, start):
+    def test_load(
+        self,
+        mnt_filepath: GRLDatasets,
+        merge_type: MergeType,
+        offset: str,
+        duration: timedelta,
+        start: datetime,
+    ):
         instance = MergeCollection(
             merge_type=merge_type,
             start=start,
@@ -87,7 +123,14 @@ class TestMergeCollection:
         # 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):
+    def test_get_items(
+        self,
+        mnt_filepath: GRLDatasets,
+        merge_type: MergeType,
+        offset: str,
+        duration: timedelta,
+        start: datetime,
+    ):
         instance = MergeCollection(
             start=start,
             finished=start + duration,
diff --git a/tests/incite/mergers/test_merge_collection_item.py b/tests/incite/mergers/test_merge_collection_item.py
index 96f8789..baf1bc4 100644
--- a/tests/incite/mergers/test_merge_collection_item.py
+++ b/tests/incite/mergers/test_merge_collection_item.py
@@ -1,17 +1,19 @@
-from datetime import datetime, timezone, timedelta
+from __future__ import annotations
+
+from datetime import timedelta
 from itertools import product
 from pathlib import PurePath
+from typing import TYPE_CHECKING
 
 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
+from generalresearch.incite.mergers.base import MergeType
+
+if TYPE_CHECKING:
+    from generalresearch.incite.mergers.base import (
+        MergeCollection,
+        MergeCollectionItem,
+    )
 
 
 @pytest.mark.parametrize(
@@ -26,7 +28,10 @@ from test_utils.incite.mergers.conftest import merge_collection
 )
 class TestMergeCollectionItem:
 
-    def test_file_naming(self, merge_collection, offset, duration, start):
+    def test_file_naming(
+        self,
+        merge_collection: MergeCollection,
+    ):
         assert len(merge_collection.items) == 25
 
         items: list[MergeCollectionItem] = merge_collection.items
@@ -41,7 +46,10 @@ class TestMergeCollectionItem:
             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):
+    def test_archives(
+        self,
+        merge_collection: MergeCollection,
+    ):
         assert len(merge_collection.items) == 25
 
         for i in merge_collection.items:
@@ -51,10 +59,13 @@ class TestMergeCollectionItem:
             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])
+        res = {i.should_archive() for i in merge_collection.items}
         assert len(res) == 1
 
-    def test_item_to_archive(self, merge_collection, offset, duration, start):
+    def test_item_to_archive(
+        self,
+        merge_collection: MergeCollection,
+    ):
         for item in merge_collection.items:
             item: MergeCollectionItem
             assert not item.has_archive()
diff --git a/tests/incite/mergers/test_pop_ledger.py b/tests/incite/mergers/test_pop_ledger.py
index 6f96108..9ec188b 100644
--- a/tests/incite/mergers/test_pop_ledger.py
+++ b/tests/incite/mergers/test_pop_ledger.py
@@ -1,18 +1,28 @@
-from datetime import timedelta, datetime, timezone
+from __future__ import annotations
+
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
 from itertools import product as iter_product
-from typing import Optional
+from typing import TYPE_CHECKING
 
 import pandas as pd
 import pytest
-from distributed.utils_test import client_no_amm
+from dask.distributed import Client as DaskClient
 
 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
+
+if TYPE_CHECKING:
+    from generalresearch.incite.base import GRLDatasets
+    from generalresearch.incite.collections.thl_web import (
+        LedgerDFCollection,
+        SessionDFCollection,
+    )
+    from generalresearch.incite.mergers.pop_ledger import PopLedgerMerge
+    from generalresearch.managers.thl.ledger_manager.thl_ledger import ThlLedgerManager
+    from generalresearch.models.thl.product import Product
+    from generalresearch.models.thl.user import User
 
 
 @pytest.mark.parametrize(
@@ -27,25 +37,25 @@ from test_utils.managers.ledger.conftest import create_main_accounts
 class TestMergePOPLedger:
 
     @pytest.fixture
-    def start(self) -> "datetime":
-        return datetime(year=2020, month=3, day=14, tzinfo=timezone.utc)
+    def start(self) -> datetime:
+        return datetime(year=2020, month=3, day=14, tzinfo=UTC)
 
     @pytest.fixture
-    def duration(self) -> Optional["timedelta"]:
+    def duration(self) -> timedelta | None:
         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,
+        client_no_amm: DaskClient,
+        ledger_collection: LedgerDFCollection,
+        pop_ledger_merge: PopLedgerMerge,
+        product: Product,
+        user_factory: Callable[..., User],
+        create_main_accounts: Callable[..., None],
+        thl_ledger_manager: ThlLedgerManager,
+        delete_df_collection: Callable[..., None],
+        incite_item_factory: Callable[..., None],
+        delete_ledger_db: Callable[..., None],
     ):
         from generalresearch.models.thl.ledger import LedgerAccount
 
@@ -79,19 +89,19 @@ class TestMergePOPLedger:
 
         # --
 
-        user_wallet_account: LedgerAccount = thl_lm.get_account_or_create_user_wallet(
-            user=u
+        thl_ledger_manager.get_account_or_create_user_wallet(user=u)
+        cash_account: LedgerAccount = thl_ledger_manager.get_account_cash()
+        rev_account: LedgerAccount = (
+            thl_ledger_manager.get_account_task_complete_revenue()
         )
-        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)
+        cash_balance: int = thl_ledger_manager.get_account_balance(account=cash_account)
+        rev_balance: int = thl_ledger_manager.get_account_balance(account=rev_account)
         assert cash_balance > rev_balance
 
         # (1) Test Cash Account
@@ -129,39 +139,42 @@ class TestMergePOPLedger:
 
     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,
+        client_no_amm: DaskClient,
+        ledger_collection: LedgerDFCollection,
+        pop_ledger_merge: PopLedgerMerge,
+        mnt_filepath: GRLDatasets,
+        product: Product,
+        user_factory: Callable[..., User],
+        create_main_accounts: Callable[..., None],
+        offset: str,
+        duration: timedelta,
+        start: datetime,
+        thl_ledger_manager: ThlLedgerManager,
+        incite_item_factory: Callable[..., None],
+        delete_df_collection: Callable[..., None],
+        delete_ledger_db: Callable[..., None],
+        session_collection: SessionDFCollection,
     ):
+        from generalresearch.models.thl.finance import ProductBalances
         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(),
+        create_main_accounts()
+
         delete_df_collection(coll=ledger_collection)
 
-        bp_account: LedgerAccount = thl_lm.get_account_or_create_bp_wallet(
+        bp_account: LedgerAccount = thl_ledger_manager.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()
+        cash_account: LedgerAccount = thl_ledger_manager.get_account_cash()
+        rev_account: LedgerAccount = (
+            thl_ledger_manager.get_account_task_complete_revenue()
+        )
 
         for item in ledger_collection.items:
             incite_item_factory(item=item, user=u)
@@ -191,8 +204,10 @@ class TestMergePOPLedger:
         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)
+        assert instance.balance == thl_ledger_manager.get_account_balance(bp_account)
+        assert df["bp_payment.CREDIT"].sum() == thl_ledger_manager.get_account_balance(
+            bp_account
+        )
 
         # (2) Filter by the Cash Account
         ddf = pop_ledger_merge.ddf(
@@ -205,7 +220,7 @@ class TestMergePOPLedger:
         )
         df: pd.DataFrame = client_no_amm.compute(collections=ddf, sync=True)
 
-        cash_balance: int = thl_lm.get_account_balance(account=cash_account)
+        cash_balance: int = thl_ledger_manager.get_account_balance(account=cash_account)
         assert df["bp_payment.CREDIT"].sum() == 0
         assert cash_balance > 0
         assert df["mp_payment.CREDIT"].sum() == 0
@@ -222,7 +237,7 @@ class TestMergePOPLedger:
         )
         df: pd.DataFrame = client_no_amm.compute(collections=ddf, sync=True)
 
-        rev_balance: int = thl_lm.get_account_balance(account=rev_account)
+        rev_balance: int = thl_ledger_manager.get_account_balance(account=rev_account)
         assert rev_balance == 0
         assert df["bp_payment.CREDIT"].sum() == 0
         assert df["mp_payment.DEBIT"].sum() == 0
@@ -230,27 +245,28 @@ class TestMergePOPLedger:
 
     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,
+        client_no_amm: DaskClient,
+        ledger_collection: LedgerDFCollection,
+        pop_ledger_merge: PopLedgerMerge,
+        mnt_filepath: GRLDatasets,
+        user_factory: Callable[..., User],
+        product: Product,
+        create_main_accounts: Callable[..., None],
+        offset: str,
+        duration: timedelta,
+        start: datetime,
+        thl_ledger_manager: ThlLedgerManager,
+        delete_df_collection: Callable[..., None],
+        incite_item_factory: Callable[..., None],
     ):
-        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)
+        bp_account = thl_ledger_manager.get_account_or_create_bp_wallet(
+            product=u1.product
+        )
 
         for item in ledger_collection.items:
             incite_item_factory(user=u1, item=item)
@@ -280,7 +296,7 @@ class TestMergePOPLedger:
         assert isinstance(df.index, pd.Index)
         assert isinstance(df.index, pd.DatetimeIndex)
 
-        bp_account_balance = thl_lm.get_account_balance(account=bp_account)
+        thl_ledger_manager.get_account_balance(account=bp_account)
 
         # Initial sum
         initial_sum = df.sum().sum()
diff --git a/tests/incite/mergers/test_ym_survey_merge.py b/tests/incite/mergers/test_ym_survey_merge.py
index 4c2df6b..d83a98c 100644
--- a/tests/incite/mergers/test_ym_survey_merge.py
+++ b/tests/incite/mergers/test_ym_survey_merge.py
@@ -1,25 +1,28 @@
-from datetime import timedelta, timezone, datetime
+from __future__ import annotations
+
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
 from itertools import product
+from typing import TYPE_CHECKING
 
 import pandas as pd
 import pytest
+from dask.distributed import Client as DaskClient
+
+if TYPE_CHECKING:
+    from generalresearch.incite.collections.thl_web import (
+        SessionDFCollection,
+        WallDFCollection,
+    )
+    from generalresearch.incite.mergers.foundations.enriched_session import (
+        EnrichedSessionMerge,
+    )
+    from generalresearch.incite.mergers.ym_survey_wall import YMSurveyWallMerge
+    from generalresearch.models.thl.product import Product
+    from generalresearch.models.thl.user import User
+    from generalresearch.pg_helper import PostgresConfig
 
 # 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(
@@ -28,11 +31,7 @@ from test_utils.incite.mergers.conftest import (
         product(
             ["12h", "3D"],
             [timedelta(days=30)],
-            [
-                (datetime.now(tz=timezone.utc) - timedelta(days=35)).replace(
-                    microsecond=0
-                )
-            ],
+            [(datetime.now(tz=UTC) - timedelta(days=35)).replace(microsecond=0)],
         )
     ),
 )
@@ -46,18 +45,17 @@ class TestYMSurveyMerge:
 
     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,
+        client_no_amm: DaskClient,
+        user_factory: Callable[..., User],
+        product: Product,
+        ym_survey_wall_merge: YMSurveyWallMerge,
+        wall_collection: WallDFCollection,
+        session_collection: SessionDFCollection,
+        enriched_session_merge: EnrichedSessionMerge,
+        delete_df_collection: Callable[..., None],
+        incite_item_factory: Callable[..., None],
+        thl_web_rr: PostgresConfig,
     ):
-        from generalresearch.models.thl.user import User
 
         delete_df_collection(coll=session_collection)
         user: User = user_factory(product=product, created=session_collection.start)
@@ -85,10 +83,10 @@ class TestYMSurveyMerge:
         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)
+        df1: pd.DataFrame | None = client_no_amm.compute(collections=ddf, sync=True)
 
-        assert isinstance(df, pd.DataFrame)
-        assert not df.empty
+        assert isinstance(df1, pd.DataFrame)
+        assert not df1.empty
 
         # --
 
@@ -102,18 +100,18 @@ class TestYMSurveyMerge:
         # --
 
         ddf = ym_survey_wall_merge.ddf()
-        df: pd.DataFrame = client_no_amm.compute(collections=ddf, sync=True)
+        df2: pd.DataFrame | None = client_no_amm.compute(collections=ddf, sync=True)
 
-        assert isinstance(df, pd.DataFrame)
-        assert not df.empty
+        assert isinstance(df2, pd.DataFrame)
+        assert not df2.empty
 
         # --
-        assert df.product_id.nunique() == 1
-        assert df.team_id.nunique() == 1
-        assert df.source.nunique() > 1
+        assert df2.product_id.nunique() == 1
+        assert df2.team_id.nunique() == 1
+        assert df2.source.nunique() > 1
 
-        started_min_ts = df.started.min()
-        started_max_ts = df.started.max()
+        started_min_ts = df2.started.min()
+        started_max_ts = df2.started.max()
 
         assert type(started_min_ts) is pd.Timestamp
         assert type(started_max_ts) is pd.Timestamp