use axum::Router;
use axum::extract::{Path, State};
use axum::http::StatusCode;
use axum::routing::{get, post};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use time::OffsetDateTime;
use time::format_description::well_known::Rfc3339;
use uuid::Uuid;
use crate::api::extract::{Json, Query};
use crate::api::response::ApiResponse;
use crate::api::targets::texto_del_tipo;
use crate::api::{contenido, del_motor};
use crate::db::entities::{ObjetivoDeRestauracion, Repositorio, TipoDeTarget};
use crate::error::AppError;
use crate::rustic::{Acceso, ResumenDeSnapshot};
use crate::state::AppState;
pub const LA_VENTANA: usize = 200;
pub fn router() -> Router<AppState> {
Router::new()
.route("/api/v1/snapshots", get(listado))
.route("/api/v1/snapshots/{id}/restore", post(restaurar))
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ConsultaDelCable {
pub repository_id: Option<String>,
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct CuerpoDeLaRestauracion {
pub repository_id: String,
}
pub async fn listado(
State(estado): State<AppState>,
Query(consulta): Query<ConsultaDelCable>,
) -> Result<ApiResponse, AppError> {
let fila = repositorio_del_explorador(&estado, consulta).await?;
let instantaneas = estado
.motor
.listar_snapshots(acceso_del_repositorio(&fila), LA_VENTANA)
.await
.map_err(del_motor)?;
let truncated = instantaneas.len() >= LA_VENTANA;
let identificadores: Vec<String> = instantaneas
.iter()
.map(|resumen| resumen.id.clone())
.collect();
let objetivos = estado
.db
.ejecuciones
.objetivos_de_las_instantaneas(&identificadores, &fila.id)
.await?;
let vista = ListadoDelCable::desde(&instantaneas, &objetivos, truncated)?;
Ok(ApiResponse::new(
StatusCode::OK,
"instantáneas listadas",
contenido(&vista)?,
))
}
pub async fn restaurar(
State(estado): State<AppState>,
Path(instantanea): Path<String>,
Json(cuerpo): Json<CuerpoDeLaRestauracion>,
) -> Result<ApiResponse, AppError> {
let id = identificador(&cuerpo.repository_id)?;
let fila = estado.db.repositorios.leer(&id).await?.ok_or_else(|| {
AppError::NotFound("no existe ningún repositorio con ese identificador".to_owned())
})?;
let objetivo = estado
.db
.ejecuciones
.objetivo_de_la_instantanea(&instantanea, &id)
.await?;
let Some(objetivo) = objetivo else {
return Err(AppError::Conflict(
"la instantánea no está asociada a ninguna ejecución ni target: no hay sobre qué \
restaurarla"
.to_owned(),
));
};
if objetivo.target_type != TipoDeTarget::Directory {
return Err(AppError::UnprocessableEntity(
"restaurar targets de base de datos todavía no está soportado en esta versión: solo se \
restauran targets de directorio"
.to_owned(),
));
}
let destino = objetivo.host_path.ok_or_else(|| {
AppError::UnprocessableEntity(
"el target de la instantánea es de directorio pero no declara su ruta en el host: no \
hay dónde restaurarla"
.to_owned(),
)
})?;
let acceso_de_la_comprobacion = acceso_del_repositorio(&fila);
if !estado
.motor
.tiene_la_instantanea(acceso_de_la_comprobacion, &instantanea)
.await
.map_err(del_motor)?
{
return Err(AppError::NotFound(
"el repositorio indicado no tiene ninguna instantánea con ese identificador: no hay \
nada que restaurar"
.to_owned(),
));
}
let reserva = estado
.motor
.reservar_el_destino(&destino)
.map_err(del_motor)?;
let execution_id = Uuid::new_v4();
let motor = estado.motor.clone();
let acceso = acceso_del_repositorio(&fila);
let instantanea_tarea = instantanea.clone();
tokio::spawn(async move {
let _reserva = reserva;
if let Err(error) = motor
.restaurar(acceso, instantanea_tarea, destino, execution_id)
.await
{
tracing::error!(%error, "la restauración en segundo plano falló");
}
});
let vista = RestauracionDelCable {
execution_id: execution_id.to_string(),
snapshot_id: instantanea,
target_id: objetivo.target_id.to_string(),
target_name: objetivo.target_name,
};
Ok(ApiResponse::new(
StatusCode::OK,
"restauración lanzada",
contenido(&vista)?,
))
}
#[derive(Debug, Serialize)]
struct RestauracionDelCable {
execution_id: String,
snapshot_id: String,
target_id: String,
target_name: String,
}
fn identificador(repository_id: &str) -> Result<Uuid, AppError> {
Uuid::parse_str(repository_id).map_err(|_| {
AppError::InvalidInput(
"el campo `repository_id` del cuerpo no tiene la forma de un UUID".to_owned(),
)
})
}
async fn repositorio_del_explorador(
estado: &AppState,
consulta: ConsultaDelCable,
) -> Result<Repositorio, AppError> {
let Some(crudo) = consulta.repository_id.as_deref() else {
return Err(AppError::InvalidInput(
"el parámetro `repository_id` de la consulta es obligatorio: el explorador mira un \
repositorio cada vez"
.to_owned(),
));
};
let id = Uuid::parse_str(crudo).map_err(|_| {
AppError::InvalidInput(
"el parámetro `repository_id` de la consulta no tiene la forma de un UUID".to_owned(),
)
})?;
estado.db.repositorios.leer(&id).await?.ok_or_else(|| {
AppError::NotFound("no existe ningún repositorio con ese identificador".to_owned())
})
}
fn acceso_del_repositorio(fila: &Repositorio) -> Acceso {
Acceso {
endpoint: fila.endpoint.clone(),
bucket: fila.bucket.clone(),
region: fila.region.clone(),
access_key_id: fila.access_key_id.clone(),
secret_access_key: fila.secret_access_key.clone(),
passphrase: fila.passphrase.clone(),
}
}
#[derive(Debug, Serialize)]
struct ListadoDelCable {
snapshots: Vec<SnapshotDelCable>,
truncated: bool,
}
impl ListadoDelCable {
fn desde(
instantaneas: &[ResumenDeSnapshot],
objetivos: &HashMap<String, ObjetivoDeRestauracion>,
truncated: bool,
) -> Result<Self, AppError> {
let snapshots = instantaneas
.iter()
.map(|resumen| SnapshotDelCable::desde(resumen, objetivos.get(&resumen.id)))
.collect::<Result<_, _>>()?;
Ok(Self {
snapshots,
truncated,
})
}
}
#[derive(Debug, Serialize)]
struct SnapshotDelCable {
id: String,
time: String,
paths: Vec<String>,
label: Option<String>,
files: u64,
bytes: u64,
target_id: Option<String>,
target_name: Option<String>,
target_type: Option<String>,
}
impl SnapshotDelCable {
fn desde(
resumen: &ResumenDeSnapshot,
objetivo: Option<&ObjetivoDeRestauracion>,
) -> Result<Self, AppError> {
Ok(Self {
id: resumen.id.clone(),
time: fecha(resumen.fecha)?,
paths: resumen.rutas.clone(),
label: resumen.etiqueta.clone(),
files: resumen.ficheros,
bytes: resumen.bytes,
target_id: objetivo.map(|objetivo| objetivo.target_id.to_string()),
target_name: objetivo.map(|objetivo| objetivo.target_name.clone()),
target_type: objetivo.map(|objetivo| texto_del_tipo(objetivo.target_type).to_owned()),
})
}
}
fn fecha(fecha: OffsetDateTime) -> Result<String, AppError> {
fecha.format(&Rfc3339).map_err(|error| {
AppError::Other(format!(
"la fecha de la instantánea no se pudo formatear: {error}"
))
})
}
#[cfg(test)]
mod tests {
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{Duration, Instant};
use axum::Router;
use axum::http::StatusCode;
use time::OffsetDateTime;
use time::format_description::well_known::Rfc3339;
use uuid::Uuid;
use crate::api::api_router;
use crate::db::entities::{AltaDeEjecucion, AltaDeTarea, AltaDeTarget, Estado, TipoDeTarget};
use crate::engine::EventoDelMotor;
use crate::state::{AppState, EstadoDeBase};
use crate::test_support::{
DirectorioDePrueba, ETIQUETA_AJENA, ETIQUETA_SEMBRADA, PASSPHRASE_DE_PRUEBA,
SEMILLA_CONTENIDO, base_con_clave, body_json, config_con_dsn, directorio_de_prueba,
estado_de_prueba, get, post_json, router_desarrollo, sembrar_snapshot_ajeno,
sembrar_snapshots,
};
const UN_UUID: &str = "00000000-0000-4000-8000-000000000000";
const LA_VENTANA: usize = 200;
fn cuerpo_de_alta(nombre: &str, ruta: &str) -> serde_json::Value {
serde_json::json!({
"name": nombre,
"endpoint": ruta,
"bucket": format!("cubo-{nombre}"),
"region": "eu-west-1",
"access_key_id": "AKIAIOSFODNN7EXAMPLE",
"secret_access_key": "AKIAIOSFODNN7EXAMPLE-secreto-del-explorador",
"passphrase": PASSPHRASE_DE_PRUEBA,
})
}
async fn alta_de_repositorio(router: &Router, nombre: &str, ruta: &str) -> String {
let respuesta = post_json(
router.clone(),
"/api/v1/repositories",
&cuerpo_de_alta(nombre, ruta),
)
.await;
let status = respuesta.status();
let cuerpo = body_json(respuesta).await;
assert_eq!(
status,
StatusCode::CREATED,
"el repositorio del explorador tiene que darse de alta —el listado sale de él y no de la \
base— y el alta contestó {status} con {cuerpo}"
);
cuerpo["data"]["id"]
.as_str()
.expect("el alta del repositorio tiene que devolver su identificador")
.to_owned()
}
fn comprobar_elemento(elemento: &serde_json::Value, donde: &str) {
assert!(
elemento["id"].as_str().is_some_and(|id| !id.is_empty()),
"el elemento {donde} no trae su identificador: {elemento}"
);
let fecha = elemento["time"]
.as_str()
.unwrap_or_else(|| panic!("el elemento {donde} no trae la fecha: {elemento}"));
OffsetDateTime::parse(fecha, &Rfc3339).unwrap_or_else(|error| {
panic!("la fecha del elemento {donde} tiene que ser RFC3339 y llegó `{fecha}`: {error}")
});
let rutas = elemento["paths"]
.as_array()
.unwrap_or_else(|| panic!("el elemento {donde} no trae sus rutas: {elemento}"));
assert!(
!rutas.is_empty() && rutas.iter().all(serde_json::Value::is_string),
"el elemento {donde} tiene que traer al menos una ruta y todas como cadenas: {elemento}"
);
assert!(
elemento["label"]
.as_str()
.is_some_and(|etiqueta| !etiqueta.is_empty()),
"la instantánea del elemento {donde} se sembró con etiqueta y llegó sin ella: {elemento}"
);
assert!(
elemento["files"]
.as_u64()
.is_some_and(|ficheros| ficheros >= 1),
"el resumen del elemento {donde} tiene que contar al menos un fichero: {elemento}"
);
assert!(
elemento["bytes"].as_u64().is_some_and(|bytes| bytes >= 1),
"el resumen del elemento {donde} tiene que contar los bytes que la copia procesó: \
{elemento}"
);
for campo in CAMPOS_DEL_TARGET {
assert!(
elemento.get(campo).is_some(),
"el elemento {donde} no trae el campo `{campo}` del target, y el contrato lo publica \
en **todos** los elementos, aunque sea a `null`: {elemento}"
);
}
}
const CAMPOS_DEL_TARGET: [&str; 3] = ["target_id", "target_name", "target_type"];
fn elemento_con_id<'a>(
listado: &'a [serde_json::Value],
instantanea: &str,
cuerpo: &serde_json::Value,
) -> &'a serde_json::Value {
listado
.iter()
.find(|elemento| elemento["id"].as_str() == Some(instantanea))
.unwrap_or_else(|| {
panic!("la instantánea `{instantanea}` no está en el listado: {cuerpo}")
})
}
#[tokio::test]
async fn sin_el_parametro_el_listado_contesta_400_nombrandolo() {
let respuesta = get(router_desarrollo().await, "/api/v1/snapshots").await;
let status = respuesta.status();
assert_ne!(
status,
StatusCode::NOT_FOUND,
"una petición a la que le falta el parámetro no es una ruta ausente"
);
assert_eq!(
status,
StatusCode::BAD_REQUEST,
"sin `repository_id` el listado no puede saber qué repositorio mirar y tiene que contestar \
400; contestó {status}"
);
let cuerpo = body_json(respuesta).await;
assert_eq!(cuerpo["status"], 400, "la envoltura del 400: {cuerpo}");
assert!(
cuerpo["message"]
.as_str()
.is_some_and(|motivo| motivo.contains("repository_id")),
"el motivo del 400 tiene que nombrar el parámetro que falta, y llegó {cuerpo}"
);
}
#[tokio::test]
async fn un_repository_id_que_no_es_un_uuid_contesta_400() {
let respuesta = get(
router_desarrollo().await,
"/api/v1/snapshots?repository_id=no-es-un-uuid",
)
.await;
let status = respuesta.status();
assert_ne!(
status,
StatusCode::NOT_FOUND,
"un identificador mal formado no es un recurso ausente"
);
assert_eq!(
status,
StatusCode::BAD_REQUEST,
"un `repository_id` que no es un UUID tiene que contestar 400; contestó {status}"
);
let cuerpo = body_json(respuesta).await;
assert_eq!(cuerpo["status"], 400, "la envoltura del 400: {cuerpo}");
assert!(
cuerpo["message"]
.as_str()
.is_some_and(|motivo| !motivo.is_empty()),
"el 400 tiene que traer un motivo: {cuerpo}"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn un_repositorio_que_no_existe_contesta_404() {
let router = router_desarrollo().await;
let destino = directorio_de_prueba();
let ruta = destino.ruta().to_string_lossy().into_owned();
sembrar_snapshots(
destino.ruta(),
PASSPHRASE_DE_PRUEBA,
&["2026-01-01T10:00:00Z"],
);
let existe = alta_de_repositorio(&router, "explorador-1-29c", &ruta).await;
let respuesta = get(
router.clone(),
&format!("/api/v1/snapshots?repository_id={existe}"),
)
.await;
let status = respuesta.status();
let cuerpo = body_json(respuesta).await;
assert_eq!(
status,
StatusCode::OK,
"el listado de un repositorio que existe tiene que contestar 200, y contestó {status} con \
{cuerpo}"
);
let respuesta = get(
router,
&format!("/api/v1/snapshots?repository_id={UN_UUID}"),
)
.await;
let status = respuesta.status();
let cuerpo = body_json(respuesta).await;
assert_ne!(
status,
StatusCode::INTERNAL_SERVER_ERROR,
"un repositorio que no existe no es un fallo del servidor: contestó {status} con {cuerpo}"
);
assert_eq!(
status,
StatusCode::NOT_FOUND,
"un repositorio que no existe es un recurso ausente y no una petición mal formada; contestó \
{status} con {cuerpo}"
);
assert_eq!(cuerpo["status"], 404, "la envoltura del 404: {cuerpo}");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cada_elemento_trae_sus_seis_datos_y_el_listado_sale_del_repositorio() {
let router = router_desarrollo().await;
let destino = directorio_de_prueba();
let ruta = destino.ruta().to_string_lossy().into_owned();
let sembradas = sembrar_snapshots(
destino.ruta(),
PASSPHRASE_DE_PRUEBA,
&["2026-01-01T10:00:00Z", "2026-01-02T10:00:00Z"],
);
let (ajena, _nodo_ajeno) = sembrar_snapshot_ajeno(destino.ruta(), PASSPHRASE_DE_PRUEBA);
let repositorio = alta_de_repositorio(&router, "explorador-1-29d", &ruta).await;
let respuesta = get(
router,
&format!("/api/v1/snapshots?repository_id={repositorio}"),
)
.await;
let status = respuesta.status();
let cuerpo = body_json(respuesta).await;
assert_eq!(
status,
StatusCode::OK,
"el listado de un repositorio sembrado tiene que contestar 200, y contestó {status} con \
{cuerpo}"
);
let listado = cuerpo["data"]["snapshots"].as_array().unwrap_or_else(|| {
panic!("el listado tiene que traer sus elementos en `data.snapshots`: {cuerpo}")
});
assert_eq!(
listado.len(),
sembradas.identificadores.len() + 1,
"el repositorio tiene las dos instantáneas sembradas y la ajena, y el listado trajo {}: \
{cuerpo}",
listado.len()
);
for (indice, elemento) in listado.iter().enumerate() {
comprobar_elemento(elemento, &format!("en la posición {indice}"));
}
for (indice, elemento) in listado.iter().enumerate() {
for campo in CAMPOS_DEL_TARGET {
assert!(
elemento[campo].is_null(),
"sin ejecuciones que anoten la instantánea, el campo `{campo}` del elemento en \
la posición {indice} tiene que llegar a `null` y llegó {elemento}"
);
}
}
let fechas: Vec<OffsetDateTime> = listado
.iter()
.map(|elemento| {
let fecha = elemento["time"]
.as_str()
.unwrap_or_else(|| panic!("un elemento llegó sin fecha: {elemento}"));
OffsetDateTime::parse(fecha, &Rfc3339).unwrap_or_else(|error| {
panic!("la fecha `{fecha}` del listado no es RFC3339: {error}")
})
})
.collect();
assert!(
fechas.windows(2).all(|par| par[0] > par[1]),
"el explorador promete el orden de la más reciente a la más antigua, y llegó {fechas:?}: \
un listado desordenado enseña las copias al revés"
);
assert_eq!(
cuerpo["data"]["truncated"], false,
"un listado por debajo de la ventana no puede venir recortado: {cuerpo}"
);
let identificadores: Vec<&str> = listado
.iter()
.filter_map(|elemento| elemento["id"].as_str())
.collect();
let etiquetas: Vec<&str> = listado
.iter()
.filter_map(|elemento| elemento["label"].as_str())
.collect();
for sembrada in &sembradas.identificadores {
assert!(
identificadores.contains(&sembrada.as_str()),
"la instantánea sembrada `{sembrada}` no está en el listado: {cuerpo}"
);
}
assert!(
identificadores.contains(&ajena.as_str()),
"la instantánea ajena `{ajena}` no está en el listado, y el listado tiene que salir del \
repositorio y no del histórico de ejecuciones: {cuerpo}"
);
assert!(
etiquetas.contains(&ETIQUETA_SEMBRADA) && etiquetas.contains(&ETIQUETA_AJENA),
"las instantáneas del repositorio tienen que llegar con su etiqueta —`{ETIQUETA_SEMBRADA}` \
y `{ETIQUETA_AJENA}`— y llegaron {etiquetas:?}"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn el_listado_publica_el_target_de_cada_instantanea_y_null_si_no_lo_tiene() {
let (estado, _base) = estado_con_base().await;
let router = api_router(estado.clone());
let destino = directorio_de_prueba();
let ruta = destino.ruta().to_string_lossy().into_owned();
let siembra = sembrar_snapshots(
destino.ruta(),
PASSPHRASE_DE_PRUEBA,
&["2026-01-01T10:00:00Z", "2026-01-02T10:00:00Z"],
);
let repositorio = alta_de_repositorio(&router, "explorador-1-29f", &ruta).await;
let repositorio_id = Uuid::parse_str(&repositorio)
.expect("el alta del repositorio devuelve un identificador con forma de UUID");
let con_vinculo = siembra.identificadores[0].clone();
let sin_vinculo = siembra.identificadores[1].clone();
let (target, _ejecucion) = sembrar_cadena(
&estado,
repositorio_id,
&con_vinculo,
TipoDeTarget::Directory,
Some("/srv/explorador-1-29f".to_owned()),
)
.await;
let respuesta = get(
router,
&format!("/api/v1/snapshots?repository_id={repositorio}"),
)
.await;
let status = respuesta.status();
let cuerpo = body_json(respuesta).await;
assert_eq!(
status,
StatusCode::OK,
"el listado de un repositorio con una instantánea con vínculo y otra sin él tiene que \
contestar 200, y contestó {status} con {cuerpo}"
);
let listado = cuerpo["data"]["snapshots"]
.as_array()
.unwrap_or_else(|| panic!("el listado tiene que traer sus elementos: {cuerpo}"));
assert_eq!(
listado.len(),
siembra.identificadores.len(),
"una instantánea sin vínculo sigue siendo una copia del repositorio y no puede \
desaparecer del listado: {cuerpo}"
);
let del_vinculo = elemento_con_id(listado, &con_vinculo, &cuerpo);
assert_eq!(
del_vinculo["target_id"].as_str(),
Some(target.to_string().as_str()),
"la instantánea con vínculo tiene que publicar el identificador de su target: {cuerpo}"
);
assert_eq!(
del_vinculo["target_name"].as_str(),
Some("target-de-la-restauracion"),
"el listado tiene que publicar el nombre del target, para que el panel lo anuncie sin \
otra consulta: {cuerpo}"
);
assert_eq!(
del_vinculo["target_type"].as_str(),
Some("directory"),
"el tipo del target viaja con el mismo texto del cable que publica `/api/v1/targets`: \
{cuerpo}"
);
let huerfana = elemento_con_id(listado, &sin_vinculo, &cuerpo);
for campo in CAMPOS_DEL_TARGET {
assert!(
huerfana[campo].is_null(),
"una instantánea sin ejecución que la anote no tiene target y su campo `{campo}` \
tiene que llegar a `null`: {cuerpo}"
);
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn el_listado_esta_acotado_a_la_ventana_e_indica_que_hay_mas() {
let router = router_desarrollo().await;
let destino = directorio_de_prueba();
let ruta = destino.ruta().to_string_lossy().into_owned();
let fechas: Vec<String> = (0..=LA_VENTANA)
.map(|indice| format!("2026-01-{:02}T{:02}:00:00Z", 1 + indice / 24, indice % 24))
.collect();
let referencias: Vec<&str> = fechas.iter().map(String::as_str).collect();
sembrar_snapshots(destino.ruta(), PASSPHRASE_DE_PRUEBA, &referencias);
let repositorio = alta_de_repositorio(&router, "explorador-1-29e", &ruta).await;
let respuesta = get(
router,
&format!("/api/v1/snapshots?repository_id={repositorio}"),
)
.await;
let status = respuesta.status();
let cuerpo = body_json(respuesta).await;
assert_eq!(
status,
StatusCode::OK,
"el listado tiene que contestar 200 aunque tenga más instantáneas que la ventana; contestó \
{status} con {cuerpo}"
);
let listado = cuerpo["data"]["snapshots"]
.as_array()
.unwrap_or_else(|| panic!("el listado tiene que traer sus elementos: {cuerpo}"));
assert!(
listado.len() <= LA_VENTANA,
"el listado devolvió {} elementos y no puede superar la ventana de {LA_VENTANA}: {cuerpo}",
listado.len()
);
assert_eq!(
listado.len(),
LA_VENTANA,
"el repositorio tiene {} instantáneas y la ventana de {LA_VENTANA} tiene que llenarse: \
{cuerpo}",
fechas.len()
);
assert_eq!(
cuerpo["data"]["truncated"], true,
"con más instantáneas que la ventana, la respuesta tiene que decir que hay más: {cuerpo}"
);
}
const PLAZO_DE_LA_RESTAURACION: Duration = Duration::from_secs(30);
async fn estado_con_base() -> (AppState, DirectorioDePrueba) {
let (directorio, pool, clave) = base_con_clave().await;
let estado = estado_de_prueba(
config_con_dsn(&directorio.dsn()),
EstadoDeBase::nuevo(pool, Arc::clone(&clave)),
);
(estado, directorio)
}
async fn sembrar_cadena(
estado: &AppState,
repositorio: Uuid,
instantanea: &str,
tipo: TipoDeTarget,
host_path: Option<String>,
) -> (Uuid, Uuid) {
let target = estado
.db
.targets
.alta(AltaDeTarget {
name: "target-de-la-restauracion".to_owned(),
target_type: tipo,
container_name: None,
db_user: None,
db_password: None,
db_name: None,
host_path,
db_path: None,
})
.await
.expect("el target de la restauración se da de alta");
let tarea = estado
.db
.tareas
.alta(AltaDeTarea {
target_id: target.id,
repository_id: repositorio,
cron_expression: "0 3 * * *".to_owned(),
keep_daily: 7,
keep_weekly: 4,
keep_monthly: 6,
enabled: true,
})
.await
.expect("la tarea de la restauración se da de alta");
let ejecucion = estado
.db
.ejecuciones
.alta(AltaDeEjecucion {
job_id: tarea.id,
status: Estado::Success,
snapshot_id: Some(instantanea.to_owned()),
bytes_processed: 1024,
bytes_uploaded: 512,
error_message: None,
started_at: OffsetDateTime::now_utc(),
finished_at: Some(OffsetDateTime::now_utc()),
})
.await
.expect("la ejecución de la restauración se da de alta");
(target.id, ejecucion.id)
}
fn no_hay_restauraciones_pendientes(
receptor: &mut tokio::sync::broadcast::Receiver<EventoDelMotor>,
) -> bool {
loop {
match receptor.try_recv() {
Ok(EventoDelMotor::Restauracion(_)) => return false,
Ok(EventoDelMotor::Ejecucion(_)) => continue,
Err(
tokio::sync::broadcast::error::TryRecvError::Empty
| tokio::sync::broadcast::error::TryRecvError::Closed,
) => return true,
Err(tokio::sync::broadcast::error::TryRecvError::Lagged(_)) => continue,
}
}
}
fn recoge_contenido(raiz: &Path, contenido: &[u8]) -> Option<PathBuf> {
for entrada in std::fs::read_dir(raiz).ok()?.flatten() {
let ruta = entrada.path();
if ruta.is_dir() {
if let Some(encontrado) = recoge_contenido(&ruta, contenido) {
return Some(encontrado);
}
} else if std::fs::read(&ruta).is_ok_and(|leido| leido == contenido) {
return Some(ruta);
}
}
None
}
async fn espera_el_contenido(
raiz: &Path,
contenido: &[u8],
plazo: Duration,
) -> Option<PathBuf> {
let limite = Instant::now() + plazo;
loop {
if let Some(encontrado) = recoge_contenido(raiz, contenido) {
return Some(encontrado);
}
if Instant::now() >= limite {
return None;
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn restaurar_una_instantanea_de_un_target_de_directorio_devuelve_su_ejecucion_y_su_target()
{
let (estado, _base) = estado_con_base().await;
let router = api_router(estado.clone());
let destino = directorio_de_prueba();
let ruta = destino.ruta().to_string_lossy().into_owned();
let siembra = sembrar_snapshots(
destino.ruta(),
PASSPHRASE_DE_PRUEBA,
&["2026-01-01T10:00:00Z"],
);
let repositorio = alta_de_repositorio(&router, "restaurador-1-39", &ruta).await;
let repositorio_id = Uuid::parse_str(&repositorio)
.expect("el alta del repositorio devuelve un identificador con forma de UUID");
let host = directorio_de_prueba();
let instantanea = siembra.identificadores[0].clone();
let (target, _ejecucion) = sembrar_cadena(
&estado,
repositorio_id,
&instantanea,
TipoDeTarget::Directory,
Some(host.ruta().to_string_lossy().into_owned()),
)
.await;
let respuesta = post_json(
router,
&format!("/api/v1/snapshots/{instantanea}/restore"),
&serde_json::json!({ "repository_id": repositorio }),
)
.await;
let status = respuesta.status();
let cuerpo = body_json(respuesta).await;
assert_eq!(
status,
StatusCode::OK,
"restaurar una instantánea de un target de directorio tiene que contestar 200, y \
contestó {status} con {cuerpo}"
);
assert!(
cuerpo["data"]["execution_id"]
.as_str()
.is_some_and(|id| !id.is_empty()),
"la respuesta tiene que traer el identificador de la ejecución de restauración: {cuerpo}"
);
assert_eq!(
cuerpo["data"]["snapshot_id"].as_str(),
Some(instantanea.as_str()),
"la respuesta tiene que identificar la instantánea que se restauró: {cuerpo}"
);
let target_publicado = target.to_string();
assert_eq!(
cuerpo["data"]["target_id"].as_str(),
Some(target_publicado.as_str()),
"la respuesta tiene que identificar el target sobre el que se restauró: {cuerpo}"
);
let encontrado =
espera_el_contenido(host.ruta(), SEMILLA_CONTENIDO, PLAZO_DE_LA_RESTAURACION).await;
assert!(
encontrado.is_some(),
"el fichero sembrado no reapareció bajo `{}` en {PLAZO_DE_LA_RESTAURACION:?}: una \
restauración que contesta 200 y no devuelve los datos no ha restaurado nada",
host.ruta().display()
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn una_instantanea_sin_ejecucion_contesta_409() {
let router = router_desarrollo().await;
let destino = directorio_de_prueba();
let ruta = destino.ruta().to_string_lossy().into_owned();
let siembra = sembrar_snapshots(
destino.ruta(),
PASSPHRASE_DE_PRUEBA,
&["2026-01-01T10:00:00Z"],
);
let repositorio = alta_de_repositorio(&router, "restaurador-1-40", &ruta).await;
let instantanea = siembra.identificadores[0].clone();
let respuesta = post_json(
router,
&format!("/api/v1/snapshots/{instantanea}/restore"),
&serde_json::json!({ "repository_id": repositorio }),
)
.await;
let status = respuesta.status();
let cuerpo = body_json(respuesta).await;
assert_ne!(
status,
StatusCode::NOT_FOUND,
"la instantánea existe en el repositorio, así que lo que falla no es un recurso \
ausente: contestó {status} con {cuerpo}"
);
assert_eq!(
status,
StatusCode::CONFLICT,
"una instantánea sin ejecución ni target no se puede restaurar y la API tiene que \
contestar 409; contestó {status} con {cuerpo}"
);
assert!(
cuerpo["message"]
.as_str()
.is_some_and(|motivo| !motivo.is_empty()),
"el 409 tiene que traer un motivo que explique que falta el vínculo: {cuerpo}"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn una_instantanea_de_un_target_de_base_de_datos_contesta_422() {
let (estado, _base) = estado_con_base().await;
let router = api_router(estado.clone());
let destino = directorio_de_prueba();
let ruta = destino.ruta().to_string_lossy().into_owned();
let siembra = sembrar_snapshots(
destino.ruta(),
PASSPHRASE_DE_PRUEBA,
&["2026-01-01T10:00:00Z"],
);
let repositorio = alta_de_repositorio(&router, "restaurador-1-41", &ruta).await;
let repositorio_id = Uuid::parse_str(&repositorio)
.expect("el alta del repositorio devuelve un identificador con forma de UUID");
let instantanea = siembra.identificadores[0].clone();
sembrar_cadena(
&estado,
repositorio_id,
&instantanea,
TipoDeTarget::Mariadb,
None,
)
.await;
let respuesta = post_json(
router,
&format!("/api/v1/snapshots/{instantanea}/restore"),
&serde_json::json!({ "repository_id": repositorio }),
)
.await;
let status = respuesta.status();
let cuerpo = body_json(respuesta).await;
assert_ne!(
status,
StatusCode::NOT_FOUND,
"la ruta de la restauración tiene que estar montada: contestó {status} con {cuerpo}"
);
assert_ne!(
status,
StatusCode::CONFLICT,
"el vínculo con el target existe, así que no es un conflicto con el estado del \
sistema: contestó {status} con {cuerpo}"
);
assert_eq!(
status,
StatusCode::UNPROCESSABLE_ENTITY,
"restaurar un target de base de datos no está soportado en esta versión y la API tiene \
que contestar 422 explicándolo; contestó {status} con {cuerpo}"
);
assert!(
cuerpo["message"]
.as_str()
.is_some_and(|motivo| !motivo.is_empty()),
"el 422 tiene que explicar la limitación: {cuerpo}"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn una_instantanea_que_no_esta_en_el_repositorio_contesta_404() {
let (estado, _base) = estado_con_base().await;
let router = api_router(estado.clone());
let destino = directorio_de_prueba();
let ruta = destino.ruta().to_string_lossy().into_owned();
sembrar_snapshots(
destino.ruta(),
PASSPHRASE_DE_PRUEBA,
&["2026-01-01T10:00:00Z"],
);
let repositorio = alta_de_repositorio(&router, "restaurador-1-43", &ruta).await;
let repositorio_id = Uuid::parse_str(&repositorio)
.expect("el alta del repositorio devuelve un identificador con forma de UUID");
let inventada = "f".repeat(64);
let host = directorio_de_prueba();
sembrar_cadena(
&estado,
repositorio_id,
&inventada,
TipoDeTarget::Directory,
Some(host.ruta().to_string_lossy().into_owned()),
)
.await;
let mut receptor = estado
.motor
.hub()
.suscribir()
.expect("el hub no está cerrado en esta prueba");
let respuesta = post_json(
router,
&format!("/api/v1/snapshots/{inventada}/restore"),
&serde_json::json!({ "repository_id": repositorio }),
)
.await;
let status = respuesta.status();
let cuerpo = body_json(respuesta).await;
assert_ne!(
status,
StatusCode::CONFLICT,
"el vínculo con el target existe, así que lo que falta no es el vínculo: contestó \
{status} con {cuerpo}"
);
assert_eq!(
status,
StatusCode::NOT_FOUND,
"una instantánea que el repositorio no tiene no se puede restaurar y la API tiene que \
contestar 404 —y no un 200 optimista—: contestó {status} con {cuerpo}"
);
let motivo = cuerpo["message"].as_str().unwrap_or_default();
assert!(
motivo.contains("instantánea"),
"el 404 de la instantánea ausente tiene que decir que lo que falta es la copia: {cuerpo}"
);
assert_ne!(
motivo, "no existe ningún repositorio con ese identificador",
"el 404 de la copia ausente y el del repositorio ausente tienen que ser distinguibles: \
el operador necesita saber cuál de los dos le falta"
);
assert!(
no_hay_restauraciones_pendientes(&mut receptor),
"un 404 no puede publicar ningún `restore_progress`: el panel daría por lanzada una \
restauración que no existe"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn dos_restauraciones_simultaneas_sobre_el_mismo_destino_no_caben() {
let (estado, _base) = estado_con_base().await;
let router = api_router(estado.clone());
let destino = directorio_de_prueba();
let ruta = destino.ruta().to_string_lossy().into_owned();
let siembra = sembrar_snapshots(
destino.ruta(),
PASSPHRASE_DE_PRUEBA,
&["2026-01-01T10:00:00Z"],
);
let instantanea = siembra.identificadores[0].clone();
let repositorio = alta_de_repositorio(&router, "restaurador-1-44", &ruta).await;
let repositorio_id = Uuid::parse_str(&repositorio)
.expect("el alta del repositorio devuelve un identificador con forma de UUID");
let host = directorio_de_prueba();
sembrar_cadena(
&estado,
repositorio_id,
&instantanea,
TipoDeTarget::Directory,
Some(host.ruta().to_string_lossy().into_owned()),
)
.await;
let uri = format!("/api/v1/snapshots/{instantanea}/restore");
let cuerpo = serde_json::json!({ "repository_id": repositorio });
let (primera, segunda) = tokio::join!(
post_json(router.clone(), &uri, &cuerpo),
post_json(router.clone(), &uri, &cuerpo),
);
let mut codigos = [primera.status(), segunda.status()];
codigos.sort_by_key(|codigo| codigo.as_u16());
assert_eq!(
codigos,
[StatusCode::OK, StatusCode::CONFLICT],
"dos restauraciones simultáneas sobre el mismo destino tienen que dejar una lanzada \
(200) y la otra rechazada (409): contestaron {codigos:?}"
);
}
}