[Git][NTPsec/ntpsec][master] Added tests for PeerSummary()
Ian Bruene
gitlab at mg.gitlab.com
Sat Sep 2 00:10:53 UTC 2017
Ian Bruene pushed to branch master at NTPsec / ntpsec
Commits:
7f0f4d80 by Ian Bruene at 2017-09-01T19:10:29-05:00
Added tests for PeerSummary()
- - - - -
3 changed files:
- pylib/util.py
- tests/pylib/jigs.py
- tests/pylib/test_util.py
Changes:
=====================================
pylib/util.py
=====================================
--- a/pylib/util.py
+++ b/pylib/util.py
@@ -784,7 +784,7 @@ class PeerSummary:
@staticmethod
def is_clock(variables):
"Does a set of variables look like it returned from a clock?"
- return "srchost" in variables and '(' in variables["srchost"]
+ return "srchost" in variables and '(' in variables["srchost"][0]
def header(self):
"Column headers for peer display"
=====================================
tests/pylib/jigs.py
=====================================
--- a/tests/pylib/jigs.py
+++ b/tests/pylib/jigs.py
@@ -221,3 +221,11 @@ class ShutilModuleJig:
self.gts_calls.append(default)
return self.gts_returns.pop(0)
+class TimeModuleJig:
+ def __init__(self):
+ self.time_calls = 0
+ self.time_returns = []
+
+ def time(self):
+ self.time_calls += 1
+ return self.time_returns.pop(0)
=====================================
tests/pylib/test_util.py
=====================================
--- a/tests/pylib/test_util.py
+++ b/tests/pylib/test_util.py
@@ -1,6 +1,8 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
+from __future__ import print_function
+
import unittest
import ntp.util
import ntp.packet
@@ -385,6 +387,7 @@ class TestPylibUtilMethods(unittest.TestCase):
c = ntp.util.Cache
monodata = []
+
def monoclock_jig():
return monodata.pop(0)
@@ -440,7 +443,8 @@ class TestPylibUtilMethods(unittest.TestCase):
fakesockmod.gni_error_count = 1
fakesockmod.gni_returns = [("www.Hastur.madness", 42)]
fakesockmod.gai_returns = [(("family", "socktype", "proto",
- "san.Hastur.madness", "42.23.%$.(#"),)]
+ "san.Hastur.madness",
+ "42.23.%$.(#"),)]
self.assertEqual(f("bar:42"), "san.hastur.madness:42")
# Test nameinfo fail, no canonname
fakesockmod.__init__()
@@ -596,6 +600,7 @@ class TestPylibUtilMethods(unittest.TestCase):
od = ntp.util.OrderedDict
termsize = (80, 24)
+
def termsize_jig():
return ntp.util.TermSize(*termsize)
@@ -847,5 +852,252 @@ class TestPylibUtilMethods(unittest.TestCase):
# Test with missing data
self.assertEqual(cls.summary(1, od()), "")
+
+class TestPeerSummary(unittest.TestCase):
+ target = ntp.util.PeerSummary
+
+ def test___init__(self):
+ cls = self.target("peers", 4, True, False)
+ self.assertEqual(cls.displaymode, "peers")
+ self.assertEqual(cls.pktversion, 4)
+ self.assertEqual(cls.showhostnames, True)
+ self.assertEqual(cls.showunits, False)
+ self.assertEqual(cls.wideremote, False)
+ self.assertEqual(cls.debug, 0)
+ self.assertEqual(cls.logfp, sys.stderr)
+ self.assertEqual(cls.termwidth, None)
+ self.assertEqual(cls.horizontal_slack, 0)
+ self.assertEqual(cls.namewidth, 15)
+ self.assertEqual(cls.refidwidth, 15)
+ self.assertEqual(cls._PeerSummary__remote, " remote ")
+ self.assertEqual(cls._PeerSummary__common,
+ "st t when poll reach delay offset ")
+ self.assertEqual(cls._PeerSummary__header, None)
+ self.assertEqual(cls.polls, [])
+
+ def test_prettyinterval(self):
+ m = self.target.prettyinterval
+
+ # Test invalid
+ self.assertEqual(m("Failure"), "-")
+ # Test <=2048
+ self.assertEqual(m(2048), "2048")
+ # Test <=300
+ self.assertEqual(m(17971), "300m")
+ # Test <=96
+ self.assertEqual(m(343831), "96h")
+ # Test final
+ self.assertEqual(m(350000), "4d")
+
+ def test_high_truncate(self):
+ m = self.target.high_truncate
+
+ # Test fit
+ self.assertEqual(m("foo", 10), "foo")
+ # Test doesn't fit
+ self.assertEqual(m("lorem ipsum", 5), "-psum")
+
+ def test_is_clock(self):
+ m = self.target.is_clock
+
+ # Test True
+ data = {"srchost": "(would)"}
+ self.assertEqual(m(data), True)
+ # Test False, not a clock
+ data = {"srchost": "wouldn't"}
+ self.assertEqual(m(data), False)
+ # Test False, no host
+ data = {"blah": "definately wouldn't"}
+ self.assertEqual(m(data), False)
+
+ def test_header(self):
+ # Test peers
+ cls = self.target("peers", 4, True, False)
+ self.assertEqual(cls.header(),
+ " remote refid st t when "
+ "poll reach delay offset jitter")
+ # Test opeers
+ cls = self.target("opeers", 4, True, False)
+ self.assertEqual(cls.header(),
+ " remote local st t when "
+ "poll reach delay offset disp")
+ # Test apeers
+ cls = self.target("apeers", 4, True, False)
+ self.assertEqual(cls.header(),
+ " remote refid assid st t when "
+ "poll reach delay offset jitter")
+
+ def test_width(self):
+ cls = self.target("peers", 4, True, False)
+ self.assertEqual(cls.width(), 79)
+ cls.horizontal_slack = 10
+ self.assertEqual(cls.width(), 89)
+
+ def test_summary(self):
+ cls = self.target("peers", 4, True, False)
+ cls.header()
+
+ cdns_jig_calls = []
+ cdns_jig_returns = []
+
+ def cdns_jig(ip): # canonicalize_dns()
+ cdns_jig_calls.append(ip)
+ return cdns_jig_returns.pop(0)
+
+ data = ntp.util.OrderedDict((("delay", (1.234567, "1.234567")),
+ ("dstadr", ("1.2.3.4", "1.2.3.4")),
+ ("dstport", ("blah0", "blah0")),
+ ("filtdelay", ("blah1", "blah1")),
+ ("filtdisp", ("blah2", "blah2")),
+ ("filtoffset", ("blah3", "blah3")),
+ ("flash", ("blah4", "blah4")),
+ ("headway", ("blah5", "blah5")),
+ ("hmode", (6, "6")),
+ ("hpoll", (12, "12")),
+ ("jitter", (3.14159, "3.14159")),
+ ("keyid", ("blah6", "blah6")),
+ ("leap", ("blah7", "blah7")),
+ ("offset", (2.71828, "2.71828")),
+ ("pmode", ("blah8", "blah8")),
+ ("ppoll", (5, "5")),
+ ("precision", ("blah9", "blah9")),
+ ("reach", (500, "500")),
+ ("refid", ("FAIL", "FAIL")),
+ ("rec", ("0x00000000.00000000",
+ "0x00000000.00000000")),
+ ("reftime", ("0x10000000.00000000",
+ "0x10000000.00000000")),
+ ("rootdelay", ("blah10", "blah10")),
+ ("rootdisp", (299792.458, "299792.458")),
+ ("srcadr", ("10.20.30.40",
+ "10.20.30.40")),
+ ("srchost", ("15.25.35.45",
+ "15.25.35.45")),
+ ("scrport", ("blah11", "blah11")),
+ ("stratum", (8, "8")),
+ ("ttl", ("blah12", "blah12")),
+ ("unreach", ("blah13", "blah13")),
+ ("xmt", ("blah14", "blah14")),
+ ("randomness!", ("foo!", "bar!"))))
+ faketimemod = jigs.TimeModuleJig()
+ try:
+ timetemp = ntp.util.time
+ ntp.util.time = faketimemod
+ cdnstemp = ntp.util.canonicalize_dns
+ ntp.util.canonicalize_dns = cdns_jig
+ # Test, no units, hmode=BCLIENTX, peers
+ cdns_jig_returns = ["clock_canon"]
+ faketimemod.time_returns = [0]
+ self.assertEqual(cls.summary(5, data, 12345),
+ " clock_canon .FAIL. 8 b -"
+ " 32 764 1.2346 2.7183 3.1416\n")
+ # Test, no units, hmode=BROADCAST, not multicast
+ data["hmode"] = (5, "5")
+ cdns_jig_returns = ["clock_canon"]
+ faketimemod.time_returns = [0]
+ self.assertEqual(cls.summary(5, data, 12345),
+ " clock_canon .FAIL. 8 B -"
+ " 32 764 1.2346 2.7183 3.1416\n")
+ # Test, no units, hmode=BROADCAST, not multicast
+ data["srcadr"] = ("224.2.3.4", "224.2.3.4")
+ cdns_jig_returns = ["clock_canon"]
+ faketimemod.time_returns = [0]
+ self.assertEqual(cls.summary(5, data, 12345),
+ " clock_canon .FAIL. 8 M -"
+ " 32 764 1.2346 2.7183 3.1416\n")
+ # Test, no units, hmode=CLIENT, local refclock
+ data["srcadr"] = ("10.20.30.40", "10.20.30.40")
+ data["hmode"] = (3, "3")
+ data["srchost"] = ("(blah)", "(blah)")
+ cdns_jig_returns = ["clock_canon"]
+ faketimemod.time_returns = [0]
+ self.assertEqual(cls.summary(5, data, 12345),
+ " clock_canon .FAIL. 8 l -"
+ " 32 764 1.2346 2.7183 3.1416\n")
+ # Test, no units, hmode=CLIENT, pool
+ data["srchost"] = ("15.25.35.45", "15.25.35.45")
+ data["refid"] = ("POOL", "POOL")
+ cdns_jig_returns = ["clock_canon"]
+ faketimemod.time_returns = [0]
+ self.assertEqual(cls.summary(5, data, 12345),
+ " clock_canon .POOL. 8 p -"
+ " 32 764 1.2346 2.7183 3.1416\n")
+ # Test, no units, hmode=CLIENT, manycast client
+ data["srcadr"] = ("224.2.3.4", "224.2.3.4")
+ data["refid"] = ("FAIL", "FAIL")
+ cdns_jig_returns = ["clock_canon"]
+ faketimemod.time_returns = [0]
+ self.assertEqual(cls.summary(5, data, 12345),
+ " clock_canon .FAIL. 8 a -"
+ " 32 764 1.2346 2.7183 3.1416\n")
+ # Test, no units, hmode=CLIENT, unicast
+ data["srcadr"] = ("10.20.30.40", "10.20.30.40")
+ cdns_jig_returns = ["clock_canon"]
+ faketimemod.time_returns = [0]
+ self.assertEqual(cls.summary(5, data, 12345),
+ " clock_canon .FAIL. 8 u -"
+ " 32 764 1.2346 2.7183 3.1416\n")
+ # Test, no units, hmode=ACTIVE
+ data["hmode"] = (1, "1")
+ cdns_jig_returns = ["clock_canon"]
+ faketimemod.time_returns = [0]
+ self.assertEqual(cls.summary(5, data, 12345),
+ " clock_canon .FAIL. 8 s -"
+ " 32 764 1.2346 2.7183 3.1416\n")
+ # Test, no units, hmode=PASSIVE
+ data["hmode"] = (2, "2")
+ cdns_jig_returns = ["clock_canon"]
+ faketimemod.time_returns = [0]
+ self.assertEqual(cls.summary(5, data, 12345),
+ " clock_canon .FAIL. 8 S -"
+ " 32 764 1.2346 2.7183 3.1416\n")
+ # Test, no units, don't show hostnames
+ cls.showhostnames = False
+ cdns_jig_returns = ["clock_canon"]
+ faketimemod.time_returns = [0]
+ self.assertEqual(cls.summary(5, data, 12345),
+ " 10.20.30.40 .FAIL. 8 S -"
+ " 32 764 1.2346 2.7183 3.1416\n")
+ # Test, no units, name crop
+ cls.showhostnames = True
+ cdns_jig_returns = ["clock_canon_blah_jabber_quantum"]
+ faketimemod.time_returns = [0]
+ self.assertEqual(cls.summary(5, data, 12345),
+ " clock_canon_bla .FAIL. 8 S -"
+ " 32 764 1.2346 2.7183 3.1416\n")
+ # Test, no units, name crop
+ cls.wideremote = True
+ cdns_jig_returns = ["clock_canon_blah_jabber_quantum"]
+ faketimemod.time_returns = [0]
+ self.assertEqual(cls.summary(5, data, 12345),
+ " clock_canon_blah_jabber_quantum\n"
+ " .FAIL. 8 S"
+ " - 32 764 1.2346 2.7183 3.1416\n")
+ # Test, with units
+ cls.showunits = True
+ cdns_jig_returns = ["clock_canon"]
+ faketimemod.time_returns = [0]
+ self.assertEqual(cls.summary(5, data, 12345),
+ " clock_canon .FAIL. 8 S -"
+ " 32 764 1.2346ms 2.7183ms 3.1416ms\n")
+ # Test, apeers
+ cls.showunits = True
+ cls.displaymode = "apeers"
+ cdns_jig_returns = ["clock_canon"]
+ faketimemod.time_returns = [0]
+ self.assertEqual(cls.summary(5, data, 12345),
+ " clock_canon .FAIL. 12345 8 S -"
+ " 32 764 1.2346ms 2.7183ms 3.1416ms\n")
+ finally:
+ ntp.util.time = timetemp
+ ntp.util.canonicalize_dns = cdnstemp
+
+ def test_intervals(self):
+ cls = self.target("peers", 4, True, False)
+ cls.polls = [1, 2, 3, 4, 5]
+ self.assertEqual(cls.intervals(), [1, 2, 3, 4, 5])
+ self.assertEqual(cls.polls, [])
+
+
if __name__ == '__main__':
unittest.main()
View it on GitLab: https://gitlab.com/NTPsec/ntpsec/commit/7f0f4d80d030ae7bc22546d2257ab4ef26b12c3e
---
View it on GitLab: https://gitlab.com/NTPsec/ntpsec/commit/7f0f4d80d030ae7bc22546d2257ab4ef26b12c3e
You're receiving this email because of your account on gitlab.com.
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <https://lists.ntpsec.org/pipermail/vc/attachments/20170902/afea3065/attachment.html>
More information about the vc
mailing list