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