reinhardt_testkit/fixtures/
server.rs1#![allow(deprecated)]
9
10use reinhardt_di::InjectionContext;
11use reinhardt_http::Handler;
12use reinhardt_http::{Request, Response};
13use reinhardt_server::{
14 HttpServer, RateLimitConfig, RateLimitHandler, ShutdownCoordinator, TimeoutHandler,
15};
16use reinhardt_urls::routers::ServerRouter as Router;
17use rstest::fixture;
18use std::net::SocketAddr;
19use std::sync::Arc;
20use std::time::Duration;
21use tokio::net::TcpListener;
22use tokio::task::JoinHandle;
23
24#[cfg(feature = "websockets")]
25use reinhardt_server::WebSocketServer;
26
27#[cfg(feature = "graphql")]
28use reinhardt_server::GraphQLHandler;
29
30pub struct TestServerGuard {
53 pub url: String,
55 pub coordinator: Arc<ShutdownCoordinator>,
57 server_task: Option<JoinHandle<()>>,
59}
60
61impl TestServerGuard {
62 async fn new(router: Router) -> Self {
74 let shutdown_timeout = Duration::from_secs(5);
75 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
77 let actual_addr = listener.local_addr().unwrap();
78 let url = format!("http://{}", actual_addr);
79
80 let coordinator = Arc::new(ShutdownCoordinator::new(shutdown_timeout));
82
83 let server_coordinator = (*coordinator).clone();
85 let handler: Arc<dyn Handler> = Arc::new(router);
86 let server = HttpServer::new(handler);
87 let mut shutdown_rx = server_coordinator.subscribe();
88 let server_task = tokio::spawn(async move {
89 loop {
90 tokio::select! {
91 result = listener.accept() => {
92 match result {
93 Ok((stream, socket_addr)) => {
94 let handler_clone = server.handler();
95 tokio::spawn(async move {
96 if let Err(e) =
97 HttpServer::handle_connection(stream, socket_addr, handler_clone, None)
98 .await
99 {
100 eprintln!("Error handling connection: {:?}", e);
101 }
102 });
103 }
104 Err(e) => {
105 eprintln!("Error accepting connection: {:?}", e);
106 break;
107 }
108 }
109 }
110 _ = shutdown_rx.recv() => {
111 break;
112 }
113 }
114 }
115 });
116
117 wait_for_server_ready(actual_addr)
120 .await
121 .expect("Test server failed to become ready");
122
123 Self {
124 url,
125 coordinator,
126 server_task: Some(server_task),
127 }
128 }
129}
130
131impl Drop for TestServerGuard {
132 fn drop(&mut self) {
133 self.coordinator.shutdown();
135
136 if let Some(task) = self.server_task.take() {
140 task.abort();
141 }
142 }
143}
144
145pub async fn test_server_guard(router: Router) -> TestServerGuard {
168 TestServerGuard::new(router).await
169}
170
171#[derive(Clone)]
177pub struct BasicHandler;
178
179#[async_trait::async_trait]
180impl Handler for BasicHandler {
181 async fn handle(&self, _request: Request) -> reinhardt_core::exception::Result<Response> {
182 Ok(Response::ok().with_body("OK"))
183 }
184}
185
186#[fixture]
210pub fn http_client() -> reqwest::Client {
211 reqwest::Client::builder()
212 .timeout(Duration::from_secs(10))
213 .build()
214 .expect("Failed to create HTTP client")
215}
216#[fixture]
238pub async fn http1_server() -> TestServer {
239 let handler = Arc::new(BasicHandler);
240 TestServer::builder()
241 .handler(handler)
242 .build()
243 .await
244 .expect("Failed to create HTTP/1.1 server")
245}
246
247#[fixture]
267pub async fn http2_server() -> TestServer {
268 let handler = Arc::new(BasicHandler);
269 TestServer::builder()
270 .handler(handler)
271 .http2(true)
272 .build()
273 .await
274 .expect("Failed to create HTTP/2 server")
275}
276
277#[fixture]
299pub async fn server_with_timeout(
300 #[default(Duration::from_secs(5))] timeout: Duration,
301) -> TestServer {
302 let handler = Arc::new(BasicHandler);
303 let timeout_handler = Arc::new(TimeoutHandler::new(handler, timeout));
304 TestServer::builder()
305 .handler(timeout_handler)
306 .build()
307 .await
308 .expect("Failed to create server with timeout")
309}
310
311#[fixture]
329pub async fn server_with_rate_limit(#[default(100)] limit: u32) -> TestServer {
330 let handler = Arc::new(BasicHandler);
331 let config = RateLimitConfig::per_minute(limit as usize);
332 let rate_limit_handler = Arc::new(RateLimitHandler::new(handler, config));
333 TestServer::builder()
334 .handler(rate_limit_handler)
335 .build()
336 .await
337 .expect("Failed to create server with rate limit")
338}
339
340#[fixture]
356pub async fn server_with_middleware_chain() -> TestServer {
357 let handler = Arc::new(BasicHandler);
358 let timeout_handler = Arc::new(TimeoutHandler::new(handler, Duration::from_secs(5)));
359 let config = RateLimitConfig::per_minute(100);
360 let rate_limit_handler = Arc::new(RateLimitHandler::new(timeout_handler, config));
361
362 TestServer::builder()
363 .handler(rate_limit_handler)
364 .build()
365 .await
366 .expect("Failed to create server with middleware chain")
367}
368
369#[fixture]
385pub async fn server_with_di() -> (TestServer, Arc<InjectionContext>) {
386 use reinhardt_di::SingletonScope;
387
388 let handler = Arc::new(BasicHandler);
389 let di_context = Arc::new(InjectionContext::builder(Arc::new(SingletonScope::new())).build());
390
391 let server = TestServer::builder()
392 .handler(handler)
393 .di_context(di_context.clone())
394 .build()
395 .await
396 .expect("Failed to create server with DI context");
397
398 (server, di_context)
399}
400
401#[cfg(feature = "websockets")]
406#[fixture]
422pub async fn websocket_server() -> TestServer {
423 use reinhardt_server::WebSocketHandler;
424
425 #[derive(Clone)]
426 struct EchoHandler;
427
428 #[async_trait::async_trait]
429 impl WebSocketHandler for EchoHandler {
430 async fn handle_message(&self, message: String) -> Result<String, String> {
431 Ok(message) }
433
434 async fn on_connect(&self) {}
435 async fn on_disconnect(&self) {}
436 }
437
438 let ws_handler = Arc::new(EchoHandler);
439 TestServer::builder()
440 .websocket_handler(ws_handler)
441 .build()
442 .await
443 .expect("Failed to create WebSocket server")
444}
445
446#[cfg(feature = "websockets")]
447#[fixture]
462pub async fn websocket_client(
463 #[from(websocket_server)]
464 #[future]
465 server: TestServer,
466) -> tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>> {
467 let server = server.await;
468 let ws_url = server.url.replace("http://", "ws://");
469 let (ws_stream, _) = tokio_tungstenite::connect_async(&ws_url)
470 .await
471 .expect("Failed to connect WebSocket");
472 ws_stream
473}
474
475#[cfg(feature = "graphql")]
480#[cfg(feature = "graphql")]
496#[fixture]
497pub async fn graphql_server() -> TestServer {
498 use async_graphql::{EmptyMutation, EmptySubscription, Object, Schema};
499
500 struct Query;
501
502 #[Object]
503 impl Query {
504 async fn hello(&self) -> &'static str {
505 "Hello, GraphQL!"
506 }
507 }
508
509 let schema = Schema::build(Query, EmptyMutation, EmptySubscription).finish();
510 let graphql_handler = Arc::new(GraphQLHandler::new(schema));
511
512 TestServer::builder()
513 .handler(graphql_handler)
514 .build()
515 .await
516 .expect("Failed to create GraphQL server")
517}
518
519pub struct TestServer {
525 pub url: String,
527 pub addr: SocketAddr,
529 pub coordinator: Arc<ShutdownCoordinator>,
531 server_task: Option<JoinHandle<()>>,
533}
534
535impl TestServer {
536 pub fn builder() -> TestServerBuilder {
538 TestServerBuilder::new()
539 }
540}
541
542impl Drop for TestServer {
543 fn drop(&mut self) {
544 self.coordinator.shutdown();
546
547 if let Some(task) = self.server_task.take() {
549 task.abort();
550 }
551 }
552}
553
554pub struct TestServerBuilder {
556 handler: Option<Arc<dyn Handler>>,
557 #[cfg(feature = "websockets")]
558 websocket_handler: Option<Arc<dyn reinhardt_server::WebSocketHandler>>,
559 di_context: Option<Arc<InjectionContext>>,
560 http2: bool,
561 shutdown_timeout: Duration,
562}
563
564impl TestServerBuilder {
565 fn new() -> Self {
566 Self {
567 handler: None,
568 #[cfg(feature = "websockets")]
569 websocket_handler: None,
570 di_context: None,
571 http2: false,
572 shutdown_timeout: Duration::from_secs(5),
573 }
574 }
575
576 pub fn handler(mut self, handler: Arc<dyn Handler>) -> Self {
578 self.handler = Some(handler);
579 self
580 }
581
582 #[cfg(feature = "websockets")]
583 pub fn websocket_handler(
585 mut self,
586 handler: Arc<dyn reinhardt_server::WebSocketHandler>,
587 ) -> Self {
588 self.websocket_handler = Some(handler);
589 self
590 }
591
592 pub fn di_context(mut self, context: Arc<InjectionContext>) -> Self {
594 self.di_context = Some(context);
595 self
596 }
597
598 pub fn http2(mut self, enabled: bool) -> Self {
600 self.http2 = enabled;
601 self
602 }
603
604 pub fn shutdown_timeout(mut self, timeout: Duration) -> Self {
606 self.shutdown_timeout = timeout;
607 self
608 }
609
610 pub async fn build(self) -> Result<TestServer, Box<dyn std::error::Error>> {
612 let listener = TcpListener::bind("127.0.0.1:0").await?;
614 let actual_addr = listener.local_addr()?;
615 let url = format!("http://{}", actual_addr);
616
617 let coordinator = Arc::new(ShutdownCoordinator::new(self.shutdown_timeout));
619
620 let server_coordinator = (*coordinator).clone();
622
623 #[cfg(feature = "websockets")]
624 let websocket_handler = self.websocket_handler;
625
626 let handler = self.handler;
627 let di_context = self.di_context;
628 let http2 = self.http2;
629
630 let server_task = tokio::spawn(async move {
631 #[cfg(feature = "websockets")]
635 if let Some(ws_handler) = websocket_handler {
636 drop(listener);
637 let server = WebSocketServer::from_arc(ws_handler);
638 let _ = server
639 .listen_with_shutdown(actual_addr, server_coordinator)
640 .await;
641 return;
642 }
643
644 if let Some(h) = handler {
645 if http2 {
646 drop(listener);
647 let server = reinhardt_server::Http2Server::new(h);
648 let _ = server
649 .listen_with_shutdown(actual_addr, server_coordinator)
650 .await;
651 } else {
652 let server = HttpServer::new(h);
654 let mut shutdown_rx = server_coordinator.subscribe();
655 loop {
656 tokio::select! {
657 result = listener.accept() => {
658 match result {
659 Ok((stream, socket_addr)) => {
660 let handler_clone = server.handler();
661 let di_ctx = di_context.clone();
662 tokio::spawn(async move {
663 if let Err(e) =
664 HttpServer::handle_connection(stream, socket_addr, handler_clone, di_ctx)
665 .await
666 {
667 eprintln!("Error handling connection: {:?}", e);
668 }
669 });
670 }
671 Err(e) => {
672 eprintln!("Error accepting connection: {:?}", e);
673 break;
674 }
675 }
676 }
677 _ = shutdown_rx.recv() => {
678 break;
679 }
680 }
681 }
682 }
683 }
684 });
685
686 wait_for_server_ready(actual_addr)
689 .await
690 .expect("Test server failed to become ready");
691
692 Ok(TestServer {
693 url,
694 addr: actual_addr,
695 coordinator,
696 server_task: Some(server_task),
697 })
698 }
699}
700
701const SERVER_READY_MAX_ATTEMPTS: u32 = 20;
707
708const SERVER_READY_PROBE_INTERVAL_MS: u64 = 50;
710
711async fn wait_for_server_ready(addr: SocketAddr) -> Result<(), std::io::Error> {
721 for attempt in 1..=SERVER_READY_MAX_ATTEMPTS {
722 match tokio::net::TcpStream::connect(addr).await {
724 Ok(_) => return Ok(()),
725 Err(_) if attempt < SERVER_READY_MAX_ATTEMPTS => {
726 tokio::time::sleep(Duration::from_millis(SERVER_READY_PROBE_INTERVAL_MS)).await;
727 }
728 Err(e) => {
729 return Err(std::io::Error::new(
730 std::io::ErrorKind::TimedOut,
731 format!(
732 "Server at {} not ready after {} attempts: {}",
733 addr, SERVER_READY_MAX_ATTEMPTS, e
734 ),
735 ));
736 }
737 }
738 }
739
740 Err(std::io::Error::new(
741 std::io::ErrorKind::TimedOut,
742 format!(
743 "Server at {} not ready after {} attempts",
744 addr, SERVER_READY_MAX_ATTEMPTS
745 ),
746 ))
747}
748
749#[cfg(test)]
750mod tests {
751 use super::*;
752 use rstest::*;
753
754 #[rstest]
755 #[tokio::test]
756 async fn test_basic_handler_returns_ok() {
757 let handler = BasicHandler;
759 let request = Request::builder()
760 .method(hyper::Method::GET)
761 .uri("/")
762 .build()
763 .expect("Failed to build request");
764
765 let response = handler.handle(request).await;
767
768 assert!(response.is_ok(), "Expected Ok response from BasicHandler");
770 let resp = response.unwrap();
771 assert_eq!(resp.status, hyper::StatusCode::OK);
772 }
773
774 #[rstest]
775 #[tokio::test]
776 async fn test_test_server_guard_starts() {
777 let router = Router::new();
779
780 let server = test_server_guard(router).await;
782
783 assert!(
785 server.url.starts_with("http://127.0.0.1:"),
786 "Expected URL to start with 'http://127.0.0.1:', got: {}",
787 server.url
788 );
789 }
790
791 #[rstest]
792 #[tokio::test]
793 async fn test_test_server_builder_default() {
794 let handler: Arc<dyn Handler> = Arc::new(BasicHandler);
796
797 let result = TestServer::builder().handler(handler).build().await;
799
800 assert!(
802 result.is_ok(),
803 "Expected TestServer::builder().handler().build() to succeed"
804 );
805 }
806
807 #[rstest]
808 #[tokio::test]
809 async fn test_test_server_url_format() {
810 let handler: Arc<dyn Handler> = Arc::new(BasicHandler);
812
813 let server = TestServer::builder()
815 .handler(handler)
816 .build()
817 .await
818 .expect("Failed to build TestServer");
819
820 assert!(
822 server.url.starts_with("http://127.0.0.1:"),
823 "Expected URL format 'http://127.0.0.1:<port>', got: {}",
824 server.url
825 );
826 assert!(
827 server.addr.port() > 0,
828 "Expected non-zero port, got: {}",
829 server.addr.port()
830 );
831 }
832
833 #[rstest]
834 #[tokio::test]
835 async fn test_test_server_responds_to_request() {
836 let handler: Arc<dyn Handler> = Arc::new(BasicHandler);
838 let server = TestServer::builder()
839 .handler(handler)
840 .build()
841 .await
842 .expect("Failed to build TestServer");
843 let client = reqwest::Client::new();
844
845 let response = client.get(&server.url).send().await;
847
848 assert!(response.is_ok(), "Expected GET request to succeed");
850 let resp = response.unwrap();
851 assert_eq!(resp.status(), reqwest::StatusCode::OK);
852 }
853
854 #[rstest]
855 fn test_http_client_fixture() {
856 let client = http_client();
858
859 let _: &reqwest::Client = &client;
862 }
863
864 #[rstest]
865 #[tokio::test]
866 async fn test_test_server_shutdown_timeout() {
867 let handler: Arc<dyn Handler> = Arc::new(BasicHandler);
869 let custom_timeout = Duration::from_secs(10);
870
871 let result = TestServer::builder()
873 .handler(handler)
874 .shutdown_timeout(custom_timeout)
875 .build()
876 .await;
877
878 assert!(
880 result.is_ok(),
881 "Expected TestServer with custom shutdown timeout to build successfully"
882 );
883 }
884
885 #[rstest]
886 #[tokio::test]
887 async fn test_wait_for_server_ready() {
888 let listener = TcpListener::bind("127.0.0.1:0")
890 .await
891 .expect("Failed to bind listener");
892 let addr = listener.local_addr().expect("Failed to get local addr");
893
894 let result = wait_for_server_ready(addr).await;
896
897 assert!(
899 result.is_ok(),
900 "Expected wait_for_server_ready to succeed for a bound address"
901 );
902 }
903}