feat: Add an E2E test for killswitches (#241)

Adds a test that ensures killswitches are integrated properly into the
middlewares or request handlers. The new test module can also be used
for asserting rate limits and other limits in the future.

Author: jan-auer

Cargo.lock

@@ -49,4 +49,5 @@ tokio = "1.47.0"
 tokio-stream = "0.1.17"
 tokio-util = "0.7.15"
 tracing = "0.1.41"
+tracing-subscriber = { version = "0.3.19", features = ["env-filter", "json"] }
 uuid = { version = "1.17.0", features = ["v4"] }

Cargo.toml

@@ -50,11 +50,12 @@ tower-http = { version = "0.6.6", default-features = false, features = [
     "trace",
 ] }
 tracing = { workspace = true }
-tracing-subscriber = { version = "0.3.19", features = ["env-filter", "json"] }
+tracing-subscriber = { workspace = true }
 uuid = { workspace = true, features = ["v7"] }
 
 [dev-dependencies]
 nix = { version = "0.30.1", features = ["signal"] }
+objectstore-test = { workspace = true }
 serde_json = { workspace = true }
 stresstest = { workspace = true }
 tempfile = { workspace = true }

objectstore-server/Cargo.toml

@@ -0,0 +1,49 @@
+//! Blackbox tests for limits and restrictions.
+//!
+//! These tests assert safety-related behavior of the objectstore-server, such as enforcing
+//! maximum object sizes, rate limiting, and killswitches.
+
+use std::collections::BTreeMap;
+
+use anyhow::Result;
+use objectstore_server::config::Config;
+use objectstore_server::killswitches::{Killswitch, Killswitches};
+use objectstore_test::server::TestServer;
+
+#[tokio::test]
+async fn test_killswitches() -> Result<()> {
+    let server = TestServer::with_config(Config {
+        killswitches: Killswitches(vec![Killswitch {
+            usecase: Some("blocked".to_string()),
+            scopes: BTreeMap::from_iter([("org".to_string(), "42".to_string())]),
+        }]),
+        ..Default::default()
+    })
+    .await;
+
+    let client = reqwest::Client::new();
+
+    // Object-level
+    let response = client
+        .get(server.url("/v1/objects/blocked/org=42;project=4711/foo"))
+        .send()
+        .await?;
+    assert_eq!(response.status(), reqwest::StatusCode::FORBIDDEN);
+
+    // Collection-level
+    let response = client
+        .post(server.url("/v1/objects/blocked/org=42;project=4711/"))
+        .body("test data")
+        .send()
+        .await?;
+    assert_eq!(response.status(), reqwest::StatusCode::FORBIDDEN);
+
+    // Sanity check: Allowed access on non-existing object
+    let response = client
+        .get(server.url("/v1/objects/allowed/org=43;project=4711/foo"))
+        .send()
+        .await?;
+    assert_eq!(response.status(), reqwest::StatusCode::NOT_FOUND);
+
+    Ok(())
+}

objectstore-server/tests/limits.rs

@@ -13,3 +13,4 @@ publish = false
 objectstore-server = { workspace = true }
 tempfile = { workspace = true }
 tokio = { workspace = true }
+tracing-subscriber = { workspace = true }

objectstore-test/Cargo.toml

@@ -4,3 +4,4 @@
 //! modules for all available utilities.
 
 pub mod server;
+pub mod tracing;

objectstore-test/src/lib.rs

@@ -31,23 +31,27 @@ pub struct TestServer {
 }
 
 impl TestServer {
-    pub async fn new() -> Self {
+    /// Spawns a new test server with the given configuration.
+    ///
+    /// Unless overridden to a different kind of backend, the long-term and high-volume storage
+    /// backends will use temporary directories.
+    pub async fn with_config(mut config: Config) -> Self {
         let addr = SocketAddr::from(([127, 0, 0, 1], 0));
         let listener = TcpListener::bind(addr).unwrap();
         listener.set_nonblocking(true).unwrap();
         let socket = listener.local_addr().unwrap();
 
+        config.logging.level = "trace".parse().unwrap();
+        crate::tracing::init();
+
         let long_term_tempdir = tempfile::tempdir().unwrap();
+        if let Storage::FileSystem { ref mut path } = config.long_term_storage {
+            *path = long_term_tempdir.path().into();
+        }
         let high_volume_tempdir = tempfile::tempdir().unwrap();
-        let config = Config {
-            long_term_storage: Storage::FileSystem {
-                path: long_term_tempdir.path().into(),
-            },
-            high_volume_storage: Storage::FileSystem {
-                path: high_volume_tempdir.path().into(),
-            },
-            ..Default::default()
-        };
+        if let Storage::FileSystem { ref mut path } = config.high_volume_storage {
+            *path = high_volume_tempdir.path().into();
+        }
 
         let state = Services::spawn(config).await.unwrap();
         let app = App::new(state);
@@ -65,6 +69,11 @@ impl TestServer {
         }
     }
 
+    /// Spawns a new test server with default configuration.
+    pub async fn new() -> Self {
+        Self::with_config(Config::default()).await
+    }
+
     /// Returns a full URL pointing to the given path.
     ///
     /// This URL uses `localhost` as hostname.

objectstore-test/src/server.rs

@@ -0,0 +1,30 @@
+use tracing_subscriber::EnvFilter;
+
+const CRATE_NAMES: &[&str] = &["objectstore", "objectstore_service", "objectstore_types"];
+
+/// Initialize the logger for testing.
+///
+/// This logs to the stdout registered by the Rust test runner, and only captures logs from the
+/// calling crate.
+///
+/// # Example
+///
+/// ```
+/// objectstore_test::tracing::init();
+/// ```
+pub fn init() {
+    let mut env_filter = EnvFilter::new("ERROR");
+
+    // Add all internal modules with maximum log-level.
+    for name in CRATE_NAMES {
+        env_filter = env_filter.add_directive(format!("{name}=TRACE").parse().unwrap());
+    }
+
+    tracing_subscriber::fmt::fmt()
+        .with_env_filter(env_filter)
+        .with_target(true)
+        .with_test_writer()
+        .compact()
+        .try_init()
+        .ok();
+}