8
8
from browser_use_sdk .lib .webhooks import (
9
9
WebhookTest ,
10
10
WebhookTestPayload ,
11
- WebhookAgentTaskStatusUpdate ,
12
- WebhookAgentTaskStatusUpdatePayload ,
13
11
create_webhook_signature ,
14
12
verify_webhook_event_signature ,
15
13
)
@@ -23,17 +21,15 @@ def test_create_webhook_signature() -> None:
23
21
24
22
signature = create_webhook_signature (payload , timestamp , secret )
25
23
26
- # Signature should be a hex string
27
24
assert isinstance (signature , str )
28
- assert len (signature ) == 64 # SHA256 hex length
25
+ assert len (signature ) == 64
29
26
30
- # Same inputs should produce same signature
31
27
signature2 = create_webhook_signature (payload , timestamp , secret )
32
28
assert signature == signature2
33
29
34
- # Different payload should produce different signature
35
30
different_payload = {"test" : "different" }
36
31
different_signature = create_webhook_signature (different_payload , timestamp , secret )
32
+
37
33
assert signature != different_signature
38
34
39
35
@@ -51,7 +47,10 @@ def test_verify_webhook_event_signature_valid() -> None:
51
47
52
48
# Verify signature
53
49
verified_webhook = verify_webhook_event_signature (
54
- body = webhook .model_dump (), secret = secret , signature = signature , timestamp = timestamp
50
+ body = webhook .model_dump (),
51
+ secret = secret ,
52
+ timestamp = timestamp ,
53
+ expected_signature = signature ,
55
54
)
56
55
57
56
assert verified_webhook is not None
@@ -68,67 +67,43 @@ def test_verify_webhook_event_signature_invalid_signature() -> None:
68
67
payload = WebhookTestPayload (test = "ok" )
69
68
webhook = WebhookTest (type = "test" , timestamp = datetime .now (timezone .utc ), payload = payload )
70
69
71
- # Use invalid signature
72
- invalid_signature = "invalid_signature_123"
73
-
74
70
verified_webhook = verify_webhook_event_signature (
75
- body = webhook .model_dump (), secret = secret , signature = invalid_signature , timestamp = timestamp
71
+ body = webhook .model_dump (),
72
+ secret = secret ,
73
+ timestamp = timestamp ,
74
+ expected_signature = "random_invalid_signature" ,
76
75
)
77
76
78
77
assert verified_webhook is None
79
78
80
79
81
80
def test_verify_webhook_event_signature_wrong_secret () -> None :
82
81
"""Test webhook signature verification with wrong secret."""
83
- secret = "test-secret-key"
84
- wrong_secret = "wrong-secret-key"
82
+
85
83
timestamp = "2023-01-01T00:00:00Z"
86
84
87
85
# Create test webhook
88
86
payload = WebhookTestPayload (test = "ok" )
89
87
webhook = WebhookTest (type = "test" , timestamp = datetime .now (timezone .utc ), payload = payload )
90
88
91
89
# Create signature with correct secret
92
- signature = create_webhook_signature (webhook .payload .model_dump (), timestamp , secret )
90
+ signature = create_webhook_signature (
91
+ payload = webhook .payload .model_dump (),
92
+ timestamp = timestamp ,
93
+ secret = "test-secret-key" ,
94
+ )
93
95
94
96
# Verify with wrong secret
95
97
verified_webhook = verify_webhook_event_signature (
96
- body = webhook .model_dump (), secret = wrong_secret , signature = signature , timestamp = timestamp
98
+ body = webhook .model_dump (),
99
+ secret = "wrong-secret-key" ,
100
+ timestamp = timestamp ,
101
+ expected_signature = signature ,
97
102
)
98
103
99
104
assert verified_webhook is None
100
105
101
106
102
- def test_verify_webhook_event_signature_agent_task_status_update () -> None :
103
- """Test webhook signature verification for agent task status update."""
104
- secret = "test-secret-key"
105
- timestamp = "2023-01-01T00:00:00Z"
106
-
107
- # Create agent task status update webhook
108
- payload = WebhookAgentTaskStatusUpdatePayload (
109
- session_id = "sess_123" , task_id = "task_456" , status = "started" , metadata = {"progress" : 25 }
110
- )
111
-
112
- webhook = WebhookAgentTaskStatusUpdate (
113
- type = "agent.task.status_update" , timestamp = datetime .now (timezone .utc ), payload = payload
114
- )
115
-
116
- # Create signature
117
- signature = create_webhook_signature (webhook .payload .model_dump (), timestamp , secret )
118
-
119
- # Verify signature
120
- verified_webhook = verify_webhook_event_signature (
121
- body = webhook .model_dump (), secret = secret , signature = signature , timestamp = timestamp
122
- )
123
-
124
- assert verified_webhook is not None
125
- assert isinstance (verified_webhook , WebhookAgentTaskStatusUpdate )
126
- assert verified_webhook .payload .session_id == "sess_123"
127
- assert verified_webhook .payload .status == "started"
128
- assert verified_webhook .payload .metadata is not None
129
- assert verified_webhook .payload .metadata ["progress" ] == 25
130
-
131
-
132
107
def test_verify_webhook_event_signature_string_body () -> None :
133
108
"""Test webhook signature verification with string body."""
134
109
secret = "test-secret-key"
@@ -143,7 +118,7 @@ def test_verify_webhook_event_signature_string_body() -> None:
143
118
144
119
# Verify with string body
145
120
verified_webhook = verify_webhook_event_signature (
146
- body = webhook .model_dump (), secret = secret , signature = signature , timestamp = timestamp
121
+ body = webhook .model_dump (), secret = secret , expected_signature = signature , timestamp = timestamp
147
122
)
148
123
149
124
assert verified_webhook is not None
@@ -159,7 +134,7 @@ def test_verify_webhook_event_signature_invalid_body() -> None:
159
134
invalid_body : Dict [str , Any ] = {"type" : "invalid_type" , "timestamp" : "invalid" , "payload" : {}}
160
135
161
136
verified_webhook = verify_webhook_event_signature (
162
- body = invalid_body , secret = secret , signature = "some_signature" , timestamp = timestamp
137
+ body = invalid_body , secret = secret , expected_signature = "some_signature" , timestamp = timestamp
163
138
)
164
139
165
140
assert verified_webhook is None
0 commit comments