Python event log analyzer tool: fix of indexing issue.
Test added to ensure it doesn't happen again. NOTRY=True Review-Url: https://codereview.webrtc.org/2316573003 Cr-Commit-Position: refs/heads/master@{#14121}
This commit is contained in:
@ -22,7 +22,6 @@ import misc
|
|||||||
|
|
||||||
|
|
||||||
class TestMisc(unittest.TestCase):
|
class TestMisc(unittest.TestCase):
|
||||||
|
|
||||||
def testUnwrapMod3(self):
|
def testUnwrapMod3(self):
|
||||||
data = [0, 1, 2, 0, -1, -2, -3, -4]
|
data = [0, 1, 2, 0, -1, -2, -3, -4]
|
||||||
unwrapped_3 = misc.unwrap(data, 3)
|
unwrapped_3 = misc.unwrap(data, 3)
|
||||||
|
|||||||
@ -280,7 +280,7 @@ def calculate_delay(start, stop, step, points):
|
|||||||
masked array, in which time points with no value are masked.
|
masked array, in which time points with no value are masked.
|
||||||
|
|
||||||
"""
|
"""
|
||||||
grouped_delays = [[] for _ in numpy.arange(start, stop, step)]
|
grouped_delays = [[] for _ in numpy.arange(start, stop + step, step)]
|
||||||
rounded_value_index = lambda x: int((x - start) / step)
|
rounded_value_index = lambda x: int((x - start) / step)
|
||||||
for point in points:
|
for point in points:
|
||||||
grouped_delays[rounded_value_index(point.real_send_time_ms)
|
grouped_delays[rounded_value_index(point.real_send_time_ms)
|
||||||
|
|||||||
62
webrtc/tools/py_event_log_analyzer/rtp_analyzer_test.py
Executable file
62
webrtc/tools/py_event_log_analyzer/rtp_analyzer_test.py
Executable file
@ -0,0 +1,62 @@
|
|||||||
|
#!/usr/bin/env python
|
||||||
|
# Copyright (c) 2016 The WebRTC project authors. All Rights Reserved.
|
||||||
|
#
|
||||||
|
# Use of this source code is governed by a BSD-style license
|
||||||
|
# that can be found in the LICENSE file in the root of the source
|
||||||
|
# tree. An additional intellectual property rights grant can be found
|
||||||
|
# in the file PATENTS. All contributing project authors may
|
||||||
|
# be found in the AUTHORS file in the root of the source tree.
|
||||||
|
|
||||||
|
"""Run the tests with
|
||||||
|
|
||||||
|
python rtp_analyzer_test.py
|
||||||
|
or
|
||||||
|
python3 rtp_analyzer_test.py
|
||||||
|
"""
|
||||||
|
|
||||||
|
import collections
|
||||||
|
import unittest
|
||||||
|
|
||||||
|
missing_numpy = False
|
||||||
|
try:
|
||||||
|
import numpy
|
||||||
|
import rtp_analyzer
|
||||||
|
except ImportError:
|
||||||
|
missing_numpy = True
|
||||||
|
|
||||||
|
FakePoint = collections.namedtuple("FakePoint",
|
||||||
|
["real_send_time_ms", "absdelay"])
|
||||||
|
|
||||||
|
|
||||||
|
class TestDelay(unittest.TestCase):
|
||||||
|
def assertMaskEqual(self, masked_array, data, mask):
|
||||||
|
self.assertEqual(list(masked_array.data), data)
|
||||||
|
|
||||||
|
if isinstance(masked_array.mask, numpy.bool_):
|
||||||
|
array_mask = masked_array.mask
|
||||||
|
else:
|
||||||
|
array_mask = list(masked_array.mask)
|
||||||
|
self.assertEqual(array_mask, mask)
|
||||||
|
|
||||||
|
def testCalculateDelaySimple(self):
|
||||||
|
points = [FakePoint(0, 0), FakePoint(1, 0)]
|
||||||
|
mask = rtp_analyzer.calculate_delay(0, 1, 1, points)
|
||||||
|
self.assertMaskEqual(mask, [0, 0], False)
|
||||||
|
|
||||||
|
def testCalculateDelayMissing(self):
|
||||||
|
points = [FakePoint(0, 0), FakePoint(2, 0)]
|
||||||
|
mask = rtp_analyzer.calculate_delay(0, 2, 1, points)
|
||||||
|
self.assertMaskEqual(mask, [0, -1, 0], [False, True, False])
|
||||||
|
|
||||||
|
def testCalculateDelayBorders(self):
|
||||||
|
points = [FakePoint(0, 0), FakePoint(2, 0)]
|
||||||
|
mask = rtp_analyzer.calculate_delay(0, 3, 2, points)
|
||||||
|
self.assertMaskEqual(mask, [0, 0, -1], [False, False, True])
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
if missing_numpy:
|
||||||
|
# pylint: disable=superfluous-parens
|
||||||
|
print("Missing numpy, skipping test.")
|
||||||
|
else:
|
||||||
|
unittest.main()
|
||||||
Reference in New Issue
Block a user