forked from BasedHardware/omi
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_webhooks.py
More file actions
169 lines (116 loc) · 6.15 KB
/
Copy pathtest_webhooks.py
File metadata and controls
169 lines (116 loc) · 6.15 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
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
"""Developer webhook route and delivery coverage."""
import asyncio
import httpx
import pytest
import database.webhook_health as webhook_health
def _configure_realtime_webhook(client, auth_headers, url="https://webhook.test/realtime"):
configured = client.post(
"/v1/users/developer/webhook/realtime_transcript",
json={"url": url},
headers=auth_headers,
)
assert configured.status_code == 200, configured.text
def _enable_realtime_webhook(client, auth_headers):
enabled = client.post("/v1/users/developer/webhook/realtime_transcript/enable", headers=auth_headers)
assert enabled.status_code == 200, enabled.text
def _disable_realtime_webhook(client, auth_headers):
disabled = client.post("/v1/users/developer/webhook/realtime_transcript/disable", headers=auth_headers)
assert disabled.status_code == 200, disabled.text
def _health(fake_redis, uid="123", wtype="realtime_transcript"):
return {k.decode(): v.decode() for k, v in fake_redis.hgetall(f"dev_webhook_health:{uid}:{wtype}").items()}
def _run_realtime_delivery(monkeypatch, handler):
import utils.webhooks as webhooks
async def exercise_webhook():
async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as fake_client:
monkeypatch.setattr(webhooks, "get_webhook_client", lambda: fake_client)
await webhooks.realtime_transcript_webhook(
"123",
[{"text": "Hermetic realtime transcript", "speaker": "SPEAKER_00", "start": 0.0, "end": 1.0}],
)
asyncio.run(exercise_webhook())
def test_realtime_webhook_config_roundtrip_and_delivery_capture(client, auth_headers, monkeypatch, fake_redis):
_configure_realtime_webhook(client, auth_headers)
read_back = client.get("/v1/users/developer/webhook/realtime_transcript", headers=auth_headers)
assert read_back.status_code == 200, read_back.text
assert read_back.json() == {"url": "https://webhook.test/realtime"}
_enable_realtime_webhook(client, auth_headers)
status = client.get("/v1/users/developer/webhooks/status", headers=auth_headers)
assert status.status_code == 200, status.text
assert "realtime_transcript" in status.json()
requests = []
async def handler(request):
requests.append(request)
return httpx.Response(200, json={"ok": True})
_run_realtime_delivery(monkeypatch, handler)
assert len(requests) == 1
assert str(requests[0].url) == "https://webhook.test/realtime?uid=123"
payload = requests[0].read()
assert b"Hermetic realtime transcript" in payload
assert _health(fake_redis)["failure_count"] == "0"
assert _health(fake_redis)["last_status"] == "200"
disabled = client.post("/v1/users/developer/webhook/realtime_transcript/disable", headers=auth_headers)
assert disabled.status_code == 200, disabled.text
def test_realtime_webhook_does_not_call_provider_when_disabled(client, auth_headers, monkeypatch, fake_redis):
_configure_realtime_webhook(client, auth_headers)
_disable_realtime_webhook(client, auth_headers)
requests = []
async def handler(request):
requests.append(request)
return httpx.Response(200, json={"ok": True})
_run_realtime_delivery(monkeypatch, handler)
assert requests == []
assert _health(fake_redis)["disabled"] == "0"
assert _health(fake_redis)["last_status"] == "200"
@pytest.mark.parametrize(
("status_code", "last_error"),
[(500, "HTTP 500"), (429, "HTTP 429")],
)
def test_realtime_webhook_records_non_2xx_failures(
client, auth_headers, monkeypatch, fake_redis, status_code, last_error
):
_configure_realtime_webhook(client, auth_headers)
_enable_realtime_webhook(client, auth_headers)
async def handler(request):
return httpx.Response(status_code, json={"ok": False})
_run_realtime_delivery(monkeypatch, handler)
health = _health(fake_redis)
assert health["failure_count"] == "1"
assert health["last_status"] == str(status_code)
assert health["last_error"] == last_error
assert health["disabled"] == "0"
def test_realtime_webhook_records_timeout_exception_without_real_network(client, auth_headers, monkeypatch, fake_redis):
_configure_realtime_webhook(client, auth_headers)
_enable_realtime_webhook(client, auth_headers)
async def handler(request):
raise httpx.ConnectTimeout("deterministic timeout")
_run_realtime_delivery(monkeypatch, handler)
health = _health(fake_redis)
assert health["failure_count"] == "1"
assert health["last_status"] == "0"
assert health["last_error"] == "ConnectTimeout"
assert health["disabled"] == "0"
def test_realtime_webhook_auto_disables_after_failure_threshold(client, auth_headers, monkeypatch, fake_redis):
_configure_realtime_webhook(client, auth_headers)
_enable_realtime_webhook(client, auth_headers)
async def handler(request):
return httpx.Response(500, json={"ok": False})
import database.webhook_health as webhook_health
monkeypatch.setattr(webhook_health, "_DEV_FAILURE_THRESHOLD", 2)
_run_realtime_delivery(monkeypatch, handler)
_run_realtime_delivery(monkeypatch, handler)
health = _health(fake_redis)
assert health["failure_count"] == "2"
assert health["disabled"] == "1"
status = client.get("/v1/users/developer/webhooks/status", headers=auth_headers)
assert status.status_code == 200, status.text
assert status.json()["realtime_transcript"] is False
def test_dev_webhook_failure_fallback_only_reports_disable_transition(monkeypatch, fake_redis):
"""The non-Lua fallback should match the Lua script's one-shot disable semantics."""
monkeypatch.setattr(webhook_health, "_DEV_FAILURE_THRESHOLD", 2)
monkeypatch.setattr(webhook_health, "r", fake_redis)
assert webhook_health.record_dev_webhook_failure("123", "realtime_transcript", 500, "HTTP 500") is False
assert webhook_health.record_dev_webhook_failure("123", "realtime_transcript", 500, "HTTP 500") is True
assert webhook_health.record_dev_webhook_failure("123", "realtime_transcript", 500, "HTTP 500") is False
health = _health(fake_redis)
assert health["failure_count"] == "2"
assert health["disabled"] == "1"