Skip to main content

max / makenotwork

6.3 KB · 222 lines History Blame Raw
1 //! SSE push notification integration tests.
2 //!
3 //! Tests that call `get_streaming` open infinite SSE connections via oneshot.
4 //! The handler tasks outlive the test, preventing the test binary from exiting.
5 //! These tests are marked `#[ignore]` and must be run individually:
6 //!
7 //! cargo test --test integration synckit_sse -- --ignored --test-threads=1
8
9 use crate::harness::TestHarness;
10 use makenotwork::db::{SyncAppId, SyncDeviceId, UserId};
11 use serde::Deserialize;
12 use serde_json::json;
13 use sqlx::PgPool;
14
15 // ── Response types ──
16
17 #[derive(Deserialize)]
18 struct AuthResponse {
19 token: String,
20 #[allow(dead_code)]
21 user_id: UserId,
22 app_id: SyncAppId,
23 }
24
25 #[derive(Deserialize)]
26 struct DeviceResponse {
27 id: SyncDeviceId,
28 }
29
30 // ── Helpers ──
31
32 async fn create_sync_app_for_user(pool: &PgPool, user_id: UserId) -> (SyncAppId, String) {
33 let api_key = "test-sse-api-key";
34 let key_hash = crate::harness::hash_api_key(api_key);
35 let key_prefix = &api_key[..8];
36 let app_id: SyncAppId = sqlx::query_scalar(
37 "INSERT INTO sync_apps (creator_id, name, api_key_hash, api_key_prefix) VALUES ($1, 'SSE Test App', $2, $3) RETURNING id",
38 )
39 .bind(user_id)
40 .bind(&key_hash)
41 .bind(key_prefix)
42 .fetch_one(pool)
43 .await
44 .expect("Failed to create sync app");
45 (app_id, api_key.to_string())
46 }
47
48 async fn setup_authenticated(h: &mut TestHarness) -> (String, SyncAppId) {
49 let user_id = h.signup("sse_user", "sse@example.com", "Password1!").await;
50 let (_app_id, api_key) = create_sync_app_for_user(&h.db, user_id).await;
51
52 let resp = h
53 .client
54 .post_json(
55 "/api/sync/auth",
56 &json!({
57 "email": "sse@example.com",
58 "password": "Password1!",
59 "api_key": api_key,
60 "key": "test-sdk-key",
61 })
62 .to_string(),
63 )
64 .await;
65 assert_eq!(resp.status, 200, "Auth failed: {}", resp.text);
66
67 let auth: AuthResponse = resp.json();
68 h.client.set_bearer_token(&auth.token);
69
70 (auth.token, auth.app_id)
71 }
72
73 async fn register_device(h: &mut TestHarness, name: &str) -> SyncDeviceId {
74 let resp = h
75 .client
76 .post_json(
77 "/api/sync/devices",
78 &json!({ "device_name": name, "platform": "macos" }).to_string(),
79 )
80 .await;
81 assert_eq!(resp.status, 200, "Register device failed: {}", resp.text);
82 let dev: DeviceResponse = resp.json();
83 dev.id
84 }
85
86 // ── Tests ──
87
88 #[tokio::test]
89 #[ignore = "opens infinite SSE stream; run separately"]
90 async fn sse_subscribe_returns_event_stream() {
91 let mut h = TestHarness::new().await;
92 let (_token, app_id) = setup_authenticated(&mut h).await;
93
94 let resp = h
95 .client
96 .get_streaming(&format!("/api/sync/subscribe?app_id={app_id}"))
97 .await;
98
99 assert_eq!(resp.status, 200, "SSE subscribe failed");
100 let content_type = resp.header("content-type").unwrap_or("");
101 assert!(
102 content_type.contains("text/event-stream"),
103 "Expected text/event-stream, got: {content_type}"
104 );
105 }
106
107 #[tokio::test]
108 async fn sse_subscribe_without_jwt_returns_401() {
109 let mut h = TestHarness::new().await;
110
111 let app_id = SyncAppId::new();
112 let resp = h
113 .client
114 .get(&format!("/api/sync/subscribe?app_id={app_id}"))
115 .await;
116 assert_eq!(resp.status, 401);
117 }
118
119 #[tokio::test]
120 async fn sse_subscribe_mismatched_app_id_rejected() {
121 let mut h = TestHarness::new().await;
122 setup_authenticated(&mut h).await;
123
124 let wrong_app_id = SyncAppId::new();
125 let resp = h
126 .client
127 .get(&format!("/api/sync/subscribe?app_id={wrong_app_id}"))
128 .await;
129 assert_eq!(
130 resp.status, 400,
131 "Expected 400 for mismatched app_id: {}",
132 resp.text
133 );
134 }
135
136 #[tokio::test]
137 #[ignore = "opens infinite SSE stream; run separately"]
138 async fn push_triggers_broadcast_notification() {
139 let mut h = TestHarness::new().await;
140 let user_id = h
141 .signup("notif_user", "notif@example.com", "Password1!")
142 .await;
143 let (app_id, api_key) = create_sync_app_for_user(&h.db, user_id).await;
144
145 let resp = h
146 .client
147 .post_json(
148 "/api/sync/auth",
149 &json!({
150 "email": "notif@example.com",
151 "password": "Password1!",
152 "api_key": api_key,
153 "key": "test-sdk-key",
154 })
155 .to_string(),
156 )
157 .await;
158 assert_eq!(resp.status, 200);
159 let auth: AuthResponse = resp.json();
160 h.client.set_bearer_token(&auth.token);
161 let device_id = register_device(&mut h, "TestDevice").await;
162
163 let resp = h
164 .client
165 .get_streaming(&format!("/api/sync/subscribe?app_id={app_id}"))
166 .await;
167 assert_eq!(resp.status, 200);
168
169 let resp = h
170 .client
171 .post_json(
172 "/api/sync/push",
173 &json!({
174 "device_id": device_id,
175 "changes": [{
176 "table": "tasks",
177 "op": "INSERT",
178 "row_id": "sse-test-1",
179 "timestamp": "2025-01-01T00:00:00Z",
180 "data": {"title": "SSE test"}
181 }]
182 })
183 .to_string(),
184 )
185 .await;
186 assert_eq!(resp.status, 200, "Push failed: {}", resp.text);
187 }
188
189 #[tokio::test]
190 #[ignore = "opens infinite SSE stream; run separately"]
191 async fn multiple_subscribers_receive_events() {
192 let mut h = TestHarness::new().await;
193 let (_token, app_id) = setup_authenticated(&mut h).await;
194
195 let resp1 = h
196 .client
197 .get_streaming(&format!("/api/sync/subscribe?app_id={app_id}"))
198 .await;
199 assert_eq!(resp1.status, 200);
200
201 let resp2 = h
202 .client
203 .get_streaming(&format!("/api/sync/subscribe?app_id={app_id}"))
204 .await;
205 assert_eq!(resp2.status, 200);
206 }
207
208 #[tokio::test]
209 #[ignore = "opens infinite SSE stream; run separately"]
210 async fn sse_stream_includes_keepalive_header() {
211 let mut h = TestHarness::new().await;
212 let (_token, app_id) = setup_authenticated(&mut h).await;
213
214 let resp = h
215 .client
216 .get_streaming(&format!("/api/sync/subscribe?app_id={app_id}"))
217 .await;
218 assert_eq!(resp.status, 200);
219 let content_type = resp.header("content-type").unwrap_or("");
220 assert!(content_type.contains("text/event-stream"));
221 }
222