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
|
from __future__ import annotations
from collections.abc import Callable
from typing import TYPE_CHECKING
import pytest
from generalresearch.models.thl.category import Category
if TYPE_CHECKING:
from generalresearch.managers.thl.category import CategoryManager
from generalresearch.pg_helper import PostgresConfig
class TestCategory:
@pytest.fixture
def beauty_fitness(self) -> Category:
return Category(
uuid="12c1e96be82c4642a07a12a90ce6f59e",
adwords_vertical_id="44",
label="Beauty & Fitness",
path="/Beauty & Fitness",
)
@pytest.fixture
def hair_care(self, beauty_fitness: Category) -> Category:
return Category(
uuid="dd76c4b565d34f198dad3687326503d6",
adwords_vertical_id="146",
label="Hair Care",
path=f"{beauty_fitness.path}/Hair Care",
)
@pytest.fixture
def hair_loss(self, hair_care: Category) -> Category:
return Category(
uuid="aacff523c8e246888215611ec3b823c0",
adwords_vertical_id="235",
label="Hair Loss",
path=f"{hair_care.path}/Hair Loss",
)
@pytest.fixture
def category_data(
self,
category_manager: CategoryManager,
thl_web_rw: PostgresConfig,
beauty_fitness: Category,
hair_care: Category,
hair_loss: Category,
) -> Callable[..., None]:
def _inner():
cats = [beauty_fitness, hair_care, hair_loss]
data = [x.model_dump(mode="json") for x in cats]
# We need the parent pk's to set the parent_id. So insert all without a parent,
# then pull back all pks and map to the parents as parsed by the parent_path
query = """
INSERT INTO marketplace_category
(uuid, adwords_vertical_id, label, path)
VALUES
(%(uuid)s, %(adwords_vertical_id)s, %(label)s, %(path)s)
ON CONFLICT (uuid) DO NOTHING;
"""
with thl_web_rw.make_connection() as conn:
with conn.cursor() as c:
c.executemany(query=query, params_seq=data)
conn.commit()
res = thl_web_rw.execute_sql_query(
"SELECT id, path FROM marketplace_category"
)
path_id = {x["path"]: x["id"] for x in res}
data = [
{"id": path_id[c.path], "parent_id": path_id[c.parent_path]}
for c in cats
if c.parent_path
]
query = """
UPDATE marketplace_category
SET parent_id = %(parent_id)s
WHERE id = %(id)s;
"""
with thl_web_rw.make_connection() as conn:
with conn.cursor() as c:
c.executemany(query=query, params_seq=data)
conn.commit()
category_manager.populate_caches()
return _inner
def test(
self,
category_data: Callable[..., None],
category_manager: CategoryManager,
beauty_fitness: Category,
):
category_data()
# category_manager on init caches the category info. This rarely/never changes so this is fine,
# but now that tests get run on a new db each time, the category_manager is inited before
# the fixtures run. so category_manager's cache needs to be rerun
# path='/Beauty & Fitness/Hair Care/Hair Loss'
c: Category = category_manager.get_by_label("Hair Loss")
# Beauty & Fitness
assert beauty_fitness.uuid == category_manager.get_top_level(c).uuid
c: Category = category_manager.categories[beauty_fitness.uuid]
# The root is itself
assert c == category_manager.get_category_root(c)
# The root is Beauty & Fitness
c: Category = category_manager.get_by_label("Hair Loss")
assert beauty_fitness.uuid == category_manager.get_category_root(c).uuid
|
