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
|
# Copyright 2016 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import unittest
from py_utils import ts_proxy_server
class TsProxyServerTest(unittest.TestCase):
def testParseTsProxyPort(self):
self.assertEquals(
ts_proxy_server.ParseTsProxyPortFromOutput(
'Started Socks5 proxy server on 127.0.0.1:54430 \n'),
54430)
self.assertEquals(
ts_proxy_server.ParseTsProxyPortFromOutput(
'Started Socks5 proxy server on foo.bar.com:430 \n'),
430)
self.assertEquals(
ts_proxy_server.ParseTsProxyPortFromOutput(
'Failed to start sock5 proxy.'),
None)
def testSmokeStartingTsProxyServer(self):
with ts_proxy_server.TsProxyServer() as server:
self.assertIsNotNone(server.port)
with ts_proxy_server.TsProxyServer(None, 37124, 37125) as server:
self.assertIsNotNone(server.port)
def testSmokeUpdatingOutboundPorts(self):
with ts_proxy_server.TsProxyServer() as server:
self.assertIsNotNone(server.port)
server.UpdateOutboundPorts(31242, 14220)
def testSmokeUpdateOutboundPortsInvalid(self):
with ts_proxy_server.TsProxyServer() as server:
self.assertIsNotNone(server.port)
with self.assertRaises(AssertionError):
server.UpdateOutboundPorts(31242, 'abcde')
def testSmokeUpdateTrafficSettings(self):
with ts_proxy_server.TsProxyServer() as server:
server.UpdateTrafficSettings(round_trip_latency_ms=100)
server.UpdateTrafficSettings(download_bandwidth_kbps=5000)
server.UpdateTrafficSettings(upload_bandwidth_kbps=2000)
self.assertEquals(server._rtt, 100)
self.assertEquals(server._inbkps, 5000)
self.assertEquals(server._outkbps, 2000)
server.UpdateTrafficSettings(
round_trip_latency_ms=200, download_bandwidth_kbps=500,
upload_bandwidth_kbps=2000)
self.assertEquals(server._rtt, 200)
self.assertEquals(server._inbkps, 500)
self.assertEquals(server._outkbps, 2000)
|
