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
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
|
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.base import (
MergeCollection,
MergeType,
)
if TYPE_CHECKING:
from generalresearch.incite.base import GRLDatasets
merge_types = [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=UTC) - timedelta(days=35)).replace(microsecond=0)],
)
),
)
class TestMergeCollection:
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)
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: GRLDatasets,
merge_type: MergeType,
offset: str,
duration: timedelta,
start: datetime,
):
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: GRLDatasets,
merge_type: MergeType,
offset: str,
duration: timedelta,
start: datetime,
):
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,
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),
)
assert isinstance(instance._schema, DataFrameSchema)
def test_load(
self,
mnt_filepath: GRLDatasets,
merge_type: MergeType,
offset: str,
duration: timedelta,
start: datetime,
):
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: GRLDatasets,
merge_type: MergeType,
offset: str,
duration: timedelta,
start: datetime,
):
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)
|
