Add additional endpoint integration tests (#383)

Add tests for filtering by msg channels and endpoint header validation/sending

server/svix-server/src/v1/endpoints/endpoint/mod.rs

@@ -166,11 +166,11 @@ pub struct RecoverIn {
     pub since: DateTime<Utc>,
 }
 
-#[derive(Clone, Debug, PartialEq, Validate, Deserialize)]
+#[derive(Clone, Debug, PartialEq, Validate, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct EndpointHeadersIn {
     #[validate]
-    headers: EndpointHeaders,
+    pub headers: EndpointHeaders,
 }
 
 impl ModelIn for EndpointHeadersIn {
@@ -181,11 +181,11 @@ impl ModelIn for EndpointHeadersIn {
     }
 }
 
-#[derive(Clone, Debug, PartialEq, Serialize, Default)]
+#[derive(Clone, Debug, PartialEq, Deserialize, Serialize, Default)]
 #[serde(rename_all = "camelCase")]
-struct EndpointHeadersOut {
-    headers: HashMap<String, String>,
-    sensitive: HashSet<String>,
+pub struct EndpointHeadersOut {
+    pub headers: HashMap<String, String>,
+    pub sensitive: HashSet<String>,
 }
 
 impl EndpointHeadersOut {
@@ -212,11 +212,11 @@ impl From<EndpointHeaders> for EndpointHeadersOut {
     }
 }
 
-#[derive(Clone, Debug, PartialEq, Validate, Deserialize)]
+#[derive(Clone, Debug, PartialEq, Validate, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
-struct EndpointHeadersPatchIn {
+pub struct EndpointHeadersPatchIn {
     #[validate]
-    headers: EndpointHeaders,
+    pub headers: EndpointHeaders,
 }
 
 impl ModelIn for EndpointHeadersPatchIn {

server/svix-server/tests/e2e_endpoint.rs

@@ -1,20 +1,26 @@
 // SPDX-FileCopyrightText: © 2022 Svix Authors
 // SPDX-License-Identifier: MIT
 
-use std::{collections::HashSet, time::Duration};
+use std::{
+    collections::{HashMap, HashSet},
+    time::Duration,
+};
 
 use anyhow::Result;
 use chrono::{DateTime, Utc};
 use reqwest::StatusCode;
 
 use svix_server::{
     core::types::{
-        ApplicationId, EndpointId, EndpointSecret, EndpointUid, EventChannel, EventChannelSet,
-        EventTypeName, EventTypeNameSet, ExpiringSigningKeys,
+        ApplicationId, EndpointHeaders, EndpointId, EndpointSecret, EndpointUid, EventChannel,
+        EventChannelSet, EventTypeName, EventTypeNameSet, ExpiringSigningKeys,
     },
     v1::{
         endpoints::{
-            endpoint::{EndpointIn, EndpointOut, EndpointSecretOut, RecoverIn},
+            endpoint::{
+                EndpointHeadersIn, EndpointHeadersOut, EndpointHeadersPatchIn, EndpointIn,
+                EndpointOut, EndpointSecretOut, RecoverIn,
+            },
             event_type::EventTypeOut,
             message::{MessageIn, MessageOut},
         },
@@ -989,3 +995,229 @@ async fn test_msg_event_types_filter() {
                 .unwrap();
     }
 }
+
+#[tokio::test]
+async fn test_msg_channels_filter() {
+    let (client, _jh) = start_svix_server();
+
+    let app_id = create_test_app(&client, "app1").await.unwrap().id;
+
+    let receiver = TestReceiver::start(StatusCode::OK);
+
+    let ec = EventChannelSet(HashSet::from([EventChannel("tag1".to_owned())]));
+
+    for channels in [Some(ec.clone()), None] {
+        let _endp = post_endpoint(
+            &client,
+            &app_id,
+            EndpointIn {
+                url: receiver.endpoint.to_owned(),
+                version: 1,
+                channels,
+                ..Default::default()
+            },
+        )
+        .await
+        .unwrap();
+    }
+
+    for (channels, expected_count) in [(Some(ec.clone()), 2), (None, 1)] {
+        let msg: MessageOut = client
+            .post(
+                &format!("api/v1/app/{}/msg/", &app_id),
+                MessageIn {
+                    channels: channels.clone(),
+                    event_type: EventTypeName("et1".to_owned()),
+                    payload: serde_json::json!({}),
+                    uid: None,
+                },
+                StatusCode::ACCEPTED,
+            )
+            .await
+            .unwrap();
+
+        tokio::time::sleep(Duration::from_millis(100)).await;
+
+        let _list =
+            get_msg_attempt_list_and_assert_count(&client, &app_id, &msg.id, expected_count)
+                .await
+                .unwrap();
+
+        let msg: MessageOut = client
+            .get(
+                &format!("api/v1/app/{}/msg/{}", &app_id, &msg.id),
+                StatusCode::OK,
+            )
+            .await
+            .unwrap();
+
+        assert_eq!(msg.channels, channels);
+    }
+}
+
+#[tokio::test]
+async fn test_endpoint_headers_manipulation() {
+    let (client, _jh) = start_svix_server();
+
+    let app_id = create_test_app(&client, "app1").await.unwrap().id;
+
+    let endp = create_test_endpoint(&client, &app_id, "http://www.example.com")
+        .await
+        .unwrap();
+
+    for bad_hdr in [
+        "content-length",
+        "some:thing",
+        "some\u{0000}thing",
+        "svix-foo",
+        "x-svix-foo",
+        "x-amzn-foo",
+    ] {
+        let _: IgnoredResponse = client
+            .put(
+                &format!("api/v1/app/{}/endpoint/{}/headers", app_id, endp.id),
+                serde_json::json!({ "headers": { bad_hdr: "123"}}),
+                StatusCode::UNPROCESSABLE_ENTITY,
+            )
+            .await
+            .unwrap();
+    }
+
+    let org_headers = EndpointHeadersIn {
+        headers: EndpointHeaders(HashMap::from([
+            ("x-test-1".to_owned(), "1".to_owned()),
+            ("x-test-2".to_owned(), "2".to_owned()),
+        ])),
+    };
+
+    let updated_headers = EndpointHeadersIn {
+        headers: EndpointHeaders(HashMap::from([
+            ("x-test-1".to_owned(), "3".to_owned()),
+            ("x-test-2".to_owned(), "2".to_owned()),
+        ])),
+    };
+
+    for hdrs in [&org_headers, &updated_headers] {
+        let _: IgnoredResponse = client
+            .put(
+                &format!("api/v1/app/{}/endpoint/{}/headers", app_id, endp.id),
+                hdrs,
+                StatusCode::NO_CONTENT,
+            )
+            .await
+            .unwrap();
+
+        let recvd_headers: EndpointHeadersOut = client
+            .get(
+                &format!("api/v1/app/{}/endpoint/{}/headers", app_id, endp.id),
+                StatusCode::OK,
+            )
+            .await
+            .unwrap();
+
+        assert_eq!(hdrs.headers.0, recvd_headers.headers);
+    }
+
+    let patched_headers_in = EndpointHeadersPatchIn {
+        headers: EndpointHeaders(HashMap::from([("x-test-3".to_owned(), "4".to_owned())])),
+    };
+
+    let _: IgnoredResponse = client
+        .patch(
+            &format!("api/v1/app/{}/endpoint/{}/headers", app_id, endp.id),
+            &patched_headers_in,
+            StatusCode::NO_CONTENT,
+        )
+        .await
+        .unwrap();
+
+    let recvd_headers: EndpointHeadersOut = client
+        .get(
+            &format!("api/v1/app/{}/endpoint/{}/headers", app_id, endp.id),
+            StatusCode::OK,
+        )
+        .await
+        .unwrap();
+
+    assert_eq!(
+        HashMap::from([
+            ("x-test-1".to_owned(), "3".to_owned()),
+            ("x-test-2".to_owned(), "2".to_owned()),
+            ("x-test-3".to_owned(), "4".to_owned()),
+        ]),
+        recvd_headers.headers
+    );
+
+    let redacted_headers = EndpointHeadersIn {
+        headers: EndpointHeaders(HashMap::from([
+            ("x-test-1".to_owned(), "1".to_owned()),
+            ("authorization".to_owned(), "secret".to_owned()),
+        ])),
+    };
+
+    let _: IgnoredResponse = client
+        .put(
+            &format!("api/v1/app/{}/endpoint/{}/headers", app_id, endp.id),
+            redacted_headers,
+            StatusCode::NO_CONTENT,
+        )
+        .await
+        .unwrap();
+
+    let recvd_headers: EndpointHeadersOut = client
+        .get(
+            &format!("api/v1/app/{}/endpoint/{}/headers", app_id, endp.id),
+            StatusCode::OK,
+        )
+        .await
+        .unwrap();
+
+    assert_eq!(
+        HashMap::from([("x-test-1".to_owned(), "1".to_owned())]),
+        recvd_headers.headers
+    );
+
+    assert_eq!(
+        HashSet::from(["authorization".to_owned()]),
+        recvd_headers.sensitive
+    );
+}
+
+#[tokio::test]
+async fn test_endpoint_headers_sending() {
+    let (client, _jh) = start_svix_server();
+
+    let app_id = create_test_app(&client, "app1").await.unwrap().id;
+
+    let mut receiver = TestReceiver::start(StatusCode::OK);
+
+    let endp = create_test_endpoint(&client, &app_id, &receiver.endpoint)
+        .await
+        .unwrap();
+
+    let headers = EndpointHeadersIn {
+        headers: EndpointHeaders(HashMap::from([
+            ("x-test-1".to_owned(), "1".to_owned()),
+            ("x-test-2".to_owned(), "2".to_owned()),
+        ])),
+    };
+
+    let _: IgnoredResponse = client
+        .put(
+            &format!("api/v1/app/{}/endpoint/{}/headers", app_id, endp.id),
+            &headers,
+            StatusCode::NO_CONTENT,
+        )
+        .await
+        .unwrap();
+
+    create_test_message(&client, &app_id, serde_json::json!({"test": "data1"}))
+        .await
+        .unwrap();
+
+    let last_headers = receiver.header_recv.recv().await.unwrap();
+
+    for (k, v) in &headers.headers.0 {
+        assert_eq!(v, last_headers.get(k).unwrap().to_str().unwrap());
+    }
+}

server/svix-server/tests/utils/mod.rs

@@ -149,6 +149,30 @@ impl TestClient {
             .await
             .context("error receiving/parsing response")
     }
+
+    pub async fn patch<I: Serialize, O: DeserializeOwned>(
+        &self,
+        endpoint: &str,
+        input: I,
+        expected_code: StatusCode,
+    ) -> Result<O> {
+        let mut req = self.client.patch(self.build_uri(endpoint));
+        req = self.add_headers(req).json(&input);
+
+        let resp = req.send().await.context("error sending request")?;
+
+        if resp.status() != expected_code {
+            anyhow::bail!(
+                "assertation failed: expected status {}, actual status {}",
+                expected_code,
+                resp.status()
+            );
+        }
+
+        resp.json()
+            .await
+            .context("error receiving/parsing response")
+    }
 }
 
 pub fn get_default_test_config() -> ConfigurationInner {