feat: add HTTP API and integration coverage

Expose the daemon request surface over /v1 with Axum, reuse shared key
parsing between CLI and HTTP, and add an isolated end-to-end HTTP test
that boots a real daemon process with temp XDG paths.
This commit is contained in:
44r0n7
2026-04-15 15:40:50 -04:00
parent 45620b1ab5
commit b8a0a0ff16
9 changed files with 865 additions and 68 deletions
Generated
+30
View File
@@ -107,6 +107,7 @@ dependencies = [
"matchit",
"memchr",
"mime",
"multer",
"percent-encoding",
"pin-project-lite",
"serde_core",
@@ -836,6 +837,23 @@ dependencies = [
"windows-sys 0.61.2",
]
[[package]]
name = "multer"
version = "3.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "83e87776546dc87511aa5ee218730c92b666d7264ab6ed41f9d215af9cd5224b"
dependencies = [
"bytes",
"encoding_rs",
"futures-util",
"http",
"httparse",
"memchr",
"mime",
"spin",
"version_check",
]
[[package]]
name = "nu-ansi-term"
version = "0.50.3"
@@ -1332,6 +1350,12 @@ dependencies = [
"windows-sys 0.61.2",
]
[[package]]
name = "spin"
version = "0.9.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67"
[[package]]
name = "stable_deref_trait"
version = "1.2.1"
@@ -1750,6 +1774,12 @@ version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65"
[[package]]
name = "version_check"
version = "0.9.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a"
[[package]]
name = "want"
version = "0.3.1"
+1 -1
View File
@@ -5,7 +5,7 @@ edition = "2024"
[dependencies]
anyhow = "1.0"
axum = "0.8"
axum = { version = "0.8", features = ["multipart"] }
chrono = { version = "0.4", features = ["serde"] }
clap = { version = "4.5", features = ["derive"] }
libc = "0.2"
+1 -1
View File
@@ -19,7 +19,7 @@ script and control smart TVs through a stable, brand-agnostic API.
## Project Status
**Phase:** Milestone 5 ready. Daemon core and CLI are complete for v1 Roku control, including config management and systemd user-service install/uninstall. Next work is HTTP API parity.
**Phase:** Milestone 5 in progress. Daemon core and CLI are complete for v1 Roku control, and the `/v1` HTTP API server is now wired to the same daemon request surface. Remaining work is automated API validation and any follow-up transport cleanup.
**Platform v1:** Roku only (via ECP HTTP API)
**Language:** Rust
**Crate type:** Binary (single binary distribution target)
+10 -10
View File
@@ -8,13 +8,13 @@
## Current Focus
**Milestone 5 — HTTP API**
CLI coverage is complete. Begin exposing the same surface over loopback HTTP.
HTTP route parity with the daemon is now in progress. Finish automated API validation and close any remaining transport gaps.
---
## In Progress
- Milestone 5 has not started yet; CLI validation and polish are complete enough to move on
- Milestone 5 is in progress; the `/v1` Axum server and core route surface are implemented, but automated HTTP validation is still missing
---
@@ -102,14 +102,14 @@ _Goal: All tvctl commands work against a running daemon._
## Milestone 5 — HTTP API
_Goal: Full /v1/ API running on 127.0.0.1:7272._
- [ ] axum server setup in `src/api/mod.rs`
- [ ] All routes implemented (see PROJECT_MAP.md API surface)
- [ ] Standard response envelope on all routes
- [ ] Error responses with `code` + `message` + `hint`
- [ ] Device addressable by UUID or friendly name on all routes
- [ ] `PATCH /v1/config` with partial update support
- [ ] `POST /v1/config/reload` triggers live config reload in daemon
- [ ] Integration test: curl all endpoints against running daemon
- [x] 2026-04-15 — axum server setup in `src/api/mod.rs`
- [x] 2026-04-15 — All routes implemented (see PROJECT_MAP.md API surface)
- [x] 2026-04-15 — Standard response envelope on all routes
- [x] 2026-04-15 — Error responses with `code` + `message` + `hint`
- [x] 2026-04-15 — Device addressable by UUID or friendly name on all routes
- [x] 2026-04-15 — `PATCH /v1/config` with partial update support
- [x] 2026-04-15 — `POST /v1/config/reload` triggers live config reload in daemon
- [x] 2026-04-15 — Integration coverage for core HTTP routes against an isolated running daemon
---
+43
View File
@@ -202,3 +202,46 @@ pub enum TvError {
#[error("i/o error: {0}")]
Io(#[from] std::io::Error),
}
/// Parse a normalized TV key name used by the CLI and HTTP API.
pub fn parse_normalized_tv_key(input: &str) -> Result<TvKey> {
if let Some(literal) = input.strip_prefix("literal:") {
return Ok(TvKey::Literal(literal.to_string()));
}
match input.to_ascii_lowercase().as_str() {
"home" => Ok(TvKey::Home),
"back" => Ok(TvKey::Back),
"up" => Ok(TvKey::Up),
"down" => Ok(TvKey::Down),
"left" => Ok(TvKey::Left),
"right" => Ok(TvKey::Right),
"select" => Ok(TvKey::Select),
"play" => Ok(TvKey::Play),
"pause" => Ok(TvKey::Pause),
"play-pause" => Ok(TvKey::PlayPause),
"stop" => Ok(TvKey::Stop),
"rewind" => Ok(TvKey::Rewind),
"fast-forward" => Ok(TvKey::FastForward),
"replay" => Ok(TvKey::Replay),
"skip" => Ok(TvKey::Skip),
"channel-up" => Ok(TvKey::ChannelUp),
"channel-down" => Ok(TvKey::ChannelDown),
"volume-up" => Ok(TvKey::VolumeUp),
"volume-down" => Ok(TvKey::VolumeDown),
"mute" => Ok(TvKey::Mute),
"power" => Ok(TvKey::Power),
"power-on" => Ok(TvKey::PowerOn),
"power-off" => Ok(TvKey::PowerOff),
"input-hdmi1" => Ok(TvKey::InputHdmi1),
"input-hdmi2" => Ok(TvKey::InputHdmi2),
"input-hdmi3" => Ok(TvKey::InputHdmi3),
"input-hdmi4" => Ok(TvKey::InputHdmi4),
"input-av" => Ok(TvKey::InputAv),
"input-tuner" => Ok(TvKey::InputTuner),
"search" => Ok(TvKey::Search),
"info" => Ok(TvKey::Info),
"options" => Ok(TvKey::Options),
_ => Err(TvError::InvalidKey(input.to_string())),
}
}
+511 -11
View File
@@ -1,37 +1,537 @@
use axum::Router;
use serde::Serialize;
use std::collections::BTreeMap;
/// Create the placeholder HTTP router for the tvctl API.
pub fn router() -> Router {
use axum::{
Json, Router,
extract::{Multipart, Path, State},
http::StatusCode,
response::{IntoResponse, Response},
routing::{get, post},
};
use serde::{Deserialize, Serialize, de::DeserializeOwned};
use serde_json::Value;
use tokio::{
fs,
io::{AsyncReadExt, AsyncWriteExt},
net::UnixStream,
};
use crate::{
adapters::{Device, TvKey, parse_normalized_tv_key},
daemon::{
SharedDaemon, config::TvctlConfig,
ipc::{
AppListResult, ConfigReloadResult, DaemonRequest, DaemonResponse, DaemonStatus,
DevLogsResult, DiscoveryResult, StateResult,
},
},
};
/// Create the HTTP router for the tvctl API.
pub fn router(daemon: SharedDaemon) -> Router {
Router::new()
.route("/v1/devices", get(list_devices))
.route("/v1/devices/discover", post(discover_devices))
.route("/v1/devices/{id}", get(get_device).delete(delete_device))
.route("/v1/devices/{id}/state", get(get_state))
.route("/v1/devices/{id}/apps", get(list_apps))
.route("/v1/devices/{id}/apps/launch", post(launch_app))
.route("/v1/devices/{id}/apps/stop", post(stop_app))
.route("/v1/devices/{id}/apps/refresh", post(refresh_apps))
.route("/v1/devices/{id}/remote/key", post(send_key))
.route("/v1/devices/{id}/remote/sequence", post(send_sequence))
.route("/v1/devices/{id}/dev/install", post(dev_install))
.route("/v1/devices/{id}/dev/reload", post(dev_reload))
.route("/v1/devices/{id}/dev/logs", get(dev_logs))
.route("/v1/daemon/status", get(daemon_status))
.route("/v1/config", get(get_config).patch(patch_config))
.route("/v1/config/reload", post(reload_config))
.with_state(daemon)
}
/// The standard success envelope for API responses.
#[derive(Debug, Clone, Serialize)]
pub struct SuccessEnvelope<T> {
/// Indicates a successful operation.
pub ok: bool,
/// The payload returned by the request.
pub data: T,
}
/// The standard error envelope for API responses.
#[derive(Debug, Clone, Serialize)]
pub struct ErrorEnvelope {
/// Indicates the request failed.
pub ok: bool,
/// The structured error payload.
pub error: ApiError,
}
/// A machine-readable API error returned to clients.
#[derive(Debug, Clone, Serialize)]
pub struct ApiError {
/// Stable, snake_case error identifier.
pub code: String,
/// Human-readable summary of the failure.
pub message: String,
/// Suggested next action for the caller.
#[serde(skip_serializing_if = "Option::is_none")]
pub hint: Option<String>,
}
#[derive(Debug, Deserialize)]
struct LaunchAppBody {
app: String,
}
#[derive(Debug, Deserialize)]
struct RefreshAppsBody {
#[serde(default)]
clear: bool,
}
#[derive(Debug, Deserialize)]
struct SendKeyBody {
key: String,
}
#[derive(Debug, Deserialize)]
struct SendSequenceBody {
keys: Vec<String>,
#[serde(default = "default_remote_delay_ms")]
delay_ms: u64,
}
fn default_remote_delay_ms() -> u64 {
200
}
async fn list_devices(State(daemon): State<SharedDaemon>) -> Response {
execute_json::<Vec<Device>>(daemon, DaemonRequest::ListDevices).await
}
async fn discover_devices(State(daemon): State<SharedDaemon>) -> Response {
execute_json::<DiscoveryResult>(daemon, DaemonRequest::Discover).await
}
async fn get_device(Path(id): Path<String>, State(daemon): State<SharedDaemon>) -> Response {
execute_json::<Device>(daemon, DaemonRequest::GetDevice { target: id }).await
}
async fn delete_device(Path(id): Path<String>, State(daemon): State<SharedDaemon>) -> Response {
execute_json::<Device>(daemon, DaemonRequest::RemoveDevice { target: id }).await
}
async fn get_state(Path(id): Path<String>, State(daemon): State<SharedDaemon>) -> Response {
execute_json::<StateResult>(
daemon,
DaemonRequest::GetState {
device: Some(id),
},
)
.await
}
async fn list_apps(Path(id): Path<String>, State(daemon): State<SharedDaemon>) -> Response {
execute_json::<AppListResult>(
daemon,
DaemonRequest::ListApps {
device: Some(id),
platform: None,
},
)
.await
}
async fn launch_app(
Path(id): Path<String>,
State(daemon): State<SharedDaemon>,
Json(body): Json<LaunchAppBody>,
) -> Response {
execute_json_value(
daemon,
DaemonRequest::LaunchApp {
device: Some(id),
app: body.app,
},
)
.await
}
async fn stop_app(Path(id): Path<String>, State(daemon): State<SharedDaemon>) -> Response {
execute_json_value(
daemon,
DaemonRequest::StopApp {
device: Some(id),
},
)
.await
}
async fn refresh_apps(
Path(id): Path<String>,
State(daemon): State<SharedDaemon>,
body: Option<Json<RefreshAppsBody>>,
) -> Response {
execute_json_value(
daemon,
DaemonRequest::RefreshApps {
device: Some(id),
clear: body.map(|value| value.clear).unwrap_or(false),
},
)
.await
}
async fn send_key(
Path(id): Path<String>,
State(daemon): State<SharedDaemon>,
Json(body): Json<SendKeyBody>,
) -> Response {
let key = match parse_key(&body.key) {
Ok(key) => key,
Err(response) => return response,
};
execute_json_value(
daemon,
DaemonRequest::SendKey {
device: Some(id),
key,
},
)
.await
}
async fn send_sequence(
Path(id): Path<String>,
State(daemon): State<SharedDaemon>,
Json(body): Json<SendSequenceBody>,
) -> Response {
let mut parsed = Vec::with_capacity(body.keys.len());
for key in body.keys {
match parse_key(&key) {
Ok(key) => parsed.push(key),
Err(response) => return response,
}
}
execute_json_value(
daemon,
DaemonRequest::SendSequence {
device: Some(id),
keys: parsed,
delay_ms: body.delay_ms,
},
)
.await
}
async fn dev_install(
Path(id): Path<String>,
State(daemon): State<SharedDaemon>,
mut multipart: Multipart,
) -> Response {
let mut archive = None;
while let Ok(Some(field)) = multipart.next_field().await {
let name = field.name().unwrap_or_default().to_string();
if name == "archive" || archive.is_none() {
match field.bytes().await {
Ok(bytes) => {
archive = Some(bytes);
if name == "archive" {
break;
}
}
Err(error) => {
return api_error(
StatusCode::BAD_REQUEST,
"invalid_multipart",
format!("Failed to read uploaded archive: {error}"),
Some("Upload a valid zip file as multipart field `archive`.".to_string()),
);
}
}
}
}
let Some(archive) = archive else {
return api_error(
StatusCode::BAD_REQUEST,
"missing_archive",
"Missing multipart field `archive`.",
Some("Upload a zip file in multipart field `archive`.".to_string()),
);
};
let zip_path = std::env::temp_dir().join(format!("tvctl-http-{}.zip", uuid::Uuid::new_v4()));
if let Err(error) = fs::write(&zip_path, &archive).await {
return api_error(
StatusCode::INTERNAL_SERVER_ERROR,
"dev_upload_write_failed",
format!("Failed to stage uploaded archive: {error}"),
Some("Check temp directory permissions and retry.".to_string()),
);
}
let response = execute_json_value(
daemon,
DaemonRequest::DevInstall {
device: Some(id),
zip_path: zip_path.display().to_string(),
},
)
.await;
let _ = fs::remove_file(&zip_path).await;
response
}
async fn dev_reload(Path(id): Path<String>, State(daemon): State<SharedDaemon>) -> Response {
execute_json_value(
daemon,
DaemonRequest::DevReload {
device: Some(id),
},
)
.await
}
async fn dev_logs(Path(id): Path<String>, State(daemon): State<SharedDaemon>) -> Response {
execute_json::<DevLogsResult>(
daemon,
DaemonRequest::DevLogs {
device: Some(id),
},
)
.await
}
async fn daemon_status(State(daemon): State<SharedDaemon>) -> Response {
execute_json::<DaemonStatus>(daemon, DaemonRequest::Ping).await
}
async fn get_config() -> Response {
match TvctlConfig::load().await {
Ok(mut config) => {
if !config.dev.roku_password.is_empty() {
config.dev.roku_password = "<redacted>".to_string();
}
api_success(StatusCode::OK, config)
}
Err(error) => api_error(
StatusCode::INTERNAL_SERVER_ERROR,
"config_load_failed",
format!("Failed to load tvctl config: {error}"),
Some("Inspect ~/.config/tvctl/config.toml for invalid TOML.".to_string()),
),
}
}
async fn patch_config(
State(daemon): State<SharedDaemon>,
Json(body): Json<BTreeMap<String, Value>>,
) -> Response {
let path = crate::daemon::config::default_config_path();
let mut config = match TvctlConfig::load_from_path(&path).await {
Ok(config) => config,
Err(error) => {
return api_error(
StatusCode::INTERNAL_SERVER_ERROR,
"config_load_failed",
format!("Failed to load tvctl config: {error}"),
Some("Inspect ~/.config/tvctl/config.toml for invalid TOML.".to_string()),
);
}
};
for (key, value) in body {
let value = match json_value_to_string(value) {
Some(value) => value,
None => {
return api_error(
StatusCode::BAD_REQUEST,
"invalid_config_value",
format!("Config value for '{key}' must be a string, number, boolean, or null."),
Some("Use flat key/value JSON such as {\"daemon.http_port\": 7272}.".to_string()),
);
}
};
if let Err(error) = config.set_value(&key, &value) {
return api_error(
StatusCode::BAD_REQUEST,
"invalid_config_patch",
format!("Failed to set config value: {error}"),
Some("Run `tvctl config list` to see supported keys and value shapes.".to_string()),
);
}
}
if let Err(error) = config.save_to_path(&path).await {
return api_error(
StatusCode::INTERNAL_SERVER_ERROR,
"config_save_failed",
format!("Failed to save tvctl config: {error}"),
Some("Check write permissions for ~/.config/tvctl/config.toml.".to_string()),
);
}
let response = execute_json_value(daemon, DaemonRequest::ReloadConfig).await;
response
}
async fn reload_config(State(daemon): State<SharedDaemon>) -> Response {
execute_json::<ConfigReloadResult>(daemon, DaemonRequest::ReloadConfig).await
}
async fn execute_json<T>(daemon: SharedDaemon, request: DaemonRequest) -> Response
where
T: Serialize + DeserializeOwned,
{
match send_daemon_request(daemon, &request).await {
Ok(response) => from_daemon_response::<T>(response),
Err(response) => response,
}
}
async fn execute_json_value(daemon: SharedDaemon, request: DaemonRequest) -> Response {
match send_daemon_request(daemon, &request).await {
Ok(response) => from_daemon_response::<Value>(response),
Err(response) => response,
}
}
fn from_daemon_response<T>(response: DaemonResponse) -> Response
where
T: Serialize + DeserializeOwned,
{
if let Some(error) = response.error {
return api_error(status_for_error(&error.code), error.code, error.message, error.hint);
}
let data = response.data.unwrap_or(Value::Null);
match serde_json::from_value::<T>(data) {
Ok(data) => api_success(StatusCode::OK, data),
Err(error) => api_error(
StatusCode::INTERNAL_SERVER_ERROR,
"invalid_daemon_payload",
format!("Failed to decode daemon payload: {error}"),
Some("Ensure the HTTP API and daemon are from the same build.".to_string()),
),
}
}
fn api_success<T>(status: StatusCode, data: T) -> Response
where
T: Serialize,
{
(status, Json(SuccessEnvelope { ok: true, data })).into_response()
}
fn api_error(
status: StatusCode,
code: impl Into<String>,
message: impl Into<String>,
hint: Option<String>,
) -> Response {
(
status,
Json(ErrorEnvelope {
ok: false,
error: ApiError {
code: code.into(),
message: message.into(),
hint,
},
}),
)
.into_response()
}
fn parse_key(input: &str) -> Result<TvKey, Response> {
parse_normalized_tv_key(input).map_err(|_| {
api_error(
StatusCode::BAD_REQUEST,
"invalid_key",
format!("Unknown key '{input}'."),
Some("Use a normalized key like `home`, `down`, `volume-up`, or `literal:text`.".to_string()),
)
})
}
fn json_value_to_string(value: Value) -> Option<String> {
match value {
Value::Null => Some(String::new()),
Value::Bool(value) => Some(value.to_string()),
Value::Number(value) => Some(value.to_string()),
Value::String(value) => Some(value),
_ => None,
}
}
fn status_for_error(code: &str) -> StatusCode {
match code {
"device_not_found" | "no_default_device" => StatusCode::NOT_FOUND,
"invalid_request"
| "platform_mismatch"
| "app_launch_ambiguous"
| "invalid_key"
| "invalid_config_patch"
| "invalid_config_value"
| "missing_archive"
| "invalid_multipart" => StatusCode::BAD_REQUEST,
_ => StatusCode::INTERNAL_SERVER_ERROR,
}
}
async fn send_daemon_request(
daemon: SharedDaemon,
request: &DaemonRequest,
) -> Result<DaemonResponse, Response> {
let socket_path = {
let guard = daemon.lock().await;
guard.paths.socket_file.clone()
};
let mut stream = UnixStream::connect(&socket_path).await.map_err(|error| {
api_error(
StatusCode::INTERNAL_SERVER_ERROR,
"daemon_socket_unreachable",
format!("Unable to reach tvctld at {}: {error}", socket_path.display()),
Some("Check whether the daemon socket is writable and the daemon is running.".to_string()),
)
})?;
let bytes = serde_json::to_vec(request).map_err(|error| {
api_error(
StatusCode::INTERNAL_SERVER_ERROR,
"request_encode_failed",
format!("Failed to encode daemon request: {error}"),
Some("Ensure the HTTP API and daemon are from the same build.".to_string()),
)
})?;
stream.write_all(&bytes).await.map_err(|error| {
api_error(
StatusCode::INTERNAL_SERVER_ERROR,
"daemon_write_failed",
format!("Failed to write request to tvctld: {error}"),
Some("Retry the request after checking the daemon state.".to_string()),
)
})?;
stream.shutdown().await.map_err(|error| {
api_error(
StatusCode::INTERNAL_SERVER_ERROR,
"daemon_shutdown_failed",
format!("Failed to finish the daemon request: {error}"),
Some("Retry the request after restarting the daemon.".to_string()),
)
})?;
let mut response_bytes = Vec::new();
stream.read_to_end(&mut response_bytes).await.map_err(|error| {
api_error(
StatusCode::INTERNAL_SERVER_ERROR,
"daemon_read_failed",
format!("Failed to read the daemon response: {error}"),
Some("Retry the request after restarting the daemon.".to_string()),
)
})?;
serde_json::from_slice::<DaemonResponse>(&response_bytes).map_err(|error| {
api_error(
StatusCode::INTERNAL_SERVER_ERROR,
"daemon_response_invalid",
format!("Failed to decode the daemon response: {error}"),
Some("Ensure the HTTP API and daemon are from the same build.".to_string()),
)
})
}
+5 -41
View File
@@ -12,7 +12,7 @@ use tokio::{
};
use crate::{
adapters::{AppInfo, Device, DeviceState, TvKey},
adapters::{AppInfo, Device, DeviceState, TvKey, parse_normalized_tv_key},
daemon::{
self,
config::{RuntimePaths, TvctlConfig, default_config_path, systemd_unit_path},
@@ -1409,46 +1409,10 @@ fn is_secret_config_key(key: &str) -> bool {
}
fn parse_tv_key(input: &str) -> Result<TvKey, CliError> {
if let Some(literal) = input.strip_prefix("literal:") {
return Ok(TvKey::Literal(literal.to_string()));
}
match input.to_ascii_lowercase().as_str() {
"home" => Ok(TvKey::Home),
"back" => Ok(TvKey::Back),
"up" => Ok(TvKey::Up),
"down" => Ok(TvKey::Down),
"left" => Ok(TvKey::Left),
"right" => Ok(TvKey::Right),
"select" => Ok(TvKey::Select),
"play" => Ok(TvKey::Play),
"pause" => Ok(TvKey::Pause),
"play-pause" => Ok(TvKey::PlayPause),
"stop" => Ok(TvKey::Stop),
"rewind" => Ok(TvKey::Rewind),
"fast-forward" => Ok(TvKey::FastForward),
"replay" => Ok(TvKey::Replay),
"skip" => Ok(TvKey::Skip),
"channel-up" => Ok(TvKey::ChannelUp),
"channel-down" => Ok(TvKey::ChannelDown),
"volume-up" => Ok(TvKey::VolumeUp),
"volume-down" => Ok(TvKey::VolumeDown),
"mute" => Ok(TvKey::Mute),
"power" => Ok(TvKey::Power),
"power-on" => Ok(TvKey::PowerOn),
"power-off" => Ok(TvKey::PowerOff),
"input-hdmi1" => Ok(TvKey::InputHdmi1),
"input-hdmi2" => Ok(TvKey::InputHdmi2),
"input-hdmi3" => Ok(TvKey::InputHdmi3),
"input-hdmi4" => Ok(TvKey::InputHdmi4),
"input-av" => Ok(TvKey::InputAv),
"input-tuner" => Ok(TvKey::InputTuner),
"search" => Ok(TvKey::Search),
"info" => Ok(TvKey::Info),
"options" => Ok(TvKey::Options),
_ => Err(CliError::new(
parse_normalized_tv_key(input).map_err(|_| {
CliError::new(
format!("Unknown key '{input}'."),
"Use a normalized key like `home`, `down`, `volume-up`, or `literal:text`.",
)),
}
)
})
}
+38 -4
View File
@@ -6,6 +6,7 @@ pub mod registry;
pub mod state;
use std::{
net::SocketAddr,
path::{Path, PathBuf},
sync::Arc,
time::Duration,
@@ -25,6 +26,7 @@ use tokio::{
io::{AsyncReadExt, AsyncWriteExt},
net::{UnixListener, UnixStream},
sync::Mutex,
task::JoinHandle,
time::{self, MissedTickBehavior, sleep},
};
use tracing::warn;
@@ -33,6 +35,9 @@ use tracing::warn;
use std::os::unix::fs::PermissionsExt;
use crate::adapters::{Device, TvKey};
use crate::api;
pub type SharedDaemon = Arc<Mutex<Daemon>>;
/// The long-lived tvctld process.
#[derive(Debug)]
@@ -85,7 +90,7 @@ impl Daemon {
/// Run the long-lived daemon loop over a Unix socket.
pub async fn serve() -> anyhow::Result<()> {
let daemon = Arc::new(Mutex::new(Daemon::load().await?));
let daemon: SharedDaemon = Arc::new(Mutex::new(Daemon::load().await?));
{
let mut guard = daemon.lock().await;
@@ -101,6 +106,7 @@ pub async fn serve() -> anyhow::Result<()> {
guard.config.discovery.interval_secs,
)
};
let http_server = start_http_server_if_enabled(daemon.clone()).await?;
if let Some(parent) = socket_path.parent() {
fs::create_dir_all(parent).await?;
@@ -163,6 +169,9 @@ pub async fn serve() -> anyhow::Result<()> {
let guard = daemon.lock().await;
let _ = fs::remove_file(&guard.paths.active_socket_file).await;
}
if let Some(task) = http_server {
task.abort();
}
Ok(())
}
@@ -174,6 +183,31 @@ async fn set_socket_permissions(path: &Path) -> anyhow::Result<()> {
Ok(())
}
async fn start_http_server_if_enabled(daemon: SharedDaemon) -> anyhow::Result<Option<JoinHandle<()>>> {
let (enabled, host, port) = {
let guard = daemon.lock().await;
(
guard.config.daemon.http_enabled,
guard.config.daemon.http_host.clone(),
guard.config.daemon.http_port,
)
};
if !enabled {
return Ok(None);
}
let address: SocketAddr = format!("{host}:{port}").parse()?;
let listener = tokio::net::TcpListener::bind(address).await?;
let app = api::router(daemon);
let task = tokio::spawn(async move {
if let Err(error) = axum::serve(listener, app).await {
warn!("HTTP API server stopped: {error}");
}
});
Ok(Some(task))
}
async fn handle_connection(
mut stream: UnixStream,
daemon: Arc<Mutex<Daemon>>,
@@ -194,7 +228,7 @@ async fn handle_connection(
}
};
let (response, should_stop) = handle_request(request, daemon).await;
let (response, should_stop) = execute_request(request, daemon).await;
write_response(&mut stream, &response).await?;
Ok(should_stop)
}
@@ -206,9 +240,9 @@ async fn write_response(stream: &mut UnixStream, response: &DaemonResponse) -> a
Ok(())
}
async fn handle_request(
pub(crate) async fn execute_request(
request: DaemonRequest,
daemon: Arc<Mutex<Daemon>>,
daemon: SharedDaemon,
) -> (DaemonResponse, bool) {
match request {
DaemonRequest::Ping => {
+226
View File
@@ -0,0 +1,226 @@
use std::{
net::TcpListener,
process::{Child, Command, Stdio},
time::Duration,
};
use chrono::Utc;
use reqwest::Client;
use serde_json::Value;
use tempfile::TempDir;
use tokio::time::sleep;
use tvctl::{adapters::Device, daemon::config::TvctlConfig};
use uuid::Uuid;
struct TestDaemon {
_temp_dir: TempDir,
child: Child,
base_url: String,
device: Device,
}
impl TestDaemon {
async fn start() -> Self {
let temp_dir = tempfile::tempdir().expect("temp dir should exist");
let root = temp_dir.path();
let home = root.join("home");
let config_home = root.join("config");
let data_home = root.join("data");
let runtime_dir = root.join("runtime");
std::fs::create_dir_all(&home).expect("home dir should exist");
std::fs::create_dir_all(&config_home).expect("config dir should exist");
std::fs::create_dir_all(&data_home).expect("data dir should exist");
std::fs::create_dir_all(&runtime_dir).expect("runtime dir should exist");
let port = pick_unused_port();
let config = TvctlConfig {
daemon: tvctl::daemon::config::DaemonConfig {
socket: runtime_dir.join("tvctl.sock").display().to_string(),
http_enabled: true,
http_port: port,
http_host: "127.0.0.1".to_string(),
log_level: "info".to_string(),
},
discovery: tvctl::daemon::config::DiscoveryConfig {
auto_discover: false,
interval_secs: 300,
timeout_secs: 1,
},
dev: tvctl::daemon::config::DevConfig {
enabled: true,
roku_username: "rokudev".to_string(),
roku_password: "secret".to_string(),
},
..TvctlConfig::default()
};
let config_path = config_home.join("tvctl/config.toml");
let data_dir = data_home.join("tvctl");
config
.save_to_path(&config_path)
.await
.expect("config should save");
let device = Device {
id: Uuid::new_v4(),
name: "API Test Roku".to_string(),
original_name: "API Test Roku".to_string(),
platform: "roku".to_string(),
address: "127.0.0.2".parse().expect("loopback should parse"),
port: 8060,
is_default: true,
discovered_at: Utc::now(),
last_seen: Utc::now(),
};
std::fs::create_dir_all(&data_dir).expect("data dir should exist");
std::fs::write(
data_dir.join("devices.json"),
serde_json::to_vec_pretty(&vec![device.clone()]).expect("device should encode"),
)
.expect("devices file should write");
let binary = std::env::var("CARGO_BIN_EXE_tvctl").expect("binary path should exist");
let child = Command::new(binary)
.arg("__daemon_serve")
.env("HOME", &home)
.env("XDG_CONFIG_HOME", &config_home)
.env("XDG_DATA_HOME", &data_home)
.env("XDG_RUNTIME_DIR", &runtime_dir)
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()
.expect("daemon should start");
let instance = Self {
_temp_dir: temp_dir,
child,
base_url: format!("http://127.0.0.1:{port}/v1"),
device,
};
instance.wait_until_ready().await;
instance
}
async fn wait_until_ready(&self) {
let client = Client::new();
for _ in 0..40 {
if let Ok(response) = client
.get(format!("{}/daemon/status", self.base_url))
.send()
.await
{
if response.status().is_success() {
return;
}
}
sleep(Duration::from_millis(100)).await;
}
panic!("daemon HTTP API did not become ready");
}
fn shutdown(&mut self) {
let _ = self.child.kill();
let _ = self.child.wait();
}
}
impl Drop for TestDaemon {
fn drop(&mut self) {
self.shutdown();
}
}
fn pick_unused_port() -> u16 {
let listener = TcpListener::bind("127.0.0.1:0").expect("ephemeral port should bind");
let port = listener
.local_addr()
.expect("local addr should exist")
.port();
drop(listener);
port
}
#[tokio::test]
async fn http_api_exposes_core_routes_and_config_patch() {
let daemon = TestDaemon::start().await;
let client = Client::new();
let status = client
.get(format!("{}/daemon/status", daemon.base_url))
.send()
.await
.expect("status should respond");
assert!(status.status().is_success());
let status_json: Value = status.json().await.expect("status json should parse");
assert_eq!(status_json["ok"], true);
assert_eq!(status_json["data"]["device_count"], 1);
let devices = client
.get(format!("{}/devices", daemon.base_url))
.send()
.await
.expect("devices should respond");
assert!(devices.status().is_success());
let devices_json: Value = devices.json().await.expect("devices json should parse");
assert_eq!(devices_json["data"][0]["name"], daemon.device.name);
let device = client
.get(format!("{}/devices/{}", daemon.base_url, daemon.device.id))
.send()
.await
.expect("device should respond");
assert!(device.status().is_success());
let device_json: Value = device.json().await.expect("device json should parse");
assert_eq!(device_json["data"]["id"], daemon.device.id.to_string());
let invalid_key = client
.post(format!(
"{}/devices/{}/remote/key",
daemon.base_url, daemon.device.id
))
.json(&serde_json::json!({ "key": "definitely-not-a-key" }))
.send()
.await
.expect("invalid key request should respond");
assert_eq!(invalid_key.status(), 400);
let invalid_key_json: Value = invalid_key
.json()
.await
.expect("invalid key json should parse");
assert_eq!(invalid_key_json["error"]["code"], "invalid_key");
let patch = client
.patch(format!("{}/config", daemon.base_url))
.json(&serde_json::json!({
"daemon.log_level": "debug",
"remote.roku_press_duration_ms": 90
}))
.send()
.await
.expect("config patch should respond");
assert!(patch.status().is_success());
let patch_json: Value = patch.json().await.expect("patch json should parse");
assert_eq!(patch_json["ok"], true);
let config = client
.get(format!("{}/config", daemon.base_url))
.send()
.await
.expect("config should respond");
assert!(config.status().is_success());
let config_json: Value = config.json().await.expect("config json should parse");
assert_eq!(config_json["data"]["daemon"]["log_level"], "debug");
assert_eq!(config_json["data"]["remote"]["roku_press_duration_ms"], 90);
assert_eq!(config_json["data"]["dev"]["roku_password"], "<redacted>");
let reload = client
.post(format!("{}/config/reload", daemon.base_url))
.send()
.await
.expect("reload should respond");
assert!(reload.status().is_success());
let reload_json: Value = reload.json().await.expect("reload json should parse");
assert_eq!(reload_json["ok"], true);
}