Implement runtime test services

Signed-off-by: Ryan Levick <ryan.levick@fermyon.com>

Cargo.toml

@@ -124,7 +124,7 @@ members = [
 anyhow = "1.0.75"
 http-body-util = "=0.1.0-rc.2"
 hyper = { version = "=1.0.0-rc.3", features = ["full"] }
-reqwest = { version = "0.11", features = ["stream"] }
+reqwest = { version = "0.11", features = ["stream", "blocking"] }
 tracing = { version = "0.1", features = ["log"] }
 
 wasi-common-preview1 = { version = "15.0.0", package = "wasi-common" }

tests/runtime-tests/.gitignore

@@ -0,0 +1 @@
+*.lock

tests/runtime-tests/Cargo.toml

@@ -5,10 +5,11 @@ edition = "2021"
 
 [dependencies]
 anyhow = { workspace = true }
-test-components = { path = "../test-components" }
 env_logger = "0.10.0"
+fslock = "0.2.1"
 log = "0.4"
 nix = "0.26.1"
 reqwest = { workspace = true }
-toml = "0.8.6"
 temp-dir = "0.1.11"
+test-components = { path = "../test-components" }
+toml = "0.8.6"

tests/runtime-tests/README.md

@@ -18,20 +18,35 @@ Runtime tests are not full end-to-end integration tests, and thus there are some
 
 ## How do I run the tests?
 
-The runtime tests can either be run as a library function (e.g., this is how they are run as part of Spin's test suite using `cargo test`) or they can be run stand alone using the `runtime-tests` crate's binary (i.e., running `cargo run` from this directory).
+The runtime tests can either be run as a library function (e.g., this is how they are run as part of Spin's test suite using `cargo test`), or they can be run stand alone using the `runtime-tests` crate's binary (i.e., running `cargo run` from this directory).
 
 ## How do I add a new test?
 
 To add a new test you must add a new folder to the `tests` directory with at least a `spin.toml` manifest. 
 
 The manifest can reference pre-built Spin compliant WebAssembly modules that can be found in the `test-components` folder in the Spin repo. It does so by using the `{{$NAME}}` where `$NAME` is substituted for the name of the test component to be used. For example `{{sqlite}}` will use the test-component named "sqlite" found in the `test-components` directory.
 
-The test directory may additionally contain an `error.txt` if the Spin application is expected to fail.
+The test directory may additionally contain:
+* an `error.txt` if the Spin application is expected to fail
+* a `services` config file (more on this below)
 
 ### The testing protocol
 
 The test runner will make a GET request against the `/` path. The component should either return a 200 if everything goes well or a 500 if there is an error. If an `error.txt` file is present, the Spin application must return a 500 with the body set to some error message that contains the contents of `error.txt`.
 
+### Services
+
+Services allow for tests to be run against external sources. The service definitions can be found in the 'services' directory. Each test directory contains a 'services' file that configures the tests services. Each line of the services file should contain the name of a services file that needs to run. For example, the following 'services' file will run the `tcp-echo.py` service:
+
+```txt
+tcp-echo.py
+```
+
+Each service is run under a file lock meaning that all other tests that require that service must wait until the current test using that service has finished.
+
+The following service types are supported:
+* Python services (a python script ending in the .py file extension)
+
 ## When do tests pass?
 
 A test will pass in the following conditions:

tests/runtime-tests/services/tcp-echo.py

@@ -0,0 +1,37 @@
+import socket
+import threading
+import os
+
+def handle_client(client_socket):
+    while True:
+        data = client_socket.recv(1024)
+        if not data:
+            break
+        # Echo the received data back to the client
+        client_socket.send(data)
+    client_socket.close()
+
+def echo_server():
+    host = "127.0.0.1" 
+    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    server_socket.bind((host, 6001))
+    server_socket.listen(5)
+    _, port = server_socket.getsockname()
+    print(f"Listening on {host}:{port}")
+
+    try:
+        while True:
+            client_socket, client_address = server_socket.accept()
+            print(f"Accepted connection from {client_address}")
+            # Handle the client in a separate thread
+            client_handler = threading.Thread(target=handle_client, args=(client_socket,))
+            client_handler.start()
+    except KeyboardInterrupt:
+        print("Server shutting down.")
+    finally:
+        # Close the server socket
+        server_socket.close()
+
+if __name__ == "__main__":
+    # Run the echo server
+    echo_server()

tests/runtime-tests/src/lib.rs

@@ -1,9 +1,10 @@
 use std::{
     io::Read,
     path::{Path, PathBuf},
+    process::{Command, Stdio},
 };
 
-use anyhow::Context;
+use anyhow::{bail, Context};
 
 /// Configuration for the test suite
 pub struct Config {
@@ -54,12 +55,92 @@ pub fn bootstrap_and_run(test_path: &Path, config: &Config) -> Result<(), anyhow
         .context("failed to produce a temporary directory to run the test in")?;
     log::trace!("Temporary directory: {}", temp.path().display());
     copy_manifest(test_path, &temp)?;
-    let spin = Spin::start(&config.spin_binary_path, temp.path())?;
+    let mut services = start_services(test_path)?;
+    let spin = Spin::start(&config.spin_binary_path, temp.path(), &mut services)?;
     log::debug!("Spin started on port {}.", spin.port());
     run_test(test_path, spin, config.on_error);
     Ok(())
 }
 
+fn start_services(test_path: &Path) -> anyhow::Result<Services> {
+    let services_config_path = test_path.join("services");
+    let children = if services_config_path.exists() {
+        let services = std::fs::read_to_string(&services_config_path)
+            .context("could not read services file")?;
+        let service_files = services.lines().filter_map(|s| {
+            let s = s.trim();
+            (!s.is_empty()).then_some(Path::new(s))
+        });
+        // TODO: make this more robust so that it is not just assumed where the services definitions are
+        let services_path = test_path
+            .parent()
+            .unwrap()
+            .parent()
+            .unwrap()
+            .join("services");
+        let mut services = Vec::new();
+        for service_file in service_files {
+            let service_name = service_file.file_stem().unwrap().to_str().unwrap();
+            let child = match service_file.extension().and_then(|e| e.to_str()) {
+                Some("py") => {
+                    let mut lock =
+                        fslock::LockFile::open(&services_path.join(format!("{service_name}.lock")))
+                            .context("failed to open service file lock")?;
+                    lock.lock().context("failed to obtain service file lock")?;
+                    let child = python()
+                        .arg(services_path.join(service_file).display().to_string())
+                        // Ignore stdout
+                        .stdout(Stdio::null())
+                        .spawn()
+                        .context("service failed to spawn")?;
+                    (child, Some(lock))
+                }
+                _ => bail!("unsupported service type found: {service_name}",),
+            };
+            services.push(child);
+        }
+        services
+    } else {
+        Vec::new()
+    };
+
+    Ok(Services { children })
+}
+
+fn python() -> Command {
+    Command::new("python3")
+}
+
+struct Services {
+    children: Vec<(std::process::Child, Option<fslock::LockFile>)>,
+}
+
+impl Services {
+    fn error(&mut self) -> std::io::Result<()> {
+        for (child, _) in &mut self.children {
+            let exit = child.try_wait()?;
+            if exit.is_some() {
+                return Err(std::io::Error::new(
+                    std::io::ErrorKind::Interrupted,
+                    "process exited early",
+                ));
+            }
+        }
+        Ok(())
+    }
+}
+
+impl Drop for Services {
+    fn drop(&mut self) {
+        for (child, lock) in &mut self.children {
+            let _ = child.kill();
+            if let Some(lock) = lock {
+                let _ = lock.unlock();
+            }
+        }
+    }
+}
+
 /// Run an individual test
 fn run_test(test_path: &Path, mut spin: Spin, on_error: OnTestError) {
     // macro which will look at `on_error` and do the right thing
@@ -68,7 +149,7 @@ fn run_test(test_path: &Path, mut spin: Spin, on_error: OnTestError) {
             match $on_error {
                 OnTestError::Panic => panic!($($arg)*),
                 OnTestError::Log => {
-                    println!($($arg)*);
+                    eprintln!($($arg)*);
                     return;
                 }
             }
@@ -131,6 +212,9 @@ fn run_test(test_path: &Path, mut spin: Spin, on_error: OnTestError) {
             error!(on_error, "Test '{}' errored: {e}", test_path.display());
         }
     }
+    if let OnTestError::Log = on_error {
+        println!("'{}' passed", test_path.display())
+    }
 }
 
 /// Copies the test dir's manifest file into the temporary directory
@@ -228,14 +312,18 @@ struct Spin {
 }
 
 impl Spin {
-    fn start(spin_binary_path: &Path, current_dir: &Path) -> Result<Self, anyhow::Error> {
+    fn start(
+        spin_binary_path: &Path,
+        current_dir: &Path,
+        services: &mut Services,
+    ) -> Result<Self, anyhow::Error> {
         let port = get_random_port()?;
-        let mut child = std::process::Command::new(spin_binary_path)
+        let mut child = Command::new(spin_binary_path)
             .arg("up")
             .current_dir(current_dir)
             .args(["--listen", &format!("127.0.0.1:{port}")])
-            .stdout(std::process::Stdio::piped())
-            .stderr(std::process::Stdio::piped())
+            .stdout(Stdio::piped())
+            .stderr(Stdio::piped())
             .spawn()?;
         let stdout = OutputStream::new(child.stdout.take().unwrap());
         let stderr = OutputStream::new(child.stderr.take().unwrap());
@@ -247,6 +335,7 @@ impl Spin {
             port,
         };
         for _ in 0..80 {
+            services.error()?;
             match std::net::TcpStream::connect(format!("127.0.0.1:{port}")) {
                 Ok(_) => return Ok(spin),
                 Err(e) => {

tests/runtime-tests/tests/tcp-sockets-ip-range/services

@@ -0,0 +1 @@
+tcp-echo.py

tests/runtime-tests/tests/tcp-sockets-ip-range/spin.toml

@@ -11,4 +11,4 @@ component = "test"
 
 [component.test]
 source = "{{tcp-sockets}}"
-allowed_outbound_hosts = ["*://127.0.0.0/24:5001"]
+allowed_outbound_hosts = ["*://127.0.0.0/24:6001"]

tests/runtime-tests/tests/tcp-sockets-no-ip-permission/spin.toml

@@ -12,4 +12,4 @@ component = "test"
 [component.test]
 source = "{{tcp-sockets}}"
 # Component expects 127.0.0.1 but we only allow 127.0.0.2
-allowed_outbound_hosts = ["*://127.0.0.2:5001"]
+allowed_outbound_hosts = ["*://127.0.0.2:6001"]

tests/runtime-tests/tests/tcp-sockets-no-port-permission/spin.toml

@@ -11,5 +11,5 @@ component = "test"
 
 [component.test]
 source = "{{tcp-sockets}}"
-# Component expects port 5001 but we allow 5002
-allowed_outbound_hosts = ["*://127.0.0.1:5002"]
+# Component expects port 5001 but we allow 6002
+allowed_outbound_hosts = ["*://127.0.0.1:6002"]

tests/runtime-tests/tests/tcp-sockets/services

@@ -0,0 +1 @@
+tcp-echo.py

tests/runtime-tests/tests/tcp-sockets/spin.toml

@@ -11,4 +11,4 @@ component = "test"
 
 [component.test]
 source = "{{tcp-sockets}}"
-allowed_outbound_hosts = ["*://127.0.0.1:5001"]
+allowed_outbound_hosts = ["*://127.0.0.1:6001"]

tests/runtime.rs

@@ -28,14 +28,13 @@ mod runtime_tests {
     test!(key_value, "key-value");
     test!(key_value_no_permission, "key-value-no-permission");
     test!(variables, "variables");
-    // TODO: reenable these tests once we have a way to run them reliably in CI
-    // test!(tcp_sockets, "tcp-sockets");
-    // test!(tcp_sockets_ip_range, "tcp-sockets-ip-range");
-    // test!(
-    //     tcp_sockets_no_port_permission,
-    //     "tcp-sockets-no-port-permission"
-    // );
-    // test!(tcp_sockets_no_ip_permission, "tcp-sockets-no-ip-permission");
+    test!(tcp_sockets, "tcp-sockets");
+    test!(tcp_sockets_ip_range, "tcp-sockets-ip-range");
+    test!(
+        tcp_sockets_no_port_permission,
+        "tcp-sockets-no-port-permission"
+    );
+    test!(tcp_sockets_no_ip_permission, "tcp-sockets-no-ip-permission");
 
     fn run(name: &str) {
         let spin_binary_path = env!("CARGO_BIN_EXE_spin").into();

tests/test-components/components/tcp-sockets/README.md

@@ -5,4 +5,4 @@ Tests the `wasi:sockets` TCP related interfaces
 ## Expectations
 
 This test component expects the following to be true:
-* It has access to a TCP echo server on 127.0.0.1:5001
+* It has access to a TCP echo server on 127.0.0.1:6001

tests/test-components/components/tcp-sockets/src/lib.rs

@@ -15,7 +15,7 @@ helper::define_component!(Component);
 
 impl Component {
     fn main() -> Result<(), String> {
-        let address = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 5001);
+        let address = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 6001);
 
         let client = ensure_ok!(tcp_create_socket::create_tcp_socket(IpAddressFamily::Ipv4));