mod common;
use axum::{
body::Body,
extract::Request,
http::header::{HeaderValue, AUTHORIZATION, CONTENT_TYPE},
response::Response,
routing::{get, post},
Router,
};
use common::{create_temp_dir, TestConfig};
use flate2::write::{DeflateEncoder, GzEncoder};
use flate2::Compression;
use ivoryvalley::{config::Config, db::SeenUriStore, proxy::create_proxy_router};
use std::io::Write;
use std::net::SocketAddr;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc as StdArc;
use tokio::net::TcpListener;
struct MockUpstream {
pub addr: SocketAddr,
shutdown_tx: Option<tokio::sync::oneshot::Sender<()>>,
}
impl MockUpstream {
async fn start() -> Self {
let app = Router::new()
.route("/api/v1/timelines/home", get(mock_timeline_handler))
.route(
"/api/v1/accounts/verify_credentials",
get(mock_verify_credentials),
)
.route("/api/v1/statuses", post(mock_post_status))
.route("/oauth/token", post(mock_oauth_token))
.fallback(mock_fallback_handler);
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel();
tokio::spawn(async move {
axum::serve(listener, app)
.with_graceful_shutdown(async {
let _ = shutdown_rx.await;
})
.await
.unwrap();
});
Self {
addr,
shutdown_tx: Some(shutdown_tx),
}
}
fn url(&self) -> String {
format!("http://{}", self.addr)
}
}
impl Drop for MockUpstream {
fn drop(&mut self) {
if let Some(tx) = self.shutdown_tx.take() {
let _ = tx.send(());
}
}
}
async fn mock_timeline_handler(req: Request<Body>) -> Response<Body> {
let auth = req
.headers()
.get("Authorization")
.and_then(|v| v.to_str().ok())
.unwrap_or("");
if auth.is_empty() {
return Response::builder()
.status(401)
.body(Body::from(r#"{"error":"unauthorized"}"#))
.unwrap();
}
Response::builder()
.status(200)
.header("Content-Type", "application/json")
.body(Body::from(r#"[{"id":"1","content":"Hello"}]"#))
.unwrap()
}
async fn mock_verify_credentials(req: Request<Body>) -> Response<Body> {
let auth = req
.headers()
.get("Authorization")
.and_then(|v| v.to_str().ok())
.unwrap_or("");
if auth.is_empty() {
return Response::builder()
.status(401)
.body(Body::from(r#"{"error":"unauthorized"}"#))
.unwrap();
}
Response::builder()
.status(200)
.header("Content-Type", "application/json")
.body(Body::from(r#"{"id":"12345","username":"testuser"}"#))
.unwrap()
}
async fn mock_post_status(req: Request<Body>) -> Response<Body> {
let auth = req
.headers()
.get("Authorization")
.and_then(|v| v.to_str().ok())
.unwrap_or("");
if auth.is_empty() {
return Response::builder()
.status(401)
.body(Body::from(r#"{"error":"unauthorized"}"#))
.unwrap();
}
Response::builder()
.status(200)
.header("Content-Type", "application/json")
.body(Body::from(r#"{"id":"999","content":"Posted!"}"#))
.unwrap()
}
async fn mock_oauth_token() -> Response<Body> {
Response::builder()
.status(200)
.header("Content-Type", "application/json")
.body(Body::from(
r#"{"access_token":"test_token","token_type":"Bearer"}"#,
))
.unwrap()
}
async fn mock_fallback_handler(req: Request<Body>) -> Response<Body> {
let path = req.uri().path().to_string();
let method = req.method().to_string();
Response::builder()
.status(200)
.header("Content-Type", "application/json")
.body(Body::from(format!(
r#"{{"path":"{}","method":"{}"}}"#,
path, method
)))
.unwrap()
}
#[test]
fn test_proxy_config_creation() {
let temp_dir = create_temp_dir();
let config = TestConfig::new()
.with_db_path(temp_dir.path().join("test.db"))
.with_upstream_url("http://mastodon.example.com");
assert!(config.db_path.is_some());
assert_eq!(config.upstream_url, "http://mastodon.example.com");
}
#[tokio::test]
async fn test_proxy_forwards_get_request() {
let upstream = MockUpstream::start().await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client
.get("/api/v1/timelines/home")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.await;
response.assert_status_ok();
let body = response.text();
assert!(body.contains("Hello"));
}
#[tokio::test]
async fn test_proxy_passes_auth_header() {
let upstream = MockUpstream::start().await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client.get("/api/v1/timelines/home").await;
response.assert_status_unauthorized();
let response = client
.get("/api/v1/timelines/home")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.await;
response.assert_status_ok();
}
#[tokio::test]
async fn test_proxy_forwards_post_request() {
let upstream = MockUpstream::start().await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client
.post("/api/v1/statuses")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.add_header(CONTENT_TYPE, HeaderValue::from_static("application/json"))
.text(r#"{"status":"Hello world"}"#)
.await;
response.assert_status_ok();
let body = response.text();
assert!(body.contains("Posted!"));
}
#[tokio::test]
async fn test_proxy_oauth_passthrough() {
let upstream = MockUpstream::start().await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client
.post("/oauth/token")
.add_header(
CONTENT_TYPE,
HeaderValue::from_static("application/x-www-form-urlencoded"),
)
.text("grant_type=authorization_code&code=test")
.await;
response.assert_status_ok();
let body = response.text();
assert!(body.contains("access_token"));
}
#[tokio::test]
async fn test_proxy_account_passthrough() {
let upstream = MockUpstream::start().await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client
.get("/api/v1/accounts/verify_credentials")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.await;
response.assert_status_ok();
let body = response.text();
assert!(body.contains("testuser"));
}
#[tokio::test]
async fn test_proxy_fallback_passthrough() {
let upstream = MockUpstream::start().await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client
.get("/api/v1/some/unknown/endpoint")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.await;
response.assert_status_ok();
let body = response.text();
assert!(body.contains("/api/v1/some/unknown/endpoint"));
assert!(body.contains("GET"));
}
struct MockTimelineUpstream {
pub addr: SocketAddr,
shutdown_tx: Option<tokio::sync::oneshot::Sender<()>>,
}
impl MockTimelineUpstream {
async fn start_with_statuses(statuses_json: &'static str) -> Self {
let app = Router::new()
.route(
"/api/v1/timelines/home",
get(move || async move {
Response::builder()
.status(200)
.header("Content-Type", "application/json")
.body(Body::from(statuses_json))
.unwrap()
}),
)
.route(
"/api/v1/timelines/public",
get(move || async move {
Response::builder()
.status(200)
.header("Content-Type", "application/json")
.body(Body::from(statuses_json))
.unwrap()
}),
)
.route(
"/api/v1/timelines/list/{list_id}",
get(move || async move {
Response::builder()
.status(200)
.header("Content-Type", "application/json")
.body(Body::from(statuses_json))
.unwrap()
}),
)
.route(
"/api/v1/timelines/tag/{hashtag}",
get(move || async move {
Response::builder()
.status(200)
.header("Content-Type", "application/json")
.body(Body::from(statuses_json))
.unwrap()
}),
)
.route(
"/api/v1/timelines/link",
get(move || async move {
Response::builder()
.status(200)
.header("Content-Type", "application/json")
.body(Body::from(statuses_json))
.unwrap()
}),
)
.route(
"/api/v1/trends/statuses",
get(move || async move {
Response::builder()
.status(200)
.header("Content-Type", "application/json")
.body(Body::from(statuses_json))
.unwrap()
}),
);
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel();
tokio::spawn(async move {
axum::serve(listener, app)
.with_graceful_shutdown(async {
let _ = shutdown_rx.await;
})
.await
.unwrap();
});
Self {
addr,
shutdown_tx: Some(shutdown_tx),
}
}
fn url(&self) -> String {
format!("http://{}", self.addr)
}
}
impl Drop for MockTimelineUpstream {
fn drop(&mut self) {
if let Some(tx) = self.shutdown_tx.take() {
let _ = tx.send(());
}
}
}
#[tokio::test]
async fn test_timeline_first_status_passes_through() {
let statuses = r#"[
{"id": "1", "uri": "https://example.com/statuses/1", "content": "<p>Hello</p>"},
{"id": "2", "uri": "https://example.com/statuses/2", "content": "<p>World</p>"}
]"#;
let upstream = MockTimelineUpstream::start_with_statuses(statuses).await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client
.get("/api/v1/timelines/home")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
let statuses = body.as_array().unwrap();
assert_eq!(statuses.len(), 2);
assert_eq!(statuses[0]["id"], "1");
assert_eq!(statuses[1]["id"], "2");
}
#[tokio::test]
async fn test_timeline_duplicates_are_filtered() {
let statuses = r#"[
{"id": "1", "uri": "https://example.com/statuses/1", "content": "<p>Hello</p>"}
]"#;
let upstream = MockTimelineUpstream::start_with_statuses(statuses).await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client
.get("/api/v1/timelines/home")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
assert_eq!(body.as_array().unwrap().len(), 1);
let response = client
.get("/api/v1/timelines/home")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
assert_eq!(body.as_array().unwrap().len(), 0);
}
#[tokio::test]
async fn test_timeline_boost_deduplication() {
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let seen_store = SeenUriStore::open(":memory:").unwrap();
seen_store
.mark_seen("https://original.com/statuses/1")
.unwrap();
let boost_statuses = r#"[
{
"id": "2",
"uri": "https://booster.com/statuses/2",
"content": "",
"reblog": {
"id": "1",
"uri": "https://original.com/statuses/1",
"content": "<p>Original</p>"
}
}
]"#;
let upstream = MockTimelineUpstream::start_with_statuses(boost_statuses).await;
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client
.get("/api/v1/timelines/home")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
assert_eq!(body.as_array().unwrap().len(), 0);
}
#[tokio::test]
async fn test_timeline_public_filtering() {
let statuses = r#"[
{"id": "1", "uri": "https://example.com/statuses/1", "content": "<p>Hello</p>"}
]"#;
let upstream = MockTimelineUpstream::start_with_statuses(statuses).await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client.get("/api/v1/timelines/public").await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
assert_eq!(body.as_array().unwrap().len(), 1);
let response = client.get("/api/v1/timelines/public").await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
assert_eq!(body.as_array().unwrap().len(), 0);
}
#[tokio::test]
async fn test_timeline_list_filtering() {
let statuses = r#"[
{"id": "1", "uri": "https://example.com/statuses/1", "content": "<p>Hello</p>"}
]"#;
let upstream = MockTimelineUpstream::start_with_statuses(statuses).await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client.get("/api/v1/timelines/list/12345").await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
assert_eq!(body.as_array().unwrap().len(), 1);
let response = client.get("/api/v1/timelines/list/12345").await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
assert_eq!(body.as_array().unwrap().len(), 0);
}
#[tokio::test]
async fn test_timeline_hashtag_filtering() {
let statuses = r#"[
{"id": "1", "uri": "https://example.com/statuses/1", "content": "<p>Hello</p>"}
]"#;
let upstream = MockTimelineUpstream::start_with_statuses(statuses).await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client.get("/api/v1/timelines/tag/rust").await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
assert_eq!(body.as_array().unwrap().len(), 1);
let response = client.get("/api/v1/timelines/tag/rust").await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
assert_eq!(body.as_array().unwrap().len(), 0);
}
#[tokio::test]
async fn test_timeline_status_without_uri_passes_through() {
let statuses = r#"[
{"id": "1", "content": "<p>No URI field</p>"}
]"#;
let upstream = MockTimelineUpstream::start_with_statuses(statuses).await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client
.get("/api/v1/timelines/home")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
assert_eq!(body.as_array().unwrap().len(), 1);
let response = client
.get("/api/v1/timelines/home")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
assert_eq!(body.as_array().unwrap().len(), 1);
}
#[tokio::test]
async fn test_body_within_limit_succeeds() {
let upstream = MockUpstream::start().await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::with_max_body_size(&upstream.url(), "0.0.0.0", 0, db_path, 1024);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let small_body = "x".repeat(500);
let response = client
.post("/api/v1/statuses")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.add_header(CONTENT_TYPE, HeaderValue::from_static("application/json"))
.text(small_body)
.await;
response.assert_status_ok();
}
#[tokio::test]
async fn test_body_exceeding_limit_returns_413() {
let upstream = MockUpstream::start().await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::with_max_body_size(&upstream.url(), "0.0.0.0", 0, db_path, 1024);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let large_body = "x".repeat(2000);
let response = client
.post("/api/v1/statuses")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.add_header(CONTENT_TYPE, HeaderValue::from_static("application/json"))
.text(large_body)
.await;
response.assert_status(axum::http::StatusCode::PAYLOAD_TOO_LARGE);
}
#[tokio::test]
async fn test_default_body_limit_allows_normal_requests() {
let upstream = MockUpstream::start().await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let normal_body = r#"{"status":"Hello world with some content"}"#;
let response = client
.post("/api/v1/statuses")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.add_header(CONTENT_TYPE, HeaderValue::from_static("application/json"))
.text(normal_body)
.await;
response.assert_status_ok();
}
#[tokio::test]
async fn test_timeline_link_filtering() {
let statuses = r#"[
{"id": "1", "uri": "https://example.com/statuses/1", "content": "<p>Hello</p>"}
]"#;
let upstream = MockTimelineUpstream::start_with_statuses(statuses).await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client.get("/api/v1/timelines/link").await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
assert_eq!(body.as_array().unwrap().len(), 1);
let response = client.get("/api/v1/timelines/link").await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
assert_eq!(body.as_array().unwrap().len(), 0);
}
#[tokio::test]
async fn test_trends_statuses_filtering() {
let statuses = r#"[
{"id": "1", "uri": "https://example.com/statuses/1", "content": "<p>Hello</p>"}
]"#;
let upstream = MockTimelineUpstream::start_with_statuses(statuses).await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client.get("/api/v1/trends/statuses").await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
assert_eq!(body.as_array().unwrap().len(), 1);
let response = client.get("/api/v1/trends/statuses").await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
assert_eq!(body.as_array().unwrap().len(), 0);
}
#[tokio::test]
async fn test_proxy_requests_compression_and_parses_json() {
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
let received_accept_encoding = Arc::new(AtomicBool::new(false));
let received_accept_encoding_clone = received_accept_encoding.clone();
let app = Router::new().route(
"/api/v1/timelines/home",
get(move |headers: axum::http::HeaderMap| async move {
let has_accept_encoding = headers
.get("accept-encoding")
.map(|v| {
let val = v.to_str().unwrap_or("");
val.contains("gzip") && val.contains("deflate")
})
.unwrap_or(false);
received_accept_encoding_clone.store(has_accept_encoding, Ordering::SeqCst);
axum::Json(serde_json::json!([
{"id": "1", "uri": "https://example.com/1", "content": "test"}
]))
}),
);
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&format!("http://{}", addr), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let proxy_app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(proxy_app).unwrap();
let response = client
.get("/api/v1/timelines/home")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test"))
.await;
response.assert_status_ok();
assert!(
received_accept_encoding.load(Ordering::SeqCst),
"Proxy should send Accept-Encoding: gzip, deflate to upstream"
);
let body: serde_json::Value = response.json();
assert_eq!(body.as_array().unwrap().len(), 1);
}
#[tokio::test]
async fn test_same_content_different_authors_not_deduplicated() {
let statuses = r#"[
{"id": "1", "uri": "https://alice.example/statuses/1", "content": "<p>Hello World</p>"},
{"id": "2", "uri": "https://bob.example/statuses/2", "content": "<p>Hello World</p>"}
]"#;
let upstream = MockTimelineUpstream::start_with_statuses(statuses).await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client
.get("/api/v1/timelines/home")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
let statuses = body.as_array().unwrap();
assert_eq!(statuses.len(), 2);
assert_eq!(statuses[0]["id"], "1");
assert_eq!(statuses[1]["id"], "2");
}
#[tokio::test]
async fn test_reply_passes_through_with_seen_parent() {
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let seen_store = SeenUriStore::open(":memory:").unwrap();
seen_store
.mark_seen("https://example.com/statuses/1")
.unwrap();
let statuses = r#"[
{"id": "2", "uri": "https://example.com/statuses/2", "in_reply_to_id": "1", "content": "<p>This is a reply</p>"}
]"#;
let upstream = MockTimelineUpstream::start_with_statuses(statuses).await;
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client
.get("/api/v1/timelines/home")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
assert_eq!(body.as_array().unwrap().len(), 1);
assert_eq!(body[0]["id"], "2");
}
#[tokio::test]
async fn test_thread_all_statuses_pass_through() {
let statuses = r#"[
{"id": "1", "uri": "https://example.com/statuses/1", "content": "<p>Original post</p>"},
{"id": "2", "uri": "https://example.com/statuses/2", "in_reply_to_id": "1", "content": "<p>First reply</p>"},
{"id": "3", "uri": "https://example.com/statuses/3", "in_reply_to_id": "2", "content": "<p>Reply to reply</p>"}
]"#;
let upstream = MockTimelineUpstream::start_with_statuses(statuses).await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client
.get("/api/v1/timelines/home")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
let statuses = body.as_array().unwrap();
assert_eq!(statuses.len(), 3);
assert_eq!(statuses[0]["id"], "1");
assert_eq!(statuses[1]["id"], "2");
assert_eq!(statuses[2]["id"], "3");
}
#[tokio::test]
async fn test_cross_instance_same_id_different_uri() {
let statuses = r#"[
{"id": "123", "uri": "https://instance-a.social/statuses/123", "content": "<p>From instance A</p>"},
{"id": "123", "uri": "https://instance-b.social/statuses/123", "content": "<p>From instance B</p>"}
]"#;
let upstream = MockTimelineUpstream::start_with_statuses(statuses).await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client
.get("/api/v1/timelines/home")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
let statuses = body.as_array().unwrap();
assert_eq!(statuses.len(), 2);
}
struct MockNonTimelineUpstream {
pub addr: SocketAddr,
shutdown_tx: Option<tokio::sync::oneshot::Sender<()>>,
}
impl MockNonTimelineUpstream {
async fn start_with_statuses(statuses_json: &'static str) -> Self {
let app = Router::new()
.route(
"/api/v1/accounts/{account_id}/statuses",
get(move || async move {
Response::builder()
.status(200)
.header("Content-Type", "application/json")
.body(Body::from(statuses_json))
.unwrap()
}),
)
.route(
"/api/v1/statuses/{status_id}/context",
get(move || async move {
let context =
format!(r#"{{"ancestors": {}, "descendants": []}}"#, statuses_json);
Response::builder()
.status(200)
.header("Content-Type", "application/json")
.body(Body::from(context))
.unwrap()
}),
)
.route(
"/api/v1/bookmarks",
get(move || async move {
Response::builder()
.status(200)
.header("Content-Type", "application/json")
.body(Body::from(statuses_json))
.unwrap()
}),
)
.route(
"/api/v1/favourites",
get(move || async move {
Response::builder()
.status(200)
.header("Content-Type", "application/json")
.body(Body::from(statuses_json))
.unwrap()
}),
);
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel();
tokio::spawn(async move {
axum::serve(listener, app)
.with_graceful_shutdown(async {
let _ = shutdown_rx.await;
})
.await
.unwrap();
});
Self {
addr,
shutdown_tx: Some(shutdown_tx),
}
}
fn url(&self) -> String {
format!("http://{}", self.addr)
}
}
impl Drop for MockNonTimelineUpstream {
fn drop(&mut self) {
if let Some(tx) = self.shutdown_tx.take() {
let _ = tx.send(());
}
}
}
#[tokio::test]
async fn test_account_statuses_not_filtered() {
let statuses = r#"[
{"id": "1", "uri": "https://example.com/statuses/1", "content": "<p>Hello</p>"}
]"#;
let upstream = MockNonTimelineUpstream::start_with_statuses(statuses).await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
seen_store
.mark_seen("https://example.com/statuses/1")
.unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client
.get("/api/v1/accounts/12345/statuses")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
assert_eq!(body.as_array().unwrap().len(), 1);
assert_eq!(body[0]["id"], "1");
let response = client
.get("/api/v1/accounts/12345/statuses")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
assert_eq!(body.as_array().unwrap().len(), 1);
}
#[tokio::test]
async fn test_thread_context_not_filtered() {
let statuses = r#"[
{"id": "1", "uri": "https://example.com/statuses/1", "content": "<p>Parent</p>"}
]"#;
let upstream = MockNonTimelineUpstream::start_with_statuses(statuses).await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
seen_store
.mark_seen("https://example.com/statuses/1")
.unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client
.get("/api/v1/statuses/999/context")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
let ancestors = body["ancestors"].as_array().unwrap();
assert_eq!(ancestors.len(), 1);
assert_eq!(ancestors[0]["id"], "1");
}
#[tokio::test]
async fn test_bookmarks_not_filtered() {
let statuses = r#"[
{"id": "1", "uri": "https://example.com/statuses/1", "content": "<p>Bookmarked</p>"}
]"#;
let upstream = MockNonTimelineUpstream::start_with_statuses(statuses).await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
seen_store
.mark_seen("https://example.com/statuses/1")
.unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client
.get("/api/v1/bookmarks")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
assert_eq!(body.as_array().unwrap().len(), 1);
let response = client
.get("/api/v1/bookmarks")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
assert_eq!(body.as_array().unwrap().len(), 1);
}
#[tokio::test]
async fn test_favourites_not_filtered() {
let statuses = r#"[
{"id": "1", "uri": "https://example.com/statuses/1", "content": "<p>Favourited</p>"}
]"#;
let upstream = MockNonTimelineUpstream::start_with_statuses(statuses).await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
seen_store
.mark_seen("https://example.com/statuses/1")
.unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client
.get("/api/v1/favourites")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
assert_eq!(body.as_array().unwrap().len(), 1);
}
#[tokio::test]
async fn test_health_endpoint_returns_ok() {
let upstream = MockUpstream::start().await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client.get("/health").await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
assert_eq!(body["status"], "healthy");
assert!(body["version"].is_string());
}
#[tokio::test]
async fn test_health_endpoint_no_auth_required() {
let upstream = MockUpstream::start().await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client.get("/health").await;
response.assert_status_ok();
}
#[tokio::test]
async fn test_health_endpoint_deep_check() {
let upstream = MockUpstream::start().await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client.get("/health?deep=true").await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
assert_eq!(body["status"], "healthy");
assert!(body["version"].is_string());
assert!(body["checks"]["database"].is_string());
}
fn gzip_compress(data: &[u8]) -> Vec<u8> {
let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
encoder.write_all(data).unwrap();
encoder.finish().unwrap()
}
fn deflate_compress(data: &[u8]) -> Vec<u8> {
let mut encoder = DeflateEncoder::new(Vec::new(), Compression::default());
encoder.write_all(data).unwrap();
encoder.finish().unwrap()
}
#[tokio::test]
async fn test_gzip_response_decompressed_and_filtered() {
let json_data =
r#"[{"id":"1","uri":"https://example.com/statuses/gzip1","content":"<p>Hello gzip</p>"}]"#;
let compressed_data = gzip_compress(json_data.as_bytes());
let app = Router::new().route(
"/api/v1/timelines/home",
get({
let data = compressed_data.clone();
move |req: Request<Body>| {
let data = data.clone();
async move {
let auth = req
.headers()
.get("Authorization")
.and_then(|v| v.to_str().ok())
.unwrap_or("");
if auth.is_empty() {
return Response::builder()
.status(401)
.body(Body::from(r#"{"error":"unauthorized"}"#))
.unwrap();
}
Response::builder()
.status(200)
.header("Content-Type", "application/json")
.header("Content-Encoding", "gzip")
.body(Body::from(data))
.unwrap()
}
}
}),
);
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let upstream_url = format!("http://{}", addr);
tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream_url, "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let proxy_app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(proxy_app).unwrap();
let response = client
.get("/api/v1/timelines/home")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test"))
.await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
let statuses = body.as_array().expect("Response should be an array");
assert_eq!(statuses.len(), 1, "First request should return 1 status");
assert_eq!(statuses[0]["id"], "1");
let response = client
.get("/api/v1/timelines/home")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test"))
.await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
let statuses = body.as_array().expect("Response should be an array");
assert_eq!(
statuses.len(),
0,
"Second request should return 0 statuses (deduplicated)"
);
}
#[tokio::test]
async fn test_deflate_response_decompressed() {
let json_data = r#"[{"id":"1","uri":"https://example.com/statuses/deflate1","content":"<p>Hello deflate</p>"}]"#;
let compressed_data = deflate_compress(json_data.as_bytes());
let app = Router::new().route(
"/api/v1/timelines/home",
get({
let data = compressed_data.clone();
move |req: Request<Body>| {
let data = data.clone();
async move {
let auth = req
.headers()
.get("Authorization")
.and_then(|v| v.to_str().ok())
.unwrap_or("");
if auth.is_empty() {
return Response::builder()
.status(401)
.body(Body::from(r#"{"error":"unauthorized"}"#))
.unwrap();
}
Response::builder()
.status(200)
.header("Content-Type", "application/json")
.header("Content-Encoding", "deflate")
.body(Body::from(data))
.unwrap()
}
}
}),
);
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let upstream_url = format!("http://{}", addr);
tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream_url, "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let proxy_app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(proxy_app).unwrap();
let response = client
.get("/api/v1/timelines/home")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test"))
.await;
response.assert_status_ok();
let body: serde_json::Value = response.json();
let statuses = body.as_array().expect("Response should be an array");
assert_eq!(statuses.len(), 1);
assert_eq!(statuses[0]["id"], "1");
}
#[tokio::test]
async fn test_malformed_gzip_returns_error() {
let invalid_gzip = vec![
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xff, 0xff,
];
let app = Router::new().route(
"/api/v1/timelines/home",
get({
let data = invalid_gzip.clone();
move |_req: Request<Body>| {
let data = data.clone();
async move {
Response::builder()
.status(200)
.header("Content-Type", "application/json")
.header("Content-Encoding", "gzip")
.body(Body::from(data))
.unwrap()
}
}
}),
);
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let upstream_url = format!("http://{}", addr);
tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream_url, "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let proxy_app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(proxy_app).unwrap();
let response = client
.get("/api/v1/timelines/home")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test"))
.await;
response.assert_status(axum::http::StatusCode::BAD_GATEWAY);
}
#[tokio::test]
async fn test_accept_encoding_sent_to_upstream() {
let received_accept_encoding = StdArc::new(AtomicBool::new(false));
let received_clone = received_accept_encoding.clone();
let app = Router::new().route(
"/api/v1/timelines/home",
get(move |req: Request<Body>| {
let received = received_clone.clone();
async move {
let has_encoding = req
.headers()
.get("accept-encoding")
.map(|v| {
let value = v.to_str().unwrap_or("");
value.contains("gzip") && value.contains("deflate")
})
.unwrap_or(false);
received.store(has_encoding, Ordering::SeqCst);
Response::builder()
.status(200)
.header("Content-Type", "application/json")
.body(Body::from(r#"[]"#))
.unwrap()
}
}),
);
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let upstream_url = format!("http://{}", addr);
tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream_url, "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let proxy_app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(proxy_app).unwrap();
let response = client
.get("/api/v1/timelines/home")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test"))
.await;
response.assert_status_ok();
assert!(
received_accept_encoding.load(Ordering::SeqCst),
"Proxy should send Accept-Encoding: gzip, deflate to upstream"
);
}
#[tokio::test]
async fn test_uncompressed_response_still_works() {
let upstream = MockUpstream::start().await;
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream.url(), "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(app).unwrap();
let response = client
.get("/api/v1/timelines/home")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test_token"))
.await;
response.assert_status_ok();
let body = response.text();
assert!(body.contains("Hello"), "Uncompressed response should work");
}
#[tokio::test]
async fn test_content_encoding_not_forwarded() {
let json_data = r#"[{"id":"1","content":"Hello"}]"#;
let compressed_data = gzip_compress(json_data.as_bytes());
let app = Router::new().route(
"/api/v1/timelines/home",
get({
let data = compressed_data.clone();
move |_req: Request<Body>| {
let data = data.clone();
async move {
Response::builder()
.status(200)
.header("Content-Type", "application/json")
.header("Content-Encoding", "gzip")
.body(Body::from(data))
.unwrap()
}
}
}),
);
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let upstream_url = format!("http://{}", addr);
tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
let temp_dir = create_temp_dir();
let db_path = temp_dir.path().join("test.db");
let config = Config::new(&upstream_url, "0.0.0.0", 0, db_path);
let seen_store = SeenUriStore::open(":memory:").unwrap();
let proxy_app = create_proxy_router(config, std::sync::Arc::new(seen_store));
let client = axum_test::TestServer::new(proxy_app).unwrap();
let response = client
.get("/api/v1/timelines/home")
.add_header(AUTHORIZATION, HeaderValue::from_static("Bearer test"))
.await;
response.assert_status_ok();
assert!(
response.headers().get("content-encoding").is_none(),
"Content-Encoding should not be forwarded to client"
);
}