]>
Commit | Line | Data |
---|---|---|
cc52de43 | 1 | #!/usr/bin/env python3 |
54007a45 | 2 | |
72f3289a YCH |
3 | # Allow direct execution |
4 | import os | |
5 | import sys | |
6 | import unittest | |
f8271158 | 7 | |
72f3289a YCH |
8 | sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) |
9 | ||
54007a45 | 10 | |
e21f17fc YCH |
11 | import random |
12 | import subprocess | |
ac668111 | 13 | import urllib.request |
72f3289a | 14 | |
ac668111 | 15 | from test.helper import FakeYDL, get_params, is_download_test |
72f3289a | 16 | |
e21f17fc | 17 | |
060ac762 | 18 | @is_download_test |
e21f17fc | 19 | class TestMultipleSocks(unittest.TestCase): |
72f3289a YCH |
20 | @staticmethod |
21 | def _check_params(attrs): | |
22 | params = get_params() | |
23 | for attr in attrs: | |
24 | if attr not in params: | |
25 | print('Missing %s. Skipping.' % attr) | |
26 | return | |
27 | return params | |
28 | ||
29 | def test_proxy_http(self): | |
30 | params = self._check_params(['primary_proxy', 'primary_server_ip']) | |
31 | if params is None: | |
32 | return | |
33 | ydl = FakeYDL({ | |
34 | 'proxy': params['primary_proxy'] | |
35 | }) | |
36 | self.assertEqual( | |
0f06bcd7 | 37 | ydl.urlopen('http://yt-dl.org/ip').read().decode(), |
72f3289a YCH |
38 | params['primary_server_ip']) |
39 | ||
40 | def test_proxy_https(self): | |
41 | params = self._check_params(['primary_proxy', 'primary_server_ip']) | |
42 | if params is None: | |
43 | return | |
44 | ydl = FakeYDL({ | |
45 | 'proxy': params['primary_proxy'] | |
46 | }) | |
47 | self.assertEqual( | |
0f06bcd7 | 48 | ydl.urlopen('https://yt-dl.org/ip').read().decode(), |
72f3289a YCH |
49 | params['primary_server_ip']) |
50 | ||
51 | def test_secondary_proxy_http(self): | |
52 | params = self._check_params(['secondary_proxy', 'secondary_server_ip']) | |
53 | if params is None: | |
54 | return | |
55 | ydl = FakeYDL() | |
ac668111 | 56 | req = urllib.request.Request('http://yt-dl.org/ip') |
72f3289a YCH |
57 | req.add_header('Ytdl-request-proxy', params['secondary_proxy']) |
58 | self.assertEqual( | |
0f06bcd7 | 59 | ydl.urlopen(req).read().decode(), |
72f3289a YCH |
60 | params['secondary_server_ip']) |
61 | ||
62 | def test_secondary_proxy_https(self): | |
63 | params = self._check_params(['secondary_proxy', 'secondary_server_ip']) | |
64 | if params is None: | |
65 | return | |
66 | ydl = FakeYDL() | |
ac668111 | 67 | req = urllib.request.Request('https://yt-dl.org/ip') |
72f3289a YCH |
68 | req.add_header('Ytdl-request-proxy', params['secondary_proxy']) |
69 | self.assertEqual( | |
0f06bcd7 | 70 | ydl.urlopen(req).read().decode(), |
72f3289a YCH |
71 | params['secondary_server_ip']) |
72 | ||
73 | ||
060ac762 | 74 | @is_download_test |
e21f17fc | 75 | class TestSocks(unittest.TestCase): |
98d560f2 YCH |
76 | _SKIP_SOCKS_TEST = True |
77 | ||
e21f17fc | 78 | def setUp(self): |
98d560f2 YCH |
79 | if self._SKIP_SOCKS_TEST: |
80 | return | |
81 | ||
c2876afa | 82 | self.port = random.randint(20000, 30000) |
e21f17fc YCH |
83 | self.server_process = subprocess.Popen([ |
84 | 'srelay', '-f', '-i', '127.0.0.1:%d' % self.port], | |
85 | stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) | |
86 | ||
87 | def tearDown(self): | |
98d560f2 YCH |
88 | if self._SKIP_SOCKS_TEST: |
89 | return | |
90 | ||
e21f17fc YCH |
91 | self.server_process.terminate() |
92 | self.server_process.communicate() | |
93 | ||
94 | def _get_ip(self, protocol): | |
98d560f2 YCH |
95 | if self._SKIP_SOCKS_TEST: |
96 | return '127.0.0.1' | |
97 | ||
e21f17fc YCH |
98 | ydl = FakeYDL({ |
99 | 'proxy': '%s://127.0.0.1:%d' % (protocol, self.port), | |
100 | }) | |
0f06bcd7 | 101 | return ydl.urlopen('http://yt-dl.org/ip').read().decode() |
e21f17fc YCH |
102 | |
103 | def test_socks4(self): | |
14f25df2 | 104 | self.assertTrue(isinstance(self._get_ip('socks4'), str)) |
e21f17fc YCH |
105 | |
106 | def test_socks4a(self): | |
14f25df2 | 107 | self.assertTrue(isinstance(self._get_ip('socks4a'), str)) |
e21f17fc YCH |
108 | |
109 | def test_socks5(self): | |
14f25df2 | 110 | self.assertTrue(isinstance(self._get_ip('socks5'), str)) |
e21f17fc YCH |
111 | |
112 | ||
72f3289a YCH |
113 | if __name__ == '__main__': |
114 | unittest.main() |