dev: torrent repository tests

packages/torrent-repository/Cargo.toml

@@ -25,6 +25,7 @@ torrust-tracker-clock = { version = "3.0.0-alpha.12-develop", path = "../clock"
 [dev-dependencies]
 criterion = { version = "0", features = ["async_tokio"] }
 rstest = "0"
+async-std = {version = "1", features = ["attributes", "tokio1"] }
 
 [[bench]]
 harness = false

packages/torrent-repository/tests/common/mod.rs

@@ -0,0 +1,3 @@
+pub mod repo;
+pub mod torrent;
+pub mod torrent_peer_builder;

packages/torrent-repository/tests/common/repo.rs

@@ -0,0 +1,147 @@
+use torrust_tracker_configuration::TrackerPolicy;
+use torrust_tracker_primitives::info_hash::InfoHash;
+use torrust_tracker_primitives::pagination::Pagination;
+use torrust_tracker_primitives::swarm_metadata::SwarmMetadata;
+use torrust_tracker_primitives::torrent_metrics::TorrentsMetrics;
+use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch, PersistentTorrents};
+use torrust_tracker_torrent_repository::repository::{Repository as _, RepositoryAsync as _};
+use torrust_tracker_torrent_repository::{
+ EntrySingle, TorrentsRwLockStd, TorrentsRwLockStdMutexStd, TorrentsRwLockStdMutexTokio, TorrentsRwLockTokio,
+ TorrentsRwLockTokioMutexStd, TorrentsRwLockTokioMutexTokio,
+};
+
+#[derive(Debug)]
+pub(crate) enum Repo {
+ Std(TorrentsRwLockStd),
+ StdMutexStd(TorrentsRwLockStdMutexStd),
+ StdMutexTokio(TorrentsRwLockStdMutexTokio),
+ Tokio(TorrentsRwLockTokio),
+ TokioMutexStd(TorrentsRwLockTokioMutexStd),
+ TokioMutexTokio(TorrentsRwLockTokioMutexTokio),
+}
+
+impl Repo {
+ pub(crate) async fn get(&self, key: &InfoHash) -> Option<EntrySingle> {
+ match self {
+ Repo::Std(repo) => repo.get(key),
+ Repo::StdMutexStd(repo) => Some(repo.get(key)?.lock().unwrap().clone()),
+ Repo::StdMutexTokio(repo) => Some(repo.get(key).await?.lock().await.clone()),
+ Repo::Tokio(repo) => repo.get(key).await,
+ Repo::TokioMutexStd(repo) => Some(repo.get(key).await?.lock().unwrap().clone()),
+ Repo::TokioMutexTokio(repo) => Some(repo.get(key).await?.lock().await.clone()),
+ }
+ }
+ pub(crate) async fn get_metrics(&self) -> TorrentsMetrics {
+ match self {
+ Repo::Std(repo) => repo.get_metrics(),
+ Repo::StdMutexStd(repo) => repo.get_metrics(),
+ Repo::StdMutexTokio(repo) => repo.get_metrics().await,
+ Repo::Tokio(repo) => repo.get_metrics().await,
+ Repo::TokioMutexStd(repo) => repo.get_metrics().await,
+ Repo::TokioMutexTokio(repo) => repo.get_metrics().await,
+ }
+ }
+ pub(crate) async fn get_paginated(&self, pagination: Option<&Pagination>) -> Vec<(InfoHash, EntrySingle)> {
+ match self {
+ Repo::Std(repo) => repo.get_paginated(pagination),
+ Repo::StdMutexStd(repo) => repo
+ .get_paginated(pagination)
+ .iter()
+ .map(|(i, t)| (*i, t.lock().expect("it should get a lock").clone()))
+ .collect(),
+ Repo::StdMutexTokio(repo) => {
+ let mut v: Vec<(InfoHash, EntrySingle)> = vec![];
+
+ for (i, t) in repo.get_paginated(pagination).await {
+ v.push((i, t.lock().await.clone()));
+ }
+ v
+ }
+ Repo::Tokio(repo) => repo.get_paginated(pagination).await,
+ Repo::TokioMutexStd(repo) => repo
+ .get_paginated(pagination)
+ .await
+ .iter()
+ .map(|(i, t)| (*i, t.lock().expect("it should get a lock").clone()))
+ .collect(),
+ Repo::TokioMutexTokio(repo) => {
+ let mut v: Vec<(InfoHash, EntrySingle)> = vec![];
+
+ for (i, t) in repo.get_paginated(pagination).await {
+ v.push((i, t.lock().await.clone()));
+ }
+ v
+ }
+ }
+ }
+ pub(crate) async fn import_persistent(&self, persistent_torrents: &PersistentTorrents) {
+ match self {
+ Repo::Std(repo) => repo.import_persistent(persistent_torrents),
+ Repo::StdMutexStd(repo) => repo.import_persistent(persistent_torrents),
+ Repo::StdMutexTokio(repo) => repo.import_persistent(persistent_torrents).await,
+ Repo::Tokio(repo) => repo.import_persistent(persistent_torrents).await,
+ Repo::TokioMutexStd(repo) => repo.import_persistent(persistent_torrents).await,
+ Repo::TokioMutexTokio(repo) => repo.import_persistent(persistent_torrents).await,
+ }
+ }
+ pub(crate) async fn remove(&self, key: &InfoHash) -> Option<EntrySingle> {
+ match self {
+ Repo::Std(repo) => repo.remove(key),
+ Repo::StdMutexStd(repo) => Some(repo.remove(key)?.lock().unwrap().clone()),
+ Repo::StdMutexTokio(repo) => Some(repo.remove(key).await?.lock().await.clone()),
+ Repo::Tokio(repo) => repo.remove(key).await,
+ Repo::TokioMutexStd(repo) => Some(repo.remove(key).await?.lock().unwrap().clone()),
+ Repo::TokioMutexTokio(repo) => Some(repo.remove(key).await?.lock().await.clone()),
+ }
+ }
+ pub(crate) async fn remove_inactive_peers(&self, current_cutoff: DurationSinceUnixEpoch) {
+ match self {
+ Repo::Std(repo) => repo.remove_inactive_peers(current_cutoff),
+ Repo::StdMutexStd(repo) => repo.remove_inactive_peers(current_cutoff),
+ Repo::StdMutexTokio(repo) => repo.remove_inactive_peers(current_cutoff).await,
+ Repo::Tokio(repo) => repo.remove_inactive_peers(current_cutoff).await,
+ Repo::TokioMutexStd(repo) => repo.remove_inactive_peers(current_cutoff).await,
+ Repo::TokioMutexTokio(repo) => repo.remove_inactive_peers(current_cutoff).await,
+ }
+ }
+ pub(crate) async fn remove_peerless_torrents(&self, policy: &TrackerPolicy) {
+ match self {
+ Repo::Std(repo) => repo.remove_peerless_torrents(policy),
+ Repo::StdMutexStd(repo) => repo.remove_peerless_torrents(policy),
+ Repo::StdMutexTokio(repo) => repo.remove_peerless_torrents(policy).await,
+ Repo::Tokio(repo) => repo.remove_peerless_torrents(policy).await,
+ Repo::TokioMutexStd(repo) => repo.remove_peerless_torrents(policy).await,
+ Repo::TokioMutexTokio(repo) => repo.remove_peerless_torrents(policy).await,
+ }
+ }
+ pub(crate) async fn update_torrent_with_peer_and_get_stats(
+ &self,
+ info_hash: &InfoHash,
+ peer: &peer::Peer,
+ ) -> (bool, SwarmMetadata) {
+ match self {
+ Repo::Std(repo) => repo.update_torrent_with_peer_and_get_stats(info_hash, peer),
+ Repo::StdMutexStd(repo) => repo.update_torrent_with_peer_and_get_stats(info_hash, peer),
+ Repo::StdMutexTokio(repo) => repo.update_torrent_with_peer_and_get_stats(info_hash, peer).await,
+ Repo::Tokio(repo) => repo.update_torrent_with_peer_and_get_stats(info_hash, peer).await,
+ Repo::TokioMutexStd(repo) => repo.update_torrent_with_peer_and_get_stats(info_hash, peer).await,
+ Repo::TokioMutexTokio(repo) => repo.update_torrent_with_peer_and_get_stats(info_hash, peer).await,
+ }
+ }
+ pub(crate) async fn insert(&self, info_hash: &InfoHash, torrent: EntrySingle) -> Option<EntrySingle> {
+ match self {
+ Repo::Std(repo) => repo.write().insert(*info_hash, torrent),
+ Repo::StdMutexStd(repo) => Some(repo.write().insert(*info_hash, torrent.into())?.lock().unwrap().clone()),
+ Repo::StdMutexTokio(repo) => {
+ let r = repo.write().insert(*info_hash, torrent.into());
+ match r {
+ Some(t) => Some(t.lock().await.clone()),
+ None => None,
+ }
+ }
+ Repo::Tokio(repo) => repo.write().await.insert(*info_hash, torrent),
+ Repo::TokioMutexStd(repo) => Some(repo.write().await.insert(*info_hash, torrent.into())?.lock().unwrap().clone()),
+ Repo::TokioMutexTokio(repo) => Some(repo.write().await.insert(*info_hash, torrent.into())?.lock().await.clone()),
+ }
+ }
+}

packages/torrent-repository/tests/common/torrent.rs

@@ -0,0 +1,89 @@
+use std::net::SocketAddr;
+use std::sync::Arc;
+
+use torrust_tracker_configuration::TrackerPolicy;
+use torrust_tracker_primitives::swarm_metadata::SwarmMetadata;
+use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch};
+use torrust_tracker_torrent_repository::entry::{Entry as _, EntryAsync as _, EntrySync as _};
+use torrust_tracker_torrent_repository::{EntryMutexStd, EntryMutexTokio, EntrySingle};
+
+#[derive(Debug, Clone)]
+pub(crate) enum Torrent {
+ Single(EntrySingle),
+ MutexStd(EntryMutexStd),
+ MutexTokio(EntryMutexTokio),
+}
+
+impl Torrent {
+ pub(crate) async fn get_stats(&self) -> SwarmMetadata {
+ match self {
+ Torrent::Single(entry) => entry.get_stats(),
+ Torrent::MutexStd(entry) => entry.get_stats(),
+ Torrent::MutexTokio(entry) => entry.clone().get_stats().await,
+ }
+ }
+
+ pub(crate) async fn is_good(&self, policy: &TrackerPolicy) -> bool {
+ match self {
+ Torrent::Single(entry) => entry.is_good(policy),
+ Torrent::MutexStd(entry) => entry.is_good(policy),
+ Torrent::MutexTokio(entry) => entry.clone().check_good(policy).await,
+ }
+ }
+
+ pub(crate) async fn peers_is_empty(&self) -> bool {
+ match self {
+ Torrent::Single(entry) => entry.peers_is_empty(),
+ Torrent::MutexStd(entry) => entry.peers_is_empty(),
+ Torrent::MutexTokio(entry) => entry.clone().peers_is_empty().await,
+ }
+ }
+
+ pub(crate) async fn get_peers_len(&self) -> usize {
+ match self {
+ Torrent::Single(entry) => entry.get_peers_len(),
+ Torrent::MutexStd(entry) => entry.get_peers_len(),
+ Torrent::MutexTokio(entry) => entry.clone().get_peers_len().await,
+ }
+ }
+
+ pub(crate) async fn get_peers(&self, limit: Option<usize>) -> Vec<Arc<peer::Peer>> {
+ match self {
+ Torrent::Single(entry) => entry.get_peers(limit),
+ Torrent::MutexStd(entry) => entry.get_peers(limit),
+ Torrent::MutexTokio(entry) => entry.clone().get_peers(limit).await,
+ }
+ }
+
+ pub(crate) async fn get_peers_for_client(&self, client: &SocketAddr, limit: Option<usize>) -> Vec<Arc<peer::Peer>> {
+ match self {
+ Torrent::Single(entry) => entry.get_peers_for_client(client, limit),
+ Torrent::MutexStd(entry) => entry.get_peers_for_client(client, limit),
+ Torrent::MutexTokio(entry) => entry.clone().get_peers_for_client(client, limit).await,
+ }
+ }
+
+ pub(crate) async fn insert_or_update_peer(&mut self, peer: &peer::Peer) -> bool {
+ match self {
+ Torrent::Single(entry) => entry.insert_or_update_peer(peer),
+ Torrent::MutexStd(entry) => entry.insert_or_update_peer(peer),
+ Torrent::MutexTokio(entry) => entry.clone().insert_or_update_peer(peer).await,
+ }
+ }
+
+ pub(crate) async fn insert_or_update_peer_and_get_stats(&mut self, peer: &peer::Peer) -> (bool, SwarmMetadata) {
+ match self {
+ Torrent::Single(entry) => entry.insert_or_update_peer_and_get_stats(peer),
+ Torrent::MutexStd(entry) => entry.insert_or_update_peer_and_get_stats(peer),
+ Torrent::MutexTokio(entry) => entry.clone().insert_or_update_peer_and_get_stats(peer).await,
+ }
+ }
+
+ pub(crate) async fn remove_inactive_peers(&mut self, current_cutoff: DurationSinceUnixEpoch) {
+ match self {
+ Torrent::Single(entry) => entry.remove_inactive_peers(current_cutoff),
+ Torrent::MutexStd(entry) => entry.remove_inactive_peers(current_cutoff),
+ Torrent::MutexTokio(entry) => entry.clone().remove_inactive_peers(current_cutoff).await,
+ }
+ }
+}

packages/torrent-repository/tests/common/torrent_peer_builder.rs

@@ -0,0 +1,88 @@
+use std::net::SocketAddr;
+
+use torrust_tracker_clock::clock::Time;
+use torrust_tracker_primitives::announce_event::AnnounceEvent;
+use torrust_tracker_primitives::{peer, DurationSinceUnixEpoch, NumberOfBytes};
+
+use crate::CurrentClock;
+
+#[derive(Debug, Default)]
+struct TorrentPeerBuilder {
+ peer: peer::Peer,
+}
+
+#[allow(dead_code)]
+impl TorrentPeerBuilder {
+ #[must_use]
+ fn new() -> Self {
+ Self {
+ peer: peer::Peer {
+ updated: CurrentClock::now(),
+ ..Default::default()
+ },
+ }
+ }
+
+ #[must_use]
+ fn with_event_completed(mut self) -> Self {
+ self.peer.event = AnnounceEvent::Completed;
+ self
+ }
+
+ #[must_use]
+ fn with_event_started(mut self) -> Self {
+ self.peer.event = AnnounceEvent::Started;
+ self
+ }
+
+ #[must_use]
+ fn with_peer_address(mut self, peer_addr: SocketAddr) -> Self {
+ self.peer.peer_addr = peer_addr;
+ self
+ }
+
+ #[must_use]
+ fn with_peer_id(mut self, peer_id: peer::Id) -> Self {
+ self.peer.peer_id = peer_id;
+ self
+ }
+
+ #[must_use]
+ fn with_number_of_bytes_left(mut self, left: i64) -> Self {
+ self.peer.left = NumberOfBytes(left);
+ self
+ }
+
+ #[must_use]
+ fn updated_at(mut self, updated: DurationSinceUnixEpoch) -> Self {
+ self.peer.updated = updated;
+ self
+ }
+
+ #[must_use]
+ fn into(self) -> peer::Peer {
+ self.peer
+ }
+}
+
+/// A torrent seeder is a peer with 0 bytes left to download which
+/// has not announced it has stopped
+#[must_use]
+pub fn a_completed_peer(id: i32) -> peer::Peer {
+ TorrentPeerBuilder::new()
+ .with_number_of_bytes_left(0)
+ .with_event_completed()
+ .with_peer_id(id.into())
+ .into()
+}
+
+/// A torrent leecher is a peer that is not a seeder.
+/// Leecher: left > 0 OR event = Stopped
+#[must_use]
+pub fn a_started_peer(id: i32) -> peer::Peer {
+ TorrentPeerBuilder::new()
+ .with_number_of_bytes_left(1)
+ .with_event_started()
+ .with_peer_id(id.into())
+ .into()
+}