fix(tests) ci

binbat · Oct 6, 2024 · f6c453f · f6c453f
1 parent 18322d7
commit f6c453f
Show file tree

Hide file tree

Showing 7 changed files with 153 additions and 20 deletions.
diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml
@@ -30,6 +30,10 @@ jobs:
         run: cargo build --all --release --verbose
       - name: Setup ffmpeg
         uses: AnimMouse/setup-ffmpeg@v1
+      - name: Show version information
+        run: |
+          ffmpeg -version
+          ffprobe -version
       - name: Run tests
         run: cargo test --all --verbose
       - name: Run e2e tests

diff --git a/livetwo/src/lib.rs b/livetwo/src/lib.rs
@@ -8,6 +8,6 @@ mod rtspclient;
 mod test;
 
 const PREFIX_LIB: &str = "WEBRTC";
-const SCHEME_RTSP_SERVER: &str = "rtsp-listen";
-const SCHEME_RTSP_CLIENT: &str = "rtsp";
-const SCHEME_RTP_SDP: &str = "sdp";
+pub const SCHEME_RTSP_SERVER: &str = "rtsp-listen";
+pub const SCHEME_RTSP_CLIENT: &str = "rtsp";
+pub const SCHEME_RTP_SDP: &str = "sdp";
diff --git a/livetwo/src/whep.rs b/livetwo/src/whep.rs
@@ -1,6 +1,7 @@
+use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
+
 use anyhow::{anyhow, Result};
 use cli::create_child;
-use core::net::{Ipv4Addr, Ipv6Addr};
 use portpicker::pick_unused_port;
 use scopeguard::defer;
 use sdp::{description::media::RangedPort, SessionDescription};
@@ -50,14 +51,17 @@ pub async fn from(
         Url::parse(&format!(
             "{}://{}:0/{}",
             SCHEME_RTP_SDP,
-            Ipv4Addr::UNSPECIFIED,
+            IpAddr::V4(Ipv4Addr::UNSPECIFIED),
             target_url
         ))
         .unwrap(),
     );
     info!("=== Received Output: {} ===", target_url);
 
-    let mut host = match input.host().unwrap() {
+    let mut host = match input
+        .host()
+        .unwrap_or_else(|| panic!("Invalid host for {}", input))
+    {
         Host::Domain(_) | Host::Ipv4(_) => Ipv4Addr::UNSPECIFIED.to_string(),
         Host::Ipv6(_) => Ipv6Addr::UNSPECIFIED.to_string(),
     };

diff --git a/livetwo/src/whip.rs b/livetwo/src/whip.rs
@@ -1,8 +1,8 @@
 use std::fs;
+use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
 use std::{sync::Arc, time::Duration, vec};
 
 use anyhow::{anyhow, Result};
-use core::net::{Ipv4Addr, Ipv6Addr};
 use scopeguard::defer;
 use tokio::{
     net::{TcpListener, UdpSocket},
@@ -23,9 +23,7 @@ use webrtc::{
     rtp_transceiver::{
         rtp_codec::RTCRtpCodecCapability, rtp_codec::RTPCodecType, rtp_sender::RTCRtpSender,
     },
-    track::track_local::{
-        track_local_static_rtp::TrackLocalStaticRTP, TrackLocal, TrackLocalWriter,
-    },
+    track::track_local::{track_local_static_rtp::TrackLocalStaticRTP, TrackLocalWriter},
     util::Unmarshal,
 };
 
@@ -48,14 +46,17 @@ pub async fn into(
         Url::parse(&format!(
             "{}://{}:0/{}",
             SCHEME_RTP_SDP,
-            Ipv4Addr::UNSPECIFIED,
+            IpAddr::V4(Ipv4Addr::UNSPECIFIED),
             target_url
         ))
         .unwrap(),
     );
     info!("=== Received Input: {} ===", input);
 
-    let mut host = match input.host().unwrap() {
+    let mut host = match input
+        .host()
+        .unwrap_or_else(|| panic!("Invalid host for {}", input))
+    {
         Host::Domain(_) | Host::Ipv4(_) => Ipv4Addr::UNSPECIFIED.to_string(),
         Host::Ipv6(_) => Ipv6Addr::UNSPECIFIED.to_string(),
     };
@@ -442,7 +443,7 @@ async fn new_peer(
             input.to_owned(),
         ));
         let _ = peer
-            .add_track(video_track.clone() as Arc<dyn TrackLocal + Send + Sync>)
+            .add_track(video_track.clone())
             .await
             .map_err(|error| anyhow!(format!("{:?}: {}", error, error)))?;
 
@@ -480,7 +481,7 @@ async fn new_peer(
             input.to_owned(),
         ));
         let _ = peer
-            .add_track(audio_track.clone() as Arc<dyn TrackLocal + Send + Sync>)
+            .add_track(audio_track.clone())
             .await
             .map_err(|error| anyhow!(format!("{:?}: {}", error, error)))?;
 

diff --git a/src/whepfrom.rs b/src/whepfrom.rs
@@ -1,16 +1,14 @@
 use clap::{ArgAction, Parser};
 use tracing::Level;
 
-const SCHEME_RTSP_SERVER: &str = "rtsp-listen";
-
 #[derive(Parser)]
 #[command(author, version, about, long_about = None)]
 struct Args {
     /// Verbose mode [default: "warn", -v "info", -vv "debug", -vvv "trace"]
     #[arg(short = 'v', action = ArgAction::Count, default_value_t = 0)]
     verbose: u8,
     /// rtsp://[username]:[password]@[ip]:[port]/[stream] Or <stream.sdp>
-    #[arg(short, long, default_value_t = format!("{}://0.0.0.0:8555", SCHEME_RTSP_SERVER))]
+    #[arg(short, long, default_value_t = format!("{}://0.0.0.0:8555", livetwo::SCHEME_RTSP_SERVER))]
     output: String,
     /// Set Listener address
     #[arg(long)]

diff --git a/src/whipinto.rs b/src/whipinto.rs
@@ -1,16 +1,14 @@
 use clap::{ArgAction, Parser};
 use tracing::Level;
 
-const SCHEME_RTSP_SERVER: &str = "rtsp-listen";
-
 #[derive(Parser)]
 #[command(author, version, about, long_about = None)]
 struct Args {
     /// Verbose mode [default: "warn", -v "info", -vv "debug", -vvv "trace"]
     #[arg(short = 'v', action = ArgAction::Count, default_value_t = 0)]
     verbose: u8,
     /// rtsp://[username]:[password]@[ip]:[port]/[stream] Or <stream.sdp>
-    #[arg(short, long, default_value_t = format!("{}://0.0.0.0:8554", SCHEME_RTSP_SERVER))]
+    #[arg(short, long, default_value_t = format!("{}://0.0.0.0:8554", livetwo::SCHEME_RTSP_SERVER))]
     input: String,
     /// Set Listener address
     #[arg(long)]

diff --git a/tests/tests.rs b/tests/tests.rs
@@ -81,12 +81,140 @@ async fn test_liveion_stream_create() {
     assert_eq!(1, body.len());
 }
 
+#[cfg(not(windows))]
 #[tokio::test]
 async fn test_liveion_stream_connect() {
     let cfg = liveion::config::Config::default();
     let ip = IpAddr::V4(Ipv4Addr::LOCALHOST);
     let port = 0;
 
+    let listener = TcpListener::bind(SocketAddr::new(ip, port)).await.unwrap();
+    let addr = listener.local_addr().unwrap();
+
+    tokio::spawn(liveion::server_up(cfg, listener, shutdown_signal()));
+
+    let res = reqwest::Client::new()
+        .post(format!("http://{addr}{}", api::path::streams("-")))
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(http::StatusCode::NO_CONTENT, res.status());
+
+    let res = reqwest::get(format!("http://{addr}{}", api::path::streams("")))
+        .await
+        .unwrap();
+
+    let body = res.json::<Vec<api::response::Stream>>().await.unwrap();
+
+    assert_eq!(1, body.len());
+
+    let tmp_path = tempfile::tempdir()
+        .unwrap()
+        .path()
+        .to_str()
+        .unwrap()
+        .to_string();
+
+    use std::io::Write;
+
+    let mut file = std::fs::File::create(tmp_path.clone()).unwrap();
+    file.write_all(
+        r#"
+v=0
+o=- 0 0 IN IP4 127.0.0.1
+s=No Name
+c=IN IP4 127.0.0.1
+t=0 0
+a=tool:libavformat 61.1.100
+m=video 0 RTP/AVP 96
+b=AS:256
+a=rtpmap:96 VP8/90000
+    "#
+        .as_bytes(),
+    )
+    .unwrap();
+
+    tokio::spawn(livetwo::whip::into(
+        tmp_path.clone(),
+        None,
+        format!("http://{addr}{}", api::path::whip("-")),
+        None,
+        None,
+    ));
+
+    let mut result = None;
+    for _ in 0..100 {
+        let res = reqwest::get(format!("http://{addr}{}", api::path::streams("")))
+            .await
+            .unwrap();
+
+        assert_eq!(http::StatusCode::OK, res.status());
+
+        let body = res.json::<Vec<api::response::Stream>>().await.unwrap();
+
+        if let Some(r) = body.into_iter().find(|i| i.id == "-") {
+            if !r.publish.sessions.is_empty() {
+                let s = r.publish.sessions[0].clone();
+                if s.state == api::response::RTCPeerConnectionState::Connected {
+                    result = Some(s);
+                    break;
+                }
+            }
+        };
+
+        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
+    }
+
+    assert!(result.is_some());
+
+    let tmp_path = tempfile::tempdir()
+        .unwrap()
+        .path()
+        .to_str()
+        .unwrap()
+        .to_string();
+    tokio::spawn(livetwo::whep::from(
+        tmp_path.clone(),
+        None,
+        format!("http://{addr}{}", api::path::whep("-")),
+        None,
+        None,
+    ));
+
+    let mut result = None;
+    for _ in 0..100 {
+        let res = reqwest::get(format!("http://{addr}{}", api::path::streams("")))
+            .await
+            .unwrap();
+
+        assert_eq!(http::StatusCode::OK, res.status());
+
+        let body = res.json::<Vec<api::response::Stream>>().await.unwrap();
+
+        if let Some(r) = body.into_iter().find(|i| i.id == "-") {
+            if !r.subscribe.sessions.is_empty() {
+                let s = r.subscribe.sessions[0].clone();
+                if s.state == api::response::RTCPeerConnectionState::Connected {
+                    result = Some(s);
+                    break;
+                }
+            }
+        };
+
+        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
+    }
+
+    assert!(result.is_some());
+}
+
+#[cfg(not(windows))]
+#[tokio::test]
+async fn test_liveion_stream_ffmpeg() {
+    let cfg = liveion::config::Config::default();
+    let ip = IpAddr::V4(Ipv4Addr::LOCALHOST);
+    let port = 0;
+
     let width = 640;
     let height = 480;