CalCalist/tests/support/mod.rs

//! Scaffolding for the integration tests: a real CalDAV server, a real feed.
//!
//! Everything here talks HTTP over a plain socket rather than through a client
//! crate. The requests involved are few and mostly unusual — `MKCALENDAR`, `PUT`
//! of an `.ics` — and writing them out makes exactly what the server is asked
//! for visible in the test.

use std::io::{Read, Write};
use std::net::{Ipv4Addr, SocketAddr, TcpListener, TcpStream};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::time::{Duration, Instant};

/// How long to wait for a server to start answering.
const STARTUP_TIMEOUT: Duration = Duration::from_secs(20);

/// How many ports to try before giving up on starting a server.
const ATTEMPTS: usize = 8;

/// The account Radicale files everything under. With authentication disabled it
/// accepts whatever name arrives, and `owner_only` rights then grant that name
/// its own tree — so the URLs below all live beneath it.
pub const USER: &str = "calcalist";

/// A binary the integration tests need, and whether it is here.
pub fn missing_binaries() -> Vec<&'static str> {
    ["radicale", "pimsync"]
        .into_iter()
        .filter(|binary| which(binary).is_none())
        .collect()
}

fn which(binary: &str) -> Option<PathBuf> {
    std::env::var_os("PATH")?
        .to_str()?
        .split(':')
        .map(|dir| Path::new(dir).join(binary))
        .find(|candidate| candidate.is_file())
}

/// A Radicale instance with its own storage, shut down when dropped.
pub struct Radicale {
    process: Child,
    pub port: u16,
    storage: PathBuf,
}

impl Radicale {
    pub fn start(root: &Path) -> Radicale {
        let storage = root.join("radicale");
        std::fs::create_dir_all(&storage).expect("create storage");

        // A free port is chosen by binding one and letting go, so two tests
        // running at once can pick the same number. The loser's radicale then
        // fails to bind and exits — and, worse, waiting for the port to answer
        // succeeds anyway, because the winner is listening on it. Tests would
        // quietly share a server. So: confirm the child we spawned is the one
        // still alive, and take a different port if it is not.
        for _ in 0..ATTEMPTS {
            let port = free_port();
            let mut process = Command::new("radicale")
                .arg("--config")
                .arg("")
                .arg("--server-hosts")
                .arg(format!("127.0.0.1:{port}"))
                .arg("--auth-type")
                .arg("none")
                .arg("--storage-filesystem-folder")
                .arg(&storage)
                .stdout(Stdio::null())
                .stderr(Stdio::null())
                .spawn()
                .expect("radicale should start");

            if listening(&mut process, port) {
                return Radicale {
                    process,
                    port,
                    storage,
                };
            }
            let _ = process.kill();
            let _ = process.wait();
        }
        panic!("radicale did not come up on a port of its own after {ATTEMPTS} attempts");
    }

    pub fn url(&self, calendar: &str) -> String {
        format!("http://127.0.0.1:{}/{USER}/{calendar}/", self.port)
    }

    /// Creates a calendar collection. pimsync deliberately never creates one,
    /// so the calendars a test syncs have to exist on the server first.
    pub fn create_calendar(&self, calendar: &str) {
        let body = "<?xml version=\"1.0\" encoding=\"utf-8\"?>\
             <C:mkcalendar xmlns:D=\"DAV:\" xmlns:C=\"urn:ietf:params:xml:ns:caldav\">\
             <D:set><D:prop><D:displayname>calendar</D:displayname></D:prop></D:set>\
             </C:mkcalendar>";
        let response = self.request(
            "MKCALENDAR",
            &format!("/{USER}/{calendar}/"),
            "application/xml; charset=utf-8",
            body,
        );
        // Radicale answers HTTP/1.0, so the status is read out of the line
        // rather than matched against a whole prefix.
        assert!(
            matches!(status_of(&response), Some(201 | 405)),
            "creating {calendar}: {response}"
        );
    }

    /// Stores an event, as a calendar client would.
    pub fn put_event(&self, calendar: &str, name: &str, ics: &str) {
        let response = self.request(
            "PUT",
            &format!("/{USER}/{calendar}/{name}"),
            "text/calendar; charset=utf-8",
            ics,
        );
        assert!(
            status_of(&response).is_some_and(|status| (200..300).contains(&status)),
            "storing {name}: {response}"
        );
    }

    /// Everything the server holds in a calendar, as stored.
    ///
    /// Read from Radicale's own storage rather than fetched back, so what is
    /// asserted on is the bytes that reached the server.
    pub fn stored(&self, calendar: &str) -> Vec<String> {
        let dir = self
            .storage
            .join("collection-root")
            .join(USER)
            .join(calendar);
        let Ok(entries) = std::fs::read_dir(dir) else {
            return Vec::new();
        };
        let mut items: Vec<String> = entries
            .flatten()
            .filter(|entry| {
                entry
                    .path()
                    .extension()
                    .is_some_and(|extension| extension == "ics")
            })
            .filter_map(|entry| std::fs::read_to_string(entry.path()).ok())
            .collect();
        items.sort();
        items
    }

    fn request(&self, method: &str, path: &str, content_type: &str, body: &str) -> String {
        // Retried because a server under parallel test load occasionally resets
        // a connection it has accepted; that says nothing about the request.
        let mut last = String::new();
        for attempt in 0..ATTEMPTS {
            match self.try_request(method, path, content_type, body) {
                Ok(response) => return response,
                Err(error) => {
                    last = error;
                    std::thread::sleep(Duration::from_millis(50 * (attempt as u64 + 1)));
                }
            }
        }
        panic!("{method} {path} never got a response: {last}");
    }

    fn try_request(
        &self,
        method: &str,
        path: &str,
        content_type: &str,
        body: &str,
    ) -> Result<String, String> {
        let address = SocketAddr::from((Ipv4Addr::LOCALHOST, self.port));
        let mut stream = TcpStream::connect_timeout(&address, Duration::from_secs(2))
            .map_err(|error| error.to_string())?;
        stream
            .set_read_timeout(Some(Duration::from_secs(10)))
            .map_err(|error| error.to_string())?;
        // With authentication disabled the credentials only pick the account.
        let authorization = base64(format!("{USER}:password").as_bytes());
        let request = format!(
            "{method} {path} HTTP/1.1\r\nHost: 127.0.0.1:{}\r\n\
             Authorization: Basic {authorization}\r\nContent-Type: {content_type}\r\n\
             Content-Length: {}\r\nConnection: close\r\n\r\n{body}",
            self.port,
            body.len()
        );
        stream
            .write_all(request.as_bytes())
            .map_err(|error| error.to_string())?;
        let mut response = String::new();
        stream
            .read_to_string(&mut response)
            .map_err(|error| error.to_string())?;
        Ok(response)
    }
}

impl Drop for Radicale {
    fn drop(&mut self) {
        let _ = self.process.kill();
        let _ = self.process.wait();
    }
}

/// A one-file HTTP server, standing in for a published iCal feed.
pub struct Feed {
    pub port: u16,
}

impl Feed {
    /// Serves `ics` at any path, for as long as the test runs.
    pub fn start(ics: String) -> Feed {
        let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0)))
            .expect("bind the feed port");
        let port = listener.local_addr().expect("feed address").port();
        std::thread::spawn(move || {
            for stream in listener.incoming().flatten() {
                serve_once(stream, &ics);
            }
        });
        Feed { port }
    }

    pub fn url(&self) -> String {
        format!("http://127.0.0.1:{}/holidays.ics", self.port)
    }
}

fn serve_once(mut stream: TcpStream, ics: &str) {
    // Enough of the request to reach the blank line; the path does not matter.
    let mut buffer = [0u8; 2048];
    let _ = stream.read(&mut buffer);
    let response = format!(
        "HTTP/1.1 200 OK\r\nContent-Type: text/calendar; charset=utf-8\r\n\
         Content-Length: {}\r\nConnection: close\r\n\r\n{ics}",
        ics.len()
    );
    let _ = stream.write_all(response.as_bytes());
}

/// Runs `calcalist` against a config and state directory of the test's own.
pub struct Calcalist {
    pub config: PathBuf,
    pub state: PathBuf,
}

/// What one `calcalist` invocation printed and returned.
pub struct Run {
    pub status: Option<i32>,
    pub stdout: String,
    pub stderr: String,
}

impl Run {
    pub fn succeeded(&self) -> bool {
        self.status == Some(0)
    }
}

impl Calcalist {
    pub fn new(root: &Path, config: &str) -> Calcalist {
        let config_path = root.join("calcalist.toml");
        std::fs::write(&config_path, config).expect("write config");
        let state = root.join("state");
        std::fs::create_dir_all(&state).expect("create state directory");
        Calcalist {
            config: config_path,
            state,
        }
    }

    /// The same configuration with the read-only feed taken out, for the test
    /// that retires an endpoint.
    pub fn config_without_feed(&self) -> String {
        let text = std::fs::read_to_string(&self.config).expect("read config");
        text.split("\n\n")
            .filter(|block| !block.contains("id = \"holidays\""))
            .map(|block| block.replace(", \"holidays\"", ""))
            .collect::<Vec<_>>()
            .join("\n\n")
    }

    pub fn run(&self, arguments: &[&str]) -> Run {
        let output = Command::new(env!("CARGO_BIN_EXE_calcalist"))
            .arg("--config")
            .arg(&self.config)
            .args(arguments)
            // The state directory is resolved through XDG, so this is what
            // keeps the test off the developer's own calendars and state.
            .env("XDG_STATE_HOME", &self.state)
            .output()
            .expect("calcalist should run");
        Run {
            status: output.status.code(),
            stdout: String::from_utf8_lossy(&output.stdout).into_owned(),
            stderr: String::from_utf8_lossy(&output.stderr).into_owned(),
        }
    }
}

/// The status code out of a response's first line, whatever HTTP version it
/// claims.
fn status_of(response: &str) -> Option<u16> {
    response
        .lines()
        .next()?
        .split_whitespace()
        .nth(1)?
        .parse()
        .ok()
}

fn free_port() -> u16 {
    let listener =
        TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0))).expect("bind a free port");
    listener.local_addr().expect("port").port()
}

/// Whether a real HTTP request gets a real HTTP response.
fn answers(address: SocketAddr) -> bool {
    let Ok(mut stream) = TcpStream::connect_timeout(&address, Duration::from_millis(200)) else {
        return false;
    };
    let _ = stream.set_read_timeout(Some(Duration::from_secs(2)));
    let request = format!("OPTIONS / HTTP/1.1\r\nHost: {address}\r\nConnection: close\r\n\r\n");
    if stream.write_all(request.as_bytes()).is_err() {
        return false;
    }
    let mut response = String::new();
    stream.read_to_string(&mut response).is_ok() && response.starts_with("HTTP/")
}

/// Waits for our own radicale to answer, giving up as soon as it has exited.
///
/// The liveness check is the point: without it a process that lost a port race
/// looks healthy, because something else is answering on that port.
fn listening(process: &mut Child, port: u16) -> bool {
    let address = SocketAddr::from((Ipv4Addr::LOCALHOST, port));
    let deadline = Instant::now() + STARTUP_TIMEOUT;
    while Instant::now() < deadline {
        match process.try_wait() {
            Ok(Some(_)) => return false,
            Ok(None) => {}
            Err(_) => return false,
        }
        // An open port is not readiness: the socket listens before the
        // application behind it can answer, and under parallel tests that gap
        // is wide enough to get a connection reset instead of a response.
        if answers(address) {
            return true;
        }
        std::thread::sleep(Duration::from_millis(50));
    }
    false
}

fn base64(input: &[u8]) -> String {
    const ALPHABET: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
    let mut out = String::new();
    for chunk in input.chunks(3) {
        let mut block = [0u8; 3];
        block[..chunk.len()].copy_from_slice(chunk);
        let packed = u32::from(block[0]) << 16 | u32::from(block[1]) << 8 | u32::from(block[2]);
        for index in 0..4 {
            if index <= chunk.len() {
                out.push(ALPHABET[(packed >> (18 - index * 6)) as usize & 0x3f] as char);
            } else {
                out.push('=');
            }
        }
    }
    out
}