1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
|
from __future__ import annotations
from collections.abc import Callable
from datetime import timedelta
from itertools import product as iter_product
import dask.dataframe as dd
import pandas as pd
import pytest
from dask.distributed import Client as DaskClient
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(
argnames="offset, duration,",
argvalues=list(
iter_product(
["12h", "3D"],
[timedelta(days=5)],
)
),
)
class TestEnrichedTaskAdjust:
@pytest.mark.skip
def test_base(
self,
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,
):
# -- 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
|
