quicknode_sdk/streams/
mod.rs

1pub mod stream;
2
3pub use stream::{
4    AddressBookConfig, AzureAttributes, ClickhouseAttributes, CreateStreamParams,
5    DestinationAttributes, EnabledCountResponse, FilterLanguage, KafkaAttributes,
6    ListStreamsParams, ListStreamsResponse, MongoAttributes, MysqlAttributes, PageInfo,
7    PostgresAttributes, ProductType, RedisAttributes, S3Attributes, SnowflakeAttributes, Stream,
8    StreamDataset, StreamDestination, StreamMetadataLocation, StreamRegion, StreamStatus,
9    TestFilterParams, TestFilterResponse, UpdateStreamParams, WebhookAttributes,
10};
11
12use crate::{config::StreamsConfig, errors::SdkError, SdkConfig};
13
14const STREAMS_BASE_URL: &str = "https://api.quicknode.com/streams/rest/v1/";
15
16pub(crate) struct ResolvedStreamsConfig {
17    pub(crate) base_url: reqwest::Url,
18}
19
20impl ResolvedStreamsConfig {
21    pub(crate) fn from_config(config: Option<&StreamsConfig>) -> Result<Self, SdkError> {
22        let url_str = config
23            .and_then(|s| s.base_url.as_deref())
24            .unwrap_or(STREAMS_BASE_URL);
25        let mut base_url = reqwest::Url::parse(url_str)?;
26        if !base_url.path().ends_with('/') {
27            base_url.set_path(&format!("{}/", base_url.path()));
28        }
29        Ok(Self { base_url })
30    }
31}
32
33// ── Client ─────────────────────────────────────────────────────────────────
34
35/// Client for the Quicknode Streams REST API. Create, list, update, and control
36/// blockchain data streams that deliver filtered on-chain events to configured
37/// destinations.
38#[derive(Debug, Clone)]
39pub struct StreamsApiClient {
40    config: SdkConfig,
41}
42
43impl StreamsApiClient {
44    pub fn new(config: SdkConfig) -> Self {
45        Self { config }
46    }
47
48    /// Creates a new Stream on a given blockchain network and dataset, delivering
49    /// batches to the configured destination. Start from a specific block for
50    /// backfills or from the tip for real-time streaming, and optionally attach
51    /// a base64-encoded JavaScript filter to transform data before delivery.
52    /// The stream can be created in an active or paused state and supports
53    /// reorg handling, distance-from-tip, elastic batching, notification emails,
54    /// and extra destinations for multi-destination delivery.
55    pub async fn create_stream(&self, params: &CreateStreamParams) -> Result<Stream, SdkError> {
56        let url = self.config.streams().base_url.join("streams")?;
57        let resp = self
58            .config
59            .http_client()
60            .post(url)
61            .json(params)
62            .send()
63            .await
64            .map_err(SdkError::Http)?;
65
66        let status = resp.status();
67        let body = resp.text().await.map_err(SdkError::Http)?;
68
69        if !status.is_success() {
70            return Err(SdkError::Api { status, body });
71        }
72        serde_json::from_str(&body).map_err(|source| SdkError::Decode { source, body })
73    }
74
75    /// Returns a paginated list of streams on the account. Each stream includes
76    /// its full configuration — identifiers, timestamps, network and dataset,
77    /// filter, block range, destination settings, and operational status — and
78    /// surfaces advanced features such as elastic batching and extra
79    /// destinations, where batches must be delivered to every configured
80    /// destination before the stream advances. Supports pagination via
81    /// `offset`/`limit` and sorting via `order_by`/`order_direction`, and can
82    /// filter by stream type.
83    pub async fn list_streams(
84        &self,
85        params: &ListStreamsParams,
86    ) -> Result<ListStreamsResponse, SdkError> {
87        let mut url = self.config.streams().base_url.join("streams")?;
88        {
89            let mut pairs = url.query_pairs_mut();
90            if let Some(t) = &params.stream_type {
91                pairs.append_pair("type", t);
92            }
93            if let Some(v) = params.offset {
94                pairs.append_pair("offset", &v.to_string());
95            }
96            if let Some(v) = params.limit {
97                pairs.append_pair("limit", &v.to_string());
98            }
99            if let Some(v) = &params.order_by {
100                pairs.append_pair("order_by", v);
101            }
102            if let Some(v) = &params.order_direction {
103                pairs.append_pair("order_direction", v);
104            }
105        }
106        let resp = self
107            .config
108            .http_client()
109            .get(url)
110            .send()
111            .await
112            .map_err(SdkError::Http)?;
113        let status = resp.status();
114        let body = resp.text().await.map_err(SdkError::Http)?;
115        if !status.is_success() {
116            return Err(SdkError::Api { status, body });
117        }
118        serde_json::from_str(&body).map_err(|source| SdkError::Decode { source, body })
119    }
120
121    /// Removes every stream on the account. Takes no filters and cannot be
122    /// undone.
123    pub async fn delete_all_streams(&self) -> Result<(), SdkError> {
124        let url = self.config.streams().base_url.join("streams")?;
125        let resp = self
126            .config
127            .http_client()
128            .delete(url)
129            .send()
130            .await
131            .map_err(SdkError::Http)?;
132        let status = resp.status();
133        if !status.is_success() {
134            let body = resp.text().await.map_err(SdkError::Http)?;
135            return Err(SdkError::Api { status, body });
136        }
137        Ok(())
138    }
139
140    /// Returns a single stream by ID, including its full configuration and
141    /// current status.
142    pub async fn get_stream(&self, id: &str) -> Result<Stream, SdkError> {
143        let url = self
144            .config
145            .streams()
146            .base_url
147            .join(&format!("streams/{id}"))?;
148        let resp = self
149            .config
150            .http_client()
151            .get(url)
152            .send()
153            .await
154            .map_err(SdkError::Http)?;
155        let status = resp.status();
156        let body = resp.text().await.map_err(SdkError::Http)?;
157        if !status.is_success() {
158            return Err(SdkError::Api { status, body });
159        }
160        serde_json::from_str(&body).map_err(|source| SdkError::Decode { source, body })
161    }
162
163    /// Updates an existing stream's configuration. Only fields present on
164    /// `params` are modified; omitted fields are left unchanged.
165    pub async fn update_stream(
166        &self,
167        id: &str,
168        params: &UpdateStreamParams,
169    ) -> Result<Stream, SdkError> {
170        let url = self
171            .config
172            .streams()
173            .base_url
174            .join(&format!("streams/{id}"))?;
175        let resp = self
176            .config
177            .http_client()
178            .patch(url)
179            .json(params)
180            .send()
181            .await
182            .map_err(SdkError::Http)?;
183        let status = resp.status();
184        let body = resp.text().await.map_err(SdkError::Http)?;
185        if !status.is_success() {
186            return Err(SdkError::Api { status, body });
187        }
188        serde_json::from_str(&body).map_err(|source| SdkError::Decode { source, body })
189    }
190
191    /// Deletes a single stream by ID.
192    pub async fn delete_stream(&self, id: &str) -> Result<(), SdkError> {
193        let url = self
194            .config
195            .streams()
196            .base_url
197            .join(&format!("streams/{id}"))?;
198        let resp = self
199            .config
200            .http_client()
201            .delete(url)
202            .send()
203            .await
204            .map_err(SdkError::Http)?;
205        let status = resp.status();
206        if !status.is_success() {
207            let body = resp.text().await.map_err(SdkError::Http)?;
208            return Err(SdkError::Api { status, body });
209        }
210        Ok(())
211    }
212
213    /// Activates a stream by ID, resuming delivery from its current position.
214    pub async fn activate_stream(&self, id: &str) -> Result<(), SdkError> {
215        let url = self
216            .config
217            .streams()
218            .base_url
219            .join(&format!("streams/{id}/activate"))?;
220        let resp = self
221            .config
222            .http_client()
223            .post(url)
224            .send()
225            .await
226            .map_err(SdkError::Http)?;
227        let status = resp.status();
228        if !status.is_success() {
229            let body = resp.text().await.map_err(SdkError::Http)?;
230            return Err(SdkError::Api { status, body });
231        }
232        Ok(())
233    }
234
235    /// Pauses a stream by ID, halting delivery until it is activated again.
236    pub async fn pause_stream(&self, id: &str) -> Result<(), SdkError> {
237        let url = self
238            .config
239            .streams()
240            .base_url
241            .join(&format!("streams/{id}/pause"))?;
242        let resp = self
243            .config
244            .http_client()
245            .post(url)
246            .send()
247            .await
248            .map_err(SdkError::Http)?;
249        let status = resp.status();
250        if !status.is_success() {
251            let body = resp.text().await.map_err(SdkError::Http)?;
252            return Err(SdkError::Api { status, body });
253        }
254        Ok(())
255    }
256
257    /// Runs a filter function against a specified block on a given network and
258    /// dataset, returning the filter's output so it can be validated before
259    /// being attached to a live stream.
260    pub async fn test_filter(
261        &self,
262        params: &TestFilterParams,
263    ) -> Result<TestFilterResponse, SdkError> {
264        let url = self.config.streams().base_url.join("streams/test_filter")?;
265        let resp = self
266            .config
267            .http_client()
268            .post(url)
269            .json(params)
270            .send()
271            .await
272            .map_err(SdkError::Http)?;
273        let status = resp.status();
274        let body = resp.text().await.map_err(SdkError::Http)?;
275        if !status.is_success() {
276            return Err(SdkError::Api { status, body });
277        }
278        serde_json::from_str(&body).map_err(|source| SdkError::Decode { source, body })
279    }
280
281    /// Returns the total count of currently enabled (active) streams on the
282    /// account, optionally filtered by stream type.
283    pub async fn get_enabled_count(
284        &self,
285        stream_type: Option<&str>,
286    ) -> Result<EnabledCountResponse, SdkError> {
287        let mut url = self
288            .config
289            .streams()
290            .base_url
291            .join("streams/enabled_count")?;
292        if let Some(t) = stream_type {
293            url.query_pairs_mut().append_pair("type", t);
294        }
295        let resp = self
296            .config
297            .http_client()
298            .get(url)
299            .send()
300            .await
301            .map_err(SdkError::Http)?;
302        let status = resp.status();
303        let body = resp.text().await.map_err(SdkError::Http)?;
304        if !status.is_success() {
305            return Err(SdkError::Api { status, body });
306        }
307        serde_json::from_str(&body).map_err(|source| SdkError::Decode { source, body })
308    }
309}
310
311// ── Tests ──────────────────────────────────────────────────────────────────
312
313#[cfg(test)]
314#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
315mod tests {
316    use super::*;
317    use crate::{QuicknodeSdk, SdkFullConfig, StreamsConfig};
318    use wiremock::matchers::{method, path};
319    use wiremock::{Mock, MockServer, ResponseTemplate};
320
321    fn make_sdk(base_url: String) -> QuicknodeSdk {
322        QuicknodeSdk::new(&SdkFullConfig {
323            api_key: "test-key".to_string(),
324            http: None,
325            admin: None,
326            streams: Some(StreamsConfig {
327                base_url: Some(base_url),
328            }),
329            webhooks: None,
330            kvstore: None,
331        })
332        .unwrap()
333    }
334
335    fn webhook_params() -> CreateStreamParams {
336        CreateStreamParams {
337            name: "test-stream".to_string(),
338            region: StreamRegion::UsaEast,
339            network: "ethereum-mainnet".to_string(),
340            dataset: StreamDataset::Block,
341            start_range: 17000000,
342            end_range: -1,
343            destination_attributes: DestinationAttributes::Webhook(WebhookAttributes {
344                url: "https://example.com/webhook".to_string(),
345                max_retry: 3,
346                retry_interval_sec: 1,
347                post_timeout_sec: 10,
348                compression: "none".to_string(),
349                security_token: None,
350            }),
351            plan: "growth_plan".to_string(),
352            threshold_fetch_buffer: 1000,
353            dataset_batch_size: None,
354            max_batch_size: None,
355            max_buffer_range_size: None,
356            max_buffer_processing_workers: None,
357            keep_distance_from_tip: None,
358            filter_function: None,
359            filter_language: None,
360            address_book_config: None,
361            include_stream_metadata: None,
362            product_type: None,
363            status: None,
364            notification_email: None,
365            charge_min_cap: None,
366            fix_block_reorgs: None,
367            elastic_batch_enabled: None,
368            extra_destinations: None,
369        }
370    }
371
372    fn stream_response_json() -> serde_json::Value {
373        serde_json::json!({
374            "id": "7d3c1a22-4f9e-4b1e-8b3d-1234567890ab",
375            "name": "test-stream",
376            "status": "active",
377            "created_at": "2026-03-19T12:00:00Z",
378            "updated_at": "2026-03-19T12:00:00Z",
379            "sequence": 0,
380            "network": "ethereum-mainnet",
381            "dataset": "block",
382            "region": "usa_east",
383            "destination": "webhook",
384            "destination_attributes": {
385                "url": "https://example.com/webhook",
386                "max_retry": 3,
387                "retry_interval_sec": 1,
388                "post_timeout_sec": 10,
389                "compression": "none"
390            },
391            "start_range": 17000000,
392            "end_range": -1
393        })
394    }
395
396    #[tokio::test]
397    async fn create_stream_success() {
398        let server = MockServer::start().await;
399
400        Mock::given(method("POST"))
401            .and(path("/streams"))
402            .respond_with(ResponseTemplate::new(200).set_body_json(stream_response_json()))
403            .mount(&server)
404            .await;
405
406        let sdk = make_sdk(format!("{}/", server.uri()));
407        let resp = sdk.streams.create_stream(&webhook_params()).await.unwrap();
408
409        assert_eq!(resp.id, "7d3c1a22-4f9e-4b1e-8b3d-1234567890ab");
410        assert_eq!(resp.name, "test-stream");
411        assert_eq!(resp.status, "active");
412        assert_eq!(resp.network, "ethereum-mainnet");
413        assert_eq!(resp.dataset, "block");
414        // Verify the full destination_attributes round-trip on the response
415        // side. Without this assertion, serde's flatten+Option silently
416        // swallows malformed destination_attributes as None.
417        match resp.destination_attributes {
418            Some(DestinationAttributes::Webhook(attrs)) => {
419                assert_eq!(attrs.url, "https://example.com/webhook");
420                assert_eq!(attrs.max_retry, 3);
421            }
422            other => panic!("expected Webhook destination, got {other:?}"),
423        }
424    }
425
426    #[tokio::test]
427    async fn create_stream_sends_typed_webhook_destination() {
428        use wiremock::matchers::body_partial_json;
429        let server = MockServer::start().await;
430
431        Mock::given(method("POST"))
432            .and(path("/streams"))
433            .and(body_partial_json(serde_json::json!({
434                "destination": "webhook",
435                "destination_attributes": {
436                    "url": "https://example.com/webhook",
437                    "max_retry": 3,
438                    "retry_interval_sec": 1,
439                    "post_timeout_sec": 10,
440                    "compression": "none"
441                }
442            })))
443            .respond_with(ResponseTemplate::new(200).set_body_json(stream_response_json()))
444            .mount(&server)
445            .await;
446
447        let sdk = make_sdk(format!("{}/", server.uri()));
448        let params = CreateStreamParams {
449            destination_attributes: DestinationAttributes::Webhook(WebhookAttributes {
450                url: "https://example.com/webhook".to_string(),
451                max_retry: 3,
452                retry_interval_sec: 1,
453                post_timeout_sec: 10,
454                compression: "none".to_string(),
455                security_token: None,
456            }),
457            ..webhook_params()
458        };
459        sdk.streams.create_stream(&params).await.unwrap();
460    }
461
462    #[tokio::test]
463    async fn create_stream_sends_extra_destinations() {
464        use wiremock::matchers::body_partial_json;
465        let server = MockServer::start().await;
466
467        Mock::given(method("POST"))
468            .and(path("/streams"))
469            .and(body_partial_json(serde_json::json!({
470                "extra_destinations": [
471                    {
472                        "destination": "webhook",
473                        "destination_attributes": {
474                            "url": "https://example.com/extra-hook",
475                            "max_retry": 5,
476                            "retry_interval_sec": 2,
477                            "post_timeout_sec": 15,
478                            "compression": "none"
479                        }
480                    },
481                    {
482                        "destination": "s3",
483                        "destination_attributes": {
484                            "endpoint": "s3.example.com",
485                            "access_key": "AKIA",
486                            "secret_key": "secret",
487                            "bucket": "my-bucket",
488                            "object_prefix": "streams/",
489                            "compression": "gzip",
490                            "file_type": ".json",
491                            "max_retry": 3,
492                            "retry_interval_sec": 1
493                        }
494                    }
495                ]
496            })))
497            .respond_with(ResponseTemplate::new(200).set_body_json(stream_response_json()))
498            .mount(&server)
499            .await;
500
501        let sdk = make_sdk(format!("{}/", server.uri()));
502        let params = CreateStreamParams {
503            extra_destinations: Some(vec![
504                DestinationAttributes::Webhook(WebhookAttributes {
505                    url: "https://example.com/extra-hook".to_string(),
506                    max_retry: 5,
507                    retry_interval_sec: 2,
508                    post_timeout_sec: 15,
509                    compression: "none".to_string(),
510                    security_token: None,
511                }),
512                DestinationAttributes::S3(S3Attributes {
513                    endpoint: "s3.example.com".to_string(),
514                    access_key: "AKIA".to_string(),
515                    secret_key: "secret".to_string(),
516                    bucket: "my-bucket".to_string(),
517                    object_prefix: "streams/".to_string(),
518                    compression: "gzip".to_string(),
519                    file_type: ".json".to_string(),
520                    max_retry: 3,
521                    retry_interval_sec: 1,
522                    use_ssl: None,
523                }),
524            ]),
525            ..webhook_params()
526        };
527        sdk.streams.create_stream(&params).await.unwrap();
528    }
529
530    #[tokio::test]
531    async fn update_stream_sends_extra_destinations() {
532        use wiremock::matchers::body_partial_json;
533        let server = MockServer::start().await;
534
535        Mock::given(method("PATCH"))
536            .and(path("/streams/test-id"))
537            .and(body_partial_json(serde_json::json!({
538                "extra_destinations": [
539                    {
540                        "destination": "webhook",
541                        "destination_attributes": {
542                            "url": "https://example.com/patched-hook",
543                            "max_retry": 1,
544                            "retry_interval_sec": 1,
545                            "post_timeout_sec": 5,
546                            "compression": "none"
547                        }
548                    }
549                ]
550            })))
551            .respond_with(ResponseTemplate::new(200).set_body_json(stream_response_json()))
552            .mount(&server)
553            .await;
554
555        let sdk = make_sdk(format!("{}/", server.uri()));
556        let params = UpdateStreamParams {
557            extra_destinations: Some(vec![DestinationAttributes::Webhook(WebhookAttributes {
558                url: "https://example.com/patched-hook".to_string(),
559                max_retry: 1,
560                retry_interval_sec: 1,
561                post_timeout_sec: 5,
562                compression: "none".to_string(),
563                security_token: None,
564            })]),
565            ..Default::default()
566        };
567        sdk.streams.update_stream("test-id", &params).await.unwrap();
568    }
569
570    #[tokio::test]
571    async fn get_stream_parses_extra_destinations() {
572        let server = MockServer::start().await;
573        let mut body = stream_response_json();
574        body["extra_destinations"] = serde_json::json!([
575            {
576                "destination": "webhook",
577                "destination_attributes": {
578                    "url": "https://example.com/extra",
579                    "max_retry": 2,
580                    "retry_interval_sec": 1,
581                    "post_timeout_sec": 10,
582                    "compression": "none"
583                }
584            }
585        ]);
586        Mock::given(method("GET"))
587            .and(path("/streams/test-id"))
588            .respond_with(ResponseTemplate::new(200).set_body_json(body))
589            .mount(&server)
590            .await;
591
592        let sdk = make_sdk(format!("{}/", server.uri()));
593        let resp = sdk.streams.get_stream("test-id").await.unwrap();
594        let extras = resp.extra_destinations.expect("extra_destinations present");
595        assert_eq!(extras.len(), 1);
596        match &extras[0] {
597            DestinationAttributes::Webhook(w) => {
598                assert_eq!(w.url, "https://example.com/extra");
599                assert_eq!(w.max_retry, 2);
600            }
601            other => panic!("expected Webhook, got {other:?}"),
602        }
603    }
604
605    #[tokio::test]
606    async fn create_stream_api_error() {
607        let server = MockServer::start().await;
608
609        Mock::given(method("POST"))
610            .and(path("/streams"))
611            .respond_with(ResponseTemplate::new(400).set_body_string("Bad Request"))
612            .mount(&server)
613            .await;
614
615        let sdk = make_sdk(format!("{}/", server.uri()));
616        let err = sdk
617            .streams
618            .create_stream(&webhook_params())
619            .await
620            .unwrap_err();
621
622        assert!(matches!(err, SdkError::Api { .. }));
623    }
624
625    #[tokio::test]
626    async fn create_stream_server_error() {
627        let server = MockServer::start().await;
628
629        Mock::given(method("POST"))
630            .and(path("/streams"))
631            .respond_with(ResponseTemplate::new(500).set_body_string("Internal Server Error"))
632            .mount(&server)
633            .await;
634
635        let sdk = make_sdk(format!("{}/", server.uri()));
636        let err = sdk
637            .streams
638            .create_stream(&webhook_params())
639            .await
640            .unwrap_err();
641
642        assert!(matches!(err, SdkError::Api { .. }));
643    }
644
645    #[tokio::test]
646    async fn list_streams_success() {
647        let server = MockServer::start().await;
648        let response = serde_json::json!({
649            "data": [stream_response_json()],
650            "pageInfo": { "limit": 100, "offset": 0, "total": 1 }
651        });
652        Mock::given(method("GET"))
653            .and(path("/streams"))
654            .respond_with(ResponseTemplate::new(200).set_body_json(response))
655            .mount(&server)
656            .await;
657        let sdk = make_sdk(format!("{}/", server.uri()));
658        let resp = sdk
659            .streams
660            .list_streams(&ListStreamsParams::default())
661            .await
662            .unwrap();
663        assert_eq!(resp.data.len(), 1);
664        assert_eq!(resp.page_info.total, 1);
665    }
666
667    #[tokio::test]
668    async fn list_streams_api_error() {
669        let server = MockServer::start().await;
670        Mock::given(method("GET"))
671            .and(path("/streams"))
672            .respond_with(ResponseTemplate::new(400).set_body_string("Bad Request"))
673            .mount(&server)
674            .await;
675        let sdk = make_sdk(format!("{}/", server.uri()));
676        let err = sdk
677            .streams
678            .list_streams(&ListStreamsParams::default())
679            .await
680            .unwrap_err();
681        assert!(matches!(err, SdkError::Api { .. }));
682    }
683
684    #[tokio::test]
685    async fn list_streams_server_error() {
686        let server = MockServer::start().await;
687        Mock::given(method("GET"))
688            .and(path("/streams"))
689            .respond_with(ResponseTemplate::new(500).set_body_string("Internal Server Error"))
690            .mount(&server)
691            .await;
692        let sdk = make_sdk(format!("{}/", server.uri()));
693        let err = sdk
694            .streams
695            .list_streams(&ListStreamsParams::default())
696            .await
697            .unwrap_err();
698        assert!(matches!(err, SdkError::Api { .. }));
699    }
700
701    #[tokio::test]
702    async fn get_stream_success() {
703        let server = MockServer::start().await;
704        Mock::given(method("GET"))
705            .and(path("/streams/test-id"))
706            .respond_with(ResponseTemplate::new(200).set_body_json(stream_response_json()))
707            .mount(&server)
708            .await;
709        let sdk = make_sdk(format!("{}/", server.uri()));
710        let resp = sdk.streams.get_stream("test-id").await.unwrap();
711        assert_eq!(resp.id, "7d3c1a22-4f9e-4b1e-8b3d-1234567890ab");
712    }
713
714    #[tokio::test]
715    async fn get_stream_not_found() {
716        let server = MockServer::start().await;
717        Mock::given(method("GET"))
718            .and(path("/streams/test-id"))
719            .respond_with(ResponseTemplate::new(404).set_body_string("Not Found"))
720            .mount(&server)
721            .await;
722        let sdk = make_sdk(format!("{}/", server.uri()));
723        let err = sdk.streams.get_stream("test-id").await.unwrap_err();
724        assert!(matches!(err, SdkError::Api { .. }));
725    }
726
727    #[tokio::test]
728    async fn get_stream_server_error() {
729        let server = MockServer::start().await;
730        Mock::given(method("GET"))
731            .and(path("/streams/test-id"))
732            .respond_with(ResponseTemplate::new(500).set_body_string("Internal Server Error"))
733            .mount(&server)
734            .await;
735        let sdk = make_sdk(format!("{}/", server.uri()));
736        let err = sdk.streams.get_stream("test-id").await.unwrap_err();
737        assert!(matches!(err, SdkError::Api { .. }));
738    }
739
740    #[tokio::test]
741    async fn update_stream_success() {
742        let server = MockServer::start().await;
743        let mut updated = stream_response_json();
744        updated["name"] = serde_json::json!("updated-name");
745        Mock::given(method("PATCH"))
746            .and(path("/streams/test-id"))
747            .respond_with(ResponseTemplate::new(200).set_body_json(updated))
748            .mount(&server)
749            .await;
750        let sdk = make_sdk(format!("{}/", server.uri()));
751        let params = UpdateStreamParams {
752            name: Some("updated-name".to_string()),
753            ..Default::default()
754        };
755        let resp = sdk.streams.update_stream("test-id", &params).await.unwrap();
756        assert_eq!(resp.name, "updated-name");
757    }
758
759    #[tokio::test]
760    async fn update_stream_api_error() {
761        let server = MockServer::start().await;
762        Mock::given(method("PATCH"))
763            .and(path("/streams/test-id"))
764            .respond_with(ResponseTemplate::new(400).set_body_string("Bad Request"))
765            .mount(&server)
766            .await;
767        let sdk = make_sdk(format!("{}/", server.uri()));
768        let params = UpdateStreamParams::default();
769        let err = sdk
770            .streams
771            .update_stream("test-id", &params)
772            .await
773            .unwrap_err();
774        assert!(matches!(err, SdkError::Api { .. }));
775    }
776
777    #[tokio::test]
778    async fn update_stream_server_error() {
779        let server = MockServer::start().await;
780        Mock::given(method("PATCH"))
781            .and(path("/streams/test-id"))
782            .respond_with(ResponseTemplate::new(500).set_body_string("Internal Server Error"))
783            .mount(&server)
784            .await;
785        let sdk = make_sdk(format!("{}/", server.uri()));
786        let params = UpdateStreamParams::default();
787        let err = sdk
788            .streams
789            .update_stream("test-id", &params)
790            .await
791            .unwrap_err();
792        assert!(matches!(err, SdkError::Api { .. }));
793    }
794
795    #[tokio::test]
796    async fn delete_stream_success() {
797        let server = MockServer::start().await;
798        Mock::given(method("DELETE"))
799            .and(path("/streams/test-id"))
800            .respond_with(ResponseTemplate::new(200))
801            .mount(&server)
802            .await;
803        let sdk = make_sdk(format!("{}/", server.uri()));
804        sdk.streams.delete_stream("test-id").await.unwrap();
805    }
806
807    #[tokio::test]
808    async fn delete_stream_not_found() {
809        let server = MockServer::start().await;
810        Mock::given(method("DELETE"))
811            .and(path("/streams/test-id"))
812            .respond_with(ResponseTemplate::new(404).set_body_string("Not Found"))
813            .mount(&server)
814            .await;
815        let sdk = make_sdk(format!("{}/", server.uri()));
816        let err = sdk.streams.delete_stream("test-id").await.unwrap_err();
817        assert!(matches!(err, SdkError::Api { .. }));
818    }
819
820    #[tokio::test]
821    async fn delete_stream_server_error() {
822        let server = MockServer::start().await;
823        Mock::given(method("DELETE"))
824            .and(path("/streams/test-id"))
825            .respond_with(ResponseTemplate::new(500).set_body_string("Internal Server Error"))
826            .mount(&server)
827            .await;
828        let sdk = make_sdk(format!("{}/", server.uri()));
829        let err = sdk.streams.delete_stream("test-id").await.unwrap_err();
830        assert!(matches!(err, SdkError::Api { .. }));
831    }
832
833    #[tokio::test]
834    async fn delete_all_streams_success() {
835        let server = MockServer::start().await;
836        Mock::given(method("DELETE"))
837            .and(path("/streams"))
838            .respond_with(ResponseTemplate::new(204))
839            .mount(&server)
840            .await;
841        let sdk = make_sdk(format!("{}/", server.uri()));
842        sdk.streams.delete_all_streams().await.unwrap();
843    }
844
845    #[tokio::test]
846    async fn delete_all_streams_not_found() {
847        let server = MockServer::start().await;
848        Mock::given(method("DELETE"))
849            .and(path("/streams"))
850            .respond_with(ResponseTemplate::new(404).set_body_string("Not Found"))
851            .mount(&server)
852            .await;
853        let sdk = make_sdk(format!("{}/", server.uri()));
854        let err = sdk.streams.delete_all_streams().await.unwrap_err();
855        assert!(matches!(err, SdkError::Api { .. }));
856    }
857
858    #[tokio::test]
859    async fn activate_stream_success() {
860        let server = MockServer::start().await;
861        Mock::given(method("POST"))
862            .and(path("/streams/test-id/activate"))
863            .respond_with(ResponseTemplate::new(201))
864            .mount(&server)
865            .await;
866        let sdk = make_sdk(format!("{}/", server.uri()));
867        sdk.streams.activate_stream("test-id").await.unwrap();
868    }
869
870    #[tokio::test]
871    async fn activate_stream_not_found() {
872        let server = MockServer::start().await;
873        Mock::given(method("POST"))
874            .and(path("/streams/test-id/activate"))
875            .respond_with(ResponseTemplate::new(404).set_body_string("Not Found"))
876            .mount(&server)
877            .await;
878        let sdk = make_sdk(format!("{}/", server.uri()));
879        let err = sdk.streams.activate_stream("test-id").await.unwrap_err();
880        assert!(matches!(err, SdkError::Api { .. }));
881    }
882
883    #[tokio::test]
884    async fn activate_stream_server_error() {
885        let server = MockServer::start().await;
886        Mock::given(method("POST"))
887            .and(path("/streams/test-id/activate"))
888            .respond_with(ResponseTemplate::new(500).set_body_string("Internal Server Error"))
889            .mount(&server)
890            .await;
891        let sdk = make_sdk(format!("{}/", server.uri()));
892        let err = sdk.streams.activate_stream("test-id").await.unwrap_err();
893        assert!(matches!(err, SdkError::Api { .. }));
894    }
895
896    #[tokio::test]
897    async fn pause_stream_success() {
898        let server = MockServer::start().await;
899        Mock::given(method("POST"))
900            .and(path("/streams/test-id/pause"))
901            .respond_with(ResponseTemplate::new(201))
902            .mount(&server)
903            .await;
904        let sdk = make_sdk(format!("{}/", server.uri()));
905        sdk.streams.pause_stream("test-id").await.unwrap();
906    }
907
908    #[tokio::test]
909    async fn pause_stream_not_found() {
910        let server = MockServer::start().await;
911        Mock::given(method("POST"))
912            .and(path("/streams/test-id/pause"))
913            .respond_with(ResponseTemplate::new(404).set_body_string("Not Found"))
914            .mount(&server)
915            .await;
916        let sdk = make_sdk(format!("{}/", server.uri()));
917        let err = sdk.streams.pause_stream("test-id").await.unwrap_err();
918        assert!(matches!(err, SdkError::Api { .. }));
919    }
920
921    #[tokio::test]
922    async fn pause_stream_server_error() {
923        let server = MockServer::start().await;
924        Mock::given(method("POST"))
925            .and(path("/streams/test-id/pause"))
926            .respond_with(ResponseTemplate::new(500).set_body_string("Internal Server Error"))
927            .mount(&server)
928            .await;
929        let sdk = make_sdk(format!("{}/", server.uri()));
930        let err = sdk.streams.pause_stream("test-id").await.unwrap_err();
931        assert!(matches!(err, SdkError::Api { .. }));
932    }
933
934    #[tokio::test]
935    async fn test_filter_success() {
936        let server = MockServer::start().await;
937        let response = serde_json::json!({ "result": {"hash": "0xabc"}, "logs": [] });
938        Mock::given(method("POST"))
939            .and(path("/streams/test_filter"))
940            .respond_with(ResponseTemplate::new(201).set_body_json(response))
941            .mount(&server)
942            .await;
943        let sdk = make_sdk(format!("{}/", server.uri()));
944        let params = TestFilterParams {
945            network: "ethereum-mainnet".to_string(),
946            dataset: StreamDataset::Block,
947            block: "17811625".to_string(),
948            filter_function: None,
949            filter_language: None,
950            address_book_config: None,
951        };
952        let resp = sdk.streams.test_filter(&params).await.unwrap();
953        assert!(resp.logs.is_empty());
954    }
955
956    #[tokio::test]
957    async fn test_filter_api_error() {
958        let server = MockServer::start().await;
959        Mock::given(method("POST"))
960            .and(path("/streams/test_filter"))
961            .respond_with(ResponseTemplate::new(400).set_body_string("Bad Request"))
962            .mount(&server)
963            .await;
964        let sdk = make_sdk(format!("{}/", server.uri()));
965        let params = TestFilterParams {
966            network: "ethereum-mainnet".to_string(),
967            dataset: StreamDataset::Block,
968            block: "17811625".to_string(),
969            filter_function: None,
970            filter_language: None,
971            address_book_config: None,
972        };
973        let err = sdk.streams.test_filter(&params).await.unwrap_err();
974        assert!(matches!(err, SdkError::Api { .. }));
975    }
976
977    #[tokio::test]
978    async fn get_enabled_count_success() {
979        let server = MockServer::start().await;
980        Mock::given(method("GET"))
981            .and(path("/streams/enabled_count"))
982            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({"total": 3})))
983            .mount(&server)
984            .await;
985        let sdk = make_sdk(format!("{}/", server.uri()));
986        let resp = sdk.streams.get_enabled_count(None).await.unwrap();
987        assert_eq!(resp.total, 3);
988    }
989
990    #[tokio::test]
991    async fn get_enabled_count_api_error() {
992        let server = MockServer::start().await;
993        Mock::given(method("GET"))
994            .and(path("/streams/enabled_count"))
995            .respond_with(ResponseTemplate::new(400).set_body_string("Bad Request"))
996            .mount(&server)
997            .await;
998        let sdk = make_sdk(format!("{}/", server.uri()));
999        let err = sdk.streams.get_enabled_count(None).await.unwrap_err();
1000        assert!(matches!(err, SdkError::Api { .. }));
1001    }
1002
1003    #[tokio::test]
1004    async fn get_enabled_count_server_error() {
1005        let server = MockServer::start().await;
1006        Mock::given(method("GET"))
1007            .and(path("/streams/enabled_count"))
1008            .respond_with(ResponseTemplate::new(500).set_body_string("Internal Server Error"))
1009            .mount(&server)
1010            .await;
1011        let sdk = make_sdk(format!("{}/", server.uri()));
1012        let err = sdk.streams.get_enabled_count(None).await.unwrap_err();
1013        assert!(matches!(err, SdkError::Api { .. }));
1014    }
1015}
quicknode_sdk/streams/mod.rs

quicknode_sdk/streams/
mod.rs
