mirror of
https://github.com/arvidn/libtorrent.git
synced 2026-07-22 03:55:51 -04:00
Arvid Norberg
136 lines
4.1 KiB
Python
136 lines
4.1 KiB
Python
# mypy: disable-error-code="unreachable"
|
|
import functools
|
|
import http.server
|
|
import logging
|
|
import os
|
|
import pathlib
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
from typing import Any
|
|
from typing import cast
|
|
from typing import TYPE_CHECKING
|
|
import unittest
|
|
|
|
import libtorrent as lt
|
|
|
|
from . import lib
|
|
from . import tdummy
|
|
|
|
if TYPE_CHECKING:
|
|
from libtorrent import _Entry
|
|
|
|
# import fails on windows
|
|
if os.name != "nt":
|
|
import pty
|
|
|
|
|
|
class Handler(http.server.BaseHTTPRequestHandler):
|
|
def __init__(self, *args: Any, serve_data: bytes, **kwargs: Any) -> None:
|
|
self.serve_data = serve_data
|
|
super().__init__(*args, **kwargs)
|
|
|
|
def do_GET(self) -> None:
|
|
logging.info("%s", self.requestline)
|
|
self.send_response(200)
|
|
self.send_header("content-length", str(len(self.serve_data)))
|
|
self.end_headers()
|
|
self.wfile.write(self.serve_data)
|
|
|
|
|
|
class TestClient(unittest.TestCase):
|
|
script_path: pathlib.Path
|
|
tempdir_path: pathlib.Path
|
|
torrent_path: pathlib.Path
|
|
pty_master: Any
|
|
pty_slave: Any
|
|
torrent: tdummy.Torrent
|
|
|
|
def setUp(self) -> None:
|
|
if not sys.platform.startswith("linux"):
|
|
return
|
|
|
|
# path relative to this file
|
|
self.script_path = pathlib.Path(__file__).parent / ".." / "client.py"
|
|
|
|
self.tempdir = tempfile.TemporaryDirectory()
|
|
self.tempdir_path = pathlib.Path(self.tempdir.name)
|
|
|
|
self.torrent = tdummy.get_default()
|
|
|
|
# set up a web seed to serve dummy torrent data
|
|
self.server = http.server.HTTPServer(
|
|
("127.0.0.1", 0), functools.partial(Handler, serve_data=self.torrent.data)
|
|
)
|
|
addr = self.server.server_address[0]
|
|
port = self.server.server_address[1]
|
|
self.server_thread = threading.Thread(target=self.server.serve_forever)
|
|
self.server_thread.start()
|
|
|
|
# construct a .torrent to point to our web seed
|
|
tdict = dict(self.torrent.dict)
|
|
tdict[b"url-list"] = b"".join(
|
|
[b"http://", str(addr).encode(), b":", str(port).encode()]
|
|
)
|
|
self.tdict = cast("_Entry", tdict)
|
|
self.torrent_path = self.tempdir_path / "dummy.torrent"
|
|
self.torrent_path.write_bytes(lt.bencode(self.tdict))
|
|
|
|
self.pty_master, self.pty_slave = pty.openpty()
|
|
|
|
def tearDown(self) -> None:
|
|
if sys.platform != "linux":
|
|
return
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
self.tempdir.cleanup()
|
|
os.close(self.pty_master)
|
|
os.close(self.pty_slave)
|
|
|
|
@unittest.skipIf(sys.platform != "linux", "pty control only works on linux")
|
|
def test_download_from_web_seed(self) -> None:
|
|
proc = subprocess.Popen(
|
|
[
|
|
sys.executable,
|
|
str(self.script_path),
|
|
"--port",
|
|
"0",
|
|
"--listen-interface",
|
|
"127.0.0.1",
|
|
str(self.torrent_path),
|
|
],
|
|
cwd=self.tempdir_path,
|
|
stdin=self.pty_slave,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.PIPE,
|
|
universal_newlines=True,
|
|
)
|
|
|
|
try:
|
|
# wait for the file to be downloaded from the web seed
|
|
file_path = self.tempdir_path / os.fsdecode(self.torrent.files[0].path)
|
|
for _ in lib.loop_until_timeout(15):
|
|
try:
|
|
if file_path.read_bytes() == self.torrent.data:
|
|
break
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
# send a 'q' command
|
|
os.write(self.pty_master, b"q")
|
|
# process should terminate on its own
|
|
returncode = proc.wait(15)
|
|
finally:
|
|
proc.kill()
|
|
|
|
# process should complete without errors or warnings
|
|
self.assertEqual(returncode, 0)
|
|
assert proc.stderr is not None # helps mypy
|
|
self.assertEqual(proc.stderr.read(), "")
|
|
proc.stderr.close()
|
|
|
|
# fastresume should be written
|
|
fastresume = pathlib.Path(f"{file_path}.fastresume")
|
|
lt.bdecode(fastresume.read_bytes())
|