rstango 0.3.0

//! A context in which servers are run and their devices reached, see [`TestContext`].

use std::sync::{Arc, Mutex};

use futures::executor::block_on;

use crate::backend::{default_backend, Backend, BackendDev, BackendServer, BackendServerConfig};
use crate::server::{Server, ServerBuilder, ServerHandle};
use crate::database::{DatabaseProxy, ExportDevInfo};
use crate::dev::{DevProxy, DATABASE_DEV_NAME};
use crate::error::{Error, ErrorKind, Result};
use crate::local::{self, LocalDevHandle};

use super::database::{TestDatabase, TestDatabaseDev};


/// The host everything a [`TestContext`] serves listens at, which keeps it to this machine.
const HOST: &str = "127.0.0.1";

/// Servers run for a test, and what they are reached through, which is what pytango's
/// `MultiDeviceTestContext` is.
///
/// A context serves a database of its own, see [`TestDatabase`], which is filled
/// through [`Self::database`] as a real one is, then the servers registered in it, each on
/// a port the system picks, see [`Self::serve`], whose devices are then found through that
/// database as a client of a real one finds them, see [`Self::dev`].
///
/// ```no_run
/// # use tango::testing::TestContext;
/// # use tango::server::{ClassDef, ServerBuilder};
/// let ctx = TestContext::new()?;
/// ctx.database().add_server("Motor/test", &[("test/motor/1", "Motor")])?;
/// ctx.serve(ServerBuilder::new("Motor", "test").class(ClassDef::<()>::new_default("Motor"))?)?;
/// ctx.dev("test/motor/1").ping()?;
/// # Ok::<_, tango::error::Error>(())
/// ```
///
/// Everything is served for real, through the default backend: the database and the
/// devices are reached by any process, at [`Self::host`] and [`Self::port`] for the
/// database, and by this one directly, as every device this process serves, see
/// [`Self::is_local`]. Any implementation of a device can also be served on its own, see
/// [`Self::serve_dev`].
///
/// Everything it serves is dropped with it, with no side effect: a server that is to be
/// stopped properly, its devices deleted, is stopped through the context, see
/// [`Self::stop`].
#[derive(Debug)]
pub struct TestContext {
    /// The servers served, each dropped with its handle.
    servers: Mutex<Vec<ServerHandle>>,
    /// The devices served on their own, see [`Self::serve_dev`], by name.
    devs: Mutex<Vec<(Arc<str>, LocalDevHandle)>>,
    /// What this process reaches the database by directly.
    _database_handle: LocalDevHandle,
    /// What serves the database and the devices served on their own.
    own: Arc<dyn BackendServer>,
    test_database: Arc<TestDatabase>,
    database: DatabaseProxy,
    /// The host the database answers at, see [`Self::host`].
    host: Arc<str>,
    backend: Arc<dyn Backend>,
}

impl TestContext {

    /// A context serving an empty database, on a port the system picks.
    pub fn new() -> Result<Self> {
        Self::with_port(0)
    }

    /// A context serving an empty database on that port, one the system picks if `0`,
    /// which is what a database that clients are told of in advance needs.
    pub fn with_port(port: u16) -> Result<Self> {

        let backend = default_backend();
        let own = block_on(backend.serve(&BackendServerConfig::new()
            .with_host(HOST)
            .with_port(port)))?;

        // Served under the name a client connects to a database by, which is the object key
        // a client of cppTango asks for too.
        let test_database = Arc::new(TestDatabase::default());
        let dev = Arc::new(TestDatabaseDev(Arc::clone(&test_database))) as Arc<dyn BackendDev>;
        let itr = own.add(DATABASE_DEV_NAME, Arc::clone(&dev))?;
        let database_handle = local::register(&dev, itr, own.host(), own.port(), Arc::from(DATABASE_DEV_NAME));

        let database = DatabaseProxy::new_with_backend(Arc::clone(&backend), &own.host(), own.port());

        Ok(Self {
            servers: Mutex::default(),
            devs: Mutex::default(),
            _database_handle: database_handle,
            host: own.host(),
            own,
            test_database,
            database,
            backend,
        })

    }

    /// The host the database answers at.
    pub fn host(&self) -> &str {
        &self.host
    }

    /// The port the database answers at.
    pub fn port(&self) -> u16 {
        self.own.port()
    }

    /// The database of this context, which is where the servers are registered before
    /// being served, with [`DatabaseProxy::add_server`], and their properties set.
    pub fn database(&self) -> &DatabaseProxy {
        &self.database
    }

    /// The database itself, which tells what it was asked and is made to fail, see
    /// [`TestDatabase`].
    pub fn test_database(&self) -> &Arc<TestDatabase> {
        &self.test_database
    }

    /// Serve that server until it is stopped, see [`Self::stop`], or this context is
    /// dropped: it is served through the backend of this context, registered in its
    /// database, and listens on this machine only, on a port the system picks, and is never
    /// stopped on a signal, whatever the builder was told of these, see
    /// [`ServerBuilder::stop_on_signal`]. Its devices are then reached with [`Self::dev`].
    ///
    /// The server given back is shared with the context, which keeps owning it: it is
    /// dropped all the same when the context is.
    pub fn serve(&self, builder: ServerBuilder) -> Result<Arc<Server>> {
        let handle = builder
            .backend(Arc::clone(&self.backend))
            .database(self.database.clone())
            .host(HOST)
            .port(0)
            .stop_on_signal(false)
            .serve()?;
        let server = handle.server_arc();
        self.servers.lock().unwrap().push(handle);
        Ok(server)
    }

    /// The server of that `exe/inst` name this context serves, if it does.
    pub fn server(&self, name: &str) -> Option<Arc<Server>> {
        self.servers.lock().unwrap().iter()
            .find(|handle| handle.server().name().eq_ignore_ascii_case(name))
            .map(ServerHandle::server_arc)
    }

    /// Stop the server of that `exe/inst` name and wait for its devices to be deleted,
    /// giving back the first error met deleting one, see [`ServerHandle::wait`]. A server
    /// that stopped on its own is only waited for.
    ///
    /// The server stays with the context, stopped, until it is dropped.
    pub fn stop(&self, name: &str) -> Result<()> {
        let server = self.server(name).ok_or_else(|| Error::new(ErrorKind::DevNotFound)
            .with_description(format!("No server {name} is served by this context")))?;
        block_on(server.stop())?;
        let servers = self.servers.lock().unwrap();
        let handle = servers.iter().find(|handle| handle.server().name().eq_ignore_ascii_case(name)).unwrap();
        handle.wait()
    }

    /// Serve that device, under the name it gives, until this context is dropped, through
    /// the backend of this context, and export it in its database, which is where
    /// [`Self::dev`] then finds it: any implementation of a device is tested that way, a
    /// client of it above all, without any server around it.
    ///
    /// The device must be defined in the database first, with [`DatabaseProxy::add_dev`],
    /// as a real database refuses to export a device it does not know. A device served
    /// again under the same name replaces the one that was.
    pub fn serve_dev(&self, dev: Arc<dyn BackendDev>) -> Result<()> {

        let name = block_on(dev.name())?;
        let mut devs = self.devs.lock().unwrap();

        if let Some(index) = devs.iter().position(|(other, _)| other.eq_ignore_ascii_case(&name)) {
            devs.remove(index);
            self.own.remove(&name);
        }

        let itr = self.own.add(&name, Arc::clone(&dev))?;
        let handle = local::register(&dev, Arc::clone(&itr), self.own.host(), self.own.port(), Arc::clone(&name));

        let exported = self.database.export_dev(ExportDevInfo {
            name: (&*name).into(),
            itr: (&*itr).into(),
            host: (&*self.own.host()).into(),
            version: self.own.version().to_string().into(),
            pid: std::process::id() as i32,
        });

        if let Err(err) = exported {
            drop(handle);
            self.own.remove(&name);
            return Err(err);
        }

        devs.push((name, handle));
        Ok(())

    }

    /// A proxy to the device of that name, found through the database of this context.
    pub fn dev(&self, name: &str) -> DevProxy {
        DevProxy::new_from_database_with_backend(Arc::clone(&self.backend), &self.database, name)
    }

    /// Whether the device of that name, as the database exports it, is reached directly by
    /// this process, which a server it serves is not when told so, see
    /// [`ServerBuilder::local`]: it is then reached through the backend, as another process
    /// reaches it.
    pub fn is_local(&self, name: &str) -> Result<bool> {
        let info = self.database.import_dev(name)?;
        Ok(local::find(info.itr.value()).is_some())
    }

    /// The device of that name, as the database exports it, reached as a client of this
    /// process reaches it: directly if this process serves it, see [`Self::is_local`], which
    /// a server it serves only does not when told so, see [`ServerBuilder::local`], and
    /// through the backend otherwise.
    pub fn backend_dev(&self, name: &str) -> Result<Arc<dyn BackendDev>> {
        let info = self.database.import_dev(name)?;
        match local::find(info.itr.value()) {
            Some(dev) => Ok(dev),
            None => block_on(self.backend.connect(info.itr.value())),
        }
    }

}

impl Drop for TestContext {

    /// Drop every server, then every device, then the database, with no side effect:
    /// whatever was in flight is never answered, see [`BackendServer`].
    fn drop(&mut self) {
        self.servers.get_mut().unwrap().clear();
        self.devs.get_mut().unwrap().clear();
    }

}

#[cfg(test)]
mod tests {

    use super::*;

    use crate::server::ClassDef;

    /// Two servers in the same context, each reached through its database, and stopped
    /// through the context.
    #[test]
    fn two_servers() {

        let ctx = TestContext::new().unwrap();
        ctx.database().add_server("Motor/test", &[("test/motor/1", "Motor"), ("test/motor/2", "Motor")]).unwrap();
        ctx.database().add_server("Slit/test", &[("test/slit/1", "Slit")]).unwrap();

        let motor = ctx.serve(ServerBuilder::new("Motor", "test").class(ClassDef::<()>::new_default("Motor")).unwrap()).unwrap();
        assert_ne!(motor.backend().unwrap().port(), 0);
        ctx.serve(ServerBuilder::new("Slit", "test").class(ClassDef::<()>::new_default("Slit")).unwrap()).unwrap();

        for name in ["test/motor/1", "test/motor/2", "test/slit/1"] {
            let dev = ctx.dev(name);
            dev.ping().unwrap();
            assert_eq!(&*dev.name().unwrap(), name);
        }

        let admin = ctx.dev("test/motor/1").admin_dev().unwrap();
        assert_eq!(admin.name().unwrap().to_lowercase(), "dserver/motor/test");

        ctx.stop("Motor/test").unwrap();
        ctx.stop("Slit/test").unwrap();
        let unexported = ctx.test_database().log().into_iter().filter(|(name, _)| name == "DbUnExportServer").count();
        assert_eq!(unexported, 2);

    }

    /// The database and the devices of a server told not to be reached directly, reached
    /// through the protocol, as another process does.
    #[test]
    fn not_local() {

        let ctx = TestContext::new().unwrap();
        ctx.database().add_server("Motor/test", &[("test/motor/1", "Motor")]).unwrap();
        ctx.serve(ServerBuilder::new("Motor", "test").local(false).class(ClassDef::<()>::new_default("Motor")).unwrap()).unwrap();

        let database = block_on(default_backend().connect_at(ctx.host(), ctx.port(), DATABASE_DEV_NAME)).unwrap();
        block_on(database.ping()).unwrap();

        let itr = ctx.database().import_dev("test/motor/1").unwrap().itr;
        assert!(local::find(itr.value()).is_none());

        let dev = ctx.dev("test/motor/1");
        assert_eq!(dev.admin_dev_name().unwrap().to_lowercase(), "dserver/motor/test");
        let remote = ctx.backend_dev("test/motor/1").unwrap();
        assert_eq!(block_on(remote.admin_dev_name()).unwrap().to_lowercase(), "dserver/motor/test");

    }

    /// A server stopped before the context, which keeps it, stopped.
    #[test]
    fn stop() {

        let ctx = TestContext::new().unwrap();
        ctx.database().add_server("Motor/test", &[("test/motor/1", "Motor")]).unwrap();
        ctx.serve(ServerBuilder::new("Motor", "test").class(ClassDef::<()>::new_default("Motor")).unwrap()).unwrap();

        ctx.stop("motor/TEST").unwrap();
        assert!(ctx.dev("test/motor/1").ping().is_err());
        assert!(ctx.server("Motor/test").is_some());
        assert!(ctx.stop("Other/test").is_err());

    }

    /// A device served on its own, which is a client of a device of the same name served by
    /// another context, in a database of its own: reached directly by this process and
    /// through the backend by any other, and no longer served once the context is dropped.
    #[test]
    fn serve_dev() {

        let other = TestContext::new().unwrap();
        other.database().add_server("Motor/other", &[("test/motor/1", "Motor")]).unwrap();
        other.serve(ServerBuilder::new("Motor", "other").class(ClassDef::<()>::new_default("Motor")).unwrap()).unwrap();
        let dev = other.backend_dev("test/motor/1").unwrap();

        let ctx = TestContext::new().unwrap();

        // Not defined, so not exported.
        assert!(ctx.serve_dev(Arc::clone(&dev)).is_err());

        ctx.database().add_dev("test/motor/1", "Motor", "Motor/test").unwrap();
        ctx.serve_dev(dev).unwrap();

        // Directly, as this process does, the device being a client of the other one.
        let proxy = ctx.dev("test/motor/1");
        proxy.ping().unwrap();
        assert_eq!(proxy.admin_dev_name().unwrap().to_lowercase(), "dserver/motor/other");

        // Through the backend, as another process does.
        let itr = ctx.database().import_dev("test/motor/1").unwrap().itr;
        let remote = block_on(default_backend().connect(itr.value())).unwrap();
        assert_eq!(block_on(remote.admin_dev_name()).unwrap().to_lowercase(), "dserver/motor/other");

        let database = ctx.database().clone();
        drop(ctx);
        assert!(local::find(itr.value()).is_none());
        assert!(database.ping().is_err());
        other.dev("test/motor/1").ping().unwrap();

    }

}