-
-
Notifications
You must be signed in to change notification settings - Fork 62
Expand file tree
/
Copy pathanalytics_rules_test.py
More file actions
150 lines (123 loc) · 4.66 KB
/
analytics_rules_test.py
File metadata and controls
150 lines (123 loc) · 4.66 KB
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
146
147
148
149
150
"""Tests for v30 Analytics Rules endpoints (client.analytics.rules)."""
import pytest
from tests.utils.version import is_v30_or_above
from typesense.sync.client import Client
from typesense.sync.analytics_rules import AnalyticsRules
from typesense.sync.analytics_rule import AnalyticsRule
from typesense.async_.api_call import AsyncApiCall
from typesense.async_.analytics_rules import AsyncAnalyticsRules
from typesense.types.analytics import AnalyticsRuleCreate
pytestmark = pytest.mark.skipif(
not is_v30_or_above(
Client(
{
"api_key": "xyz",
"nodes": [{"host": "localhost", "port": 8108, "protocol": "http"}],
}
)
),
reason="Run v30 analytics tests only on v30+",
)
def test_rules_init(fake_api_call) -> None:
rules = AnalyticsRules(fake_api_call)
assert rules.rules == {}
def test_rule_getitem(fake_api_call) -> None:
rules = AnalyticsRules(fake_api_call)
rule = rules["company_analytics_rule"]
assert isinstance(rule, AnalyticsRule)
assert rule._endpoint_path == "/analytics/rules/company_analytics_rule"
def test_actual_create(
actual_analytics_rules: AnalyticsRules,
delete_all: None,
delete_all_analytics_rules: None,
create_collection: None,
create_query_collection: None,
) -> None:
body: AnalyticsRuleCreate = {
"name": "company_analytics_rule",
"type": "nohits_queries",
"collection": "companies",
"event_type": "search",
"params": {"destination_collection": "companies_queries", "limit": 1000},
}
resp = actual_analytics_rules.create(rule=body)
assert resp["name"] == "company_analytics_rule"
assert resp["params"]["destination_collection"] == "companies_queries"
def test_actual_update(
actual_analytics_rules: AnalyticsRules,
delete_all: None,
delete_all_analytics_rules: None,
create_analytics_rule: None,
) -> None:
resp = actual_analytics_rules.upsert(
"company_analytics_rule",
{
"params": {
"destination_collection": "companies_queries",
"limit": 500,
},
},
)
assert resp["name"] == "company_analytics_rule"
def test_actual_retrieve(
actual_analytics_rules: AnalyticsRules,
delete_all: None,
delete_all_analytics_rules: None,
create_analytics_rule: None,
) -> None:
rules = actual_analytics_rules.retrieve()
assert isinstance(rules, list)
assert any(r.get("name") == "company_analytics_rule" for r in rules)
def test_rules_init_async(fake_async_api_call: AsyncApiCall) -> None:
from typesense.async_.analytics_rules import AsyncAnalyticsRules
rules = AsyncAnalyticsRules(fake_async_api_call)
assert rules.rules == {}
def test_rule_getitem_async(fake_async_api_call: AsyncApiCall) -> None:
from typesense.async_.analytics_rules import AsyncAnalyticsRules
from typesense.async_.analytics_rule import AsyncAnalyticsRule
rules = AsyncAnalyticsRules(fake_async_api_call)
rule = rules["company_analytics_rule"]
assert isinstance(rule, AsyncAnalyticsRule)
assert rule._endpoint_path == "/analytics/rules/company_analytics_rule"
async def test_actual_create_async(
actual_async_analytics_rules: AsyncAnalyticsRules,
delete_all: None,
delete_all_analytics_rules: None,
create_collection: None,
create_query_collection: None,
) -> None:
body: AnalyticsRuleCreate = {
"name": "company_analytics_rule",
"type": "nohits_queries",
"collection": "companies",
"event_type": "search",
"params": {"destination_collection": "companies_queries", "limit": 1000},
}
resp = await actual_async_analytics_rules.create(rule=body)
assert resp["name"] == "company_analytics_rule"
assert resp["params"]["destination_collection"] == "companies_queries"
async def test_actual_update_async(
actual_async_analytics_rules: AsyncAnalyticsRules,
delete_all: None,
delete_all_analytics_rules: None,
create_analytics_rule: None,
) -> None:
resp = await actual_async_analytics_rules.upsert(
"company_analytics_rule",
{
"params": {
"destination_collection": "companies_queries",
"limit": 500,
},
},
)
assert resp["name"] == "company_analytics_rule"
async def test_actual_retrieve_async(
actual_async_analytics_rules: AsyncAnalyticsRules,
delete_all: None,
delete_all_analytics_rules: None,
create_analytics_rule: None,
) -> None:
rules = await actual_async_analytics_rules.retrieve()
assert isinstance(rules, list)
assert any(r.get("name") == "company_analytics_rule" for r in rules)