refactor(cli): hide ApiClient behind Client facade

Wrap the remaining CLI server API calls in server_client::Client,
remove the api/connect_api_client escape hatches, and migrate
model/install/tests to the new facade.
This commit is contained in:
Bryan Helmkamp 2026-04-20 09:29:02 -04:00
parent aa8cdd6986
commit 63cd749dcb
No known key found for this signature in database
15 changed files with 276 additions and 237 deletions

View file

@ -312,7 +312,7 @@ pub(crate) async fn run_doctor(
}
};
if let Err(err) = server.api().get_health().send().await {
if let Err(err) = server.get_health().await {
report.sections.push(CheckSection {
title: "Server".to_string(),
checks: vec![CheckResult {
@ -349,9 +349,8 @@ pub(crate) async fn run_doctor(
}],
});
match server.api().run_diagnostics().send().await {
Ok(response) => {
let diagnostics = response.into_inner();
match server.run_diagnostics().await {
Ok(diagnostics) => {
report.sections[0]
.checks
.push(check_version_parity(&diagnostics.version));

View file

@ -1094,19 +1094,17 @@ async fn setup_github_app(
}
async fn persist_vault_secrets_via_server(
client: &fabro_api::ApiClient,
client: &server_client::Client,
secrets: &[CreateSecretRequest],
) -> Result<()> {
for secret in secrets {
client
.create_secret()
.body(CreateSecretRequest {
.create_secret(CreateSecretRequest {
name: secret.name.clone(),
value: secret.value.clone(),
type_: secret.type_,
description: secret.description.clone(),
})
.send()
.await?;
}
@ -1117,14 +1115,14 @@ async fn persist_vault_secrets_with(
storage_dir: &Path,
secrets: &[CreateSecretRequest],
server_was_running: bool,
connect_api_client: impl for<'a> Fn(&'a Path) -> BoxFuture<'a, Result<fabro_api::ApiClient>>,
connect_server: impl for<'a> Fn(&'a Path) -> BoxFuture<'a, Result<server_client::Client>>,
stop_server: impl for<'a> Fn(&'a Path, Duration) -> BoxFuture<'a, bool>,
) -> Result<()> {
if secrets.is_empty() {
return Ok(());
}
let client = match connect_api_client(storage_dir).await {
let client = match connect_server(storage_dir).await {
Ok(client) => client,
Err(err) => {
if !server_was_running {
@ -1173,7 +1171,7 @@ async fn persist_install_outputs(
vault_secrets,
settings_write,
server_was_running,
|path| Box::pin(server_client::connect_api_client(path)),
|path| Box::pin(server_client::connect_server(path)),
|path, timeout| {
Box::pin(async move { stop::stop_server(path, timeout).await.unwrap_or(false) })
},
@ -1303,7 +1301,7 @@ async fn persist_install_outputs_with_settings(
vault_secrets: &[CreateSecretRequest],
settings_write: Option<PendingSettingsWrite<'_>>,
server_was_running: bool,
connect_api_client: impl for<'a> Fn(&'a Path) -> BoxFuture<'a, Result<fabro_api::ApiClient>>,
connect_server: impl for<'a> Fn(&'a Path) -> BoxFuture<'a, Result<server_client::Client>>,
stop_server: impl for<'a> Fn(&'a Path, Duration) -> BoxFuture<'a, bool>,
) -> Result<()> {
persist_server_env_secrets(storage_dir, server_env_secrets)?;
@ -1317,7 +1315,7 @@ async fn persist_install_outputs_with_settings(
storage_dir,
vault_secrets,
server_was_running,
connect_api_client,
connect_server,
stop_server,
)
.await;
@ -2486,10 +2484,7 @@ client_id = "client-id"
&vault_secrets,
false,
|_| {
let client = fabro_api::ApiClient::new_with_client(
&server.base_url(),
fabro_test::test_http_client(),
);
let client = server_client::Client::new_no_proxy(&server.base_url()).unwrap();
Box::pin(async move { Ok(client) })
},
{
@ -2548,10 +2543,7 @@ client_id = "client-id"
&vault_secrets,
true,
|_| {
let client = fabro_api::ApiClient::new_with_client(
&server.base_url(),
fabro_test::test_http_client(),
);
let client = server_client::Client::new_no_proxy(&server.base_url()).unwrap();
Box::pin(async move { Ok(client) })
},
{

View file

@ -1,14 +1,13 @@
use anyhow::{Context, Result, bail};
use cli_table::format::{Border, Justify, Separator};
use cli_table::{Cell, CellStruct, Color, Style, Table};
use fabro_api::{self, types as api_types};
use fabro_api::types as api_types;
use fabro_model::{Catalog, Model, Provider};
use fabro_types::settings::CliSettings;
use fabro_types::settings::cli::{CliLayer, OutputFormat};
use fabro_util::printer::Printer;
use fabro_util::terminal::Styles;
use serde::Serialize;
use serde::de::DeserializeOwned;
use crate::args::{ModelListArgs, ModelTestArgs, ModelsCommand};
use crate::command_context::CommandContext;
@ -55,12 +54,7 @@ pub(crate) async fn execute(
let ctx = CommandContext::for_target(target_args, printer, cli.clone(), cli_layer)?;
let server = ctx.server().await?;
run_models(
command,
server.api(),
cli.output.format == OutputFormat::Json,
)
.await
run_models(command, &server, cli.output.format == OutputFormat::Json).await
}
fn format_context_window(tokens: i64) -> String {
@ -198,64 +192,13 @@ fn model_test_row_from_status(model: &Model, status: &str, result_color: Color)
}
}
fn convert_type<TInput, TOutput>(value: TInput) -> Result<TOutput>
where
TInput: serde::Serialize,
TOutput: DeserializeOwned,
{
serde_json::from_value(serde_json::to_value(value)?).map_err(Into::into)
}
async fn fetch_models_from_server(
client: &fabro_api::ApiClient,
provider: Option<&str>,
query: Option<&str>,
) -> Result<Vec<Model>> {
let mut offset = 0u64;
let mut models = Vec::new();
loop {
let mut request = client.list_models().page_limit(100u64).page_offset(offset);
if let Some(provider) = provider {
request = request.provider(provider.to_string());
}
if let Some(query) = query {
request = request.query(query.to_string());
}
let response = request.send().await.map_err(server_client::map_api_error)?;
let parsed = response.into_inner();
let count = parsed.data.len() as u64;
models.extend(convert_type::<_, Vec<Model>>(parsed.data)?);
if !parsed.meta.has_more {
break;
}
offset += count;
}
Ok(models)
}
async fn test_model_via_server(
client: &fabro_api::ApiClient,
model_id: &str,
mode: Option<api_types::ModelTestMode>,
) -> Result<api_types::ModelTestResult> {
let mut request = client.test_model().id(model_id.to_string());
if let Some(mode) = mode {
request = request.mode(mode);
}
let response = request.send().await.map_err(server_client::map_api_error)?;
Ok(response.into_inner())
}
#[allow(
clippy::print_stdout,
clippy::print_stderr,
reason = "Progress goes to stderr while tables or JSON results go to stdout."
)]
async fn test_models_via_server(
client: &fabro_api::ApiClient,
client: &server_client::Client,
provider: Option<&str>,
model: Option<&str>,
deep: bool,
@ -277,7 +220,7 @@ async fn test_models_via_server(
if !json_output {
eprint!("Testing {model_id}...");
}
let result = test_model_via_server(client, model_id, request_mode).await;
let result = client.test_model(model_id, request_mode).await;
if !json_output {
eprintln!(" done");
}
@ -327,7 +270,7 @@ async fn test_models_via_server(
rows.push(row);
json_rows.push(model_test_row_from_status(&info, &status, result_color));
} else {
let models_to_test = fetch_models_from_server(client, provider, None).await?;
let models_to_test = client.list_models(provider, None).await?;
if models_to_test.is_empty() {
bail!("No models found");
}
@ -336,7 +279,7 @@ async fn test_models_via_server(
if !json_output {
eprint!("Testing {}...", info.id);
}
let result = test_model_via_server(client, &info.id, request_mode).await;
let result = client.test_model(&info.id, request_mode).await;
if !json_output {
eprintln!(" done");
}
@ -432,7 +375,7 @@ async fn test_models_via_server(
)]
async fn run_models(
command: ModelsCommand,
client: &fabro_api::ApiClient,
client: &server_client::Client,
json_output: bool,
) -> Result<()> {
let styles = Styles::detect_stdout();
@ -441,8 +384,9 @@ async fn run_models(
ModelsCommand::List(ModelListArgs {
provider, query, ..
}) => {
let models =
fetch_models_from_server(client, provider.as_deref(), query.as_deref()).await?;
let models = client
.list_models(provider.as_deref(), query.as_deref())
.await?;
if json_output {
println!("{}", serde_json::to_string_pretty(&models)?);
@ -483,8 +427,8 @@ mod tests {
use super::*;
fn test_api_client(api_url: &str) -> fabro_api::ApiClient {
fabro_api::ApiClient::new_with_client(api_url, fabro_test::test_http_client())
fn test_api_client(api_url: &str) -> server_client::Client {
server_client::Client::new_no_proxy(api_url).unwrap()
}
fn test_model_json(id: &str, provider: Provider) -> serde_json::Value {
@ -591,9 +535,7 @@ mod tests {
.await;
let client = test_api_client(&server.url(""));
let response = test_model_via_server(&client, "test-model", None)
.await
.unwrap();
let response = client.test_model("test-model", None).await.unwrap();
assert_eq!(response.status, api_types::ModelTestResultStatus::Ok);
assert!(response.error_message.is_none());
@ -621,10 +563,10 @@ mod tests {
.await;
let client = test_api_client(&server.url(""));
let response =
test_model_via_server(&client, "test-model", Some(api_types::ModelTestMode::Deep))
.await
.unwrap();
let response = client
.test_model("test-model", Some(api_types::ModelTestMode::Deep))
.await
.unwrap();
assert_eq!(response.status, api_types::ModelTestResultStatus::Error);
assert_eq!(response.error_message.as_deref(), Some("timeout"));
@ -649,9 +591,7 @@ mod tests {
.await;
let client = test_api_client(&server.url(""));
let response = test_model_via_server(&client, "kimi-k2.5", None)
.await
.unwrap();
let response = client.test_model("kimi-k2.5", None).await.unwrap();
assert_eq!(response.status, api_types::ModelTestResultStatus::Skip);
assert!(response.error_message.is_none());
@ -675,7 +615,7 @@ mod tests {
.await;
let client = test_api_client(&server.url(""));
let result = test_model_via_server(&client, "bad-model", None).await;
let result = client.test_model("bad-model", None).await;
assert!(result.is_err());
assert!(result.unwrap_err().to_string().contains("Model not found"));
}
@ -702,7 +642,7 @@ mod tests {
.await;
let client = test_api_client(&server.url(""));
let models = fetch_models_from_server(&client, None, None).await.unwrap();
let models = client.list_models(None, None).await.unwrap();
mock.assert_async().await;
assert_eq!(models.len(), 1);
@ -733,9 +673,7 @@ mod tests {
.await;
let client = test_api_client(&server.url(""));
let models = fetch_models_from_server(&client, Some("anthropic"), None)
.await
.unwrap();
let models = client.list_models(Some("anthropic"), None).await.unwrap();
assert_eq!(models.len(), 1);
assert_eq!(models[0].id, "model-a");
@ -764,9 +702,7 @@ mod tests {
.await;
let client = test_api_client(&server.url(""));
let models = fetch_models_from_server(&client, None, Some("sonnet"))
.await
.unwrap();
let models = client.list_models(None, Some("sonnet")).await.unwrap();
mock.assert_async().await;
assert_eq!(models.len(), 1);
@ -812,7 +748,7 @@ mod tests {
.await;
let client = test_api_client(&server.url(""));
let models = fetch_models_from_server(&client, None, None).await.unwrap();
let models = client.list_models(None, None).await.unwrap();
first_page.assert_async().await;
second_page.assert_async().await;
@ -835,7 +771,7 @@ mod tests {
.await;
let client = test_api_client(&server.url(""));
let result = fetch_models_from_server(&client, None, None).await;
let result = client.list_models(None, None).await;
assert!(result.is_err());
}
}

View file

@ -36,15 +36,12 @@ pub(super) async fn login_command(
let value = serde_json::to_string(&credential)?;
server
.api()
.create_secret()
.body(types::CreateSecretRequest {
.create_secret(types::CreateSecretRequest {
name: credential_id.clone(),
value,
type_: types::SecretType::Credential,
description: None,
})
.send()
.await?;
fabro_util::printerr!(
printer,

View file

@ -222,15 +222,8 @@ async fn check_github_app_installation(
}
};
let check = match server
.api()
.get_github_repo()
.owner(owner.clone())
.name(repo.clone())
.send()
.await
{
Ok(response) => response.into_inner(),
let check = match server.get_github_repo(&owner, &repo).await {
Ok(response) => response,
Err(err) => {
fabro_util::printerr!(printer, "\n Warning: could not check GitHub access: {err}");
return;
@ -270,16 +263,8 @@ async fn check_github_app_installation(
})
.await;
match server
.api()
.get_github_repo()
.owner(owner.clone())
.name(repo.clone())
.send()
.await
{
match server.get_github_repo(&owner, &repo).await {
Ok(response) => {
let response = response.into_inner();
if response.accessible {
let green = console::Style::new().green();
fabro_util::printerr!(

View file

@ -2,14 +2,13 @@ use anyhow::Result;
use chrono::{DateTime, Utc};
use cli_table::format::{Border, Separator};
use cli_table::{Cell, CellStruct, Style, Table};
use fabro_api::ApiClient;
use fabro_types::settings::CliSettings;
use fabro_types::settings::cli::OutputFormat;
use fabro_util::printer::Printer;
use fabro_util::terminal::Styles;
use crate::args::SecretListArgs;
use crate::server_client;
use crate::server_client::Client;
use crate::shared::print_json_pretty;
fn format_age(dt: DateTime<Utc>, now: DateTime<Utc>) -> String {
@ -24,17 +23,12 @@ fn format_age(dt: DateTime<Utc>, now: DateTime<Utc>) -> String {
}
pub(super) async fn list_command(
client: &ApiClient,
client: &Client,
_args: &SecretListArgs,
cli: &CliSettings,
printer: Printer,
) -> Result<()> {
let response = client
.list_secrets()
.send()
.await
.map_err(server_client::map_api_error)?;
let secrets = response.into_inner().data;
let secrets = client.list_secrets().await?;
if cli.output.format == OutputFormat::Json {
print_json_pretty(&secrets)?;
return Ok(());

View file

@ -19,8 +19,8 @@ pub(crate) async fn dispatch(
let ctx = CommandContext::for_target(&ns.target, printer, cli.clone(), cli_layer)?;
let server = ctx.server().await?;
match ns.command {
SecretCommand::List(args) => list::list_command(server.api(), &args, cli, printer).await,
SecretCommand::Rm(args) => rm::rm_command(server.api(), &args, cli, printer).await,
SecretCommand::Set(args) => set::set_command(server.api(), &args, cli, printer).await,
SecretCommand::List(args) => list::list_command(&server, &args, cli, printer).await,
SecretCommand::Rm(args) => rm::rm_command(&server, &args, cli, printer).await,
SecretCommand::Set(args) => set::set_command(&server, &args, cli, printer).await,
}
}

View file

@ -1,27 +1,19 @@
use anyhow::Result;
use fabro_api::{ApiClient, types};
use fabro_types::settings::CliSettings;
use fabro_types::settings::cli::OutputFormat;
use fabro_util::printer::Printer;
use crate::args::SecretRmArgs;
use crate::server_client;
use crate::server_client::Client;
use crate::shared::print_json_pretty;
pub(super) async fn rm_command(
client: &ApiClient,
client: &Client,
args: &SecretRmArgs,
cli: &CliSettings,
printer: Printer,
) -> Result<()> {
client
.delete_secret_by_name()
.body(types::DeleteSecretRequest {
name: args.key.clone(),
})
.send()
.await
.map_err(server_client::map_api_error)?;
client.delete_secret_by_name(&args.key).await?;
if cli.output.format == OutputFormat::Json {
print_json_pretty(&serde_json::json!({ "key": args.key }))?;
} else {

View file

@ -10,14 +10,14 @@
use std::io::{IsTerminal, Read as _};
use anyhow::{Context as _, Result, bail};
use fabro_api::{ApiClient, types};
use fabro_api::types;
use fabro_types::settings::CliSettings;
use fabro_types::settings::cli::OutputFormat;
use fabro_util::printer::Printer;
use tokio::task::spawn_blocking;
use crate::args::{SecretSetArgs, SecretTypeArg};
use crate::server_client;
use crate::server_client::Client;
use crate::shared::print_json_pretty;
use crate::shared::provider_auth::prompt_password;
@ -58,24 +58,20 @@ async fn resolve_value(args: &SecretSetArgs) -> Result<String> {
}
pub(super) async fn set_command(
client: &ApiClient,
client: &Client,
args: &SecretSetArgs,
cli: &CliSettings,
printer: Printer,
) -> Result<()> {
let value = resolve_value(args).await?;
let meta = client
.create_secret()
.body(types::CreateSecretRequest {
.create_secret(types::CreateSecretRequest {
name: args.key.clone(),
value,
type_: api_secret_type(args.r#type),
description: args.description.clone(),
})
.send()
.await
.map_err(server_client::map_api_error)?
.into_inner();
.await?;
if cli.output.format == OutputFormat::Json {
print_json_pretty(&meta)?;
} else {

View file

@ -9,7 +9,6 @@ use fabro_util::printer::Printer;
use crate::args::DfArgs;
use crate::command_context::CommandContext;
use crate::server_client;
use crate::shared::{format_size, print_json_pretty};
pub(super) async fn df_command(
@ -20,27 +19,13 @@ pub(super) async fn df_command(
) -> Result<()> {
let ctx = CommandContext::for_connection(&args.connection, printer, cli.clone(), cli_layer)?;
let server = ctx.server().await?;
let output = server
.api()
.get_system_disk_usage()
.verbose(args.verbose)
.send()
.await
.map_err(server_client::map_api_error)?
.into_inner();
let output = server.get_system_disk_usage(args.verbose).await?;
let json = cli.output.format == OutputFormat::Json;
let storage_dir = if json {
None
} else {
server
.api()
.get_system_info()
.send()
.await
.map_err(server_client::map_api_error)?
.into_inner()
.storage_dir
server.get_system_info().await?.storage_dir
};
df_from(&output, storage_dir.as_deref(), json)

View file

@ -6,7 +6,7 @@ use futures::StreamExt;
use crate::args::SystemEventsArgs;
use crate::command_context::CommandContext;
use crate::{server_client, sse};
use crate::sse;
pub(super) async fn events_command(
args: &SystemEventsArgs,
@ -16,14 +16,7 @@ pub(super) async fn events_command(
) -> Result<()> {
let ctx = CommandContext::for_connection(&args.connection, printer, cli.clone(), cli_layer)?;
let server = ctx.server().await?;
let mut request = server.api().attach_events();
if !args.run_ids.is_empty() {
request = request.run_id(args.run_ids.join(","));
}
let response = request.send().await.map_err(server_client::map_api_error)?;
let mut stream = response.into_inner();
let mut stream = server.attach_events(&args.run_ids).await?;
let mut pending = Vec::new();
let json = cli.output.format == OutputFormat::Json;

View file

@ -5,7 +5,6 @@ use fabro_util::printer::Printer;
use crate::args::SystemInfoArgs;
use crate::command_context::CommandContext;
use crate::server_client;
use crate::shared::print_json_pretty;
pub(super) async fn info_command(
@ -16,13 +15,7 @@ pub(super) async fn info_command(
) -> Result<()> {
let ctx = CommandContext::for_connection(&args.connection, printer, cli.clone(), cli_layer)?;
let server = ctx.server().await?;
let response = server
.api()
.get_system_info()
.send()
.await
.map_err(server_client::map_api_error)?
.into_inner();
let response = server.get_system_info().await?;
if cli.output.format == OutputFormat::Json {
print_json_pretty(&response)?;

View file

@ -9,7 +9,6 @@ use tracing::{debug, info};
use crate::args::RunsPruneArgs;
use crate::command_context::CommandContext;
use crate::server_client;
use crate::shared::{format_size, print_json_pretty};
pub(super) async fn prune_command(
@ -21,9 +20,7 @@ pub(super) async fn prune_command(
let ctx = CommandContext::for_connection(&args.connection, printer, cli.clone(), cli_layer)?;
let server = ctx.server().await?;
let response = server
.api()
.prune_runs()
.body(types::PruneRunsRequest {
.prune_runs(types::PruneRunsRequest {
before: args.filter.before.clone(),
dry_run: !args.yes,
labels: parse_label_filters(&args.filter.label),
@ -31,10 +28,7 @@ pub(super) async fn prune_command(
orphans: args.filter.orphans,
workflow: args.filter.workflow.clone(),
})
.send()
.await
.map_err(server_client::map_api_error)?
.into_inner();
.await?;
prune_from(&response, cli.output.format == OutputFormat::Json, printer)
}

View file

@ -13,7 +13,6 @@ use serde_json::{Map, Value, json};
use crate::args::VersionArgs;
use crate::command_context::CommandContext;
use crate::server_client;
use crate::shared::print_json_pretty;
use crate::user_config::{self, ServerTarget};
@ -28,26 +27,17 @@ pub(crate) async fn version_command(
let server_target = user_config::resolve_server_target(&args.target, ctx.machine_settings())?;
let server_address = format_server_target(&server_target);
let server_info = match ctx.server().await {
Ok(server) => match server
.api()
.get_system_info()
.send()
.await
.map_err(server_client::map_api_error)
{
Ok(response) => {
let response = response.into_inner();
ServerVersionInfo::Success {
address: server_address,
version: response.version,
git_sha: response.git_sha,
build_date: response.build_date,
profile: response.profile,
os: response.os,
arch: response.arch,
uptime_secs: response.uptime_secs,
}
}
Ok(server) => match server.get_system_info().await {
Ok(response) => ServerVersionInfo::Success {
address: server_address,
version: response.version,
git_sha: response.git_sha,
build_date: response.build_date,
profile: response.profile,
os: response.os,
arch: response.arch,
uptime_secs: response.uptime_secs,
},
Err(err) => ServerVersionInfo::Error {
address: server_address,
error: err.to_string(),

View file

@ -10,6 +10,7 @@ use fabro_api::types;
use fabro_config::Storage;
use fabro_http::header::{AUTHORIZATION, CONTENT_LENGTH, CONTENT_TYPE};
use fabro_http::multipart::{Form, Part};
use fabro_model::Model;
use fabro_server::bind::Bind;
use fabro_store::{EventEnvelope, RunSummary, StageId};
use fabro_types::settings::SettingsLayer;
@ -89,7 +90,6 @@ impl RunAttachEventStream {
pub(crate) use fabro_store::RunProjection;
#[cfg(test)]
pub(crate) async fn connect_server(storage_dir: &Path) -> Result<Client> {
connect_api_client_bundle(storage_dir).await
}
@ -141,12 +141,6 @@ async fn connect_api_client_bundle(storage_dir: &Path) -> Result<Client> {
}
}
pub(crate) async fn connect_api_client(storage_dir: &Path) -> Result<fabro_api::ApiClient> {
connect_api_client_bundle(storage_dir)
.await
.map(|client| client.client)
}
async fn connect_target_api_client_bundle(
target: &user_config::ServerTarget,
runtime: &LocalServerRuntime,
@ -474,10 +468,6 @@ impl Client {
self.clone()
}
pub(crate) fn api(&self) -> &fabro_api::ApiClient {
&self.client
}
#[allow(
dead_code,
reason = "This accessor is kept for tests and pending callers."
@ -551,6 +541,174 @@ impl Client {
.map_err(|err| anyhow!("invalid run ID from server: {err}"))
}
pub(crate) async fn list_secrets(&self) -> Result<Vec<types::SecretMetadata>> {
let response = self
.client
.list_secrets()
.send()
.await
.map_err(map_api_error)?;
Ok(response.into_inner().data)
}
pub(crate) async fn create_secret(
&self,
body: types::CreateSecretRequest,
) -> Result<types::SecretMetadata> {
let response = self
.client
.create_secret()
.body(body)
.send()
.await
.map_err(map_api_error)?;
Ok(response.into_inner())
}
pub(crate) async fn delete_secret_by_name(&self, name: &str) -> Result<()> {
self.client
.delete_secret_by_name()
.body(types::DeleteSecretRequest {
name: name.to_string(),
})
.send()
.await
.map_err(map_api_error)?;
Ok(())
}
pub(crate) async fn list_models(
&self,
provider: Option<&str>,
query: Option<&str>,
) -> Result<Vec<Model>> {
let mut offset = 0u64;
let mut models = Vec::new();
loop {
let mut request = self
.client
.list_models()
.page_limit(100u64)
.page_offset(offset);
if let Some(provider) = provider {
request = request.provider(provider.to_string());
}
if let Some(query) = query {
request = request.query(query.to_string());
}
let response = request.send().await.map_err(map_api_error)?;
let parsed = response.into_inner();
let count = parsed.data.len() as u64;
models.extend(convert_type::<_, Vec<Model>>(parsed.data)?);
if !parsed.meta.has_more {
break;
}
offset += count;
}
Ok(models)
}
pub(crate) async fn test_model(
&self,
id: &str,
mode: Option<types::ModelTestMode>,
) -> Result<types::ModelTestResult> {
let mut request = self.client.test_model().id(id.to_string());
if let Some(mode) = mode {
request = request.mode(mode);
}
let response = request.send().await.map_err(map_api_error)?;
Ok(response.into_inner())
}
pub(crate) async fn attach_events(
&self,
run_ids: &[String],
) -> Result<progenitor_client::ByteStream> {
let mut request = self.client.attach_events();
if !run_ids.is_empty() {
request = request.run_id(run_ids.join(","));
}
let response = request.send().await.map_err(map_api_error)?;
Ok(response.into_inner())
}
pub(crate) async fn get_system_info(&self) -> Result<types::SystemInfoResponse> {
let response = self
.client
.get_system_info()
.send()
.await
.map_err(map_api_error)?;
Ok(response.into_inner())
}
pub(crate) async fn get_system_disk_usage(
&self,
verbose: bool,
) -> Result<types::DiskUsageResponse> {
let response = self
.client
.get_system_disk_usage()
.verbose(verbose)
.send()
.await
.map_err(map_api_error)?;
Ok(response.into_inner())
}
pub(crate) async fn prune_runs(
&self,
body: types::PruneRunsRequest,
) -> Result<types::PruneRunsResponse> {
let response = self
.client
.prune_runs()
.body(body)
.send()
.await
.map_err(map_api_error)?;
Ok(response.into_inner())
}
pub(crate) async fn get_health(&self) -> Result<()> {
self.client
.get_health()
.send()
.await
.map_err(map_api_error)?;
Ok(())
}
pub(crate) async fn run_diagnostics(&self) -> Result<types::DiagnosticsReport> {
let response = self
.client
.run_diagnostics()
.send()
.await
.map_err(map_api_error)?;
Ok(response.into_inner())
}
pub(crate) async fn get_github_repo(
&self,
owner: &str,
name: &str,
) -> Result<types::RepoCheckResponse> {
let response = self
.client
.get_github_repo()
.owner(owner.to_string())
.name(name.to_string())
.send()
.await
.map_err(map_api_error)?;
Ok(response.into_inner())
}
pub(crate) async fn run_preflight(
&self,
manifest: types::RunManifest,
@ -1190,6 +1348,10 @@ fn non_zero_u64_from_usize(value: usize) -> Option<NonZeroU64> {
reason = "server-client tests stage local dev-token fixtures with sync std::fs::write"
)]
mod tests {
use futures::StreamExt as _;
use httpmock::Method::GET;
use httpmock::MockServer;
use super::*;
#[test]
@ -1299,4 +1461,35 @@ mod tests {
assert!(remote_url_targets_local_host("http://0.0.0.0:32276"));
assert!(!remote_url_targets_local_host("https://example.com"));
}
#[tokio::test]
async fn attach_events_returns_raw_sse_bytes() {
let server = MockServer::start_async().await;
let payload = serde_json::json!({
"seq": 1,
"payload": {
"event": "run.started",
"run_id": "run_123"
}
});
let body = format!("data: {payload}\n\n");
let mock = server
.mock_async(|when, then| {
when.method(GET).path("/api/v1/attach");
then.status(200)
.header("Content-Type", "text/event-stream")
.body(body.clone());
})
.await;
let client = Client::new_no_proxy(&server.base_url()).unwrap();
let mut stream = client.attach_events(&[]).await.unwrap();
let mut bytes = Vec::new();
while let Some(chunk) = stream.next().await {
bytes.extend_from_slice(&chunk.unwrap());
}
mock.assert_async().await;
assert_eq!(String::from_utf8(bytes).unwrap(), body);
}
}