GKaszewski/movies-diary
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: Upload extractor, 429 refusal test, stricter contract test

Browse Source
This commit is contained in:
Gabriel Kaszewski
2026-10-25 21:59:31 +01:00
parent c8d9138349
commit 2ac1d4e0cf
10 changed files with 139 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
crates/adapters/http-axum/src/api/import/sessions.rs
View File

@@ -1,4 +1,4 @@
use crate::extractors::{Body, PathId};
use crate::extractors::{Body, PathId, Upload};
use crate::support::str_to_domain_field;
use crate::{errors::ApiError, extractors::AuthenticatedUser, state::AppState};
use api_types::{
@@ -14,10 +14,7 @@ use application::import::{
create_session as create_import_session, execute as execute_import, get_preview_stage,
get_session_state,
};
use axum::{
extract::{Multipart, State},
response::IntoResponse,
};
use axum::{extract::State, response::IntoResponse};
use domain::errors::DomainError;
use domain::models::{
AnnotatedRow, FieldMapping, FileFormat,
@@ -39,7 +36,7 @@ use domain::value_objects::ImportSessionId;
pub async fn api_post_session(
State(state): State<AppState>,
AuthenticatedUser(user_id): AuthenticatedUser,
mut multipart: Multipart,
Upload(mut multipart): Upload,
) -> Result<impl IntoResponse, ApiError> {
let mut file_bytes: Option<Vec<u8>> = None;
let mut format_str = "csv".to_string();

11
crates/adapters/http-axum/src/api/users.rs
View File

@@ -1,15 +1,10 @@
use crate::extractors::Body;
use crate::extractors::{Body, Upload};
use crate::{errors::ApiError, extractors::AuthenticatedUser, state::AppState};
use api_types::{ProfileResponse, UserSummaryDto, UsersResponse};
use application::users::{
get_users, queries::GetUsersQuery, update_profile, update_profile_fields,
};
use axum::{
Json,
extract::{Multipart, State},
http::StatusCode,
response::IntoResponse,
};
use axum::{Json, extract::State, http::StatusCode, response::IntoResponse};
#[utoipa::path(
get, path = "/api/v1/profile",
@@ -66,7 +61,7 @@ pub async fn get_profile(
pub async fn update_profile_handler(
State(state): State<AppState>,
AuthenticatedUser(user_id): AuthenticatedUser,
multipart: Multipart,
Upload(multipart): Upload,
) -> impl IntoResponse {
let data = crate::support::parse_profile_multipart(multipart).await;

6
crates/adapters/http-axum/src/api/webhook.rs
View File

@@ -1,4 +1,4 @@
use crate::extractors::{Body, Params, PathId};
use crate::extractors::{Body, Params, PathId, Upload};
use crate::{errors::ApiError, extractors::AuthenticatedUser, state::AppState};
use api_types::{
ConfirmWatchRequest, ConfirmWatchResponse, DismissWatchRequest, DismissWatchResponse,
@@ -17,7 +17,7 @@ use application::integrations::{
};
use axum::{
Json,
extract::{Multipart, State},
extract::State,
http::{HeaderMap, StatusCode},
response::IntoResponse,
};
@@ -98,7 +98,7 @@ pub async fn post_plex_webhook(
State(state): State<AppState>,
headers: HeaderMap,
Params(query): Params<WebhookTokenQuery>,
mut multipart: Multipart,
Upload(mut multipart): Upload,
) -> impl IntoResponse {
let token = match extract_webhook_token(&headers, &query) {
Some(t) => t,

2
crates/adapters/http-axum/src/api/wrapup.rs
View File

@@ -116,7 +116,7 @@ pub async fn get_status(
get, path = "/api/v1/wrapups/{id}/report",
params(("id" = Uuid, Path, description = "Wrap-up ID")),
responses(
(status = 200, description = "Report JSON", content_type = "application/json"),
(status = 200, description = "Report JSON", body = inline(serde_json::Value)),
(status = 202, description = "Still generating"),
(status = 401, description = "Unauthorized", body = api_types::ErrorResponse),
(status = 404, description = "Not found", body = api_types::ErrorResponse),

26
crates/adapters/http-axum/src/extractors/json.rs
View File

@@ -2,7 +2,8 @@
use axum::{
extract::{
FromRequest, FromRequestParts, Path, Query, Request,
FromRequest, FromRequestParts, Multipart, Path, Query, Request,
multipart::MultipartRejection,
rejection::{JsonRejection, PathRejection, QueryRejection},
},
http::{StatusCode, request::Parts},
@@ -52,6 +53,29 @@ impl From<PathRejection> for MalformedRequest {
}
}
impl From<MultipartRejection> for MalformedRequest {
fn from(r: MultipartRejection) -> Self {
Self {
status: r.status(),
message: r.body_text(),
}
}
}
/// `axum::extract::Multipart` with refusals in the one error shape.
pub struct Upload(pub Multipart);
impl<S> FromRequest<S> for Upload
where
S: Send + Sync,
{
type Rejection = MalformedRequest;
async fn from_request(req: Request, state: &S) -> Result<Self, Self::Rejection> {
Ok(Self(Multipart::from_request(req, state).await?))
}
}
/// `axum::Json` with refusals in the one error shape.
pub struct Body<T>(pub T);

2
crates/adapters/http-axum/src/extractors/mod.rs
View File

@@ -1,5 +1,5 @@
mod json;
pub use json::{Body, MalformedRequest, Params, PathId};
pub use json::{Body, MalformedRequest, Params, PathId, Upload};
use axum::{
extract::{FromRef, FromRequestParts},

70
crates/adapters/http-axum/src/router.rs
View File

@@ -64,6 +64,17 @@ fn slow_down(reason: RejectionReason) -> axum::response::Response {
}
}
/// Every `/api/v1` limiter is built here so its refusals use the one error shape.
fn api_governor(quota: Quota) -> axum_governor::GovernorConfig<std::net::IpAddr> {
GovernorConfigBuilder::default()
.with_extractor(PeerIp::default())
.expect_connect_info()
.quota_default(quota)
.error_handler(slow_down)
.finish()
.unwrap()
}
fn per_minute(n: u64) -> Quota {
let n = NonZeroU32::new(n.clamp(1, u32::MAX as u64) as u32).unwrap();
Quota::requests_per_minute(n)
@@ -130,23 +141,11 @@ fn cors_layer(origins: &str) -> CorsLayer {
}
fn api_routes(rate_limit: u64, cors_origins: &str) -> Router<AppState> {
let cfg = GovernorConfigBuilder::default()
.with_extractor(PeerIp::default())
.expect_connect_info()
.quota_default(per_minute(rate_limit))
.error_handler(slow_down)
.finish()
.unwrap();
let cfg = api_governor(per_minute(rate_limit));
let base = api::router().merge(federation::api_routes());
let webhook_cfg = GovernorConfigBuilder::default()
.with_extractor(PeerIp::default())
.expect_connect_info()
.quota_default(per_minute(rate_limit / 4))
.error_handler(slow_down)
.finish()
.unwrap();
let webhook_cfg = api_governor(per_minute(rate_limit / 4));
let webhook_routes = api::webhook_router().layer(GovernorLayer::new(webhook_cfg));
Router::new()
@@ -154,3 +153,46 @@ fn api_routes(rate_limit: u64, cors_origins: &str) -> Router<AppState> {
.nest("/api/v1", webhook_routes)
.layer(cors_layer(cors_origins))
}
#[cfg(test)]
mod tests {
use super::*;
use axum::{body::Body, http::Request};
use tower::ServiceExt;
async fn body_json(resp: axum::response::Response) -> serde_json::Value {
let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
.await
.unwrap();
serde_json::from_slice(&bytes).unwrap()
}
#[tokio::test]
async fn api_rate_limit_refusal_is_the_one_error_shape() {
let app = Router::new()
.route("/x", routing::get(|| async { "ok" }))
.layer(GovernorLayer::new(api_governor(per_minute(1))));
let addr: std::net::SocketAddr = "127.0.0.1:1".parse().unwrap();
let mut last = None;
for _ in 0..3 {
let mut req = Request::builder().uri("/x").body(Body::empty()).unwrap();
req.extensions_mut()
.insert(axum::extract::ConnectInfo::<std::net::SocketAddr>(addr));
last = Some(app.clone().oneshot(req).await.unwrap());
}
let resp = last.unwrap();
assert_eq!(resp.status(), StatusCode::TOO_MANY_REQUESTS);
let body = body_json(resp).await;
assert_eq!(body["code"], "RATE_LIMITED");
assert!(body["error"].is_string());
}
#[tokio::test]
async fn unidentifiable_caller_is_refused_in_the_one_error_shape() {
let resp = slow_down(RejectionReason::KeyExtractionFailed(
axum_governor::ExtractionError::MissingConnectInfo,
));
assert_eq!(resp.status(), StatusCode::INTERNAL_SERVER_ERROR);
assert_eq!(body_json(resp).await["code"], "INTERNAL");
}
}

25
crates/adapters/http-axum/src/tests/api_handlers.rs
View File

@@ -529,3 +529,28 @@ async fn a_json_body_of_the_wrong_shape_is_refused_as_unprocessable() {
assert_eq!(resp.status(), StatusCode::UNPROCESSABLE_ENTITY);
assert_eq!(refusal_body(resp).await["code"], "INVALID_INPUT");
}
#[tokio::test]
async fn a_non_multipart_upload_is_refused_with_invalid_input_code() {
let app = Router::new()
.route(
"/api/v1/webhooks/plex",
axum::routing::post(crate::api::webhook::post_plex_webhook),
)
.with_state(test_app_state());
let resp = app
.oneshot(
Request::builder()
.method("POST")
.uri("/api/v1/webhooks/plex?token=t")
.header("content-type", "text/plain")
.body(Body::from("hello"))
.unwrap(),
)
.await
.unwrap();
assert!(resp.status().is_client_error());
assert_eq!(refusal_body(resp).await["code"], "INVALID_INPUT");
}

35
crates/adapters/http-axum/tests/openapi_contract_test.rs
View File

@@ -18,30 +18,31 @@ fn operations(spec: &serde_json::Value) -> Vec<(String, String, serde_json::Valu
}
#[test]
fn every_operation_that_can_succeed_with_a_body_declares_its_shape() {
fn the_spec_is_not_vacuous() {
let spec = spec();
assert!(!operations(&spec).is_empty());
assert!(spec["components"]["schemas"]["ErrorResponse"].is_object());
}
#[test]
fn every_2xx_json_response_declares_a_schema() {
let spec = spec();
let mut untyped = Vec::new();
for (verb, path, op) in operations(&spec) {
let responses = op["responses"].as_object().unwrap();
let typed = responses.iter().any(|(code, r)| {
code.starts_with('2') && r["content"]["application/json"]["schema"].is_object()
});
let bodyless = responses
.iter()
.any(|(code, r)| code.starts_with('2') && r.get("content").is_none());
let binary = responses.iter().any(|(code, r)| {
code.starts_with('2')
&& r["content"]
.as_object()
.is_some_and(|c| !c.contains_key("application/json"))
});
if !typed && !bodyless && !binary {
untyped.push(format!("{verb} {path}"));
for (code, r) in op["responses"].as_object().unwrap() {
if !code.starts_with('2') {
continue;
}
if let Some(json) = r["content"].get("application/json")
&& !json["schema"].is_object()
{
untyped.push(format!("{verb} {path} -> {code}"));
}
}
}
assert!(untyped.is_empty(), "2xx without a schema: {untyped:?}");
assert!(untyped.is_empty(), "2xx JSON without a schema: {untyped:?}");
}
#[test]