This repository has been archived by the owner on Sep 23, 2020. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 82
/
request_test.py
executable file
·110 lines (86 loc) · 3.23 KB
/
request_test.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
import string
import random
import os
import sys
import nose.tools
import sys
import time
import unittest
import tempfile
import filecmp
import uuid
import traceback
from pylantorrent.client import *
from pylantorrent.server import *
from pylantorrent.request import *
class TestRequestXfer(unittest.TestCase):
def setUp(self):
self.src_file = "/etc/group"
self.src_size = os.path.getsize(self.src_file)
self.ports_a = os.environ['LANTORRENT_TEST_PORTS'].split(",")
self.files = []
def tearDown(self):
while len(self.files) > 0:
f = self.files.pop(0)
os.remove(f)
def _get_temp_file(self):
(osf, fname) = tempfile.mkstemp()
os.close(osf)
self.files.append(fname)
return fname
def _t_file_compare(self, f):
rc = filecmp.cmp(self.src_file, f)
self.assertTrue(rc)
def test_request_one(self):
port = int(self.ports_a[0])
host = "localhost:%d" % (port)
fname = self._get_temp_file()
rc = pylantorrent.request.main([self.src_file, fname, str(uuid.uuid1()), host])
self.assertEqual(rc, 0, "rc should be 0 but is %d" % (rc))
self._t_file_compare(fname)
def test_request_many_block(self):
port = int(self.ports_a[0])
host = "localhost:%d" % (port)
for i in range(0, 7):
fname = self._get_temp_file()
rc = pylantorrent.request.main([self.src_file, fname, str(uuid.uuid1()), host])
self.assertEqual(rc, 0, "rc should be 0 but is %d" % (rc))
self._t_file_compare(fname)
def test_request_same_host_nonblock(self):
port = int(self.ports_a[0])
host = "localhost:%d" % (port)
print "requesting all the files"
rids = []
for i in range(0, 10):
fname = self._get_temp_file()
rid = str(uuid.uuid1())
rc = pylantorrent.request.main(["-n", self.src_file, fname, rid, host])
self.assertEqual(rc, 0, "rc should be 0 but is %d" % (rc))
rids.append(rid)
print "waiting on all the files"
for rid in rids:
print "waiting on %s" % (rid)
rc = pylantorrent.request.main(["-a", rid])
self.assertEqual(rc, 0, "rc should be 0 but is %d" % (rc))
print "checking all the files"
for f in self.files:
self._t_file_compare(f)
def test_request_many_host_nonblock(self):
print "requesting all the files"
rids = []
for i in range(0, len(self.ports_a) * 2):
port = int(self.ports_a[i%len(self.ports_a)])
host = "localhost:%d" % (port)
fname = self._get_temp_file()
rid = str(uuid.uuid1())
rc = pylantorrent.request.main(["-n", self.src_file, fname, rid, host])
self.assertEqual(rc, 0, "rc should be 0 but is %d" % (rc))
rids.append(rid)
print "waiting on all the files"
for rid in rids:
print "waiting on %s" % (rid)
rc = pylantorrent.request.main(["-a", rid])
self.assertEqual(rc, 0, "rc should be 0 but is %d" % (rc))
print "checking all the files"
for f in self.files:
self._t_file_compare(f)