telemetry/third_party/webpagereplay/proxyshaper_test.py - Issue 2210063003: Rename third_party/webpagereplay to third_party/web-page-replay

Side by Side Diff: telemetry/third_party/webpagereplay/proxyshaper_test.py

Issue 2210063003: Rename third_party/webpagereplay to third_party/web-page-replay (Closed) Base URL: https://github.com/catapult-project/catapult@master

Patch Set: Created 4 years, 4 months ago

Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.

Jump to:

View unified diff | Download patch

OLD	NEW
	(Empty)
1 #!/usr/bin/env python

2 # Copyright 2012 Google Inc. All Rights Reserved.

3 #

4 # Licensed under the Apache License, Version 2.0 (the "License");

5 # you may not use this file except in compliance with the License.

6 # You may obtain a copy of the License at

7 #

8 # http://www.apache.org/licenses/LICENSE-2.0

9 #

10 # Unless required by applicable law or agreed to in writing, software

11 # distributed under the License is distributed on an "AS IS" BASIS,

12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

13 # See the License for the specific language governing permissions and

14 # limitations under the License.

15

16 """Unit tests for proxyshaper.

17

18 Usage:

19 $ ./proxyshaper_test.py

20 """

21

22 import proxyshaper

23 import StringIO

24 import unittest

25

26

27 # pylint: disable=bad-whitespace

28 VALID_RATES = (

29 # input, expected_bps

30 ( '384Kbit/s', 384000),

31 ('1536Kbit/s', 1536000),

32 ( '1Mbit/s', 1000000),

33 ( '5Mbit/s', 5000000),

34 ( '2MByte/s', 16000000),

35 ( '0', 0),

36 ( '5', 5),

37 ( 384000, 384000),

38 )

39

40 ERROR_RATES = (

41 '1536KBit/s', # Older versions of dummynet used capital 'B' for bytes.

42 '1Mbyte/s', # Require capital 'B' for bytes.

43 '5bps',

44 )

45

46

47 class TimedTestCase(unittest.TestCase):

48 def assertValuesAlmostEqual(self, expected, actual, tolerance=0.05):

49 """Like the following with nicer default message:

50 assertTrue(expected <= actual + tolerance &&

51 expected >= actual - tolerance)

52 """

53 delta = tolerance * expected

54 if actual > expected + delta or actual < expected - delta:

55 self.fail('%s is not equal to expected %s +/- %s%%' % (

56 actual, expected, 100 * tolerance))

57

58

59 class RateLimitedFileTest(TimedTestCase):

60 def testReadLimitedBasic(self):

61 num_bytes = 1024

62 bps = 384000

63 request_counter = lambda: 1

64 f = StringIO.StringIO(' ' * num_bytes)

65 limited_f = proxyshaper.RateLimitedFile(request_counter, f, bps)

66 start = proxyshaper.TIMER()

67 self.assertEqual(num_bytes, len(limited_f.read()))

68 expected_ms = 8.0 * num_bytes / bps * 1000.0

69 actual_ms = (proxyshaper.TIMER() - start) * 1000.0

70 self.assertValuesAlmostEqual(expected_ms, actual_ms)

71

72 def testReadlineLimitedBasic(self):

73 num_bytes = 1024 * 8 + 512

74 bps = 384000

75 request_counter = lambda: 1

76 f = StringIO.StringIO(' ' * num_bytes)

77 limited_f = proxyshaper.RateLimitedFile(request_counter, f, bps)

78 start = proxyshaper.TIMER()

79 self.assertEqual(num_bytes, len(limited_f.readline()))

80 expected_ms = 8.0 * num_bytes / bps * 1000.0

81 actual_ms = (proxyshaper.TIMER() - start) * 1000.0

82 self.assertValuesAlmostEqual(expected_ms, actual_ms)

83

84 def testReadLimitedSlowedByMultipleRequests(self):

85 num_bytes = 1024

86 bps = 384000

87 request_count = 2

88 request_counter = lambda: request_count

89 f = StringIO.StringIO(' ' * num_bytes)

90 limited_f = proxyshaper.RateLimitedFile(request_counter, f, bps)

91 start = proxyshaper.TIMER()

92 num_read_bytes = limited_f.read()

93 self.assertEqual(num_bytes, len(num_read_bytes))

94 expected_ms = 8.0 * num_bytes / (bps / float(request_count)) * 1000.0

95 actual_ms = (proxyshaper.TIMER() - start) * 1000.0

96 self.assertValuesAlmostEqual(expected_ms, actual_ms)

97

98 def testWriteLimitedBasic(self):

99 num_bytes = 1024 * 10 + 350

100 bps = 384000

101 request_counter = lambda: 1

102 f = StringIO.StringIO()

103 limited_f = proxyshaper.RateLimitedFile(request_counter, f, bps)

104 start = proxyshaper.TIMER()

105 limited_f.write(' ' * num_bytes)

106 self.assertEqual(num_bytes, len(limited_f.getvalue()))

107 expected_ms = 8.0 * num_bytes / bps * 1000.0

108 actual_ms = (proxyshaper.TIMER() - start) * 1000.0

109 self.assertValuesAlmostEqual(expected_ms, actual_ms)

110

111 def testWriteLimitedSlowedByMultipleRequests(self):

112 num_bytes = 1024 * 10

113 bps = 384000

114 request_count = 2

115 request_counter = lambda: request_count

116 f = StringIO.StringIO(' ' * num_bytes)

117 limited_f = proxyshaper.RateLimitedFile(request_counter, f, bps)

118 start = proxyshaper.TIMER()

119 limited_f.write(' ' * num_bytes)

120 self.assertEqual(num_bytes, len(limited_f.getvalue()))

121 expected_ms = 8.0 * num_bytes / (bps / float(request_count)) * 1000.0

122 actual_ms = (proxyshaper.TIMER() - start) * 1000.0

123 self.assertValuesAlmostEqual(expected_ms, actual_ms)

124

125

126 class GetBitsPerSecondTest(unittest.TestCase):

127 def testConvertsValidValues(self):

128 for dummynet_option, expected_bps in VALID_RATES:

129 bps = proxyshaper.GetBitsPerSecond(dummynet_option)

130 self.assertEqual(

131 expected_bps, bps, 'Unexpected result for %s: %s != %s' % (

132 dummynet_option, expected_bps, bps))

133

134 def testRaisesOnUnexpectedValues(self):

135 for dummynet_option in ERROR_RATES:

136 self.assertRaises(proxyshaper.BandwidthValueError,

137 proxyshaper.GetBitsPerSecond, dummynet_option)

138

139

140 if __name__ == '__main__':

141 unittest.main()

OLD	NEW

« no previous file with comments | « telemetry/third_party/webpagereplay/proxyshaper.py ('k') | telemetry/third_party/webpagereplay/pylintrc » ('j') | no next file with comments »