runa/crates/runa-server/tests/local_client.rs

use runa_protocol::Value;
use std::io::{BufRead, BufReader, Write};
use std::net::{Shutdown, SocketAddr, TcpStream};
use std::process::{Child, Command as ProcessCommand, Stdio};
use std::time::Duration;

use runa_client::{LocalClient, TcpClient};
use runa_protocol::{Command, PROTOCOL_VERSION, Reply, Request, Response};

struct ServerProcess(Child);

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

#[test]
fn local_client_launches_server_and_operates_the_same_live_image() {
    let mut client = LocalClient::spawn(env!("CARGO_BIN_EXE_runa-server"), &[] as &[&str]).unwrap();

    let published = client
        .request(Command::PublishBehavior {
            name: "counter".to_owned(),
            expected_revision: None,
            source: "pub fn handle(state, message) { (state + message.value, []) }".to_owned(),
        })
        .unwrap();
    assert_eq!(published.request_id, "runa-client-1");
    assert!(matches!(published.result, Ok(Reply::Behavior(_))));

    let spawned = client
        .request(Command::Spawn {
            behavior_name: "counter".to_owned(),
            state: Value::Integer(5),
            mailbox_capacity: 2,
        })
        .unwrap();
    assert_eq!(spawned.request_id, "runa-client-2");
    let actor_ref = match spawned.result {
        Ok(Reply::ActorSpawned { actor_ref }) => actor_ref,
        other => panic!("unexpected spawn response: {other:?}"),
    };
    assert_eq!(actor_ref.len(), 21);

    client
        .request(Command::Send {
            actor_ref: actor_ref.clone(),
            message: test_message(0, 7, None),
        })
        .unwrap();
    client
        .request(Command::Tick {
            ticks: 100,
            instruction_budget: 10,
        })
        .unwrap();
    let inspected = client.request(Command::InspectActor { actor_ref }).unwrap();
    assert!(matches!(
        inspected.result,
        Ok(Reply::Actor(actor)) if actor.state == Value::Integer(12)
    ));
    assert!(client.shutdown().unwrap().success());
}

#[test]
fn tcp_server_accepts_local_protocol_requests() {
    let child = ProcessCommand::new(env!("CARGO_BIN_EXE_runa-server"))
        .args(["--listen", "127.0.0.1:0"])
        .stdout(Stdio::null())
        .stderr(Stdio::piped())
        .spawn()
        .unwrap();
    let mut server = ServerProcess(child);
    let stderr = server.0.stderr.take().unwrap();
    let mut stderr = BufReader::new(stderr);
    let mut startup = String::new();
    stderr.read_line(&mut startup).unwrap();
    let address = startup
        .strip_prefix("runa-server: listening on ")
        .expect("server should report its bound address")
        .trim()
        .parse::<SocketAddr>()
        .unwrap();

    let mut client = TcpStream::connect_timeout(&address, Duration::from_secs(2)).unwrap();
    let request = Request {
        version: PROTOCOL_VERSION,
        request_id: "tcp-process-test".to_owned(),
        command: Command::ListActors,
    };
    writeln!(client, "{}", serde_json::to_string(&request).unwrap()).unwrap();
    client.shutdown(Shutdown::Write).unwrap();

    let mut response_line = String::new();
    BufReader::new(client)
        .read_line(&mut response_line)
        .unwrap();
    let response: Response = serde_json::from_str(&response_line).unwrap();
    assert_eq!(response.request_id, "tcp-process-test");
    assert_eq!(response.result, Ok(Reply::Actors(Vec::new())));

    let mut first = TcpClient::connect(address).unwrap();
    let mut second = TcpClient::connect(address).unwrap();
    let actor_ref = match first
        .request(Command::Spawn {
            behavior_name: "counter".to_owned(),
            state: Value::Integer(5),
            mailbox_capacity: 2,
        })
        .unwrap()
        .result
        .unwrap()
    {
        Reply::ActorSpawned { actor_ref } => actor_ref,
        other => panic!("unexpected spawn response: {other:?}"),
    };
    assert!(matches!(second.request(Command::InspectActor {
        actor_ref: actor_ref.clone(),
    }).unwrap().result, Ok(Reply::Actor(actor)) if actor.state == Value::Integer(5)));
    assert!(matches!(
        first
            .request(Command::RollbackBehavior {
                name: "counter".to_owned(),
                expected_revision: 1,
                target_revision: 1,
            })
            .unwrap()
            .result,
        Ok(Reply::Behavior(behavior)) if behavior.revision == 2
    ));
    drop(first);
    drop(second);

    let mut reconnected = TcpClient::connect(address).unwrap();
    assert!(matches!(
        reconnected
            .request(Command::GetBehavior {
                name: "counter".to_owned(),
            })
            .unwrap()
            .result,
        Ok(Reply::Behavior(behavior)) if behavior.revision == 2
    ));
    reconnected
        .request(Command::Send {
            actor_ref: actor_ref.clone(),
            message: Value::Record(std::collections::BTreeMap::from([
                ("op".to_owned(), Value::String("add".to_owned())),
                ("value".to_owned(), Value::Integer(7)),
            ])),
        })
        .unwrap()
        .result
        .unwrap();
    reconnected
        .request(Command::Tick {
            ticks: 100,
            instruction_budget: 10,
        })
        .unwrap()
        .result
        .unwrap();
    assert!(matches!(reconnected.request(Command::InspectActor {
        actor_ref: actor_ref.clone(),
    }).unwrap().result, Ok(Reply::Actor(actor)) if actor.state == Value::Integer(12)));

    reconnected
        .request(Command::PublishBehavior {
            name: "echo".to_owned(),
            expected_revision: None,
            source: "pub fn handle(state, message) { (message, []) }".to_owned(),
        })
        .unwrap()
        .result
        .unwrap();
    let nested = Value::Record(std::collections::BTreeMap::from([(
        "items".to_owned(),
        Value::List(vec![
            Value::Null,
            Value::Bool(true),
            Value::Integer(-42),
            Value::String("hello world".to_owned()),
            Value::ActorRef(runa_actor::ActorRef::parse(&actor_ref).unwrap()),
        ]),
    )]));
    let echo = match reconnected
        .request(Command::Spawn {
            behavior_name: "echo".to_owned(),
            state: nested.clone(),
            mailbox_capacity: 2,
        })
        .unwrap()
        .result
        .unwrap()
    {
        Reply::ActorSpawned { actor_ref } => actor_ref,
        other => panic!("unexpected spawn: {other:?}"),
    };
    assert!(matches!(reconnected.request(Command::InspectActor {
        actor_ref: echo.clone(),
    }).unwrap().result, Ok(Reply::Actor(actor)) if actor.state == nested));
    reconnected
        .request(Command::Send {
            actor_ref: echo.clone(),
            message: nested.clone(),
        })
        .unwrap()
        .result
        .unwrap();
    reconnected
        .request(Command::Tick {
            ticks: 100,
            instruction_budget: 10,
        })
        .unwrap()
        .result
        .unwrap();
    assert!(matches!(reconnected.request(Command::InspectActor {
        actor_ref: echo,
    }).unwrap().result, Ok(Reply::Actor(actor)) if actor.state == nested && actor.reserved.is_none()));
}

fn test_message(kind: i64, value: i64, target: Option<runa_actor::ActorRef>) -> Value {
    Value::Record(std::collections::BTreeMap::from([
        ("kind".to_owned(), Value::Integer(kind)),
        ("value".to_owned(), Value::Integer(value)),
        (
            "target".to_owned(),
            target.map_or(Value::Null, Value::ActorRef),
        ),
    ]))
}