dev: finish repo tests

packages/primitives/src/info_hash.rs

@@ -89,6 +89,13 @@ impl std::convert::From<&DefaultHasher> for InfoHash {
  }
 }
 
+impl std::convert::From<&i32> for InfoHash {
+ fn from(n: &i32) -> InfoHash {
+ let n = n.to_le_bytes();
+ InfoHash([0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, n[0], n[1], n[2], n[3]])
+ }
+}
+
 impl std::convert::From<[u8; 20]> for InfoHash {
  fn from(val: [u8; 20]) -> Self {
  InfoHash(val)

packages/primitives/src/pagination.rs

@@ -1,7 +1,8 @@
+use derive_more::Constructor;
 use serde::Deserialize;
 
 /// A struct to keep information about the page when results are being paginated
-#[derive(Deserialize, Copy, Clone, Debug, PartialEq)]
+#[derive(Deserialize, Copy, Clone, Debug, PartialEq, Constructor)]
 pub struct Pagination {
  /// The page number, starting at 0
  pub offset: u32,
@@ -10,11 +11,6 @@ pub struct Pagination {
 }
 
 impl Pagination {
- #[must_use]
- pub fn new(offset: u32, limit: u32) -> Self {
- Self { offset, limit }
- }
-
  #[must_use]
  pub fn new_with_options(offset_option: Option<u32>, limit_option: Option<u32>) -> Self {
  let offset = match offset_option {

packages/torrent-repository/src/entry/mod.rs

@@ -65,7 +65,7 @@ pub trait EntrySync {
 #[allow(clippy::module_name_repetitions)]
 pub trait EntryAsync {
  fn get_stats(&self) -> impl std::future::Future<Output = SwarmMetadata> + Send;
- fn is_good(&self, policy: &TrackerPolicy) -> impl std::future::Future<Output = bool> + Send;
+ fn check_good(self, policy: &TrackerPolicy) -> impl std::future::Future<Output = bool> + Send;
  fn peers_is_empty(&self) -> impl std::future::Future<Output = bool> + Send;
  fn get_peers_len(&self) -> impl std::future::Future<Output = usize> + Send;
  fn get_peers(&self, limit: Option<usize>) -> impl std::future::Future<Output = Vec<Arc<peer::Peer>>> + Send;

packages/torrent-repository/src/entry/mutex_tokio.rs

@@ -13,7 +13,7 @@ impl EntryAsync for EntryMutexTokio {
  self.lock().await.get_stats()
  }
 
- async fn is_good(&self, policy: &TrackerPolicy) -> bool {
+ async fn check_good(self, policy: &TrackerPolicy) -> bool {
  self.lock().await.is_good(policy)
  }
 

packages/torrent-repository/src/repository/rw_lock_std_mutex_tokio.rs

@@ -1,4 +1,5 @@
 use std::collections::BTreeMap;
+use std::iter::zip;
 use std::pin::Pin;
 use std::sync::Arc;
 
@@ -124,8 +125,27 @@ where
  }
 
  async fn remove_peerless_torrents(&self, policy: &TrackerPolicy) {
+ let handles: Vec<Pin<Box<dyn Future<Output = Option<InfoHash>> + Send>>>;
+
+ {
+ let db = self.get_torrents();
+
+ handles = zip(db.keys().copied(), db.values().cloned())
+ .map(|(infohash, torrent)| {
+ torrent
+ .check_good(policy)
+ .map(move |good| if good { None } else { Some(infohash) })
+ .boxed()
+ })
+ .collect::<Vec<_>>();
+ }
+
+ let not_good = join_all(handles).await;
+
  let mut db = self.get_torrents_mut();
 
- db.retain(|_, e| e.blocking_lock().is_good(policy));
+ for remove in not_good.into_iter().flatten() {
+ drop(db.remove(&remove));
+ }
  }
 }

packages/torrent-repository/src/repository/rw_lock_tokio_mutex_tokio.rs

@@ -119,6 +119,16 @@ where
  async fn remove_peerless_torrents(&self, policy: &TrackerPolicy) {
  let mut db = self.get_torrents_mut().await;
 
- db.retain(|_, e| e.blocking_lock().is_good(policy));
+ let mut not_good = Vec::<InfoHash>::default();
+
+ for (&infohash, torrent) in db.iter() {
+ if !torrent.clone().check_good(policy).await {
+ not_good.push(infohash);
+ }
+ }
+
+ for remove in not_good {
+ drop(db.remove(&remove));
+ }
  }
 }

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

@@ -20,7 +20,6 @@ pub(crate) enum Repo {
  TokioMutexTokio(TorrentsRwLockTokioMutexTokio),
 }
 
-#[allow(dead_code)]
 impl Repo {
  pub(crate) async fn get(&self, key: &InfoHash) -> Option<EntrySingle> {
  match self {

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

@@ -27,7 +27,7 @@ impl Torrent {
  match self {
  Torrent::Single(entry) => entry.is_good(policy),
  Torrent::MutexStd(entry) => entry.is_good(policy),
- Torrent::MutexTokio(entry) => entry.clone().is_good(policy).await,
+ Torrent::MutexTokio(entry) => entry.clone().check_good(policy).await,
  }
  }
 

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

@@ -1,8 +1,11 @@
+use std::collections::{BTreeMap, HashSet};
 use std::hash::{DefaultHasher, Hash, Hasher};
 
 use rstest::{fixture, rstest};
+use torrust_tracker_configuration::TrackerPolicy;
 use torrust_tracker_primitives::announce_event::AnnounceEvent;
 use torrust_tracker_primitives::info_hash::InfoHash;
+use torrust_tracker_primitives::pagination::Pagination;
 use torrust_tracker_primitives::{NumberOfBytes, PersistentTorrents};
 use torrust_tracker_torrent_repository::entry::Entry as _;
 use torrust_tracker_torrent_repository::repository::{RwLockStd, RwLockTokio};
@@ -104,6 +107,39 @@ fn three() -> Entries {
  ]
 }
 
+#[fixture]
+fn many_out_of_order() -> Entries {
+ let mut entries: HashSet<(InfoHash, EntrySingle)> = HashSet::default();
+
+ for i in 0..408 {
+ let mut entry = EntrySingle::default();
+ entry.insert_or_update_peer(&a_started_peer(i));
+
+ entries.insert((InfoHash::from(&i), entry));
+ }
+
+ // we keep the random order from the hashed set for the vector.
+ entries.iter().map(|(i, e)| (*i, e.clone())).collect()
+}
+
+#[fixture]
+fn many_hashed_in_order() -> Entries {
+ let mut entries: BTreeMap<InfoHash, EntrySingle> = BTreeMap::default();
+
+ for i in 0..408 {
+ let mut entry = EntrySingle::default();
+ entry.insert_or_update_peer(&a_started_peer(i));
+
+ let hash: &mut DefaultHasher = &mut DefaultHasher::default();
+ hash.write_i32(i);
+
+ entries.insert(InfoHash::from(&hash.clone()), entry);
+ }
+
+ // We return the entries in-order from from the b-tree map.
+ entries.iter().map(|(i, e)| (*i, e.clone())).collect()
+}
+
 #[fixture]
 fn persistent_empty() -> PersistentTorrents {
  PersistentTorrents::default()
@@ -141,13 +177,50 @@ async fn make(repo: &Repo, entries: &Entries) {
  }
 }
 
+#[fixture]
+fn paginated_limit_zero() -> Pagination {
+ Pagination::new(0, 0)
+}
+
+#[fixture]
+fn paginated_limit_one() -> Pagination {
+ Pagination::new(0, 1)
+}
+
+#[fixture]
+fn paginated_limit_one_offset_one() -> Pagination {
+ Pagination::new(1, 1)
+}
+
+#[fixture]
+fn policy_none() -> TrackerPolicy {
+ TrackerPolicy::new(false, 0, false)
+}
+
+#[fixture]
+fn policy_persist() -> TrackerPolicy {
+ TrackerPolicy::new(false, 0, true)
+}
+
+#[fixture]
+fn policy_remove() -> TrackerPolicy {
+ TrackerPolicy::new(true, 0, false)
+}
+
+#[fixture]
+fn policy_remove_persist() -> TrackerPolicy {
+ TrackerPolicy::new(true, 0, true)
+}
+
 #[rstest]
 #[case::empty(empty())]
 #[case::default(default())]
 #[case::started(started())]
 #[case::completed(completed())]
 #[case::downloaded(downloaded())]
 #[case::three(three())]
+#[case::out_of_order(many_out_of_order())]
+#[case::in_order(many_hashed_in_order())]
 #[tokio::test]
 async fn it_should_get_a_torrent_entry(
  #[values(standard(), standard_mutex(), standard_tokio(), tokio_std(), tokio_mutex(), tokio_tokio())] repo: Repo,
@@ -169,17 +242,77 @@ async fn it_should_get_a_torrent_entry(
 #[case::completed(completed())]
 #[case::downloaded(downloaded())]
 #[case::three(three())]
+#[case::out_of_order(many_out_of_order())]
+#[case::in_order(many_hashed_in_order())]
 #[tokio::test]
-async fn it_should_get_entries(
+async fn it_should_get_paginated_entries_in_a_stable_or_sorted_order(
  #[values(standard(), standard_mutex(), standard_tokio(), tokio_std(), tokio_mutex(), tokio_tokio())] repo: Repo,
  #[case] entries: Entries,
+ many_out_of_order: Entries,
 ) {
  make(&repo, &entries).await;
 
- if entries.first().is_some() {
- assert!(entries.contains(repo.get_paginated(None).await.first().expect("it should have at least one")));
- } else {
- assert!(repo.get_paginated(None).await.is_empty());
+ let entries_a = repo.get_paginated(None).await.iter().map(|(i, _)| *i).collect::<Vec<_>>();
+
+ make(&repo, &many_out_of_order).await;
+
+ let entries_b = repo.get_paginated(None).await.iter().map(|(i, _)| *i).collect::<Vec<_>>();
+
+ let is_equal = entries_b.iter().take(entries_a.len()).copied().collect::<Vec<_>>() == entries_a;
+
+ let is_sorted = entries_b.windows(2).all(|w| w[0] <= w[1]);
+
+ assert!(
+ is_equal || is_sorted,
+ "The order is unstable: {is_equal}, or is sorted {is_sorted}."
+ );
+}
+
+#[rstest]
+#[case::empty(empty())]
+#[case::default(default())]
+#[case::started(started())]
+#[case::completed(completed())]
+#[case::downloaded(downloaded())]
+#[case::three(three())]
+#[case::out_of_order(many_out_of_order())]
+#[case::in_order(many_hashed_in_order())]
+#[tokio::test]
+async fn it_should_get_paginated(
+ #[values(standard(), standard_mutex(), standard_tokio(), tokio_std(), tokio_mutex(), tokio_tokio())] repo: Repo,
+ #[case] entries: Entries,
+ #[values(paginated_limit_zero(), paginated_limit_one(), paginated_limit_one_offset_one())] paginated: Pagination,
+) {
+ make(&repo, &entries).await;
+
+ let mut info_hashes = repo.get_paginated(None).await.iter().map(|(i, _)| *i).collect::<Vec<_>>();
+ info_hashes.sort();
+
+ match paginated {
+ // it should return empty if limit is zero.
+ Pagination { limit: 0, .. } => assert_eq!(repo.get_paginated(Some(&paginated)).await, vec![]),
+
+ // it should return a single entry if the limit is one.
+ Pagination { limit: 1, offset: 0 } => {
+ if info_hashes.is_empty() {
+ assert_eq!(repo.get_paginated(Some(&paginated)).await.len(), 0);
+ } else {
+ let page = repo.get_paginated(Some(&paginated)).await;
+ assert_eq!(page.len(), 1);
+ assert_eq!(page.first().map(|(i, _)| i), info_hashes.first());
+ }
+ }
+
+ // it should return the only the second entry if both the limit and the offset are one.
+ Pagination { limit: 1, offset: 1 } => {
+ if info_hashes.len() > 1 {
+ let page = repo.get_paginated(Some(&paginated)).await;
+ assert_eq!(page.len(), 1);
+ assert_eq!(page[0].0, info_hashes[1]);
+ }
+ }
+ // the other cases are not yet tested.
+ _ => {}
  }
 }
 
@@ -190,6 +323,8 @@ async fn it_should_get_entries(
 #[case::completed(completed())]
 #[case::downloaded(downloaded())]
 #[case::three(three())]
+#[case::out_of_order(many_out_of_order())]
+#[case::in_order(many_hashed_in_order())]
 #[tokio::test]
 async fn it_should_get_metrics(
  #[values(standard(), standard_mutex(), standard_tokio(), tokio_std(), tokio_mutex(), tokio_tokio())] repo: Repo,
@@ -220,6 +355,8 @@ async fn it_should_get_metrics(
 #[case::completed(completed())]
 #[case::downloaded(downloaded())]
 #[case::three(three())]
+#[case::out_of_order(many_out_of_order())]
+#[case::in_order(many_hashed_in_order())]
 #[tokio::test]
 async fn it_should_import_persistent_torrents(
  #[values(standard(), standard_mutex(), standard_tokio(), tokio_std(), tokio_mutex(), tokio_tokio())] repo: Repo,
@@ -247,6 +384,8 @@ async fn it_should_import_persistent_torrents(
 #[case::completed(completed())]
 #[case::downloaded(downloaded())]
 #[case::three(three())]
+#[case::out_of_order(many_out_of_order())]
+#[case::in_order(many_hashed_in_order())]
 #[tokio::test]
 async fn it_should_remove_an_entry(
  #[values(standard(), standard_mutex(), standard_tokio(), tokio_std(), tokio_mutex(), tokio_tokio())] repo: Repo,
@@ -272,6 +411,8 @@ async fn it_should_remove_an_entry(
 #[case::completed(completed())]
 #[case::downloaded(downloaded())]
 #[case::three(three())]
+#[case::out_of_order(many_out_of_order())]
+#[case::in_order(many_hashed_in_order())]
 #[tokio::test]
 async fn it_should_remove_inactive_peers(
  #[values(standard(), standard_mutex(), standard_tokio(), tokio_std(), tokio_mutex(), tokio_tokio())] repo: Repo,
@@ -335,3 +476,29 @@ async fn it_should_remove_inactive_peers(
  assert!(!entry.get_peers(None).contains(&peer.into()));
  }
 }
+
+#[rstest]
+#[case::empty(empty())]
+#[case::default(default())]
+#[case::started(started())]
+#[case::completed(completed())]
+#[case::downloaded(downloaded())]
+#[case::three(three())]
+#[case::out_of_order(many_out_of_order())]
+#[case::in_order(many_hashed_in_order())]
+#[tokio::test]
+async fn it_should_remove_peerless_torrents(
+ #[values(standard(), standard_mutex(), standard_tokio(), tokio_std(), tokio_mutex(), tokio_tokio())] repo: Repo,
+ #[case] entries: Entries,
+ #[values(policy_none(), policy_persist(), policy_remove(), policy_remove_persist())] policy: TrackerPolicy,
+) {
+ make(&repo, &entries).await;
+
+ repo.remove_peerless_torrents(&policy).await;
+
+ let torrents = repo.get_paginated(None).await;
+
+ for (_, entry) in torrents {
+ assert!(entry.is_good(&policy));
+ }
+}