mirror of
https://github.com/arvidn/libtorrent.git
synced 2026-07-24 13:01:13 -04:00
Arvid Norberg
1742 lines
71 KiB
Python
1742 lines
71 KiB
Python
import os
|
|
import pathlib
|
|
import socket
|
|
import sys
|
|
import tempfile
|
|
from typing import Any
|
|
from typing import cast
|
|
from typing import Dict
|
|
from typing import TYPE_CHECKING
|
|
import unittest
|
|
|
|
import libtorrent as lt
|
|
|
|
from . import lib
|
|
from . import tdummy
|
|
|
|
if TYPE_CHECKING:
|
|
from libtorrent import _Entry
|
|
from libtorrent import AddTorrentParamsdict
|
|
from libtorrent import load_torrent_limits
|
|
from libtorrent import settings_pack
|
|
from libtorrent import TorrentFileDict
|
|
else:
|
|
AddTorrentParamsdict, TorrentFileDict, load_torrent_limits, settings_pack = (
|
|
dict,
|
|
dict,
|
|
dict,
|
|
dict,
|
|
)
|
|
|
|
|
|
class ReadWriteSessionParamsTest(unittest.TestCase):
|
|
def test_read_bdecoded(self) -> None:
|
|
params = lt.session_params({"alert_mask": 123})
|
|
bdecoded = lt.write_session_params(params)
|
|
|
|
read_params = lt.read_session_params(bdecoded)
|
|
self.assertIsInstance(read_params, lt.session_params)
|
|
# self.assertEqual(read_params, params)
|
|
self.assertEqual(read_params.settings, params.settings)
|
|
|
|
read_params = lt.read_session_params(bdecoded, flags=0)
|
|
self.assertIsInstance(read_params, lt.session_params)
|
|
# self.assertEqual(read_params, lt.session_params())
|
|
self.assertEqual(read_params.settings, lt.session_params().settings)
|
|
|
|
def test_read_bencoded(self) -> None:
|
|
params = lt.session_params({"alert_mask": 123})
|
|
bencoded = lt.bencode(lt.write_session_params(params))
|
|
|
|
read_params = lt.read_session_params(bencoded)
|
|
self.assertIsInstance(read_params, lt.session_params)
|
|
# self.assertEqual(read_params, params)
|
|
self.assertEqual(read_params.settings, params.settings)
|
|
|
|
read_params = lt.read_session_params(bencoded, flags=0)
|
|
self.assertIsInstance(read_params, lt.session_params)
|
|
# self.assertEqual(read_params, lt.session_params())
|
|
self.assertEqual(read_params.settings, lt.session_params().settings)
|
|
|
|
def test_write(self) -> None:
|
|
params = lt.session_params({"alert_mask": 123})
|
|
|
|
bdecoded = lt.write_session_params(params)
|
|
self.assertIsInstance(bdecoded, dict)
|
|
self.assertEqual(bdecoded, lt.write_session_params(params))
|
|
|
|
bdecoded = lt.write_session_params(
|
|
params, flags=lt.save_state_flags_t.save_settings
|
|
)
|
|
self.assertEqual(set(bdecoded.keys()), {b"settings"})
|
|
|
|
def test_write_empty(self) -> None:
|
|
bdecoded = lt.write_session_params(lt.session_params(), flags=0)
|
|
self.assertIsInstance(bdecoded, dict)
|
|
|
|
def test_round_trip(self) -> None:
|
|
params = lt.session_params()
|
|
params.settings = {"alert_mask": 123}
|
|
|
|
read_params = lt.read_session_params(lt.write_session_params(params))
|
|
self.assertIsInstance(read_params, lt.session_params)
|
|
# self.assertEqual(params, read_params)
|
|
self.assertEqual(params.settings, read_params.settings)
|
|
|
|
|
|
class SessionParamsTest(unittest.TestCase):
|
|
def test_constructor(self) -> None:
|
|
params = lt.session_params()
|
|
# self.assertEqual(params, lt.session_params())
|
|
self.assertEqual(params.settings, lt.session_params().settings)
|
|
|
|
params = lt.session_params({"alert_mask": 123})
|
|
expected = lt.session_params().settings
|
|
expected["alert_mask"] = 123
|
|
self.assertEqual(params.settings, expected)
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/6140")
|
|
def test_equality(self) -> None:
|
|
self.assertEqual(lt.session_params(), lt.session_params())
|
|
self.assertEqual(
|
|
lt.session_params({"alert_mask": 123}),
|
|
lt.session_params({"alert_mask": 123}),
|
|
)
|
|
|
|
def test_settings(self) -> None:
|
|
params = lt.session_params()
|
|
params.settings = {"alert_mask": 123}
|
|
expected = lt.session_params().settings
|
|
expected["alert_mask"] = 123
|
|
self.assertEqual(params.settings, expected)
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/6140")
|
|
def test_settings_reference(self) -> None:
|
|
params = lt.session_params()
|
|
params.settings["alert_mask"] = 123
|
|
self.assertEqual(params.settings["alert_mask"], 123)
|
|
|
|
def test_ip_filter(self) -> None:
|
|
params = lt.session_params()
|
|
self.assertEqual(
|
|
params.ip_filter.export_filter(),
|
|
(
|
|
[("0.0.0.0", "255.255.255.255")],
|
|
[("::", "ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff")],
|
|
),
|
|
)
|
|
ipf = lt.ip_filter()
|
|
ipf.add_rule("128.0.0.0", "128.0.0.1", 123)
|
|
params.ip_filter = ipf
|
|
self.assertEqual(
|
|
params.ip_filter.export_filter(),
|
|
(
|
|
[
|
|
("0.0.0.0", "127.255.255.255"),
|
|
("128.0.0.0", "128.0.0.1"),
|
|
("128.0.0.2", "255.255.255.255"),
|
|
],
|
|
[("::", "ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff")],
|
|
),
|
|
)
|
|
|
|
def test_ip_filter_reference(self) -> None:
|
|
params = lt.session_params()
|
|
self.assertEqual(
|
|
params.ip_filter.export_filter(),
|
|
(
|
|
[("0.0.0.0", "255.255.255.255")],
|
|
[("::", "ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff")],
|
|
),
|
|
)
|
|
params.ip_filter.add_rule("128.0.0.0", "128.0.0.1", 123)
|
|
self.assertEqual(
|
|
params.ip_filter.export_filter(),
|
|
(
|
|
[
|
|
("0.0.0.0", "127.255.255.255"),
|
|
("128.0.0.0", "128.0.0.1"),
|
|
("128.0.0.2", "255.255.255.255"),
|
|
],
|
|
[("::", "ffff:ffff:ffff:ffff:ffff:ffff:ffff:ffff")],
|
|
),
|
|
)
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/6140")
|
|
def test_dht_state(self) -> None:
|
|
params = lt.session_params()
|
|
self.assertEqual(params.dht_state, lt.dht_state())
|
|
state = lt.dht_state()
|
|
state.nodes = [("127.0.0.1", 1234)] # type: ignore[misc]
|
|
params.dht_state = state
|
|
self.assertEqual(params.dht_state, state)
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/6140")
|
|
def test_dht_state_reference(self) -> None:
|
|
params = lt.session_params()
|
|
self.assertEqual(params.dht_state, lt.dht_state())
|
|
params.dht_state.nodes = [("127.0.0.1", 1234)] # type: ignore[misc]
|
|
self.assertEqual(params.dht_state.nodes, [("127.0.0.1", 1234)])
|
|
|
|
|
|
class DhtStateTest(unittest.TestCase):
|
|
def test_constructor(self) -> None:
|
|
state = lt.dht_state()
|
|
self.assertIsInstance(state, lt.dht_state)
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/6140")
|
|
def test_equality(self) -> None:
|
|
a = lt.dht_state()
|
|
b = lt.dht_state()
|
|
self.assertEqual(a, b)
|
|
a.nodes = [("127.0.0.1", 1234)] # type: ignore[misc]
|
|
self.assertNotEqual(a, b)
|
|
b.nodes = [("127.0.0.1", 1234)] # type: ignore[misc]
|
|
self.assertNotEqual(a, b)
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/6140")
|
|
def test_nids(self) -> None:
|
|
state = lt.dht_state()
|
|
self.assertEqual(state.nids, [])
|
|
state.nids = [("127.0.0.1", lt.sha1_hash(b"a" * 20))] # type: ignore[misc]
|
|
self.assertEqual(state.nids, [("127.0.0.1", lt.sha1_hash(b"a" * 20))])
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/6140")
|
|
def test_nodes(self) -> None:
|
|
state = lt.dht_state()
|
|
self.assertEqual(state.nodes, [])
|
|
state.nodes = [("127.0.0.1", 1234)] # type: ignore[misc]
|
|
self.assertEqual(state.nodes, [("127.0.0.1", 1234)])
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/6140")
|
|
def test_nodes6(self) -> None:
|
|
state = lt.dht_state()
|
|
self.assertEqual(state.nodes6, [])
|
|
state.nodes6 = [("::1", 1234)] # type: ignore[misc]
|
|
self.assertEqual(state.nodes6, [("::1", 1234)])
|
|
|
|
|
|
class SessionStatusTest(unittest.TestCase):
|
|
def test_fields(self) -> None:
|
|
if lt.api_version < 2:
|
|
with self.assertWarns(DeprecationWarning):
|
|
stat = lt.session_status()
|
|
|
|
self.assertIsInstance(stat.has_incoming_connections, bool)
|
|
|
|
self.assertIsInstance(stat.upload_rate, int)
|
|
self.assertIsInstance(stat.download_rate, int)
|
|
self.assertIsInstance(stat.total_download, int)
|
|
self.assertIsInstance(stat.total_upload, int)
|
|
|
|
self.assertIsInstance(stat.payload_upload_rate, int)
|
|
self.assertIsInstance(stat.payload_download_rate, int)
|
|
self.assertIsInstance(stat.total_payload_download, int)
|
|
self.assertIsInstance(stat.total_payload_upload, int)
|
|
|
|
self.assertIsInstance(stat.ip_overhead_upload_rate, int)
|
|
self.assertIsInstance(stat.ip_overhead_download_rate, int)
|
|
self.assertIsInstance(stat.total_ip_overhead_download, int)
|
|
self.assertIsInstance(stat.total_ip_overhead_upload, int)
|
|
|
|
self.assertIsInstance(stat.dht_upload_rate, int)
|
|
self.assertIsInstance(stat.dht_download_rate, int)
|
|
self.assertIsInstance(stat.total_dht_download, int)
|
|
self.assertIsInstance(stat.total_dht_upload, int)
|
|
|
|
self.assertIsInstance(stat.tracker_upload_rate, int)
|
|
self.assertIsInstance(stat.tracker_download_rate, int)
|
|
self.assertIsInstance(stat.total_tracker_download, int)
|
|
self.assertIsInstance(stat.total_tracker_upload, int)
|
|
|
|
self.assertIsInstance(stat.total_redundant_bytes, int)
|
|
self.assertIsInstance(stat.total_failed_bytes, int)
|
|
|
|
self.assertIsInstance(stat.num_peers, int)
|
|
self.assertIsInstance(stat.num_unchoked, int)
|
|
self.assertIsInstance(stat.allowed_upload_slots, int)
|
|
|
|
self.assertIsInstance(stat.up_bandwidth_queue, int)
|
|
self.assertIsInstance(stat.down_bandwidth_queue, int)
|
|
|
|
self.assertIsInstance(stat.up_bandwidth_bytes_queue, int)
|
|
self.assertIsInstance(stat.down_bandwidth_bytes_queue, int)
|
|
|
|
self.assertIsInstance(stat.optimistic_unchoke_counter, int)
|
|
self.assertIsInstance(stat.unchoke_counter, int)
|
|
|
|
self.assertIsInstance(stat.dht_nodes, int)
|
|
self.assertIsInstance(stat.dht_node_cache, int)
|
|
self.assertIsInstance(stat.dht_torrents, int)
|
|
self.assertIsInstance(stat.dht_global_nodes, int)
|
|
self.assertEqual(stat.active_requests, [])
|
|
self.assertIsInstance(stat.dht_total_allocations, int)
|
|
|
|
with self.assertWarns(DeprecationWarning):
|
|
utp = stat.utp_stats
|
|
self.assertIsInstance(utp["num_idle"], int)
|
|
self.assertIsInstance(utp["num_syn_sent"], int)
|
|
self.assertIsInstance(utp["num_connected"], int)
|
|
self.assertIsInstance(utp["num_fin_sent"], int)
|
|
self.assertIsInstance(utp["num_close_wait"], int)
|
|
|
|
def test_from_session(self) -> None:
|
|
session = lt.session(lib.get_isolated_settings())
|
|
if lt.api_version < 2:
|
|
with self.assertWarns(DeprecationWarning):
|
|
stat = session.status()
|
|
self.assertIsInstance(stat, lt.session_status)
|
|
|
|
|
|
class AddTorrentParamsTest(unittest.TestCase):
|
|
def test_fields(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
|
|
atp.version = 123
|
|
self.assertEqual(atp.version, 123)
|
|
atp.ti = lt.torrent_info(lt.sha1_hash())
|
|
self.assertEqual(atp.ti.info_hashes().v1, lt.sha1_hash())
|
|
atp.trackers = ["http://example.com/tr"]
|
|
self.assertEqual(atp.trackers, ["http://example.com/tr"])
|
|
atp.tracker_tiers = [1]
|
|
self.assertEqual(atp.tracker_tiers, [1])
|
|
atp.dht_nodes = [("0.1.2.3", 1234)]
|
|
self.assertEqual(atp.dht_nodes, [("0.1.2.3", 1234)])
|
|
atp.name = "test.txt"
|
|
self.assertEqual(atp.name, "test.txt")
|
|
atp.save_path = "."
|
|
self.assertEqual(atp.save_path, ".")
|
|
atp.storage_mode = lt.storage_mode_t.storage_mode_allocate
|
|
self.assertEqual(atp.storage_mode, lt.storage_mode_t.storage_mode_allocate)
|
|
atp.file_priorities = [1]
|
|
self.assertEqual(atp.file_priorities, [1])
|
|
atp.trackerid = "trackerid"
|
|
self.assertEqual(atp.trackerid, "trackerid")
|
|
atp.flags = lt.torrent_flags.default_flags
|
|
self.assertEqual(atp.flags, lt.torrent_flags.default_flags)
|
|
atp.max_uploads = 1
|
|
self.assertEqual(atp.max_uploads, 1)
|
|
atp.max_connections = 1
|
|
self.assertEqual(atp.max_connections, 1)
|
|
atp.upload_limit = 1024
|
|
self.assertEqual(atp.upload_limit, 1024)
|
|
atp.download_limit = 1024
|
|
self.assertEqual(atp.download_limit, 1024)
|
|
atp.total_uploaded = 1024
|
|
self.assertEqual(atp.total_uploaded, 1024)
|
|
atp.total_downloaded = 1024
|
|
self.assertEqual(atp.total_downloaded, 1024)
|
|
atp.active_time = 1234
|
|
self.assertEqual(atp.active_time, 1234)
|
|
atp.finished_time = 1234
|
|
self.assertEqual(atp.finished_time, 1234)
|
|
atp.seeding_time = 1234
|
|
self.assertEqual(atp.seeding_time, 1234)
|
|
atp.added_time = 1234
|
|
self.assertEqual(atp.added_time, 1234)
|
|
atp.completed_time = 1234
|
|
self.assertEqual(atp.completed_time, 1234)
|
|
atp.last_seen_complete = 1234
|
|
self.assertEqual(atp.last_seen_complete, 1234)
|
|
atp.last_download = 1234
|
|
self.assertEqual(atp.last_download, 1234)
|
|
atp.last_upload = 1234
|
|
self.assertEqual(atp.last_upload, 1234)
|
|
atp.num_complete = 10
|
|
self.assertEqual(atp.num_complete, 10)
|
|
atp.num_incomplete = 10
|
|
self.assertEqual(atp.num_incomplete, 10)
|
|
atp.num_downloaded = 10
|
|
self.assertEqual(atp.num_downloaded, 10)
|
|
atp.info_hashes = lt.info_hash_t(lt.sha1_hash())
|
|
self.assertEqual(atp.info_hashes.v1, lt.sha1_hash())
|
|
atp.url_seeds = ["http://example.com/seed"]
|
|
self.assertEqual(atp.url_seeds, ["http://example.com/seed"])
|
|
atp.peers = [("1.2.3.4", 4321)]
|
|
self.assertEqual(atp.peers, [("1.2.3.4", 4321)])
|
|
atp.banned_peers = [("2.3.4.5", 4321)]
|
|
self.assertEqual(atp.banned_peers, [("2.3.4.5", 4321)])
|
|
atp.unfinished_pieces = {}
|
|
self.assertEqual(atp.unfinished_pieces, {})
|
|
atp.have_pieces = [True, False]
|
|
self.assertEqual(atp.have_pieces, [True, False])
|
|
atp.verified_pieces = [True, False]
|
|
self.assertEqual(atp.verified_pieces, [True, False])
|
|
atp.piece_priorities = [1]
|
|
self.assertEqual(atp.piece_priorities, [1])
|
|
atp.renamed_files = {}
|
|
self.assertEqual(atp.renamed_files, {})
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/5988")
|
|
def test_name_assign_bytes_deprecated(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
|
|
with self.assertWarns(DeprecationWarning):
|
|
atp.name = b"test.txt" # type: ignore
|
|
|
|
def test_name_assign_bytes(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
|
|
atp.name = b"test.txt" # type: ignore
|
|
self.assertEqual(atp.name, "test.txt")
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/5988")
|
|
def test_trackerid_assign_bytes_deprecated(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
with self.assertWarns(DeprecationWarning):
|
|
atp.trackerid = b"trackerid" # type: ignore
|
|
|
|
def test_trackerid_assign_bytes(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.trackerid = b"trackerid" # type: ignore
|
|
self.assertEqual(atp.trackerid, "trackerid")
|
|
|
|
def test_save_path_ascii_str(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.save_path = "test"
|
|
self.assertEqual(atp.save_path, "test")
|
|
self.assertEqual(lt.write_resume_data(atp)[b"save_path"], b"test")
|
|
|
|
def test_save_path_ascii_bytes(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.save_path = b"test" # type: ignore
|
|
self.assertEqual(atp.save_path, "test")
|
|
self.assertEqual(lt.write_resume_data(atp)[b"save_path"], b"test")
|
|
|
|
def test_save_path_non_ascii_str(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.save_path = "\u1234"
|
|
self.assertEqual(atp.save_path, "\u1234")
|
|
self.assertEqual(lt.write_resume_data(atp)[b"save_path"], os.fsencode("\u1234"))
|
|
|
|
def test_save_path_non_ascii_bytes(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.save_path = os.fsencode("\u1234") # type: ignore
|
|
self.assertEqual(atp.save_path, "\u1234")
|
|
self.assertEqual(lt.write_resume_data(atp)[b"save_path"], os.fsencode("\u1234"))
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/5984")
|
|
@lib.uses_surrogate_paths()
|
|
def test_save_path_surrogate_str(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.save_path = "\udcff"
|
|
self.assertEqual(atp.save_path, "\udcff")
|
|
self.assertEqual(lt.write_resume_data(atp)[b"save_path"], os.fsencode("\udcff"))
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/5984")
|
|
@lib.uses_surrogate_paths()
|
|
def test_save_path_surrogate_bytes(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.save_path = os.fsencode("\udcff") # type: ignore
|
|
self.assertEqual(atp.save_path, "\udcff")
|
|
self.assertEqual(lt.write_resume_data(atp)[b"save_path"], os.fsencode("\udcff"))
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/5984")
|
|
@lib.uses_non_unicode_paths()
|
|
def test_save_path_non_unicode_str(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.save_path = os.fsdecode(b"\xff")
|
|
self.assertEqual(atp.save_path, os.fsdecode(b"\xff"))
|
|
self.assertEqual(lt.write_resume_data(atp)[b"save_path"], b"\xff")
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/5984")
|
|
@lib.uses_non_unicode_paths()
|
|
def test_save_path_non_unicode_bytes(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.save_path = b"\xff" # type: ignore
|
|
self.assertEqual(atp.save_path, os.fsdecode(b"\xff"))
|
|
self.assertEqual(lt.write_resume_data(atp)[b"save_path"], b"\xff")
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/5988")
|
|
def test_info_hash_deprecated(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertTrue(atp.info_hash.is_all_zeros())
|
|
with self.assertWarns(DeprecationWarning):
|
|
atp.info_hash = lt.sha1_hash(lib.get_random_bytes(20))
|
|
|
|
def test_info_hash(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
if lt.api_version < 4:
|
|
self.assertTrue(atp.info_hash.is_all_zeros())
|
|
self.assertTrue(atp.info_hashes.v1.is_all_zeros())
|
|
self.assertTrue(atp.info_hashes.v2.is_all_zeros())
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/5988")
|
|
def test_http_seeds_deprecated(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertEqual(atp.http_seeds, [])
|
|
with self.assertWarns(DeprecationWarning):
|
|
atp.http_seeds = ["http://example.com/seed"]
|
|
|
|
def test_http_seeds(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.http_seeds = ["http://example.com/seed"]
|
|
self.assertEqual(atp.http_seeds, ["http://example.com/seed"])
|
|
|
|
def test_unfinished_pieces(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.unfinished_pieces = {}
|
|
atp.unfinished_pieces = {1: [True, False]}
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/5988")
|
|
def test_merkle_tree_deprecated(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertEqual(atp.merkle_tree, [])
|
|
with self.assertWarns(DeprecationWarning):
|
|
atp.merkle_tree = [lt.sha1_hash()]
|
|
|
|
def test_merkle_tree(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.merkle_tree = [lt.sha1_hash()]
|
|
self.assertEqual(atp.merkle_tree, [lt.sha1_hash()])
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/5988")
|
|
def test_url_deprecated(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertEqual(atp.url, "")
|
|
with self.assertWarns(DeprecationWarning):
|
|
atp.url = "http://example.com/torrent"
|
|
|
|
def test_url(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.url = "http://example.com/torrent"
|
|
self.assertEqual(atp.url, "http://example.com/torrent")
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/5988")
|
|
def test_resume_data_deprecated(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertEqual(atp.resume_data, [])
|
|
with self.assertWarns(DeprecationWarning):
|
|
atp.resume_data = ["a"]
|
|
|
|
def test_resume_data(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.resume_data = ["a"]
|
|
self.assertEqual(atp.resume_data, ["a"])
|
|
|
|
def test_renamed_files_ascii_str(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.renamed_files = {0: "test.txt"}
|
|
self.assertEqual(atp.renamed_files, {0: "test.txt"})
|
|
|
|
def test_renamed_files_ascii_bytes(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.renamed_files = {0: b"test.txt"} # type: ignore
|
|
self.assertEqual(atp.renamed_files, {0: "test.txt"})
|
|
|
|
def test_renamed_files_non_ascii_str(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.renamed_files = {0: "\u1234.txt"}
|
|
self.assertEqual(atp.renamed_files, {0: "\u1234.txt"})
|
|
|
|
def test_renamed_files_non_ascii_bytes(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.renamed_files = {0: os.fsencode("\u1234.txt")} # type: ignore
|
|
self.assertEqual(atp.renamed_files, {0: "\u1234.txt"})
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/5984")
|
|
@lib.uses_surrogate_paths()
|
|
def test_renamed_files_surrogate_str(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.renamed_files = {0: "\udcff.txt"}
|
|
self.assertEqual(atp.renamed_files, {0: "\udcff.txt"})
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/5984")
|
|
@lib.uses_surrogate_paths()
|
|
def test_renamed_files_surrogate_bytes(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.renamed_files = {0: os.fsencode("\udcff.txt")} # type: ignore
|
|
self.assertEqual(atp.renamed_files, {0: "\udcff.txt"})
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/5984")
|
|
@lib.uses_non_unicode_paths()
|
|
def test_renamed_files_non_unicode_str(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.renamed_files = {0: os.fsdecode(b"\xff.txt")}
|
|
self.assertEqual(atp.renamed_files, {0: os.fsdecode("\xff.txt")})
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/5984")
|
|
@lib.uses_non_unicode_paths()
|
|
def test_renamed_files_non_unicode_bytes(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.renamed_files = {0: b"\xff.txt"} # type: ignore
|
|
self.assertEqual(atp.renamed_files, {0: os.fsdecode("\xff.txt")})
|
|
|
|
|
|
class EnumsTest(unittest.TestCase):
|
|
def test_storage_mode_t(self) -> None:
|
|
self.assertIsInstance(lt.storage_mode_t.storage_mode_allocate, int)
|
|
self.assertIsInstance(lt.storage_mode_t.storage_mode_sparse, int)
|
|
|
|
def test_options_t(self) -> None:
|
|
self.assertIsInstance(lt.options_t.delete_files, int)
|
|
|
|
def test_session_flags_t(self) -> None:
|
|
self.assertIsInstance(lt.session_flags_t.paused, int)
|
|
if lt.api_version < 3:
|
|
self.assertIsInstance(lt.session_flags_t.add_default_plugins, int)
|
|
if lt.api_version < 2:
|
|
self.assertIsInstance(lt.session_flags_t.start_default_features, int)
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/5988")
|
|
def test_session_flags_t_deprecated(self) -> None:
|
|
if lt.api_version < 3:
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(lt.session_flags_t.add_default_plugins, int)
|
|
if lt.api_version < 2:
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(lt.session_flags_t.start_default_features, int)
|
|
|
|
def test_torrent_flags(self) -> None:
|
|
self.assertIsInstance(lt.torrent_flags.seed_mode, int)
|
|
self.assertIsInstance(lt.torrent_flags.upload_mode, int)
|
|
self.assertIsInstance(lt.torrent_flags.share_mode, int)
|
|
self.assertIsInstance(lt.torrent_flags.apply_ip_filter, int)
|
|
self.assertIsInstance(lt.torrent_flags.paused, int)
|
|
self.assertIsInstance(lt.torrent_flags.auto_managed, int)
|
|
self.assertIsInstance(lt.torrent_flags.duplicate_is_error, int)
|
|
self.assertIsInstance(lt.torrent_flags.update_subscribe, int)
|
|
self.assertIsInstance(lt.torrent_flags.super_seeding, int)
|
|
self.assertIsInstance(lt.torrent_flags.sequential_download, int)
|
|
self.assertIsInstance(lt.torrent_flags.stop_when_ready, int)
|
|
if lt.api_version < 4:
|
|
self.assertIsInstance(lt.torrent_flags.override_trackers, int)
|
|
self.assertIsInstance(lt.torrent_flags.override_web_seeds, int)
|
|
self.assertIsInstance(lt.torrent_flags.disable_dht, int)
|
|
self.assertIsInstance(lt.torrent_flags.disable_lsd, int)
|
|
self.assertIsInstance(lt.torrent_flags.disable_pex, int)
|
|
self.assertIsInstance(lt.torrent_flags.no_verify_files, int)
|
|
self.assertIsInstance(lt.torrent_flags.default_flags, int)
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/5988")
|
|
def test_atp_flags_t_deprecated(self) -> None:
|
|
if lt.api_version < 2:
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(lt.add_torrent_params_flags_t.default_flags, int)
|
|
|
|
def test_atp_flags_t(self) -> None:
|
|
if lt.api_version < 2:
|
|
self.assertIsInstance(lt.add_torrent_params_flags_t.flag_seed_mode, int)
|
|
self.assertIsInstance(lt.add_torrent_params_flags_t.flag_upload_mode, int)
|
|
self.assertIsInstance(lt.add_torrent_params_flags_t.flag_share_mode, int)
|
|
self.assertIsInstance(
|
|
lt.add_torrent_params_flags_t.flag_apply_ip_filter, int
|
|
)
|
|
self.assertIsInstance(lt.add_torrent_params_flags_t.flag_paused, int)
|
|
self.assertIsInstance(lt.add_torrent_params_flags_t.flag_auto_managed, int)
|
|
self.assertIsInstance(
|
|
lt.add_torrent_params_flags_t.flag_duplicate_is_error, int
|
|
)
|
|
self.assertIsInstance(
|
|
lt.add_torrent_params_flags_t.flag_update_subscribe, int
|
|
)
|
|
self.assertIsInstance(lt.add_torrent_params_flags_t.flag_super_seeding, int)
|
|
self.assertIsInstance(
|
|
lt.add_torrent_params_flags_t.flag_sequential_download, int
|
|
)
|
|
self.assertIsInstance(
|
|
lt.add_torrent_params_flags_t.flag_stop_when_ready, int
|
|
)
|
|
self.assertIsInstance(
|
|
lt.add_torrent_params_flags_t.flag_override_trackers, int
|
|
)
|
|
self.assertIsInstance(
|
|
lt.add_torrent_params_flags_t.flag_override_web_seeds, int
|
|
)
|
|
self.assertIsInstance(lt.add_torrent_params_flags_t.flag_pinned, int)
|
|
self.assertIsInstance(
|
|
lt.add_torrent_params_flags_t.flag_override_resume_data, int
|
|
)
|
|
self.assertIsInstance(
|
|
lt.add_torrent_params_flags_t.flag_merge_resume_trackers, int
|
|
)
|
|
self.assertIsInstance(
|
|
lt.add_torrent_params_flags_t.flag_use_resume_save_path, int
|
|
)
|
|
self.assertIsInstance(
|
|
lt.add_torrent_params_flags_t.flag_merge_resume_http_seeds, int
|
|
)
|
|
self.assertIsInstance(lt.add_torrent_params_flags_t.default_flags, int)
|
|
|
|
def test_portmap_protocol(self) -> None:
|
|
self.assertIsInstance(lt.portmap_protocol.none, int)
|
|
self.assertIsInstance(lt.portmap_protocol.udp, int)
|
|
self.assertIsInstance(lt.portmap_protocol.tcp, int)
|
|
|
|
def test_portmap_transport(self) -> None:
|
|
self.assertIsInstance(lt.portmap_transport.natpmp, int)
|
|
self.assertIsInstance(lt.portmap_transport.upnp, int)
|
|
|
|
def test_peer_class_type_filter_socket_type_t(self) -> None:
|
|
pctfst = lt.peer_class_type_filter_socket_type_t
|
|
self.assertIsInstance(pctfst.tcp_socket, int)
|
|
self.assertIsInstance(pctfst.utp_socket, int)
|
|
self.assertIsInstance(pctfst.ssl_tcp_socket, int)
|
|
self.assertIsInstance(pctfst.ssl_utp_socket, int)
|
|
self.assertIsInstance(pctfst.i2p_socket, int)
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/5988")
|
|
def test_protocol_type_deprecated(self) -> None:
|
|
if lt.api_version < 2:
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(lt.protocol_type.udp, int)
|
|
|
|
def test_protocol_type(self) -> None:
|
|
if lt.api_version < 2:
|
|
self.assertIsInstance(lt.protocol_type.udp, int)
|
|
self.assertIsInstance(lt.protocol_type.tcp, int)
|
|
|
|
def test_save_state_flags_t(self) -> None:
|
|
self.assertIsInstance(lt.save_state_flags_t.all, int)
|
|
self.assertIsInstance(lt.save_state_flags_t.save_settings, int)
|
|
self.assertIsInstance(lt.save_state_flags_t.save_dht_state, int)
|
|
if lt.api_version < 3:
|
|
self.assertIsInstance(lt.save_state_flags_t.save_dht_settings, int)
|
|
if lt.api_version < 2:
|
|
self.assertIsInstance(lt.save_state_flags_t.save_encryption_settings, int)
|
|
self.assertIsInstance(lt.save_state_flags_t.save_as_map, int)
|
|
self.assertIsInstance(lt.save_state_flags_t.save_i2p_proxy, int)
|
|
self.assertIsInstance(lt.save_state_flags_t.save_proxy, int)
|
|
self.assertIsInstance(lt.save_state_flags_t.save_dht_proxy, int)
|
|
self.assertIsInstance(lt.save_state_flags_t.save_peer_proxy, int)
|
|
self.assertIsInstance(lt.save_state_flags_t.save_web_proxy, int)
|
|
self.assertIsInstance(lt.save_state_flags_t.save_tracker_proxy, int)
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/5988")
|
|
def test_save_state_flags_t_deprecated(self) -> None:
|
|
if lt.api_version < 3:
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(lt.save_state_flags_t.save_dht_settings, int)
|
|
if lt.api_version < 2:
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(
|
|
lt.save_state_flags_t.save_encryption_settings, int
|
|
)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(lt.save_state_flags_t.save_as_map, int)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(lt.save_state_flags_t.save_i2p_proxy, int)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(lt.save_state_flags_t.save_proxy, int)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(lt.save_state_flags_t.save_dht_proxy, int)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(lt.save_state_flags_t.save_peer_proxy, int)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(lt.save_state_flags_t.save_web_proxy, int)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(lt.save_state_flags_t.save_tracker_proxy, int)
|
|
|
|
def test_listen_on_flags_t(self) -> None:
|
|
if lt.api_version < 2:
|
|
self.assertIsInstance(lt.listen_on_flags_t.listen_reuse_address, int)
|
|
self.assertIsInstance(lt.listen_on_flags_t.listen_no_system_port, int)
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/5988")
|
|
def test_listen_on_flags_t_deprecated(self) -> None:
|
|
if lt.api_version < 2:
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(lt.listen_on_flags_t.listen_reuse_address, int)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(lt.listen_on_flags_t.listen_no_system_port, int)
|
|
|
|
def test_metric_type_t(self) -> None:
|
|
self.assertIsInstance(lt.metric_type_t.counter, int)
|
|
self.assertIsInstance(lt.metric_type_t.gauge, int)
|
|
|
|
def test_session_static_vars(self) -> None:
|
|
self.assertIsInstance(lt.session.tcp, int)
|
|
self.assertIsInstance(lt.session.udp, int)
|
|
|
|
self.assertIsInstance(lt.session.global_peer_class_id, int)
|
|
self.assertIsInstance(lt.session.tcp_peer_class_id, int)
|
|
self.assertIsInstance(lt.session.local_peer_class_id, int)
|
|
|
|
self.assertIsInstance(lt.session.reopen_map_ports, int)
|
|
|
|
self.assertIsInstance(lt.session.delete_files, int)
|
|
self.assertIsInstance(lt.session.delete_partfile, int)
|
|
|
|
def test_announce_flags_t(self) -> None:
|
|
self.assertIsInstance(lt.announce_flags_t.seed, int)
|
|
self.assertIsInstance(lt.announce_flags_t.implied_port, int)
|
|
self.assertIsInstance(lt.announce_flags_t.ssl_torrent, int)
|
|
|
|
|
|
class PeerClassTypeFilterTest(unittest.TestCase):
|
|
def test_filter(self) -> None:
|
|
pctf = lt.peer_class_type_filter()
|
|
tcp_socket = lt.peer_class_type_filter_socket_type_t.tcp_socket
|
|
pctf.add(tcp_socket, 1)
|
|
pctf.remove(tcp_socket, 1)
|
|
pctf.allow(tcp_socket, 1)
|
|
pctf.disallow(tcp_socket, 1)
|
|
self.assertEqual(pctf.apply(tcp_socket, 1), 1)
|
|
|
|
def test_enum_values(self) -> None:
|
|
self.assertIsInstance(lt.peer_class_type_filter.tcp_socket, int)
|
|
self.assertIsInstance(lt.peer_class_type_filter.utp_socket, int)
|
|
self.assertIsInstance(lt.peer_class_type_filter.ssl_tcp_socket, int)
|
|
self.assertIsInstance(lt.peer_class_type_filter.ssl_utp_socket, int)
|
|
self.assertIsInstance(lt.peer_class_type_filter.i2p_socket, int)
|
|
|
|
|
|
class ResumeDataTest(unittest.TestCase):
|
|
def do_test_round_trip(self, atp: lt.add_torrent_params) -> None:
|
|
first = lt.write_resume_data(atp)
|
|
second = lt.write_resume_data(
|
|
lt.read_resume_data(lt.write_resume_data_buf(atp))
|
|
)
|
|
|
|
self.assertEqual(first, second)
|
|
|
|
def test_round_trip(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.name = "test"
|
|
self.do_test_round_trip(atp)
|
|
|
|
def test_limit_decode_depth(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
buf = lt.write_resume_data_buf(atp)
|
|
with self.assertRaises(RuntimeError):
|
|
lt.read_resume_data(buf, load_torrent_limits({"max_decode_depth": 1}))
|
|
|
|
def test_limit_decode_tokens(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
buf = lt.write_resume_data_buf(atp)
|
|
with self.assertRaises(RuntimeError):
|
|
lt.read_resume_data(buf, load_torrent_limits({"max_decode_tokens": 1}))
|
|
|
|
def test_limit_pieces(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.ti = lt.torrent_info(
|
|
TorrentFileDict(
|
|
{
|
|
b"info": {
|
|
b"name": b"test.txt",
|
|
b"length": 1234000,
|
|
b"piece length": 16384,
|
|
b"pieces": b"aaaaaaaaaaaaaaaaaaaa" * (1234000 // 16384 + 1),
|
|
}
|
|
}
|
|
)
|
|
)
|
|
buf = lt.write_resume_data_buf(atp)
|
|
with self.assertRaises(RuntimeError):
|
|
lt.read_resume_data(buf, load_torrent_limits({"max_pieces": 1}))
|
|
|
|
|
|
class ConstructorTest(unittest.TestCase):
|
|
# A bunch of these technically break isolation, but they need to be tested.
|
|
# We compensate by immediately setting isolated settings
|
|
|
|
def test_args(self) -> None:
|
|
# no-args
|
|
|
|
session = lt.session()
|
|
session.apply_settings(lib.get_isolated_settings())
|
|
|
|
# positional args
|
|
|
|
lt.session(lib.get_isolated_settings())
|
|
lt.session(lib.get_isolated_settings(), 0)
|
|
|
|
# kwargs
|
|
|
|
lt.session(settings=lib.get_isolated_settings())
|
|
|
|
if lt.api_version < 2:
|
|
session = lt.session(flags=0)
|
|
else:
|
|
session = lt.session()
|
|
session.apply_settings(lib.get_isolated_settings())
|
|
|
|
lt.session(settings=lib.get_isolated_settings(), flags=0)
|
|
|
|
def test_invalid_settings(self) -> None:
|
|
with self.assertRaises(TypeError):
|
|
lt.session({"alert_mask": "not-an-int"}) # type: ignore[call-overload]
|
|
with self.assertRaises(KeyError):
|
|
lt.session({"not-a-setting": 123}) # type: ignore[call-overload]
|
|
|
|
def test_predefined_settings_packs(self) -> None:
|
|
session = lt.session(lt.default_settings())
|
|
session.apply_settings(lib.get_isolated_settings())
|
|
|
|
session = lt.session(lt.high_performance_seed())
|
|
session.apply_settings(lib.get_isolated_settings())
|
|
|
|
session = lt.session(lt.min_memory_usage())
|
|
session.apply_settings(lib.get_isolated_settings())
|
|
|
|
def test_fingerprint(self) -> None:
|
|
if lt.api_version < 2:
|
|
with self.assertWarns(DeprecationWarning):
|
|
fingerprint = lt.fingerprint("AB", 1, 2, 3, 4)
|
|
|
|
session = lt.session(fingerprint)
|
|
session.apply_settings(lib.get_isolated_settings())
|
|
|
|
session = lt.session(fingerprint, 0)
|
|
session.apply_settings(lib.get_isolated_settings())
|
|
|
|
session = lt.session(fingerprint, 0, 0)
|
|
session.apply_settings(lib.get_isolated_settings())
|
|
|
|
session = lt.session(fingerprint=fingerprint)
|
|
session.apply_settings(lib.get_isolated_settings())
|
|
|
|
session = lt.session(fingerprint=fingerprint, flags=0)
|
|
session.apply_settings(lib.get_isolated_settings())
|
|
|
|
session = lt.session(fingerprint=fingerprint, flags=0, alert_mask=0)
|
|
session.apply_settings(lib.get_isolated_settings())
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/5988")
|
|
def test_fingerprint_deprecated(self) -> None:
|
|
if lt.api_version < 2:
|
|
fingerprint = lt.fingerprint("AB", 1, 2, 3, 4)
|
|
with self.assertWarns(DeprecationWarning):
|
|
session = lt.session(fingerprint)
|
|
session.apply_settings(lib.get_isolated_settings())
|
|
with self.assertWarns(DeprecationWarning):
|
|
session = lt.session(fingerprint, 0)
|
|
session.apply_settings(lib.get_isolated_settings())
|
|
with self.assertWarns(DeprecationWarning):
|
|
session = lt.session(fingerprint, 0, 0)
|
|
session.apply_settings(lib.get_isolated_settings())
|
|
with self.assertWarns(DeprecationWarning):
|
|
session = lt.session(fingerprint=fingerprint)
|
|
session.apply_settings(lib.get_isolated_settings())
|
|
with self.assertWarns(DeprecationWarning):
|
|
session = lt.session(fingerprint=fingerprint, flags=0)
|
|
session.apply_settings(lib.get_isolated_settings())
|
|
with self.assertWarns(DeprecationWarning):
|
|
session = lt.session(fingerprint=fingerprint, flags=0, alert_mask=0)
|
|
session.apply_settings(lib.get_isolated_settings())
|
|
|
|
|
|
class DhtTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.session = lt.session(lib.get_isolated_settings())
|
|
|
|
def test_functions(self) -> None:
|
|
self.assertFalse(self.session.is_dht_running())
|
|
|
|
# Should be "pretty safe" for isolation purposes
|
|
endpoint = ("127.1.2.3", 65535)
|
|
|
|
self.session.add_dht_node(endpoint)
|
|
|
|
sha1 = lt.sha1_hash(b"a" * 20)
|
|
|
|
self.session.dht_get_immutable_item(sha1)
|
|
self.session.dht_get_mutable_item(b"a" * 32, b"salt")
|
|
self.assertIsInstance(
|
|
self.session.dht_put_immutable_item(b"test"), lt.sha1_hash
|
|
)
|
|
self.assertIsInstance(self.session.dht_put_immutable_item(12345), lt.sha1_hash)
|
|
self.assertIsInstance(
|
|
self.session.dht_put_immutable_item(cast("_Entry", {b"a": 1})), lt.sha1_hash
|
|
)
|
|
self.assertIsInstance(
|
|
self.session.dht_put_immutable_item([1, 2, 3]), lt.sha1_hash
|
|
)
|
|
self.session.dht_put_mutable_item(b"a" * 64, b"b" * 32, b"data", b"salt")
|
|
self.session.dht_get_peers(sha1)
|
|
self.session.dht_announce(sha1)
|
|
self.session.dht_announce(sha1, 0, 0)
|
|
self.session.dht_announce(sha1, port=0, flags=lt.announce_flags_t.seed)
|
|
self.session.dht_live_nodes(sha1)
|
|
self.session.dht_sample_infohashes(endpoint, sha1)
|
|
|
|
if lt.api_version < 2:
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.add_dht_router(*endpoint)
|
|
|
|
if lt.api_version < 3:
|
|
dht_settings = self.session.get_dht_settings()
|
|
self.session.set_dht_settings(dht_settings)
|
|
|
|
def test_bad_args(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
self.session.dht_get_mutable_item(b"short", b"salt")
|
|
with self.assertRaises(ValueError):
|
|
self.session.dht_put_mutable_item(b"short", b"b" * 32, b"data", b"salt")
|
|
with self.assertRaises(ValueError):
|
|
self.session.dht_put_mutable_item(b"a" * 64, b"short", b"data", b"salt")
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/5988")
|
|
def test_deprecated(self) -> None:
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.get_dht_settings()
|
|
with self.assertWarns(DeprecationWarning):
|
|
lt.dht_settings()
|
|
dht_settings = lt.dht_settings()
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.set_dht_settings(dht_settings)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.dht_get_mutable_item("a" * 32, "salt")
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.dht_put_mutable_item(
|
|
"a" * 64,
|
|
"b" * 32,
|
|
"data",
|
|
"salt",
|
|
)
|
|
|
|
def test_dht_lookup(self) -> None:
|
|
if lt.api_version < 2:
|
|
lookup = lt.dht_lookup()
|
|
self.assertIsInstance(lookup.branch_factor, int)
|
|
self.assertIsInstance(lookup.outstanding_requests, int)
|
|
self.assertIsInstance(lookup.response, int)
|
|
self.assertIsInstance(lookup.timeouts, int)
|
|
self.assertIsNone(lookup.type) # Should be a str, for real lookups
|
|
|
|
|
|
class AlertHandlingTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
settings = lib.get_isolated_settings()
|
|
settings["alert_mask"] = 0
|
|
self.session = lt.session(settings)
|
|
|
|
def test_wait_and_pop(self) -> None:
|
|
# wait_for_alert() should report no pending alerts
|
|
self.assertFalse(self.session.wait_for_alert(0))
|
|
|
|
# Force an alert to fire
|
|
self.session.post_torrent_updates()
|
|
self.assertTrue(self.session.wait_for_alert(10000))
|
|
alerts = self.session.pop_alerts()
|
|
self.assertEqual(len(alerts), 1)
|
|
self.assertIsInstance(alerts[0], lt.state_update_alert)
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/5988")
|
|
def test_set_alert_notify_deprecated(self) -> None:
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.set_alert_notify(lambda: None)
|
|
|
|
@unittest.skipIf(sys.platform == "win32", "windows doesn't support pipes")
|
|
def test_set_alert_fd_pipe(self) -> None:
|
|
# Redundant sys.platform checks are to help mypy
|
|
r, w = os.pipe()
|
|
# Should always be non-blocking, or we'll block the event loop
|
|
if sys.platform != "win32":
|
|
os.set_blocking(w, False)
|
|
|
|
self.session.set_alert_fd(w)
|
|
|
|
# Pipe should initially be empty
|
|
if sys.platform != "win32":
|
|
os.set_blocking(r, False)
|
|
with self.assertRaises(BlockingIOError):
|
|
os.read(r, 1024)
|
|
if sys.platform != "win32":
|
|
os.set_blocking(r, True)
|
|
|
|
# Force an alert to fire
|
|
self.session.post_torrent_updates()
|
|
|
|
# We should have data shortly
|
|
data = os.read(r, 1024)
|
|
self.assertGreater(len(data), 0)
|
|
|
|
def test_set_alert_fd_socket(self) -> None:
|
|
r, w = socket.socketpair()
|
|
# Should always be non-blocking, or we'll block the event loop
|
|
w.setblocking(False)
|
|
|
|
self.session.set_alert_fd(w.fileno())
|
|
|
|
# Pipe should initially be empty
|
|
r.setblocking(False)
|
|
with self.assertRaises(BlockingIOError):
|
|
r.recv(1024)
|
|
r.setblocking(True)
|
|
|
|
# Force an alert to fire
|
|
self.session.post_torrent_updates()
|
|
|
|
# Should now have data
|
|
data = r.recv(1024)
|
|
self.assertGreater(len(data), 0)
|
|
|
|
# Explicit cleanup to avoid ResourceWarning
|
|
r.close()
|
|
w.close()
|
|
|
|
|
|
class Test5155(unittest.TestCase):
|
|
# We attempt to force a torrent_info to be allocated in python, but
|
|
# released from C++.
|
|
# See https://github.com/arvidn/libtorrent/issues/5155
|
|
def setUp(self) -> None:
|
|
self.dir = tempfile.TemporaryDirectory()
|
|
settings = lib.get_isolated_settings()
|
|
settings["alert_mask"] = 0
|
|
self.session = lt.session(settings)
|
|
|
|
def tearDown(self) -> None:
|
|
lib.cleanup_with_windows_fix(self.dir, timeout=5)
|
|
|
|
def test_5155(self) -> None:
|
|
# Ensure the torrent_info only is referenced by function locals
|
|
def do_add() -> lt.torrent_handle:
|
|
torrent = tdummy.Torrent.single_file(
|
|
piece_length=16384, name=b"test.txt", length=16384 * 9 + 1000
|
|
)
|
|
atp = lt.add_torrent_params()
|
|
atp.ti = torrent.torrent_info()
|
|
atp.save_path = "."
|
|
return self.session.add_torrent(atp)
|
|
|
|
handle = do_add()
|
|
# The bug depends on this exact sequence of alert management. I don't
|
|
# have a good explanation for why
|
|
self.session.wait_for_alert(10000) # add_torrent_alert
|
|
self.session.pop_alerts()
|
|
self.session.remove_torrent(handle)
|
|
self.session.wait_for_alert(10000) # torrent_removed_alert
|
|
self.session.pop_alerts()
|
|
|
|
|
|
class PostAlertsTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
settings = lib.get_isolated_settings()
|
|
settings["alert_mask"] = 0
|
|
self.session = lt.session(settings)
|
|
|
|
def _assert_pending_alert(self, alert_type: type) -> None:
|
|
self.assertTrue(self.session.wait_for_alert(10000))
|
|
alerts = self.session.pop_alerts()
|
|
self.assertTrue(any(isinstance(a, alert_type) for a in alerts))
|
|
|
|
def test_post_torrent_updates(self) -> None:
|
|
# no args
|
|
self.session.post_torrent_updates()
|
|
self._assert_pending_alert(lt.state_update_alert)
|
|
|
|
# positional args
|
|
self.session.post_torrent_updates(0)
|
|
self._assert_pending_alert(lt.state_update_alert)
|
|
|
|
# kwargs
|
|
self.session.post_torrent_updates(flags=0)
|
|
self._assert_pending_alert(lt.state_update_alert)
|
|
|
|
def test_post_dht_stats(self) -> None:
|
|
self.session.post_dht_stats()
|
|
self._assert_pending_alert(lt.dht_stats_alert)
|
|
|
|
|
|
class AddTorrentTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.session = lt.session(lib.get_isolated_settings())
|
|
self.torrent = tdummy.get_default()
|
|
self.dir = tempfile.TemporaryDirectory()
|
|
|
|
def tearDown(self) -> None:
|
|
lib.cleanup_with_windows_fix(self.dir, timeout=5)
|
|
|
|
def test_old_style_with_wrong_args(self) -> None:
|
|
with self.assertRaises(TypeError):
|
|
self.session.add_torrent( # type: ignore
|
|
self.torrent.torrent_info(),
|
|
resume_data=None,
|
|
)
|
|
|
|
def test_old_style(self) -> None:
|
|
if lt.api_version < 2:
|
|
ti = self.torrent.torrent_info()
|
|
# positional args
|
|
with self.assertWarns(DeprecationWarning):
|
|
handle = self.session.add_torrent(ti, self.dir.name)
|
|
self.assertIsInstance(handle, lt.torrent_handle)
|
|
self.assertTrue(handle.is_valid())
|
|
self.assertEqual(handle.status().save_path, self.dir.name)
|
|
|
|
with self.assertWarns(DeprecationWarning):
|
|
handle = self.session.add_torrent(
|
|
ti,
|
|
self.dir.name,
|
|
None,
|
|
lt.storage_mode_t.storage_mode_sparse,
|
|
False,
|
|
)
|
|
self.assertIsInstance(handle, lt.torrent_handle)
|
|
self.assertTrue(handle.is_valid())
|
|
|
|
# kwargs
|
|
with self.assertWarns(DeprecationWarning):
|
|
handle = self.session.add_torrent(
|
|
ti,
|
|
self.dir.name,
|
|
resume_data=None,
|
|
storage_mode=lt.storage_mode_t.storage_mode_sparse,
|
|
paused=False,
|
|
)
|
|
self.assertIsInstance(handle, lt.torrent_handle)
|
|
self.assertTrue(handle.is_valid())
|
|
|
|
def test_atp(self) -> None:
|
|
atp = self.torrent.atp()
|
|
atp.save_path = self.dir.name
|
|
|
|
self.session.async_add_torrent(atp)
|
|
handle = self.session.add_torrent(atp)
|
|
self.assertIsInstance(handle, lt.torrent_handle)
|
|
self.assertTrue(handle.is_valid())
|
|
|
|
def test_dict_deprecated(self) -> None:
|
|
with self.assertWarns(DeprecationWarning):
|
|
atp = {"save_path": ".", "ti": self.torrent.torrent_info()}
|
|
self.session.add_torrent(atp) # type: ignore
|
|
with self.assertWarns(DeprecationWarning):
|
|
atp = {"save_path": ".", "ti": self.torrent.torrent_info()}
|
|
self.session.async_add_torrent(atp) # type: ignore
|
|
|
|
def do_test_dict(self, params: AddTorrentParamsdict) -> lt.torrent_handle:
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.async_add_torrent(params)
|
|
with self.assertWarns(DeprecationWarning):
|
|
handle = self.session.add_torrent(params)
|
|
self.assertIsInstance(handle, lt.torrent_handle)
|
|
self.assertTrue(handle.is_valid())
|
|
return handle
|
|
|
|
def test_dict(self) -> None:
|
|
ti = self.torrent.torrent_info()
|
|
atp = AddTorrentParamsdict(
|
|
{
|
|
"ti": ti,
|
|
"info_hashes": ti.info_hashes().v1.to_bytes(),
|
|
"save_path": self.dir.name,
|
|
"storage_mode": lt.storage_mode_t.storage_mode_allocate,
|
|
"trackers": ["http://127.1.2.1/tr"],
|
|
"url_seeds": ["http://127.1.2.2/us"],
|
|
"http_seeds": ["http://127.1.2.3/hs"],
|
|
"dht_nodes": [("127.1.2.4", 1234)],
|
|
"banned_peers": [("127.1.2.5", 1234)],
|
|
"peers": [("127.1.2.6", 1234)],
|
|
"flags": lt.torrent_flags.sequential_download,
|
|
"trackerid": "trackerid",
|
|
"renamed_files": {0: "renamed.txt"},
|
|
"file_priorities": [2],
|
|
}
|
|
)
|
|
|
|
if lt.api_version < 4:
|
|
atp["info_hash"] = ti.info_hashes().v1.to_bytes()
|
|
|
|
if lt.api_version < 2:
|
|
atp["url"] = "http://127.1.2.7/u"
|
|
|
|
handle = self.do_test_dict(atp)
|
|
|
|
status = handle.status()
|
|
self.assertEqual(handle.status().save_path, self.dir.name)
|
|
self.assertEqual(status.storage_mode, lt.storage_mode_t.storage_mode_allocate)
|
|
self.assertEqual(
|
|
[tr["url"] for tr in handle.trackers()], ["http://127.1.2.1/tr"]
|
|
)
|
|
self.assertEqual(handle.url_seeds(), ["http://127.1.2.2/us"])
|
|
# TODO: can we test dht nodes?
|
|
# TODO: can we test banned_nodes?
|
|
# TODO: why is this sometimes 2, sometimes 1?
|
|
self.assertIn(status.list_peers, (1, 2))
|
|
self.assertEqual(handle.flags(), lt.torrent_flags.sequential_download)
|
|
# TODO: can we test trackerid?
|
|
torrent_file = handle.torrent_file()
|
|
assert torrent_file is not None
|
|
# the torrent_info object is immutable now, and renamed files are
|
|
# recorded in a separate object
|
|
# self.assertEqual(torrent_file.files().file_path(0), "renamed.txt")
|
|
self.assertEqual(handle.get_file_priorities(), [2])
|
|
|
|
def test_dict_no_torrent_info_old(self) -> None:
|
|
self.do_test_dict(
|
|
AddTorrentParamsdict({"info_hashes": b"a" * 20, "save_path": self.dir.name})
|
|
)
|
|
|
|
def test_no_torrent_info_old(self) -> None:
|
|
atp = lt.add_torrent_params()
|
|
atp.info_hashes = lt.info_hash_t(lt.sha1_hash(b"a" * 20))
|
|
atp.save_path = self.dir.name
|
|
self.session.add_torrent(atp)
|
|
|
|
def test_dict_name(self) -> None:
|
|
# This can only be tested *without* torrent info
|
|
handle = self.do_test_dict(
|
|
AddTorrentParamsdict(
|
|
{
|
|
"info_hashes": b"a" * 20,
|
|
"save_path": self.dir.name,
|
|
"name": "test-name",
|
|
}
|
|
)
|
|
)
|
|
|
|
self.assertEqual(handle.status().name, "test-name")
|
|
|
|
def test_dict_no_torrent_info_sha1(self) -> None:
|
|
handle = self.do_test_dict(
|
|
AddTorrentParamsdict({"info_hashes": b"a" * 20, "save_path": self.dir.name})
|
|
)
|
|
self.assertEqual(handle.info_hashes().v1.to_bytes(), b"a" * 20)
|
|
|
|
def test_dict_no_torrent_info_sha256(self) -> None:
|
|
handle = self.do_test_dict(
|
|
AddTorrentParamsdict({"info_hashes": b"a" * 32, "save_path": self.dir.name})
|
|
)
|
|
|
|
self.assertEqual(handle.info_hashes().v2.to_bytes(), b"a" * 32)
|
|
|
|
def test_dict_errors(self) -> None:
|
|
with self.assertWarns(DeprecationWarning):
|
|
with self.assertRaises(KeyError):
|
|
self.session.add_torrent({"invalid-key": None}) # type: ignore
|
|
with self.assertWarns(DeprecationWarning):
|
|
with self.assertRaises(KeyError):
|
|
self.session.async_add_torrent({"invalid-key": None}) # type: ignore
|
|
|
|
def test_errors(self) -> None:
|
|
atp = self.torrent.atp()
|
|
atp.save_path = self.dir.name
|
|
atp.flags |= lt.torrent_flags.duplicate_is_error
|
|
|
|
self.session.add_torrent(atp)
|
|
with self.assertRaises(RuntimeError):
|
|
self.session.add_torrent(atp)
|
|
|
|
|
|
class StateTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.session = lt.session(lib.get_isolated_settings())
|
|
|
|
def check_state(self, state: Dict[bytes, Any]) -> None:
|
|
self.assertIsInstance(state[b"settings"], dict)
|
|
self.assertIsInstance(state[b"dht"], dict)
|
|
# we disable dht, so we don't expect b"dht state"
|
|
|
|
def test_save(self) -> None:
|
|
if lt.api_version < 3:
|
|
self.check_state(self.session.save_state())
|
|
self.check_state(self.session.save_state(flags=2**32 - 1))
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/5988")
|
|
def test_deprecated(self) -> None:
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.save_state()
|
|
state = self.session.save_state()
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.load_state(state)
|
|
|
|
def test_load(self) -> None:
|
|
state = self.session.save_state()
|
|
self.session.load_state(state)
|
|
self.session.load_state(state, flags=2**32 - 1)
|
|
|
|
|
|
class GetTorrentsTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.session = lt.session(lib.get_isolated_settings())
|
|
self.dir = tempfile.TemporaryDirectory()
|
|
self.torrent = tdummy.get_default()
|
|
atp = self.torrent.atp()
|
|
atp.save_path = self.dir.name
|
|
self.handle = self.session.add_torrent(atp)
|
|
|
|
def tearDown(self) -> None:
|
|
lib.cleanup_with_windows_fix(self.dir, timeout=5)
|
|
|
|
def test_get_torrents(self) -> None:
|
|
torrents = self.session.get_torrents()
|
|
self.assertEqual(torrents, [self.handle])
|
|
|
|
def test_find_torrent_sha1(self) -> None:
|
|
handle = self.session.find_torrent(self.torrent.sha1_hash)
|
|
self.assertEqual(handle, self.handle)
|
|
|
|
|
|
class PauseTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.session = lt.session(lib.get_isolated_settings())
|
|
|
|
def test_pause(self) -> None:
|
|
self.assertFalse(self.session.is_paused())
|
|
self.session.pause()
|
|
self.assertTrue(self.session.is_paused())
|
|
self.session.resume()
|
|
self.assertFalse(self.session.is_paused())
|
|
|
|
|
|
class ComponentsTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.session = lt.session(lib.get_isolated_settings())
|
|
|
|
def test_start_stop(self) -> None:
|
|
if lt.api_version < 2:
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.start_upnp()
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.stop_upnp()
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.start_lsd()
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.stop_lsd()
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.start_natpmp()
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.stop_natpmp()
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.start_dht()
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.stop_dht()
|
|
|
|
|
|
class PortsTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.session = lt.session(lib.get_isolated_settings())
|
|
|
|
def test_ports(self) -> None:
|
|
# NB: this typically is a no-op with our isolated settings, so we
|
|
# may not test the case where it actually works
|
|
tags = self.session.add_port_mapping(lt.portmap_protocol.tcp, 65535, 65535)
|
|
for tag in tags:
|
|
self.session.delete_port_mapping(tag)
|
|
self.session.delete_port_mapping(12345)
|
|
self.session.reopen_network_sockets(0)
|
|
self.assertTrue(self.session.is_listening())
|
|
self.assertIsInstance(self.session.listen_port(), int)
|
|
|
|
if lt.api_version < 2:
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.outgoing_ports(1024, 65535)
|
|
|
|
# not sure how to correctly test str arg, because it raises
|
|
# RuntimeError on an invalid interface
|
|
|
|
# positional args
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.listen_on(1024, 65535)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.listen_on(1024, 65535, None)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.listen_on(
|
|
1024, 65535, None, lt.listen_on_flags_t.listen_no_system_port
|
|
)
|
|
|
|
# kwargs
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.listen_on(1024, 65535)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.listen_on(1024, 65535, interface=None)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.listen_on(
|
|
1024, 65535, flags=lt.listen_on_flags_t.listen_no_system_port
|
|
)
|
|
|
|
with self.assertWarns(DeprecationWarning):
|
|
with self.assertRaises(RuntimeError):
|
|
self.session.listen_on(1024, 65535, "interface-does-not-exist")
|
|
|
|
|
|
class SettingsTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.session = lt.session(lib.get_isolated_settings())
|
|
|
|
def do_test_settings_pack(self, settings: settings_pack) -> None:
|
|
self.session.apply_settings({**settings, **lib.get_isolated_settings()})
|
|
|
|
def test_settings_packs(self) -> None:
|
|
self.do_test_settings_pack(lt.default_settings())
|
|
self.do_test_settings_pack(lt.high_performance_seed())
|
|
self.do_test_settings_pack(lt.min_memory_usage())
|
|
|
|
self.assertIsInstance(self.session.get_settings(), dict)
|
|
|
|
def test_old_settings(self) -> None:
|
|
if lt.api_version < 2:
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(self.session.local_download_rate_limit(), int)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.set_local_download_rate_limit(0)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(self.session.local_upload_rate_limit(), int)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.set_local_upload_rate_limit(0)
|
|
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(self.session.download_rate_limit(), int)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.set_download_rate_limit(0)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(self.session.upload_rate_limit(), int)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.set_upload_rate_limit(0)
|
|
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(self.session.max_connections(), int)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.set_max_connections(0)
|
|
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.set_max_uploads(0)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.set_max_half_open_connections(0)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.set_alert_queue_size_limit(0)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.set_alert_mask(0)
|
|
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(
|
|
self.session.proxy(), lt.proxy_type_t.proxy_settings
|
|
)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.set_proxy(lt.proxy_type_t.proxy_settings())
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(
|
|
self.session.dht_proxy(), lt.proxy_type_t.proxy_settings
|
|
)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.set_dht_proxy(lt.proxy_type_t.proxy_settings())
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(
|
|
self.session.peer_proxy(), lt.proxy_type_t.proxy_settings
|
|
)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.set_peer_proxy(lt.proxy_type_t.proxy_settings())
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(
|
|
self.session.tracker_proxy(), lt.proxy_type_t.proxy_settings
|
|
)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.set_tracker_proxy(lt.proxy_type_t.proxy_settings())
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(
|
|
self.session.web_seed_proxy(), lt.proxy_type_t.proxy_settings
|
|
)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.set_web_seed_proxy(lt.proxy_type_t.proxy_settings())
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(
|
|
self.session.i2p_proxy(), lt.proxy_type_t.proxy_settings
|
|
)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.set_i2p_proxy(lt.proxy_type_t.proxy_settings())
|
|
|
|
|
|
class PeSettingsTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.session = lt.session(lib.get_isolated_settings())
|
|
|
|
def test_pe_settings(self) -> None:
|
|
if lt.api_version < 2:
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(self.session.get_pe_settings(), lt.pe_settings)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.set_pe_settings(lt.pe_settings())
|
|
|
|
|
|
class IpFilterTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.session = lt.session(lib.get_isolated_settings())
|
|
|
|
def test_ip_filter(self) -> None:
|
|
self.assertIsInstance(self.session.get_ip_filter(), lt.ip_filter)
|
|
self.session.set_ip_filter(lt.ip_filter())
|
|
|
|
|
|
def unlink_all_files(path: str) -> None:
|
|
for dirpath, _, filenames in os.walk(path):
|
|
for filename in filenames:
|
|
os.unlink(os.path.join(dirpath, filename))
|
|
|
|
|
|
class RemoveTorrentTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
settings = lib.get_isolated_settings()
|
|
settings["alert_mask"] = 0
|
|
self.session = lt.session(settings)
|
|
self.dir = tempfile.TemporaryDirectory()
|
|
self.torrent = tdummy.get_default()
|
|
self.file_path = pathlib.Path(self.dir.name) / os.fsdecode(
|
|
self.torrent.files[0].path
|
|
)
|
|
atp = self.torrent.atp()
|
|
atp.save_path = self.dir.name
|
|
self.handle = self.session.add_torrent(atp)
|
|
|
|
# add_piece() does not work in the checking_* states
|
|
for _ in lib.loop_until_timeout(5, msg="checking"):
|
|
if self.handle.status().state not in (
|
|
lt.torrent_status.checking_files,
|
|
lt.torrent_status.checking_resume_data,
|
|
):
|
|
break
|
|
|
|
# add synthetic data
|
|
for i, data in enumerate(self.torrent.pieces):
|
|
self.handle.add_piece(i, data, 0)
|
|
|
|
# wait until data is written to disk
|
|
for _ in lib.loop_until_timeout(5, msg="file write"):
|
|
if not self.file_path.is_file():
|
|
continue
|
|
if self.file_path.read_bytes() == self.torrent.data:
|
|
break
|
|
|
|
def tearDown(self) -> None:
|
|
lib.cleanup_with_windows_fix(self.dir, timeout=5)
|
|
|
|
def test_remove(self) -> None:
|
|
self.session.remove_torrent(self.handle)
|
|
# There's no good way to synchronize so that we test the data doesn't
|
|
# *eventually* get deleted, but this at least makes the test more
|
|
# readable
|
|
self.assertEqual(self.file_path.read_bytes(), self.torrent.data)
|
|
|
|
def test_remove_data(self) -> None:
|
|
self.session.remove_torrent(self.handle, option=self.session.delete_files)
|
|
for _ in lib.loop_until_timeout(5, msg="file delete"):
|
|
if not self.file_path.is_file():
|
|
break
|
|
|
|
|
|
class TorrentStatusTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.session = lt.session(lib.get_isolated_settings())
|
|
self.dir = tempfile.TemporaryDirectory()
|
|
self.torrent = tdummy.get_default()
|
|
atp = self.torrent.atp()
|
|
atp.save_path = self.dir.name
|
|
self.handle = self.session.add_torrent(atp)
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/6008")
|
|
def test_get_torrents_status(self) -> None:
|
|
# NB: the predicate function may suffer from being being "owned" by
|
|
# libtorrent and inappropriately freed. Test against this by _not_
|
|
# keeping a reference to the functions here
|
|
|
|
status_list = self.session.get_torrent_status(lambda _: True)
|
|
self.assertEqual(len(status_list), 1)
|
|
status = status_list[0]
|
|
# flags should default to 0, so no pieces
|
|
self.assertEqual(status.pieces, [])
|
|
|
|
status_list = self.session.get_torrent_status(
|
|
lambda _: True, flags=lt.status_flags_t.query_pieces
|
|
)
|
|
status = status_list[0]
|
|
self.assertGreater(len(status.pieces), 0)
|
|
|
|
status_list = self.session.get_torrent_status(lambda _: False)
|
|
self.assertEqual(len(status_list), 0)
|
|
|
|
with self.assertRaises(TypeError):
|
|
self.session.get_torrent_status(None) # type: ignore
|
|
with self.assertRaises(TypeError):
|
|
self.session.get_torrent_status(lambda: True) # type: ignore
|
|
|
|
@unittest.skip("https://github.com/arvidn/libtorrent/issues/6008")
|
|
def test_refresh_torrent_status(self) -> None:
|
|
status_list = self.session.get_torrent_status(lambda _: True)
|
|
updated_list = self.session.refresh_torrent_status(status_list)
|
|
self.assertEqual(len(updated_list), 1)
|
|
|
|
|
|
class FieldsTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.session = lt.session(lib.get_isolated_settings())
|
|
|
|
def test_peer_id(self) -> None:
|
|
sha1 = lt.sha1_hash(b"a" * 20)
|
|
if lt.api_version < 2:
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.session.set_peer_id(sha1)
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertEqual(self.session.id(), sha1)
|
|
|
|
def test_num_connections(self) -> None:
|
|
if lt.api_version < 2:
|
|
with self.assertWarns(DeprecationWarning):
|
|
self.assertIsInstance(self.session.num_connections(), int)
|
|
|
|
|
|
class PeerClassTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.session = lt.session(lib.get_isolated_settings())
|
|
|
|
def test_peer_class_filter(self) -> None:
|
|
self.session.set_peer_class_filter(lt.ip_filter())
|
|
|
|
def test_peer_class_type_filter(self) -> None:
|
|
self.session.set_peer_class_type_filter(lt.peer_class_type_filter())
|
|
|
|
def test_create_delete_peer_class(self) -> None:
|
|
pc = self.session.create_peer_class("test")
|
|
self.session.delete_peer_class(pc)
|
|
|
|
def test_get_set_peer_class(self) -> None:
|
|
pci = self.session.get_peer_class(0)
|
|
self.assertIsInstance(pci["ignore_unchoke_slots"], bool)
|
|
self.assertIsInstance(pci["connection_limit_factor"], int)
|
|
self.assertIsInstance(pci["label"], str)
|
|
self.assertIsInstance(pci["upload_limit"], int)
|
|
self.assertIsInstance(pci["download_limit"], int)
|
|
self.assertIsInstance(pci["upload_priority"], int)
|
|
self.assertIsInstance(pci["download_priority"], int)
|
|
|
|
self.session.set_peer_class(0, pci)
|
|
|
|
|
|
class ExtensionTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.session = lt.session(lib.get_isolated_settings())
|
|
|
|
def test_extensions(self) -> None:
|
|
self.session.add_extension(lt.create_smart_ban_plugin)
|
|
self.session.add_extension(lt.create_ut_metadata_plugin)
|
|
self.session.add_extension(lt.create_ut_pex_plugin)
|
|
|
|
# Should not raise an error
|
|
self.session.add_extension("does-not-exist") # type: ignore
|
|
|
|
# TODO: can we test whether extensions are loaded?
|
|
|
|
|
|
class SessionStatsMetricsTest(unittest.TestCase):
|
|
def test_session_stats_metrics(self) -> None:
|
|
metrics = lt.session_stats_metrics()
|
|
self.assertGreater(len(metrics), 0)
|
|
metric = metrics[0]
|
|
self.assertIsInstance(metric.name, str)
|
|
self.assertIsInstance(metric.type, lt.metric_type_t)
|
|
self.assertIsInstance(metric.value_index, int)
|
|
|
|
self.assertEqual(lt.find_metric_idx(metric.name), 0)
|
|
|
|
self.assertLess(lt.find_metric_idx("does-not-exist"), 0)
|
|
|
|
|
|
class SessionStateTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.settings = lib.get_isolated_settings()
|
|
self.session = lt.session(self.settings)
|
|
|
|
def test_session_state(self) -> None:
|
|
# By default, everything should be included
|
|
params = self.session.session_state()
|
|
self.assertLessEqual(set(self.settings.items()), set(params.settings.items()))
|
|
|
|
# With flags=0, nothing should be included
|
|
params = self.session.session_state(flags=0)
|
|
self.assertEqual(params.settings, lt.session_params().settings)
|