test: run integration test

Author: ymgyt

Cargo.lock

@@ -21,6 +21,7 @@ clap           = { version = "4.4" }
 feed-rs        = { version = "1.4" }
 futures-util   = { version = "0.3.30" }
 graphql_client = { version = "0.13.0", default-features = false }
+headers        = { version = "0.4.0" }
 http           = { version = "0.2" }                              # request use 0.2
 kvsd           = { version = "0.1.2", default-features = false }
 # kvsd                               = { path = "../kvsd" }

Cargo.toml

@@ -24,6 +24,7 @@ clap               = { workspace = true, features = ["derive"] }
 feed-rs            = { workspace = true }
 futures-util       = { workspace = true }
 graphql_client     = { workspace = true }
+headers            = { workspace = true }
 kvsd               = { workspace = true }
 moka               = { workspace = true, features = ["future"] }
 pin-project        = "1.1.4"

crates/synd_api/Cargo.toml

@@ -9,4 +9,5 @@ pub mod principal;
 pub mod repository;
 pub mod serve;
 pub mod service;
+pub mod shutdown;
 pub mod usecase;

crates/synd_api/src/lib.rs

@@ -1,7 +1,7 @@
 use synd_o11y::{opentelemetry::OpenTelemetryGuard, tracing_subscriber::otel_metrics};
 use tracing::{error, info};
 
-use synd_api::{args, config, dependency::Dependency, serve::listen_and_serve};
+use synd_api::{args, config, dependency::Dependency, serve::listen_and_serve, shutdown::Shutdown};
 
 fn init_tracing() -> Option<OpenTelemetryGuard> {
     use synd_o11y::{
@@ -72,10 +72,11 @@ async fn main() {
     let _guard = init_tracing();
 
     let dep = Dependency::new(args.kvsd).await.unwrap();
+    let shutdown = Shutdown::watch_signal();
 
     info!(version = config::VERSION, "Runinng...");
 
-    if let Err(err) = listen_and_serve(dep).await {
+    if let Err(err) = listen_and_serve(dep, shutdown.notify()).await {
         error!("{err:?}");
         std::process::exit(1);
     }

crates/synd_api/src/main.rs

@@ -1,4 +1,5 @@
 use async_trait::async_trait;
+use futures_util::TryFutureExt;
 use kvsd::{
     client::{tcp::Client, Api},
     Key, Value,
@@ -17,20 +18,23 @@ pub struct KvsdClient {
 }
 
 impl KvsdClient {
+    pub fn new(client: Client<TcpStream>) -> Self {
+        Self {
+            client: Mutex::new(client),
+        }
+    }
+
     pub async fn connect(
         host: impl AsRef<str>,
         port: u16,
         username: String,
         password: String,
     ) -> anyhow::Result<Self> {
-        let client =
-            kvsd::client::tcp::UnauthenticatedClient::insecure_from_addr(host, port).await?;
-
-        let client = client.authenticate(username, password).await?;
-
-        Ok(Self {
-            client: Mutex::new(client),
-        })
+        kvsd::client::tcp::UnauthenticatedClient::insecure_from_addr(host, port)
+            .and_then(|client| client.authenticate(username, password))
+            .await
+            .map(Self::new)
+            .map_err(Into::into)
     }
 
     async fn get<'a, T>(

crates/synd_api/src/repository/kvsd.rs

@@ -22,8 +22,6 @@ impl<B> tower_http::trace::MakeSpan<B> for MakeSpan {
             %request_id,
         );
 
-        tracing::info!("Request headers: {:#?}", request.headers());
-
         span.set_parent(cx);
         span
     }

crates/synd_api/src/serve/layer/trace.rs

@@ -6,6 +6,7 @@ use axum::{
     routing::{get, post},
     BoxError, Extension, Router,
 };
+use futures_util::Future;
 use tokio::net::TcpListener;
 use tower::{limit::ConcurrencyLimitLayer, timeout::TimeoutLayer, ServiceBuilder};
 use tower_http::{
@@ -26,18 +27,24 @@ mod probe;
 pub mod layer;
 
 /// Bind tcp listener and serve.
-pub async fn listen_and_serve(dep: Dependency) -> anyhow::Result<()> {
+pub async fn listen_and_serve<Fut>(dep: Dependency, shutdown: Fut) -> anyhow::Result<()>
+where
+    Fut: Future<Output = ()> + Send + 'static,
+{
     // should 127.0.0.1?
     let addr = ("0.0.0.0", config::PORT);
     let listener = TcpListener::bind(addr).await?;
 
     info!(ip = addr.0, port = addr.1, "Listening...");
 
-    serve(listener, dep).await
+    serve(listener, dep, shutdown).await
 }
 
 /// Start api server
-pub async fn serve(listener: TcpListener, dep: Dependency) -> anyhow::Result<()> {
+pub async fn serve<Fut>(listener: TcpListener, dep: Dependency, shutdown: Fut) -> anyhow::Result<()>
+where
+    Fut: Future<Output = ()> + Send + 'static,
+{
     let Dependency {
         authenticator,
         runtime,
@@ -64,8 +71,12 @@ pub async fn serve(listener: TcpListener, dep: Dependency) -> anyhow::Result<()>
         )
         .route("/healthcheck", get(probe::healthcheck));
 
-    // TODO: graceful shutdown
-    axum::serve(listener, service).await?;
+    axum::serve(listener, service)
+        .with_graceful_shutdown(shutdown)
+        .await?;
+
+    tracing::info!("Shutdown complete");
+
     Ok(())
 }
 

crates/synd_api/src/serve/mod.rs

@@ -0,0 +1,35 @@
+use tokio::sync::broadcast::{self, Receiver, Sender};
+
+pub struct Shutdown {
+    tx: Sender<()>,
+    rx: Receiver<()>,
+}
+
+impl Shutdown {
+    pub fn watch_signal() -> Self {
+        let (tx, rx) = broadcast::channel(2);
+
+        let tx2 = tx.clone();
+        tokio::spawn(async move {
+            match tokio::signal::ctrl_c().await {
+                Ok(()) => tracing::info!("Received ctrl-c signal"),
+                Err(err) => tracing::error!("Failed to handle signal {err}"),
+            }
+            tx2.send(()).ok();
+        });
+
+        Self { tx, rx }
+    }
+
+    pub async fn notify(mut self) {
+        self.rx.recv().await.ok();
+    }
+}
+
+impl Clone for Shutdown {
+    fn clone(&self) -> Self {
+        let rx = self.tx.subscribe();
+        let tx = self.tx.clone();
+        Self { tx, rx }
+    }
+}

crates/synd_api/src/shutdown.rs

@@ -119,8 +119,6 @@ impl DeviceFlow {
                 .send()
                 .await?;
 
-            debug!("{:?}", response.status());
-
             match response.status() {
                 StatusCode::OK => {
                     let full = response.bytes().await?;

crates/synd_authn/src/device_flow/github.rs

@@ -10,7 +10,9 @@ mod test {
         style::{Modifier, Style},
     };
     use serial_test::file_serial;
+
     use synd_authn::device_flow::github::DeviceFlow;
+
     use synd_term::{
         application::{Application, Config},
         client::Client,
@@ -33,14 +35,16 @@ mod test {
 
         tracing::info!("TEST hello_world run");
 
-        let _kvsd_client = helper::run_kvsd().await?;
-
         let mock_port = 6000;
+        let api_port = 6001;
         let oauth_addr = ("127.0.0.1", mock_port);
         let oauth_listener = TcpListener::bind(oauth_addr).await?;
-        tokio::spawn(synd_test::oauth::serve(oauth_listener));
+        tokio::spawn(synd_test::mock::serve(oauth_listener));
+        helper::serve_api(mock_port, api_port).await?;
 
-        let endpoint = "http://localhost:5961/graphql".parse().unwrap();
+        let endpoint = format!("http://localhost:{api_port}/graphql")
+            .parse()
+            .unwrap();
         let terminal = helper::new_test_terminal();
         let client = Client::new(endpoint).unwrap();
         let config = Config {

crates/synd_term/tests/integration.rs

@@ -1,8 +1,15 @@
-use std::future::pending;
+use std::{future::pending, sync::Arc, time::Duration};
 
-use kvsd::client::Api;
+use futures_util::TryFutureExt;
 use ratatui::backend::TestBackend;
-use synd_api::{client::github::GithubClient, serve::auth::Authenticator};
+use synd_api::{
+    client::github::GithubClient,
+    dependency::Dependency,
+    repository::kvsd::KvsdClient,
+    serve::auth::Authenticator,
+    usecase::{authorize::Authorizer, MakeUsecase, Runtime},
+};
+use synd_feed::feed::{cache::CacheLayer, parser::FeedService};
 use synd_term::terminal::Terminal;
 use tokio::net::{TcpListener, TcpStream};
 
@@ -12,25 +19,33 @@ pub fn new_test_terminal() -> Terminal {
     Terminal::with(terminal)
 }
 
-// Dependency
-// * serve
-//   * tcp listener
-//   * dependency
-//     * authenticator
-//       * github client
-//         * github endpoint
-//     * usecase runtime
-//       * make usecase
-//         * datastore
-//         * fetch cached feed
-//       * authorizer
-#[allow(unused)]
-pub fn serve_api(mock_port: u16) -> anyhow::Result<()> {
+pub async fn serve_api(mock_port: u16, api_port: u16) -> anyhow::Result<()> {
     let github_endpoint: &'static str =
         format!("http://localhost:{mock_port}/github/graphql").leak();
     let github_client = GithubClient::new()?.with_endpoint(github_endpoint);
     let authenticator = Authenticator::new()?.with_client(github_client);
 
+    let kvsd_client = run_kvsd().await.map(KvsdClient::new)?;
+    let feed_service = FeedService::new("synd_term_test", 1024 * 1024);
+    let feed_service = CacheLayer::new(feed_service);
+    let make_usecase = MakeUsecase {
+        subscription_repo: Arc::new(kvsd_client),
+        fetch_feed: Arc::new(feed_service),
+    };
+    let authorizer = Authorizer::new();
+    let runtime = Runtime::new(make_usecase, authorizer);
+    let dep = Dependency {
+        authenticator,
+        runtime,
+    };
+    let listener = TcpListener::bind(("localhost", api_port)).await?;
+
+    tokio::spawn(synd_api::serve::serve(
+        listener,
+        dep,
+        std::future::pending::<()>(),
+    ));
+
     Ok(())
 }
 
@@ -57,17 +72,19 @@ pub async fn run_kvsd() -> anyhow::Result<kvsd::client::tcp::Client<TcpStream>>
 
     let _server_handler = tokio::spawn(initializer.run_kvsd(pending::<()>()));
 
-    // TODO: retry
-    let mut client = kvsd::client::tcp::UnauthenticatedClient::insecure_from_addr(addr.0, addr.1)
-        .await
-        .unwrap()
-        .authenticate("test", "test")
-        .await
-        .unwrap();
-
-    // Ping
-    let ping_duration = client.ping().await.unwrap();
-    assert!(ping_duration.num_nanoseconds().unwrap() > 0);
+    let handshake = async {
+        loop {
+            match kvsd::client::tcp::UnauthenticatedClient::insecure_from_addr(addr.0, addr.1)
+                .and_then(|client| client.authenticate("test", "test"))
+                .await
+            {
+                Ok(client) => break client,
+                Err(_) => tokio::time::sleep(Duration::from_millis(500)).await,
+            }
+        }
+    };
+
+    let client = tokio::time::timeout(Duration::from_secs(5), handshake).await?;
 
     Ok(client)
 }

crates/synd_term/tests/test/helper.rs

@@ -14,10 +14,12 @@ version = "0.1.0"
 [dependencies]
 synd_authn = { path = "../synd_authn" }
 
-anyhow  = { workspace = true }
-axum    = { workspace = true }
-tokio   = { workspace = true, features = ["rt-multi-thread", "net", "time"] }
-tracing = { workspace = true }
+anyhow     = { workspace = true }
+axum       = { workspace = true }
+headers    = { workspace = true }
+serde_json = { workspace = true }
+tokio      = { workspace = true, features = ["rt-multi-thread", "net", "time"] }
+tracing    = { workspace = true }
 
 [lints]
 workspace = true

crates/synd_test/Cargo.toml

@@ -1 +1 @@
-pub mod oauth;
+pub mod mock;