feat: support running tests in parallel in multiple clusters

README.md

@@ -16,6 +16,7 @@
     - [Логирование](#логирование)
     - [Файлы логов кластера](#файлы-логов-кластера)
     - [Совместимость с `rstest`](#совместимость-с-rstest)
+    - [Несколько кластеров с разными топологиями](#несколько-кластеров-с-разными-топологиями)
     - [Применение конфигурации плагина к запущенному кластеру Picodata](#применение-конфигурации-плагина-к-запущенному-кластеру-picodata)
   - [Модульное тестирование](#модульное-тестирование)
     - [Требования к конфигурации](#требования-к-конфигурации)
@@ -228,6 +229,55 @@ mod test_mod {
     }
 ```
 
+### Несколько кластеров с разными топологиями
+
+По умолчанию все тесты запускаются на одном кластере `cluster` с топологией из `topology.toml`.
+С помощью фикстуры и функции `cluster_with_topology`, принимающей `TopologyPath`, можно
+поднять кластер с альтернативной топологией. Путь указывается относительно
+корневой директории плагина; значение по умолчанию, `TopologyPath::default()`, соответствует
+`topology.toml` в корне плагина.
+
+```rust
+use picotest::*;
+
+#[fixture]
+fn single_node() -> &'static Cluster {
+    cluster_with_topology("topologies/single_node.toml".into())
+}
+
+#[picotest]
+mod single_node_tests {
+    fn test_instances(single_node: &Cluster) {
+        assert_eq!(single_node.instances().len(), 1);
+    }
+}
+```
+
+Кластер создаётся при первом обращении и переиспользуется всеми тестами с такой же топологией
+(`cluster_with_topology` с одинаковой топологией возвращает один и тот же кластер).
+Кластеры с разными топологиями работают одновременно, поэтому тесты на них выполняются параллельно.
+
+Чтобы кластеры не мешали друг другу, каждый получает собственный адрес из подсети `127.0.0.0/8`
+(`127.0.0.1`, `127.0.0.2` и так далее) со стандартными портами. Адреса не пересекаются и между
+процессами, поэтому тесты можно запускать параллельно в нескольких процессах, например через `cargo nextest`.
+
+Адрес инстанса не следует задавать в тесте вручную, его нужно получать из кластера:
+
+```rust
+#[picotest]
+fn test_http_request(single_node: &Cluster) {
+    let instance = single_node.main();
+
+    instance.http_addr();  // адрес HTTP сервера, например 127.0.0.2:8001
+    instance.pg_addr();    // адрес для подключения по Postgres протоколу
+    instance.bin_addr();   // адрес для подключения по iproto
+    single_node.host();    // адрес, на котором работают все инстансы кластера
+}
+```
+
+Если на машине доступен только адрес `127.0.0.1`, следующие кластеры
+запускаются на нём же со сдвигом портов на 100, 200 и так далее.
+
 ### Применение конфигурации плагина к запущенному кластеру Picodata
 
 Picotest позволяет менять конфигурацию сервисов плагина во время исполнения интеграционного теста.

picotest/src/lib.rs

@@ -4,11 +4,16 @@ pub use picotest_helpers::{
 };
 pub use picotest_macros::*;
 pub use rstest::*;
+use std::collections::BTreeMap;
+use std::sync::{Mutex, PoisonError};
 pub use std::{panic, path::PathBuf, sync::OnceLock, time::Duration};
 
 pub mod internal;
 
-pub static SESSION_CLUSTER: OnceLock<Cluster> = OnceLock::new();
+type ClusterKey = (PathBuf, String);
+
+static SESSION_CLUSTERS: Mutex<BTreeMap<ClusterKey, &'static OnceLock<Cluster>>> =
+    Mutex::new(BTreeMap::new());
 
 pub type PluginConfigMap = picotest_helpers::PluginConfigMap;
 
@@ -17,18 +22,76 @@ pub fn cluster() -> &'static Cluster {
     get_or_create_session_cluster(None)
 }
 
+/// Path to the cluster topology file.
+#[derive(Clone, Debug, Default)]
+pub struct TopologyPath(Option<PathBuf>);
+
+impl TopologyPath {
+    pub fn new(path: impl Into<PathBuf>) -> Self {
+        Self(Some(path.into()))
+    }
+
+    fn load(&self) -> PluginTopology {
+        let plugin_root = internal::plugin_root_dir();
+        let path = match &self.0 {
+            Some(path) => plugin_root.join(path),
+            None => internal::plugin_topology_path(&plugin_root),
+        };
+
+        picotest_helpers::topology::parse_topology(&path)
+            .unwrap_or_else(|err| panic!("Failed to load cluster topology: {err:#}"))
+    }
+}
+
+impl From<&str> for TopologyPath {
+    fn from(path: &str) -> Self {
+        Self::new(path)
+    }
+}
+
+impl From<PathBuf> for TopologyPath {
+    fn from(path: PathBuf) -> Self {
+        Self::new(path)
+    }
+}
+
+#[fixture]
+pub fn cluster_with_topology(
+    #[default(TopologyPath::default())] topology: TopologyPath,
+) -> &'static Cluster {
+    get_or_create_session_cluster(Some(&topology.load()))
+}
+
 pub fn get_or_create_session_cluster(plugin_topology: Option<&PluginTopology>) -> &'static Cluster {
-    SESSION_CLUSTER.get_or_init(|| {
-        env_logger::init();
-        let plugin_topology = plugin_topology.cloned();
+    let plugin_topology =
+        plugin_topology.map_or_else(|| TopologyPath::default().load(), Clone::clone);
+
+    let key = (internal::plugin_root_dir(), format!("{plugin_topology:?}"));
+    let cell = *SESSION_CLUSTERS
+        .lock()
+        .unwrap_or_else(PoisonError::into_inner)
+        .entry(key)
+        .or_insert_with(|| Box::leak(Box::default()));
 
-        internal::create_cluster(plugin_topology)
+    cell.get_or_init(|| {
+        let _ = env_logger::try_init();
+
+        internal::create_cluster(Some(plugin_topology))
     })
 }
 
 #[dtor]
 unsafe fn tear_down() {
-    if let Some(cluster) = SESSION_CLUSTER.get() {
-        cluster.stop().expect("Failed to stop the cluster");
+    let clusters = SESSION_CLUSTERS
+        .lock()
+        .unwrap_or_else(PoisonError::into_inner);
+
+    let mut failed = false;
+    for cluster in clusters.values().filter_map(|cell| cell.get()) {
+        if let Err(err) = cluster.stop() {
+            eprintln!("Failed to stop the cluster: {err:#}");
+            failed = true;
+        }
     }
+    assert!(!failed, "Failed to stop the cluster");
 }

picotest/tests/assets/single_node_topology.toml

@@ -0,0 +1,11 @@
+[tier.default]
+replicasets = 1
+replication_factor = 1
+
+[plugin.test_plugin]
+migration_context = [
+    { name = "example_name", value = "example_value" },
+]
+
+[plugin.test_plugin.service.example_service]
+tiers = ["default"]

picotest/tests/integration_suite/integration.rs

@@ -100,6 +100,42 @@ mod test_mod {
     }
 }
 
+const SINGLE_NODE_TOPOLOGY: &str = "../../picotest/tests/assets/single_node_topology.toml";
+
+#[fixture]
+fn single_node() -> &'static Cluster {
+    cluster_with_topology(SINGLE_NODE_TOPOLOGY.into())
+}
+
+#[picotest]
+fn test_cluster_with_custom_topology(plugin: &TestPlugin, single_node: &Cluster) {
+    assert_eq!(single_node.instances().len(), 1);
+    assert_eq!(cluster.instances().len(), 4);
+
+    // Clusters run side by side on their own addresses.
+    assert_ne!(single_node.uuid, cluster.uuid);
+    assert_ne!(single_node.host(), cluster.host());
+    assert_eq!(single_node.main().host, single_node.host());
+
+    for cluster in [single_node, cluster] {
+        let enabled = cluster.run_query(format!(
+            r#"SELECT enabled FROM _pico_plugin WHERE name = '{}';"#,
+            plugin.name
+        ));
+        assert!(enabled.is_ok_and(|enabled| enabled.contains("true")));
+    }
+}
+
+#[picotest]
+fn test_cluster_is_shared_within_topology(
+    single_node: &Cluster,
+    #[from(cluster_with_topology)]
+    #[with(TopologyPath::new(SINGLE_NODE_TOPOLOGY))]
+    same_cluster: &Cluster,
+) {
+    assert_eq!(single_node.uuid, same_cluster.uuid);
+}
+
 #[picotest]
 fn test_select_from_missing_table() {
     let result = cluster.run_query("SELECT * FROM table1");