lavina/crates/projection-irc/tests/lib.rs

use std::io::ErrorKind;
use std::time::Duration;

use anyhow::{anyhow, Result};
use chrono::{DateTime, SecondsFormat};
use prometheus::Registry as MetricsRegistry;
use tokio::io::{AsyncReadExt, AsyncWriteExt, BufReader};
use tokio::net::tcp::{ReadHalf, WriteHalf};
use tokio::net::TcpStream;

use lavina_core::auth::Authenticator;
use lavina_core::clustering::{ClusterConfig, ClusterMetadata};
use lavina_core::player::{JoinResult, PlayerId, SendMessageResult};
use lavina_core::repo::{Storage, StorageConfig};
use lavina_core::room::RoomId;
use lavina_core::LavinaCore;
use projection_irc::APP_VERSION;
use projection_irc::{launch, read_irc_message, RunningServer, ServerConfig};

struct TestScope<'a> {
    reader: BufReader<ReadHalf<'a>>,
    writer: WriteHalf<'a>,
    buffer: Vec<u8>,
    pub timeout: Duration,
}

impl<'a> TestScope<'a> {
    fn new(stream: &mut TcpStream) -> TestScope<'_> {
        let (reader, writer) = stream.split();
        let reader = BufReader::new(reader);
        let buffer = vec![];
        let timeout = Duration::from_millis(1000);
        TestScope {
            reader,
            writer,
            buffer,
            timeout,
        }
    }

    async fn send(&mut self, str: &(impl AsRef<str> + ?Sized)) -> Result<()> {
        self.writer.write_all(str.as_ref().as_bytes()).await?;
        self.writer.write_all(b"\r\n").await?;
        self.writer.flush().await?;
        Ok(())
    }

    async fn expect(&mut self, str: &str) -> Result<()> {
        tracing::debug!("Expecting {}", str);
        let len = tokio::time::timeout(self.timeout, read_irc_message(&mut self.reader, &mut self.buffer)).await??;
        assert_eq!(std::str::from_utf8(&self.buffer[..len - 2])?, str);
        self.buffer.clear();
        Ok(())
    }

    async fn expect_that(&mut self, validate: impl FnOnce(&str) -> bool) -> Result<()> {
        let len = tokio::time::timeout(self.timeout, read_irc_message(&mut self.reader, &mut self.buffer)).await??;
        let msg = std::str::from_utf8(&self.buffer[..len - 2])?;
        if !validate(msg) {
            return Err(anyhow!("unexpected message: {:?}", msg));
        }
        self.buffer.clear();
        Ok(())
    }

    async fn expect_server_introduction(&mut self, nick: &str) -> Result<()> {
        self.expect(&format!(":testserver 001 {nick} :Welcome to testserver Server")).await?;
        self.expect(&format!(":testserver 002 {nick} :Welcome to testserver Server")).await?;
        self.expect(&format!(":testserver 003 {nick} :Welcome to testserver Server")).await?;
        self.expect(&format!(
            ":testserver 004 {nick} testserver {APP_VERSION} r CFILPQbcefgijklmnopqrstvz"
        ))
        .await?;
        self.expect(&format!(
            ":testserver 005 {nick} CHANTYPES=# :are supported by this server"
        ))
        .await?;
        Ok(())
    }

    async fn expect_eof(&mut self) -> Result<()> {
        let mut buf = [0; 1];
        let len = tokio::time::timeout(self.timeout, self.reader.read(&mut buf)).await??;
        if len != 0 {
            return Err(anyhow!("not a eof"));
        }
        Ok(())
    }

    async fn expect_nothing(&mut self) -> Result<()> {
        let mut buf = [0; 1];
        match tokio::time::timeout(self.timeout, self.reader.read(&mut buf)).await {
            Ok(res) => Err(anyhow!("received something: {:?}", res)),
            Err(_) => Ok(()),
        }
    }

    async fn expect_cap_ls(&mut self) -> Result<()> {
        self.expect(":testserver CAP * LS :sasl=PLAIN server-time").await?;
        Ok(())
    }
}

struct TestServer {
    metrics: MetricsRegistry,
    storage: Storage,
    core: LavinaCore,
    server: RunningServer,
}
impl TestServer {
    async fn start() -> Result<TestServer> {
        let _ = tracing_subscriber::fmt::try_init();
        let config = ServerConfig {
            listen_on: "127.0.0.1:0".parse().unwrap(),
            server_name: "testserver".into(),
        };
        let metrics = MetricsRegistry::new();
        let storage = Storage::open(StorageConfig {
            db_path: ":memory:".into(),
        })
        .await?;
        let cluster_config = ClusterConfig {
            addresses: vec![],
            metadata: ClusterMetadata {
                node_id: 0,
                main_owner: 0,
                rooms: Default::default(),
            },
        };
        let core = LavinaCore::new(metrics.clone(), cluster_config, storage.clone()).await?;
        let server = launch(config, core.clone(), metrics.clone()).await.unwrap();
        Ok(TestServer {
            metrics,
            storage,
            core,
            server,
        })
    }

    async fn reboot(self) -> Result<TestServer> {
        let config = ServerConfig {
            listen_on: "127.0.0.1:0".parse().unwrap(),
            server_name: "testserver".into(),
        };
        let cluster_config = ClusterConfig {
            addresses: vec![],
            metadata: ClusterMetadata {
                node_id: 0,
                main_owner: 0,
                rooms: Default::default(),
            },
        };
        let TestServer {
            metrics: _,
            storage,
            core,
            server,
        } = self;
        server.terminate().await?;
        core.shutdown().await?;
        let metrics = MetricsRegistry::new();
        let core = LavinaCore::new(metrics.clone(), cluster_config, storage.clone()).await?;
        let server = launch(config, core.clone(), metrics.clone()).await.unwrap();
        Ok(TestServer {
            metrics,
            storage,
            core,
            server,
        })
    }

    async fn shutdown(self) -> Result<()> {
        self.server.terminate().await?;
        self.core.shutdown().await?;
        self.storage.close().await?;
        Ok(())
    }
}

#[tokio::test]
async fn scenario_basic() -> Result<()> {
    let server = TestServer::start().await?;

    // test scenario

    server.storage.create_user("tester").await?;
    Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;

    let mut stream = TcpStream::connect(server.server.addr).await?;
    let mut s = TestScope::new(&mut stream);

    s.send("PASS password").await?;
    s.send("NICK tester").await?;
    s.send("USER UserName 0 * :Real Name").await?;
    s.expect_server_introduction("tester").await?;
    s.expect_nothing().await?;
    s.send("QUIT :Leaving").await?;
    s.expect(":testserver ERROR :Leaving the server").await?;
    s.expect_eof().await?;

    stream.shutdown().await?;

    // wrap up

    server.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn scenario_join_and_reboot() -> Result<()> {
    let server = TestServer::start().await?;

    // test scenario

    server.storage.create_user("tester").await?;
    Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;

    let mut stream = TcpStream::connect(server.server.addr).await?;
    let mut s = TestScope::new(&mut stream);

    // Open a connection and join a channel

    s.send("PASS password").await?;
    s.send("NICK tester").await?;
    s.send("USER UserName 0 * :Real Name").await?;
    s.expect_server_introduction("tester").await?;
    s.expect_nothing().await?;
    s.send("JOIN #test").await?;
    s.expect(":tester JOIN #test").await?;
    s.expect(":testserver 332 tester #test :New room").await?;
    s.expect(":testserver 353 tester = #test :tester").await?;
    s.expect(":testserver 366 tester #test :End of /NAMES list").await?;
    s.send("PRIVMSG #test :Hello").await?;
    s.send("QUIT :Leaving").await?;
    s.expect(":testserver ERROR :Leaving the server").await?;
    s.expect_eof().await?;
    stream.shutdown().await?;

    // Open a new connection and expect to be force-joined to the channel

    let mut stream = TcpStream::connect(server.server.addr).await?;
    let mut s = TestScope::new(&mut stream);

    async fn test(s: &mut TestScope<'_>) -> Result<()> {
        s.send("PASS password").await?;
        s.send("NICK tester").await?;
        s.send("USER UserName 0 * :Real Name").await?;
        s.expect_server_introduction("tester").await?;
        s.expect(":tester JOIN #test").await?;
        s.expect(":testserver 332 tester #test :New room").await?;
        s.expect(":testserver 353 tester = #test :tester").await?;
        s.expect(":testserver 366 tester #test :End of /NAMES list").await?;
        s.send("QUIT :Leaving").await?;
        s.expect(":testserver ERROR :Leaving the server").await?;
        s.expect_eof().await?;
        Ok(())
    }
    test(&mut s).await?;
    stream.shutdown().await?;

    // Reboot the server

    let server = server.reboot().await?;

    // Open a new connection and expect to be force-joined to the channel

    let mut stream = TcpStream::connect(server.server.addr).await?;
    let mut s = TestScope::new(&mut stream);

    test(&mut s).await?;
    stream.shutdown().await?;

    // wrap up

    server.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn scenario_force_join_msg() -> Result<()> {
    let server = TestServer::start().await?;

    // test scenario

    server.storage.create_user("tester").await?;
    Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;

    let mut stream1 = TcpStream::connect(server.server.addr).await?;
    let mut s1 = TestScope::new(&mut stream1);
    let mut stream2 = TcpStream::connect(server.server.addr).await?;
    let mut s2 = TestScope::new(&mut stream2);

    s1.send("PASS password").await?;
    s1.send("NICK tester").await?;
    s1.send("USER UserName 0 * :Real Name").await?;
    s1.expect_server_introduction("tester").await?;
    s1.expect_nothing().await?;

    s2.send("PASS password").await?;
    s2.send("NICK tester").await?;
    s2.send("USER UserName 0 * :Real Name").await?;
    s2.expect_server_introduction("tester").await?;
    s2.expect_nothing().await?;

    // We join the channel from the first connection
    s1.send("JOIN #test").await?;
    s1.expect(":tester JOIN #test").await?;
    s1.expect(":testserver 332 tester #test :New room").await?;
    s1.expect(":testserver 353 tester = #test :tester").await?;
    s1.expect(":testserver 366 tester #test :End of /NAMES list").await?;

    // And the second connection should receive the JOIN message (forced JOIN)
    s2.expect(":tester JOIN #test").await?;
    s2.expect(":testserver 332 tester #test :New room").await?;
    s2.expect(":testserver 353 tester = #test :tester").await?;
    s2.expect(":testserver 366 tester #test :End of /NAMES list").await?;

    // We send a message to the channel from the second connection
    s2.send("PRIVMSG #test :Hello").await?;
    // We should not receive an acknowledgement from the server
    s2.expect_nothing().await?;
    // But we should receive this message from the first connection
    s1.expect(":tester PRIVMSG #test :Hello").await?;

    s1.send("QUIT :Leaving").await?;
    s1.expect(":testserver ERROR :Leaving the server").await?;
    s1.expect_eof().await?;

    // Closing a connection does not kick you from the channel on a different connection
    s2.expect_nothing().await?;

    s2.send("QUIT :Leaving").await?;
    s2.expect(":testserver ERROR :Leaving the server").await?;
    s2.expect_eof().await?;

    stream1.shutdown().await?;
    stream2.shutdown().await?;

    // wrap up

    server.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn scenario_two_users() -> Result<()> {
    let server = TestServer::start().await?;

    // test scenario

    server.storage.create_user("tester1").await?;
    Authenticator::new(server.storage.clone()).set_password("tester1", "password").await?;
    server.storage.create_user("tester2").await?;
    Authenticator::new(server.storage.clone()).set_password("tester2", "password").await?;

    let mut stream1 = TcpStream::connect(server.server.addr).await?;
    let mut s1 = TestScope::new(&mut stream1);
    let mut stream2 = TcpStream::connect(server.server.addr).await?;
    let mut s2 = TestScope::new(&mut stream2);

    s1.send("PASS password").await?;
    s1.send("NICK tester1").await?;
    s1.send("USER UserName 0 * :Real Name").await?;
    s1.expect_server_introduction("tester1").await?;
    s1.expect_nothing().await?;

    s2.send("PASS password").await?;
    s2.send("NICK tester2").await?;
    s2.send("USER UserName 0 * :Real Name").await?;
    s2.expect_server_introduction("tester2").await?;
    s2.expect_nothing().await?;

    // Join the channel from the first user
    s1.send("JOIN #test").await?;
    s1.expect(":tester1 JOIN #test").await?;
    s1.expect(":testserver 332 tester1 #test :New room").await?;
    s1.expect(":testserver 353 tester1 = #test :tester1").await?;
    s1.expect(":testserver 366 tester1 #test :End of /NAMES list").await?;
    // Then join the channel from the second user
    s2.send("JOIN #test").await?;
    s2.expect(":tester2 JOIN #test").await?;
    s2.expect(":testserver 332 tester2 #test :New room").await?;
    s2.expect_that(|msg| {
        msg == ":testserver 353 tester2 = #test :tester1 tester2"
            || msg == ":testserver 353 tester2 = #test :tester2 tester1"
    })
    .await?;
    s2.expect(":testserver 366 tester2 #test :End of /NAMES list").await?;
    // The first user should receive the JOIN message from the second user
    s1.expect(":tester2 JOIN #test").await?;
    s1.expect_nothing().await?;
    s2.expect_nothing().await?;
    // Send a message from the second user
    s2.send("PRIVMSG #test :Hello").await?;
    // The first user should receive the message
    s1.expect(":tester2 PRIVMSG #test :Hello").await?;
    // Leave the channel from the first user
    s1.send("PART #test").await?;
    s1.expect(":tester1 PART #test").await?;
    // The second user should receive the PART message
    s2.expect(":tester1 PART #test").await?;

    s1.send("WHOIS tester2").await?;
    s1.expect(":testserver 318 tester1 tester2 :End of /WHOIS list").await?;

    stream1.shutdown().await?;
    s2.send("WHOIS tester3").await?;
    s2.expect(":testserver 401 tester2 tester3 :No such nick/channel").await?;
    s2.expect(":testserver 318 tester2 tester3 :End of /WHOIS list").await?;

    stream2.shutdown().await?;

    server.shutdown().await?;
    Ok(())
}

/*
IRC SASL doc: https://ircv3.net/specs/extensions/sasl-3.1.html
AUTHENTICATE doc: https://modern.ircdocs.horse/#authenticate-message
*/
#[tokio::test]
async fn scenario_cap_full_negotiation() -> Result<()> {
    let server = TestServer::start().await?;

    // test scenario

    server.storage.create_user("tester").await?;
    Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;

    let mut stream = TcpStream::connect(server.server.addr).await?;
    let mut s = TestScope::new(&mut stream);

    s.send("CAP LS 302").await?;
    s.send("NICK tester").await?;
    s.send("USER UserName 0 * :Real Name").await?;
    s.expect_cap_ls().await?;
    s.send("CAP REQ :sasl").await?;
    s.expect(":testserver CAP tester ACK :sasl").await?;
    s.send("AUTHENTICATE PLAIN").await?;
    s.expect(":testserver AUTHENTICATE +").await?;
    s.send("AUTHENTICATE dGVzdGVyAHRlc3RlcgBwYXNzd29yZA==").await?; // base64-encoded 'tester\x00tester\x00password'
    s.expect(":testserver 900 tester tester tester :You are now logged in as tester").await?;
    s.expect(":testserver 903 tester :SASL authentication successful").await?;

    s.send("CAP END").await?;

    s.expect_server_introduction("tester").await?;
    s.expect_nothing().await?;
    s.send("QUIT :Leaving").await?;
    s.expect(":testserver ERROR :Leaving the server").await?;
    s.expect_eof().await?;

    stream.shutdown().await?;

    // wrap up

    server.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn scenario_cap_full_negotiation_nick_last() -> Result<()> {
    let server = TestServer::start().await?;

    // test scenario

    server.storage.create_user("tester").await?;
    Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;

    let mut stream = TcpStream::connect(server.server.addr).await?;
    let mut s = TestScope::new(&mut stream);

    s.send("CAP LS 302").await?;
    s.expect_cap_ls().await?;
    s.send("CAP REQ :sasl").await?;
    s.expect(":testserver CAP * ACK :sasl").await?;
    s.send("AUTHENTICATE PLAIN").await?;
    s.expect(":testserver AUTHENTICATE +").await?;
    s.send("AUTHENTICATE dGVzdGVyAHRlc3RlcgBwYXNzd29yZA==").await?; // base64-encoded 'tester\x00tester\x00password'
    s.expect(":testserver 900 tester tester tester :You are now logged in as tester").await?;
    s.expect(":testserver 903 tester :SASL authentication successful").await?;
    s.send("CAP END").await?;
    s.send("USER UserName 0 * :Real Name").await?;
    s.send("NICK tester").await?;

    s.expect_server_introduction("tester").await?;
    s.expect_nothing().await?;
    s.send("QUIT :Leaving").await?;
    s.expect(":testserver ERROR :Leaving the server").await?;
    s.expect_eof().await?;

    stream.shutdown().await?;

    // wrap up

    server.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn scenario_cap_short_negotiation() -> Result<()> {
    let server = TestServer::start().await?;

    // test scenario

    server.storage.create_user("tester").await?;
    Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;

    let mut stream = TcpStream::connect(server.server.addr).await?;
    let mut s = TestScope::new(&mut stream);

    s.send("NICK tester").await?;
    s.send("CAP REQ :sasl").await?;
    s.send("USER UserName 0 * :Real Name").await?;
    s.expect(":testserver CAP tester ACK :sasl").await?;
    s.send("AUTHENTICATE PLAIN").await?;
    s.expect(":testserver AUTHENTICATE +").await?;
    s.send("AUTHENTICATE dGVzdGVyAHRlc3RlcgBwYXNzd29yZA==").await?; // base64-encoded 'tester\x00tester\x00password'
    s.expect(":testserver 900 tester tester tester :You are now logged in as tester").await?;
    s.expect(":testserver 903 tester :SASL authentication successful").await?;

    s.send("CAP END").await?;

    s.expect_server_introduction("tester").await?;
    s.expect_nothing().await?;
    s.send("QUIT :Leaving").await?;
    s.expect(":testserver ERROR :Leaving the server").await?;
    s.expect_eof().await?;

    stream.shutdown().await?;

    // wrap up

    server.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn scenario_cap_sasl_fail() -> Result<()> {
    let server = TestServer::start().await?;

    // test scenario

    server.storage.create_user("tester").await?;
    Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;

    let mut stream = TcpStream::connect(server.server.addr).await?;
    let mut s = TestScope::new(&mut stream);

    s.send("CAP LS 302").await?;
    s.send("NICK tester").await?;
    s.send("USER UserName 0 * :Real Name").await?;
    s.expect_cap_ls().await?;
    s.send("CAP REQ :sasl").await?;
    s.expect(":testserver CAP tester ACK :sasl").await?;
    s.send("AUTHENTICATE SHA256").await?;
    s.expect(":testserver 904 tester :Unsupported mechanism").await?;
    s.send("AUTHENTICATE PLAIN").await?;
    s.expect(":testserver AUTHENTICATE +").await?;
    s.send("AUTHENTICATE dGVzdGVyAHRlc3RlcgBwYXNzd29yZDE=").await?;
    s.expect(":testserver 904 tester :Bad credentials").await?;
    s.send("AUTHENTICATE dGVzdGVyAHRlc3RlcgBwYXNzd29yZA==").await?; // base64-encoded 'tester\x00tester\x00password'
    s.expect(":testserver 900 tester tester tester :You are now logged in as tester").await?;
    s.expect(":testserver 903 tester :SASL authentication successful").await?;

    s.send("CAP END").await?;

    s.expect_server_introduction("tester").await?;
    s.expect_nothing().await?;
    s.send("QUIT :Leaving").await?;
    s.expect(":testserver ERROR :Leaving the server").await?;
    s.expect_eof().await?;

    stream.shutdown().await?;

    // wrap up

    server.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn terminate_socket_scenario() -> Result<()> {
    let server = TestServer::start().await?;

    // test scenario

    server.storage.create_user("tester").await?;
    Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;

    let mut stream = TcpStream::connect(server.server.addr).await?;
    let mut s = TestScope::new(&mut stream);

    s.send("NICK tester").await?;
    s.send("CAP REQ :sasl").await?;
    s.send("USER UserName 0 * :Real Name").await?;
    s.expect(":testserver CAP tester ACK :sasl").await?;
    s.send("AUTHENTICATE PLAIN").await?;
    s.expect(":testserver AUTHENTICATE +").await?;

    server.shutdown().await?;
    assert_eq!(stream.read_u8().await.unwrap_err().kind(), ErrorKind::UnexpectedEof);

    Ok(())
}

#[tokio::test]
async fn server_time_capability() -> Result<()> {
    let server = TestServer::start().await?;

    // test scenario

    server.storage.create_user("tester").await?;
    Authenticator::new(server.storage.clone()).set_password("tester", "password").await?;

    let mut stream = TcpStream::connect(server.server.addr).await?;
    let mut s = TestScope::new(&mut stream);

    s.send("CAP LS 302").await?;
    s.send("NICK tester").await?;
    s.send("USER UserName 0 * :Real Name").await?;
    s.expect_cap_ls().await?;
    s.send("CAP REQ :sasl server-time").await?;
    s.expect(":testserver CAP tester ACK :sasl server-time").await?;
    s.send("AUTHENTICATE PLAIN").await?;
    s.expect(":testserver AUTHENTICATE +").await?;
    s.send("AUTHENTICATE dGVzdGVyAHRlc3RlcgBwYXNzd29yZA==").await?; // base64-encoded 'tester\x00tester\x00password'
    s.expect(":testserver 900 tester tester tester :You are now logged in as tester").await?;
    s.expect(":testserver 903 tester :SASL authentication successful").await?;
    s.send("CAP END").await?;
    s.expect_server_introduction("tester").await?;
    s.expect_nothing().await?;
    s.send("JOIN #test").await?;
    s.expect(":tester JOIN #test").await?;
    s.expect(":testserver 332 tester #test :New room").await?;
    s.expect(":testserver 353 tester = #test :tester").await?;
    s.expect(":testserver 366 tester #test :End of /NAMES list").await?;

    server.storage.create_user("some_guy").await?;
    let mut conn = server.core.players.connect_to_player(&PlayerId::from("some_guy").unwrap()).await;
    let res = conn.join_room(RoomId::from("test").unwrap()).await?;
    let JoinResult::Success(_) = res else {
        panic!("Failed to join room");
    };

    s.expect(":some_guy JOIN #test").await?;

    let SendMessageResult::Success(res) = conn.send_message(RoomId::from("test").unwrap(), "Hello".into()).await?
    else {
        panic!("Failed to send message");
    };
    s.expect(&format!(
        "@time={} :some_guy PRIVMSG #test :Hello",
        res.to_rfc3339_opts(SecondsFormat::Millis, true)
    ))
    .await?;

    // formatting check
    assert_eq!(
        DateTime::parse_from_rfc3339(&"2024-01-01T10:00:32.123Z").unwrap().to_rfc3339_opts(SecondsFormat::Millis, true),
        "2024-01-01T10:00:32.123Z"
    );

    s.send("QUIT :Leaving").await?;
    s.expect(":testserver ERROR :Leaving the server").await?;
    s.expect_eof().await?;

    stream.shutdown().await?;

    // wrap up

    server.shutdown().await?;
    Ok(())
}

#[tokio::test]
async fn scenario_two_players_dialog() -> Result<()> {
    let server = TestServer::start().await?;

    // test scenario

    server.storage.create_user("tester1").await?;
    server.storage.set_password("tester1", "password").await?;
    server.storage.create_user("tester2").await?;
    server.storage.set_password("tester2", "password").await?;

    let mut stream1 = TcpStream::connect(server.server.addr).await?;
    let mut s1 = TestScope::new(&mut stream1);

    let mut stream2 = TcpStream::connect(server.server.addr).await?;
    let mut s2 = TestScope::new(&mut stream2);

    s1.send("CAP LS 302").await?;
    s1.send("NICK tester1").await?;
    s1.send("USER UserName 0 * :Real Name").await?;
    s1.expect_cap_ls().await?;
    s1.send("CAP REQ :sasl").await?;
    s1.expect(":testserver CAP tester1 ACK :sasl").await?;
    s1.send("AUTHENTICATE PLAIN").await?;
    s1.expect(":testserver AUTHENTICATE +").await?;
    s1.send("AUTHENTICATE dGVzdGVyMQB0ZXN0ZXIxAHBhc3N3b3Jk").await?; // base64-encoded 'tester1\x00tester1\x00password'
    s1.expect(":testserver 900 tester1 tester1 tester1 :You are now logged in as tester1").await?;
    s1.expect(":testserver 903 tester1 :SASL authentication successful").await?;
    s1.send("CAP END").await?;
    s1.expect_server_introduction("tester1").await?;
    s1.expect_nothing().await?;

    s2.send("CAP LS 302").await?;
    s2.send("NICK tester2").await?;
    s2.send("USER UserName 0 * :Real Name").await?;
    s2.expect_cap_ls().await?;
    s2.send("CAP REQ :sasl").await?;
    s2.expect(":testserver CAP tester2 ACK :sasl").await?;
    s2.send("AUTHENTICATE PLAIN").await?;
    s2.expect(":testserver AUTHENTICATE +").await?;
    s2.send("AUTHENTICATE dGVzdGVyMgB0ZXN0ZXIyAHBhc3N3b3Jk").await?; // base64-encoded 'tester2\x00tester2\x00password'
    s2.expect(":testserver 900 tester2 tester2 tester2 :You are now logged in as tester2").await?;
    s2.expect(":testserver 903 tester2 :SASL authentication successful").await?;
    s2.send("CAP END").await?;
    s2.expect_server_introduction("tester2").await?;
    s2.expect_nothing().await?;

    s1.send("PRIVMSG tester2 :Henlo! How are you?").await?;
    s1.expect_nothing().await?;
    s2.expect(":tester1 PRIVMSG tester2 :Henlo! How are you?").await?;
    s2.expect_nothing().await?;

    s2.send("PRIVMSG tester1 good").await?;
    s2.expect_nothing().await?;
    s1.expect(":tester2 PRIVMSG tester1 :good").await?;
    s1.expect_nothing().await?;

    stream1.shutdown().await?;
    stream2.shutdown().await?;

    server.shutdown().await?;

    Ok(())
}