use std::borrow::Cow;
use std::cell::RefCell;
use std::collections::HashMap;
use std::io;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::{Duration, Instant};
use axum::Router;
use axum::body::Body;
use axum::http::{HeaderName, Method, Request, header};
use axum::response::Response;
use bollard::container::{CreateContainerOptions, RemoveContainerOptions, StartContainerOptions};
use http_body_util::BodyExt;
use rustic_backend::local::LocalBackend;
use rustic_core::repofile::{Node, SnapshotFile, SnapshotId};
use rustic_core::{
BackupOptions, ConfigOptions, Credentials, Excludes, KeyOptions, LocalSource,
LocalSourceFilterOptions, LocalSourceSaveOptions, OpenStatus, Repository, RepositoryBackends,
RepositoryOptions, SnapshotOptions, WriteBackend,
jiff::{Timestamp, Zoned, tz::TimeZone},
};
use sqlx::SqlitePool;
use sqlx::migrate::{Migration, MigrationType, Migrator};
use tower::ServiceExt;
use uuid::Uuid;
use crate::config::{Config, Lookup};
use crate::db::crypto::Clave;
use crate::docker::Cliente;
use crate::engine::{Ajustes, Dependencias, Motor};
use crate::rustic::BarrasDelMotor;
use crate::secret::SecretString;
use crate::state::{AppState, AuthState, EstadoDeBase};
pub(crate) mod proveedor_falso;
pub(crate) const MASTER_KEY: &str =
"clave-maestra-distinguible-000000000000000000000000000000000000";
pub(crate) const SESSION_SECRET: &str =
"clave-de-firma-distinguible-000000000000000000000000000000000000";
pub(crate) const OTRA_SESSION_SECRET: &str =
"otra-clave-de-firma-000000000000000000000000000000000000000000";
pub(crate) const OIDC_ISSUER_URL: &str = "https://idp.example.test/realms/kesto";
pub(crate) const OIDC_CLIENT_ID: &str = "kesto-panel";
pub(crate) const OIDC_CLIENT_SECRET: &str = "secreto-de-cliente-oidc-distinguible";
pub(crate) const OIDC_REDIRECT_URL: &str = "http://localhost:8080/api/v1/auth/callback";
pub(crate) const SUB: &str = "usuario-distinguible-1";
pub(crate) const EMAIL: &str = "operador-distinguible@example.test";
pub(crate) const BIND_ADDRESS: &str = "127.0.0.1:0";
pub(crate) fn lookup_from(pares: &[(&str, &str)]) -> impl Lookup + use<> {
let variables: HashMap<String, String> = pares
.iter()
.map(|(nombre, valor)| ((*nombre).to_owned(), (*valor).to_owned()))
.collect();
move |nombre: &str| variables.get(nombre).cloned()
}
pub(crate) const FICHERO_DE_LA_BASE: &str = "kesto.db";
pub(crate) struct DirectorioDePrueba {
ruta: PathBuf,
}
impl DirectorioDePrueba {
pub(crate) fn ruta(&self) -> &Path {
&self.ruta
}
pub(crate) fn ruta_de_la_base(&self) -> PathBuf {
self.ruta.join(FICHERO_DE_LA_BASE)
}
pub(crate) fn dsn(&self) -> String {
format!("sqlite://{}", self.ruta_de_la_base().display())
}
}
impl Drop for DirectorioDePrueba {
fn drop(&mut self) {
let _ = std::fs::remove_dir_all(&self.ruta);
}
}
pub(crate) fn directorio_de_prueba() -> DirectorioDePrueba {
let ruta = std::env::temp_dir().join(format!("kesto-prueba-{}", Uuid::new_v4()));
std::fs::create_dir_all(&ruta).expect("el directorio temporal de la prueba debe crearse");
DirectorioDePrueba { ruta }
}
pub(crate) const PASSPHRASE_DE_PRUEBA: &str =
"passphrase-de-prueba-000000000000000000000000000000000000";
pub(crate) const OTRA_PASSPHRASE_DE_PRUEBA: &str =
"otra-passphrase-de-prueba-0000000000000000000000000000000000";
pub(crate) const CLAVE_SECRETA_DE_PRUEBA: &str =
"clave-secreta-de-prueba-distinguible-000000000000000000000000000000000000";
pub(crate) const ETIQUETA_SEMBRADA: &str = "tarea-sembrada-de-prueba";
pub(crate) const ETIQUETA_AJENA: &str = "instantanea-de-otra-herramienta";
pub(crate) const HOST_AJENO: &str = "host-ajeno";
pub(crate) const AJENA: &str = "paisaje.txt";
pub(crate) const AJENA_CONTENIDO: &[u8] = b"contenido-de-la-instantanea-ajena\n";
pub(crate) const SEMILLA: &str = "semilla.txt";
pub(crate) const SEMILLA_CONTENIDO: &[u8] = b"contenido-de-la-semilla\n";
pub(crate) fn acceso_local(ruta: &Path, passphrase: &str) -> crate::rustic::Acceso {
crate::rustic::Acceso {
endpoint: ruta.to_string_lossy().into_owned(),
bucket: "cubo-de-prueba".to_owned(),
region: "region-de-prueba".to_owned(),
access_key_id: "clave-de-acceso-de-prueba".to_owned(),
secret_access_key: SecretString::new(CLAVE_SECRETA_DE_PRUEBA.to_owned()),
passphrase: SecretString::new(passphrase.to_owned()),
}
}
pub(crate) fn backends_locales(ruta: &Path) -> RepositoryBackends {
let backend: Arc<dyn WriteBackend> = Arc::new(
LocalBackend::new(ruta.to_string_lossy().as_ref(), Vec::new())
.expect("el backend local del arnés se construye sobre cualquier ruta"),
);
RepositoryBackends::new(backend, None)
}
pub(crate) fn barras_mudas() -> BarrasDelMotor {
BarrasDelMotor::mudas()
}
pub(crate) fn abridor_local(
acceso: &crate::rustic::Acceso,
barras: BarrasDelMotor,
) -> Result<(crate::rustic::Repositorio, crate::rustic::PuestaEnMarcha), crate::rustic::Error> {
crate::rustic::poner_en_marcha_con(
acceso,
backends_locales(Path::new(&acceso.endpoint)),
barras,
)
}
pub(crate) fn destino_de_s3_que_rechaza() -> String {
let oyente = std::net::TcpListener::bind("127.0.0.1:0")
.expect("el oyente local de la prueba se abre en un puerto libre");
let puerto = oyente
.local_addr()
.expect("el oyente local tiene dirección")
.port();
std::thread::spawn(move || {
for conexion in oyente.incoming() {
let Ok(mut flujo) = conexion else { break };
let mut peticion = [0_u8; 4096];
let _ = std::io::Read::read(&mut flujo, &mut peticion);
let cuerpo = r#"<?xml version="1.0" encoding="UTF-8"?><Error><Code>AccessDenied</Code><Message>Access Denied</Message><RequestId>prueba</RequestId></Error>"#;
let respuesta = format!(
"HTTP/1.1 403 Forbidden\r\nContent-Type: application/xml\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{cuerpo}",
cuerpo.len()
);
let _ = std::io::Write::write_all(&mut flujo, respuesta.as_bytes());
}
});
format!("http://127.0.0.1:{puerto}")
}
pub(crate) fn destino_local_de_prueba(nombre: &str) -> String {
let directorio = directorio_de_prueba();
let ruta = directorio.ruta().join(nombre);
retener(directorio);
ruta.to_string_lossy().into_owned()
}
pub(crate) fn abrir_local(ruta: &Path, passphrase: &str) -> Repository<OpenStatus> {
Repository::new(&RepositoryOptions::default(), &backends_locales(ruta))
.expect("el repositorio local se construye")
.open(&Credentials::Password(passphrase.to_owned()))
.expect("el repositorio local de prueba se abre con su passphrase")
}
pub(crate) fn snapshot_guardado(ruta: &Path, passphrase: &str, id: &str) -> SnapshotFile {
let repositorio = abrir_local(ruta, passphrase);
repositorio
.get_snapshots(&[id])
.expect("el snapshot del identificador dado se lee del repositorio")
.into_iter()
.next()
.expect("el identificador tiene que resolver exactamente un snapshot")
}
pub(crate) fn nodo_del_snapshot(ruta: &Path, passphrase: &str, id: &str, nombre: &str) -> Node {
let repositorio = abrir_local(ruta, passphrase);
let snapshot = repositorio
.get_snapshots(&[id])
.expect("el snapshot del identificador dado se lee del repositorio")
.into_iter()
.next()
.expect("el identificador tiene que resolver exactamente un snapshot");
let indexado = repositorio
.to_indexed()
.expect("el índice del repositorio de prueba se lee");
indexado
.node_from_snapshot_and_path(&snapshot, nombre)
.expect("el snapshot tiene que tener ese nodo dentro")
}
pub(crate) fn volcado_guardado(
ruta: &Path,
passphrase: &str,
id: &str,
nombre: &str,
) -> Option<Vec<u8>> {
let repositorio = abrir_local(ruta, passphrase);
let snapshot = repositorio
.get_snapshots(&[id])
.expect("el snapshot del identificador dado se lee del repositorio")
.into_iter()
.next()
.expect("el identificador tiene que resolver exactamente un snapshot");
let indexado = repositorio
.to_indexed()
.expect("el índice del repositorio de prueba se lee");
let Ok(nodo) = indexado.node_from_snapshot_and_path(&snapshot, nombre) else {
return None;
};
let mut salida = Vec::new();
indexado
.dump(&nodo, &mut salida)
.expect("un nodo de fichero se vuelca al buffer");
Some(salida)
}
fn instante(fecha: &str) -> Zoned {
fecha
.parse::<Timestamp>()
.expect("la fecha de la prueba es un instante en RFC 3339")
.to_zoned(TimeZone::UTC)
}
pub(crate) struct Siembra {
pub identificadores: Vec<String>,
pub nodo: String,
}
pub(crate) fn sembrar_snapshots(ruta: &Path, passphrase: &str, fechas: &[&str]) -> Siembra {
let backends = backends_locales(ruta);
let credenciales = Credentials::Password(passphrase.to_owned());
let repositorio = Repository::new(&RepositoryOptions::default(), &backends)
.expect("el repositorio local se construye")
.init(
&credenciales,
&KeyOptions::default(),
&ConfigOptions::default(),
)
.expect("un destino vacío se inicializa");
let fuente = directorio_de_prueba();
let fichero = fuente.ruta().join(SEMILLA);
std::fs::write(&fichero, SEMILLA_CONTENIDO).expect("el fichero de la semilla se escribe");
let nodo = fichero.to_string_lossy().trim_start_matches('/').to_owned();
let instantanea = SnapshotFile::from_options(
&SnapshotOptions::default()
.label(ETIQUETA_SEMBRADA.to_owned())
.host("host-sembrado".to_owned())
.command("comando-sembrado".to_owned()),
)
.expect("la instantánea sembrada se construye");
let rutas = vec![fuente.ruta().to_path_buf()];
let origen = LocalSource::new(
LocalSourceSaveOptions::default(),
&Excludes::default(),
&LocalSourceFilterOptions::default(),
&rutas,
)
.expect("la fuente local de la semilla se construye");
let indexado = repositorio
.to_indexed()
.expect("el índice del repositorio recién creado se lee");
let modelo = indexado
.archive(&BackupOptions::default(), &origen, instantanea, &rutas)
.expect("la semilla se archiva");
let copias: Vec<SnapshotFile> = fechas
.iter()
.map(|fecha| {
let mut copia = modelo.clone();
copia.time = instante(fecha);
copia.id = SnapshotId::default();
copia
})
.collect();
indexado
.save_snapshots(copias)
.expect("las copias sembradas se guardan");
indexado
.delete_snapshots(&[modelo.id])
.expect("la instantánea que sirvió de modelo se retira");
let identificadores: Vec<String> = indexado
.list::<SnapshotId>()
.expect("el repositorio lista sus snapshots")
.map(|id| id.to_hex().as_str().to_owned())
.collect();
let guardados = indexado
.get_snapshots(&identificadores)
.expect("los snapshots sembrados se leen del repositorio");
let identificadores = fechas
.iter()
.map(|fecha| {
let buscada = instante(fecha);
guardados
.iter()
.find(|snapshot| snapshot.time == buscada)
.expect("la fecha sembrada tiene que estar en el repositorio")
.id
.to_hex()
.as_str()
.to_owned()
})
.collect();
Siembra {
identificadores,
nodo,
}
}
pub(crate) fn sembrar_snapshot_ajeno(ruta: &Path, passphrase: &str) -> (String, String) {
let repositorio = abrir_local(ruta, passphrase)
.to_indexed_ids()
.expect("el índice reducido del repositorio sembrado se lee");
let fuente = directorio_de_prueba();
let fichero = fuente.ruta().join(AJENA);
std::fs::write(&fichero, AJENA_CONTENIDO).expect("el fichero ajeno se escribe");
let nodo = fichero.to_string_lossy().trim_start_matches('/').to_owned();
let instantanea = SnapshotFile::from_options(
&SnapshotOptions::default()
.label(ETIQUETA_AJENA.to_owned())
.host(HOST_AJENO.to_owned())
.command("rustic -r otro-sitio backup /otro/sitio".to_owned()),
)
.expect("la instantánea ajena se construye");
let rutas = vec![fuente.ruta().to_path_buf()];
let origen = LocalSource::new(
LocalSourceSaveOptions::default(),
&Excludes::default(),
&LocalSourceFilterOptions::default(),
&rutas,
)
.expect("la fuente local de la instantánea ajena se construye");
let guardada = repositorio
.archive(&BackupOptions::default(), &origen, instantanea, &rutas)
.expect("la instantánea ajena se archiva");
(guardada.id.to_hex().as_str().to_owned(), nodo)
}
pub(crate) fn config_con_dsn(dsn: &str) -> Config {
Config::from_lookup(&lookup_from(&[
("KESTO_MASTER_KEY", MASTER_KEY),
("BIND_ADDRESS", BIND_ADDRESS),
("DATABASE_URL", dsn),
]))
.expect("la configuración de prueba con base de datos es válida")
}
pub(crate) async fn base_con_clave() -> (DirectorioDePrueba, SqlitePool, Arc<Clave>) {
let directorio = directorio_de_prueba();
let pool = crate::db::preparar(&config_con_dsn(&directorio.dsn()))
.await
.expect("el esquema del §3 debe aplicarse sobre la base de prueba");
let clave = Arc::new(Clave::desde_maestra(&SecretString::new(
MASTER_KEY.to_owned(),
)));
(directorio, pool, clave)
}
static GUARDAS_DEL_ARNES: OnceLock<Mutex<Vec<DirectorioDePrueba>>> = OnceLock::new();
fn retener(directorio: DirectorioDePrueba) {
GUARDAS_DEL_ARNES
.get_or_init(|| Mutex::new(Vec::new()))
.lock()
.expect("el registro de guardas del arnés no puede envenenarse")
.push(directorio);
}
async fn base_del_arnes() -> (String, SqlitePool, Arc<Clave>) {
let (directorio, pool, clave) = base_con_clave().await;
let dsn = directorio.dsn();
retener(directorio);
(dsn, pool, clave)
}
pub(crate) async fn estado_con_base(config: Config) -> AppState {
let (_dsn, pool, clave) = base_del_arnes().await;
estado_de_prueba(config, EstadoDeBase::nuevo(pool, clave))
}
const SOCKET_MUERTO_DEL_MOTOR: &str = "unix:///no/existe/kesto-prueba.sock";
fn motor_de_pruebas(db: &EstadoDeBase) -> Motor {
let ajustes = Ajustes::nuevos(SOCKET_MUERTO_DEL_MOTOR).con_abridor(abridor_local);
let dependencias = Dependencias {
repositorios: db.repositorios.clone(),
targets: db.targets.clone(),
tareas: db.tareas.clone(),
ejecuciones: db.ejecuciones.clone(),
};
Motor::arrancar(ajustes, dependencias)
.expect("el motor de prueba se monta desde dentro del runtime de la prueba")
}
pub(crate) fn estado_de_prueba(config: Config, db: EstadoDeBase) -> AppState {
let motor = motor_de_pruebas(&db);
AppState::new(config, db, motor).expect("el estado de prueba se construye")
}
pub(crate) fn migrador_roto() -> Migrator {
Migrator {
migrations: Cow::Owned(vec![Migration::new(
1,
Cow::Borrowed("rota"),
MigrationType::Simple,
Cow::Borrowed("ESTO NO ES SQL"),
false,
)]),
ignore_missing: false,
locking: false,
no_tx: false,
}
}
pub(crate) async fn estado_desarrollo() -> AppState {
let (dsn, pool, clave) = base_del_arnes().await;
let config = Config::from_lookup(&lookup_from(&[
("KESTO_MASTER_KEY", MASTER_KEY),
("BIND_ADDRESS", BIND_ADDRESS),
("DATABASE_URL", dsn.as_str()),
]))
.expect("la configuración de desarrollo es válida");
let db = EstadoDeBase::nuevo(pool, clave);
estado_de_prueba(config, db)
}
pub(crate) async fn estado_oidc() -> AppState {
estado_oidc_con_emisor(&emisor_caido()).await
}
pub(crate) fn lookup_oidc_con_emisor(emisor: &str, dsn: &str) -> impl Lookup + use<> {
let variables: HashMap<String, String> = [
("KESTO_MASTER_KEY", MASTER_KEY),
("KESTO_SECRET", SESSION_SECRET),
("OIDC_ISSUER_URL", emisor),
("OIDC_CLIENT_ID", OIDC_CLIENT_ID),
("OIDC_CLIENT_SECRET", OIDC_CLIENT_SECRET),
("BIND_ADDRESS", BIND_ADDRESS),
("DATABASE_URL", dsn),
]
.into_iter()
.map(|(nombre, valor)| (nombre.to_owned(), valor.to_owned()))
.collect();
move |nombre: &str| variables.get(nombre).cloned()
}
pub(crate) async fn estado_oidc_con_emisor(emisor: &str) -> AppState {
let (dsn, pool, clave) = base_del_arnes().await;
let config = Config::from_lookup(&lookup_oidc_con_emisor(emisor, &dsn))
.expect("la configuración OIDC es válida");
let db = EstadoDeBase::nuevo(pool, clave);
estado_de_prueba(config, db)
}
pub(crate) async fn router_oidc_con_emisor(emisor: &str) -> Router {
crate::api::api_router(estado_oidc_con_emisor(emisor).await)
}
pub(crate) fn emisor_caido() -> String {
let listener = std::net::TcpListener::bind("127.0.0.1:0")
.expect("la reserva de un puerto efímero no puede fallar");
let puerto = listener
.local_addr()
.expect("el puerto reservado debe conocerse")
.port();
std::thread::spawn(move || {
for stream in listener.incoming().flatten() {
drop(stream);
}
});
format!("http://127.0.0.1:{puerto}")
}
pub(crate) fn parametros_de_consulta(url: &str) -> HashMap<String, String> {
let consulta = url
.split_once('?')
.map(|(_, consulta)| consulta)
.unwrap_or("");
consulta
.split('&')
.filter_map(|par| par.split_once('='))
.map(|(clave, valor)| {
let decodificar = |texto: &str| {
urlencoding::decode(texto)
.map(|decodificada| decodificada.into_owned())
.unwrap_or_else(|_| texto.to_owned())
};
(decodificar(clave), decodificar(valor))
})
.collect()
}
pub(crate) async fn auth_oidc() -> AuthState {
estado_oidc()
.await
.auth
.expect("el modo Oidc construye estado de autenticación")
}
pub(crate) async fn router_desarrollo() -> Router {
crate::api::api_router(estado_desarrollo().await)
}
pub(crate) async fn router_oidc() -> Router {
crate::api::api_router(estado_oidc().await)
}
pub(crate) fn session_cookie_header(token: &str) -> String {
format!("{}={token}", crate::auth::session::SESSION_COOKIE_NAME)
}
pub(crate) fn token_de_set_cookie(response: &Response) -> Option<String> {
header_text(response, header::SET_COOKIE).and_then(|cabecera| {
cabecera
.split(';')
.next()
.and_then(|par| {
par.trim()
.strip_prefix(&format!("{}=", crate::auth::session::SESSION_COOKIE_NAME))
})
.map(ToOwned::to_owned)
})
}
pub(crate) fn callback_url(state: &str, code: &str) -> String {
format!(
"/api/v1/auth/callback?code={}&state={}",
urlencoding::encode(code),
urlencoding::encode(state)
)
}
pub(crate) async fn login_state_y_nonce(router: Router) -> (String, String) {
let respuesta = get(router, "/api/v1/auth/login").await;
assert_eq!(
respuesta.status(),
axum::http::StatusCode::FOUND,
"el login debe redirigir para emitir el state y el nonce"
);
let location =
header_text(&respuesta, header::LOCATION).expect("el 302 del login trae Location");
let parametros = parametros_de_consulta(&location);
(
parametros
.get("state")
.expect("la redirección del login debe traer state")
.clone(),
parametros
.get("nonce")
.expect("la redirección del login debe traer nonce")
.clone(),
)
}
pub(crate) fn cookie_de_sesion() -> String {
let token = crate::auth::session::issue(
&SecretString::new(SESSION_SECRET.to_owned()),
SUB.to_owned(),
EMAIL.to_owned(),
)
.expect("la emisión de la sesión de prueba no puede fallar");
session_cookie_header(&token)
}
async fn request(router: Router, metodo: Method, uri: &str) -> Response {
let peticion = Request::builder()
.method(metodo)
.uri(uri)
.body(Body::empty())
.expect("petición de prueba válida");
router
.oneshot(peticion)
.await
.expect("el router no debe fallar")
}
pub(crate) async fn get(router: Router, uri: &str) -> Response {
request(router, Method::GET, uri).await
}
pub(crate) async fn post(router: Router, uri: &str) -> Response {
request(router, Method::POST, uri).await
}
pub(crate) async fn post_json(router: Router, uri: &str, cuerpo: &serde_json::Value) -> Response {
let peticion = Request::builder()
.method(Method::POST)
.uri(uri)
.header(header::CONTENT_TYPE, "application/json")
.body(Body::from(cuerpo.to_string()))
.expect("petición de prueba válida");
router
.oneshot(peticion)
.await
.expect("el router no debe fallar")
}
pub(crate) async fn put_json(router: Router, uri: &str, cuerpo: &serde_json::Value) -> Response {
let peticion = Request::builder()
.method(Method::PUT)
.uri(uri)
.header(header::CONTENT_TYPE, "application/json")
.body(Body::from(cuerpo.to_string()))
.expect("petición de prueba válida");
router
.oneshot(peticion)
.await
.expect("el router no debe fallar")
}
pub(crate) async fn delete(router: Router, uri: &str) -> Response {
request(router, Method::DELETE, uri).await
}
pub(crate) async fn post_crudo(router: Router, uri: &str, cuerpo: &str) -> Response {
let peticion = Request::builder()
.method(Method::POST)
.uri(uri)
.header(header::CONTENT_TYPE, "application/json")
.body(Body::from(cuerpo.to_owned()))
.expect("petición de prueba válida");
router
.oneshot(peticion)
.await
.expect("el router no debe fallar")
}
async fn request_with_cookie(router: Router, metodo: Method, uri: &str, cookie: &str) -> Response {
let peticion = Request::builder()
.method(metodo)
.uri(uri)
.header(header::COOKIE, cookie)
.body(Body::empty())
.expect("petición de prueba válida");
router
.oneshot(peticion)
.await
.expect("el router no debe fallar")
}
pub(crate) async fn get_with_cookie(router: Router, uri: &str, cookie: &str) -> Response {
request_with_cookie(router, Method::GET, uri, cookie).await
}
pub(crate) async fn post_with_cookie(router: Router, uri: &str, cookie: &str) -> Response {
request_with_cookie(router, Method::POST, uri, cookie).await
}
pub(crate) async fn body_text(response: Response) -> String {
let bytes = response
.into_body()
.collect()
.await
.expect("el cuerpo de la respuesta debe poder leerse")
.to_bytes();
String::from_utf8(bytes.to_vec()).expect("el cuerpo debe ser UTF-8")
}
pub(crate) async fn body_json(response: Response) -> serde_json::Value {
let texto = body_text(response).await;
serde_json::from_str(&texto).expect("el cuerpo debe ser JSON")
}
pub(crate) async fn siguiente_evento(cuerpo: &mut Body, plazo: Duration) -> Option<String> {
let limite = Instant::now() + plazo;
let mut acumulado = String::new();
loop {
let restante = limite.saturating_duration_since(Instant::now());
let Ok(Some(Ok(tramo))) = tokio::time::timeout(restante, cuerpo.frame()).await else {
return None;
};
let Ok(datos) = tramo.into_data() else {
continue;
};
acumulado.push_str(&String::from_utf8_lossy(&datos));
if acumulado.contains("\n\n") {
return Some(acumulado);
}
}
}
pub(crate) fn datos_del_evento(evento: &str) -> serde_json::Value {
let crudo = evento
.lines()
.find_map(|linea| linea.strip_prefix("data:"))
.unwrap_or_else(|| panic!("el evento no trae ninguna línea `data:`: {evento:?}"));
serde_json::from_str(crudo.trim())
.unwrap_or_else(|error| panic!("el `data` del evento no es JSON ({error}): {crudo:?}"))
}
pub(crate) fn header_text(response: &Response, name: HeaderName) -> Option<String> {
response
.headers()
.get(name)
.and_then(|valor| valor.to_str().ok())
.map(ToOwned::to_owned)
}
#[derive(Clone, Default)]
pub(crate) struct BufferDeLog(Arc<Mutex<Vec<u8>>>);
impl BufferDeLog {
pub(crate) fn texto(&self) -> String {
let bytes = self
.0
.lock()
.expect("el buffer de log no puede envenenarse");
String::from_utf8_lossy(&bytes).into_owned()
}
}
impl io::Write for BufferDeLog {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
self.0
.lock()
.expect("el buffer de log no puede envenenarse")
.extend_from_slice(buf);
Ok(buf.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for BufferDeLog {
type Writer = BufferDeLog;
fn make_writer(&'a self) -> Self::Writer {
self.clone()
}
}
thread_local! {
static BUFFER_DEL_HILO: RefCell<Option<BufferDeLog>> = const { RefCell::new(None) };
}
static SUSCRIPTOR_GLOBAL: OnceLock<()> = OnceLock::new();
struct EscritorDeCaptura(Option<BufferDeLog>);
impl io::Write for EscritorDeCaptura {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
match &mut self.0 {
Some(buffer) => buffer.write(buf),
None => Ok(buf.len()),
}
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
struct CapturaGlobal;
impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for CapturaGlobal {
type Writer = EscritorDeCaptura;
fn make_writer(&'a self) -> Self::Writer {
EscritorDeCaptura(BUFFER_DEL_HILO.with(|celda| celda.borrow().clone()))
}
}
fn instalar_suscriptor_global() {
SUSCRIPTOR_GLOBAL.get_or_init(|| {
let suscriptor = tracing_subscriber::fmt()
.with_max_level(tracing::Level::TRACE)
.with_ansi(false)
.without_time()
.with_writer(CapturaGlobal)
.finish();
if let Err(error) = tracing::subscriber::set_global_default(suscriptor) {
panic!(
"el arnés de captura no pudo instalar su suscriptor global: {error}. \
Este arnés es el único que instala un suscriptor en el binario de \
pruebas: quita el que lo pisa —el único del crate está en \
`src/main.rs:99`, y usa `try_init()`—, o la captura de registro \
deja de ser fiable."
);
}
tracing::callsite::rebuild_interest_cache();
});
}
#[must_use = "soltar la guardia deja de capturar: manténla viva mientras leas el buffer"]
pub(crate) struct GuardiaDeCaptura(Option<BufferDeLog>);
impl Drop for GuardiaDeCaptura {
fn drop(&mut self) {
BUFFER_DEL_HILO.with(|celda| *celda.borrow_mut() = self.0.take());
}
}
pub(crate) fn capturar_log() -> (BufferDeLog, GuardiaDeCaptura) {
instalar_suscriptor_global();
let buffer = BufferDeLog::default();
let anterior = BUFFER_DEL_HILO.with(|celda| celda.borrow_mut().replace(buffer.clone()));
(buffer, GuardiaDeCaptura(anterior))
}
pub(crate) fn lector_del_volcado(
origen: &crate::rustic::OrigenDelVolcado,
) -> crate::rustic::LectorDeVolcado {
use rustic_core::{ReadSource as _, ReadSourceOpen as _};
let mut entradas = origen.entries();
let entrada = entradas
.next()
.expect("el origen de un volcado entrega una entrada")
.expect("la entrada de un volcado no puede ser un error");
entrada
.open
.expect("la entrada del volcado tiene que traer cómo abrirse")
.open()
.expect("el lector del volcado tiene que abrirse")
}
const IMAGEN_DE_PRUEBA: &str = "docker.io/library/alpine:3.23";
const COMANDO_DE_PRUEBA: &[&str] = &["sleep", "600"];
pub(crate) const TIMEOUT_DEL_MOTOR: Duration = Duration::from_secs(300);
pub(crate) fn socket_del_motor() -> Option<String> {
let candidato = std::env::var("DOCKER_HOST")
.ok()
.filter(|valor| !valor.trim().is_empty())
.unwrap_or_else(ruta_de_podman);
let ruta = candidato.trim_start_matches("unix://");
Path::new(ruta).exists().then_some(candidato)
}
fn ruta_de_podman() -> String {
match std::env::var("XDG_RUNTIME_DIR") {
Ok(dir) if !dir.is_empty() => format!("unix://{dir}/podman/podman.sock"),
_ => "unix:///run/user/1000/podman/podman.sock".to_owned(),
}
}
pub(crate) fn cliente_del_motor() -> Option<Cliente> {
let socket = socket_del_motor()?;
Cliente::conectar(&socket, TIMEOUT_DEL_MOTOR).ok()
}
pub(crate) struct ContenedorDePrueba {
motor: bollard::Docker,
nombre: String,
socket: String,
}
impl ContenedorDePrueba {
pub(crate) async fn nuevo(prueba: &str) -> Self {
let socket = socket_del_motor()
.expect("las pruebas del motor necesitan un socket de Docker o Podman");
let motor = bollard::Docker::connect_with_socket(
&socket,
TIMEOUT_DEL_MOTOR.as_secs(),
bollard::API_DEFAULT_VERSION,
)
.expect("el socket existe, así que el cliente del arnés se construye");
let nombre = format!("kesto-prueba-{prueba}");
let _ = motor
.remove_container(&nombre, Some(opciones_de_borrado()))
.await;
motor
.create_container(
Some(CreateContainerOptions {
name: nombre.clone(),
platform: None,
}),
bollard::container::Config {
image: Some(IMAGEN_DE_PRUEBA.to_owned()),
cmd: Some(
COMANDO_DE_PRUEBA
.iter()
.map(|arg| (*arg).to_owned())
.collect(),
),
..Default::default()
},
)
.await
.expect("el contenedor de prueba debe poder crearse");
let contenedor = Self {
motor,
nombre,
socket,
};
contenedor
.motor
.start_container(&contenedor.nombre, None::<StartContainerOptions<String>>)
.await
.expect("el contenedor de prueba debe poder arrancarse");
contenedor
}
pub(crate) fn nombre(&self) -> &str {
&self.nombre
}
pub(crate) async fn retirar(&self) {
if let Err(error) = self
.motor
.remove_container(&self.nombre, Some(opciones_de_borrado()))
.await
{
eprintln!(
"el contenedor de prueba {} no se pudo retirar por la API: {error}",
self.nombre
);
}
}
}
fn opciones_de_borrado() -> RemoveContainerOptions {
RemoveContainerOptions {
force: true,
..Default::default()
}
}
impl Drop for ContenedorDePrueba {
fn drop(&mut self) {
let _ = std::process::Command::new("podman")
.args(["--remote", "--url", &self.socket, "rm", "-f", &self.nombre])
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.status();
}
}
#[cfg(test)]
mod tests {
use super::capturar_log;
const MENSAJE_DE_DEBUG: &str = "mensaje del callsite de depuración del arnés";
const MENSAJE_DE_ERROR: &str = "mensaje del callsite de error del arnés";
fn emitir_por_debug() {
tracing::debug!("{MENSAJE_DE_DEBUG}");
}
fn emitir_por_error() {
tracing::error!("{MENSAJE_DE_ERROR}");
}
#[test]
fn un_callsite_estrenado_sin_captura_llega_al_buffer() {
let (buffer, _guard) = capturar_log();
std::thread::spawn(emitir_por_error)
.join()
.expect("el hilo que estrena el callsite no puede entrar en pánico");
emitir_por_error();
emitir_por_debug();
let log = buffer.texto();
assert!(
log.contains(MENSAJE_DE_ERROR),
"el registro de un callsite estrenado sin captura se perdió: {log}"
);
assert!(
log.contains(MENSAJE_DE_DEBUG),
"el registro del callsite de `debug!` no llegó al buffer: {log}"
);
}
}