[Git][NTPsec/ntpsec][master] 5 commits: Removed extraneous debugging print.
Ian Bruene
gitlab at mg.gitlab.com
Fri Jan 12 20:38:23 UTC 2018
Ian Bruene pushed to branch master at NTPsec / ntpsec
Commits:
4430a634 by Ian Bruene at 2018-01-11T09:52:01-06:00
Removed extraneous debugging print.
- - - - -
af3dd0c8 by Ian Bruene at 2018-01-11T11:49:59-06:00
Brought statfiles.py tests to 100% coverage, greately simplified tests.
- - - - -
5a37efa6 by Ian Bruene at 2018-01-11T11:51:11-06:00
Removed python 2/3 shims from coverage readouts.
- - - - -
6909572b by Ian Bruene at 2018-01-12T13:56:16-06:00
Brought util.py tests up to 100% coverage.
- - - - -
79f95763 by Ian Bruene at 2018-01-12T14:37:26-06:00
Brought agentx.py tests to 100% coverage.
- - - - -
7 changed files:
- pylib/agentx.py
- pylib/statfiles.py
- pylib/util.py
- tests/pylib/test_agentx.py
- tests/pylib/test_packet.py
- tests/pylib/test_statfiles.py
- tests/pylib/test_util.py
Changes:
=====================================
pylib/agentx.py
=====================================
--- a/pylib/agentx.py
+++ b/pylib/agentx.py
@@ -9,7 +9,8 @@ from ntp.util import slicedata
master_encoding = 'latin-1'
-if str is bytes: # Python 2
+if str is bytes: # pragma: no cover
+ # Python 2
polystr = str
polybytes = bytes
polyord = ord
@@ -19,7 +20,8 @@ if str is bytes: # Python 2
def string_escape(s):
return s.decode('string_escape')
-else: # Python 3
+else: # pragma: no cover
+ # Python 3
import io
def polystr(o):
@@ -255,7 +257,7 @@ def decode_ClosePDU(data, header):
class ClosePDU(AgentXPDU):
def __init__(self, bigEndian, sID, tactID, pktID, reason):
AgentXPDU.__init__(self, PDU_CLOSE, bigEndian, sID, tactID, pktID)
- if reason not in definedReasons:
+ if reason not in definedReasons: # pragma: no cover
raise ValueError("Close reason %s not in defined types" % reason)
self.reason = reason
@@ -1045,7 +1047,7 @@ def decode_unsigned32(data, header):
return (num, data)
-def sanity_unsigned32(data):
+def sanity_unsigned32(data): # pragma: no cover
if data != (data & 0xFFFFFFFF):
raise ValueError
@@ -1284,7 +1286,7 @@ def decode_packet(data):
decoder = definedPDUTypes[pktType]
try:
parsedPkt = decoder(packetSlice, header)
- except Exception:
+ except Exception: # pragma: no cover
err = ParseError("Body parsing error", header, packetSlice, newData)
raise err
return parsedPkt, newData
=====================================
pylib/statfiles.py
=====================================
--- a/pylib/statfiles.py
+++ b/pylib/statfiles.py
@@ -129,7 +129,7 @@ class NTPStats:
# boring, use hostname
self.sitename = socket.getfqdn()
- if not os.path.isdir(statsdir):
+ if not os.path.isdir(statsdir): # pragma: no cover
sys.stderr.write("ntpviz: ERROR: %s is not a directory\n"
% statsdir)
raise SystemExit(1)
@@ -161,7 +161,7 @@ class NTPStats:
lines += gzip.open(logpart, 'rt').readlines()
else:
lines += open(logpart, 'r').readlines()
- except IOError:
+ except IOError: # pragma: no cover
sys.stderr.write("ntpviz: WARNING: could not read %s\n"
% logpart)
pass
@@ -212,7 +212,7 @@ class NTPStats:
if ip not in self.peermap:
self.peermap[ip] = []
self.peermap[ip].append(row)
- except IndexError:
+ except IndexError: # pragma: no cover
# ignore corrupted rows
pass
return self.peermap
@@ -226,7 +226,7 @@ class NTPStats:
if source not in gpsmap:
gpsmap[source] = []
gpsmap[source].append(row)
- except IndexError:
+ except IndexError: # pragma: no cover
# ignore corrupted rows
pass
return gpsmap
@@ -240,7 +240,7 @@ class NTPStats:
if source not in tempsmap:
tempsmap[source] = []
tempsmap[source].append(row)
- except IndexError:
+ except IndexError: # pragma: no cover
# ignore corrupted rows
pass
return tempsmap
=====================================
pylib/util.py
=====================================
--- a/pylib/util.py
+++ b/pylib/util.py
@@ -47,13 +47,16 @@ UNITS_PPX = [UNIT_PPT, UNIT_PPB, UNIT_PPM, UNIT_PPK]
unitgroups = (UNITS_SEC, UNITS_PPX)
-def check_unicode():
+# These two functions are not tested because they will muck up the module
+# for everything else, and they are simple.
+
+def check_unicode(): # pragma: no cover
if "UTF-8" != sys.stdout.encoding:
deunicode_units()
return True # needed by ntpmon
return False
-def deunicode_units():
+def deunicode_units(): # pragma: no cover
"Under certain conditions it is not possible to force unicode output, "
"this overwrites units that contain unicode with safe versions"
global UNIT_US
@@ -61,7 +64,7 @@ def deunicode_units():
# Replacement units
new_us = "us"
new_ppk = "ppk"
- # Replace units in unit group
+ # Replace units in unit groups
UNITS_SEC[UNITS_SEC.index(UNIT_US)] = new_us
UNITS_PPX[UNITS_PPX.index(UNIT_PPK)] = new_ppk
# Replace the units themselves
@@ -600,7 +603,7 @@ if str is bytes: # We are on python 2.x
import struct
-def termsize():
+def termsize(): # pragma: no cover
"Return the current terminal size."
# Alternatives at http://stackoverflow.com/questions/566746
# The way this is used makes it not a big deal if the default is wrong.
@@ -1067,7 +1070,7 @@ class PeerSummary:
clock_name = canonicalize_dns(srcadr)
if self.debug:
self.logfp.write("DNS lookup ends.\n")
- except TypeError:
+ except TypeError: # pragma: no cover
return ''
else:
clock_name = srcadr
@@ -1305,7 +1308,7 @@ class IfstatsSummary:
address))
if bcast:
s += " %s\n" % bcast
- except TypeError:
+ except TypeError: # pragma: no cover
# Can happen when ntpd ships a corrupted response
return ''
@@ -1319,7 +1322,7 @@ class IfstatsSummary:
try:
from collections import OrderedDict
-except ImportError:
+except ImportError: # pragma: no cover
class OrderedDict(dict):
"A stupid simple implementation in order to be back-portable to 2.6"
=====================================
tests/pylib/test_agentx.py
=====================================
--- a/tests/pylib/test_agentx.py
+++ b/tests/pylib/test_agentx.py
@@ -2083,6 +2083,20 @@ class TestNtpclientsNtpsnmpd(unittest.TestCase):
b"\x01\x00\x00\x00\x02\x00\x00\x00\x03\x00\x00\x00"
b"\x03\x00\x00\x00\x10\x00\x00\x00"
b"\x2A\x00\x00\x00\x00\x01\x00\x00")
+ # Test __eq__
+ one = target(a.VALUE_INTEGER, (1, 2, 3), 1)
+ two = target(a.VALUE_INTEGER, (1, 2, 3), 1)
+ # Test equal
+ self.assertEqual(one == two, True)
+ # Test different type
+ two = target(a.VALUE_GAUGE32, (1, 2, 3), 1)
+ self.assertEqual(one == two, False)
+ # Test different OID
+ two = target(a.VALUE_INTEGER, (1, 2, 3, 4), 1)
+ self.assertEqual(one == two, False)
+ # Test different payload
+ two = target(a.VALUE_INTEGER, (1, 2, 3), 2)
+ self.assertEqual(one == two, False)
def test_decode_varbind(self):
f = ntp.agentx.decode_Varbind
@@ -2184,6 +2198,45 @@ class TestNtpclientsNtpsnmpd(unittest.TestCase):
42),
b""))
+ def test_xcode_varbindlist(self):
+ a = ntp.agentx
+ enc = a.encode_varbindlist
+ dec = a.decode_varbindlist
+ vb = a.Varbind
+
+ # Test encode empty
+ self.assertEqual(enc(True, []), b"")
+ # Test encode big endian
+ big = enc(True,
+ [vb(a.VALUE_INTEGER, (1, 2), 1),
+ vb(a.VALUE_INTEGER, (3, 4), 2)])
+ self.assertEqual(big, b"\x00\x02\x00\x00"
+ "\x02\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x02"
+ "\x00\x00\x00\x01"
+ "\x00\x02\x00\x00"
+ "\x02\x00\x00\x00\x00\x00\x00\x03\x00\x00\x00\x04"
+ "\x00\x00\x00\x02")
+ # Test encode little endian
+ little = enc(False,
+ [vb(a.VALUE_INTEGER, (1, 2), 1),
+ vb(a.VALUE_INTEGER, (3, 4), 2)])
+ self.assertEqual(little, b"\x02\x00\x00\x00"
+ "\x02\x00\x00\x00\x01\x00\x00\x00\x02\x00\x00\x00"
+ "\x01\x00\x00\x00"
+ "\x02\x00\x00\x00"
+ "\x02\x00\x00\x00\x03\x00\x00\x00\x04\x00\x00\x00"
+ "\x02\x00\x00\x00")
+ # Test decode empty
+ self.assertEqual(dec(b"", standardFlags), None)
+ # Test decode big endian
+ self.assertEqual(dec(big, standardFlags),
+ (vb(a.VALUE_INTEGER, (1, 2), 1),
+ vb(a.VALUE_INTEGER, (3, 4), 2)))
+ # Test decode little endian
+ self.assertEqual(dec(little, lilEndianFlags),
+ (vb(a.VALUE_INTEGER, (1, 2), 1),
+ vb(a.VALUE_INTEGER, (3, 4), 2)))
+
def test_encode_flagbyte(self):
f = ntp.agentx.encode_flagbyte
self.assertEqual(f(makeflags(False, True, False, True, False)), 0x0A)
=====================================
tests/pylib/test_packet.py
=====================================
--- a/tests/pylib/test_packet.py
+++ b/tests/pylib/test_packet.py
@@ -1597,7 +1597,6 @@ class TestControlSession(unittest.TestCase):
"frags": 24,
"resall":5})
self.assertEqual(nonce_fetch_count, [4])
- print(repr(queries))
self.assertEqual(queries,
[(10, 0, "nonce=foo, frags=24, resall=0x5", False),
(10, 0,
=====================================
tests/pylib/test_statfiles.py
=====================================
--- a/tests/pylib/test_statfiles.py
+++ b/tests/pylib/test_statfiles.py
@@ -131,10 +131,17 @@ class TestNTPStats(unittest.TestCase):
faketimemod = jigs.TimeModuleJig()
fakeosmod = jigs.OSModuleJig()
fakeglobmod = jigs.GlobModuleJig()
+ load_args = []
+ load_returns = []
+ def loadjig(self, statsdir, stem):
+ load_args.append((statsdir, stem))
+ return load_returns.pop(0)
+ process_args = []
+ def processjig(self, stem, lines):
+ process_args.append((stem, lines))
+ return [stem + " " + line for line in lines]
try:
# Splice in jigs
- gziptemp = ntp.statfiles.gzip
- ntp.statfiles.gzip = fakegzipmod
socktemp = ntp.statfiles.socket
ntp.statfiles.socket = fakesockmod
errtemp = sys.stderr
@@ -143,52 +150,17 @@ class TestNTPStats(unittest.TestCase):
ntp.statfiles.time = faketimemod
ostemp = ntp.statfiles.os
ntp.statfiles.os = fakeosmod
- globtemp = ntp.statfiles.glob
- ntp.statfiles.glob = fakeglobmod
- opentemp = open
- ntp.statfiles.open = self.open_jig
+ stemtemp = self.target._NTPStats__load_stem
+ self.target._NTPStats__load_stem = loadjig
+ processtemp = self.target._NTPStats__process_stem
+ self.target._NTPStats__process_stem = processjig
# Test simplest
TDP = self.target.DefaultPeriod
faketimemod.time_returns = [TDP * 2]
fakeosmod.path.isdir_returns = [True]
- fakeosmod.path.getmtime_returns = [TDP+1, TDP+2, TDP+3, TDP+4,
- TDP+5, TDP+6, TDP+7, TDP+8,
- TDP+9, TDP+10, TDP+11, TDP-1]
- self.open_returns = [jigs.FileJig(["40594 10\n", "40594 11"]),
- jigs.FileJig(["40594 31\n",
- "50594 -12\n",
- "40594 30\n"]),
- jigs.FileJig(["40594 40\n", "40594 41"]),
- jigs.FileJig(["40594 50\n", "40594 51"]),
- jigs.FileJig(["40594 60\n", "40594 61"]),
- jigs.FileJig(["40594 70\n", "40594 71"]),
- jigs.FileJig(["40594 80\n", "40594 81"]),
- jigs.FileJig(["604801.25 40594 90\n",
- "#blah",
- "604802.25 40594 91"]),
- jigs.FileJig(["604803.25 40594 100\n",
- "#blah",
- "604804.25 40594 101"]),
- jigs.FileJig(["604805.25 40594 110\n",
- "#blah",
- "604806.25 40594 111"]),
- jigs.FileJig(["604807.25 40594 120\n",
- "#blah",
- "604808.25 40594 121"])]
- fakegzipmod.files_returned = [jigs.FileJig(["40594 20\n",
- "40594 21"])]
- fakeglobmod.glob_returns = [("/foo/bar/clockstats.0",
- "/foo/bar/clockstats.1gz"),
- ("/foo/bar/peerstats.0",
- "/foo/bar/peerstats.1"),
- ("/foo/bar/loopstats.0",
- "/foo/bar/loopstats.1"),
- ("/foo/bar/rawstats.0",
- "/foo/bar/rawstats.1"),
- ("/foo/bar/temps0",
- "/foo/bar/temps1"),
- ("/foo/bar/gpsd0",
- "/foo/bar/gpsd1")] # time kicked
+ load_returns = [["clock0", "clock1"], ["peer0", "peer1"],
+ ["loop0", "loop1"], ["raw0", "raw1"],
+ ["temp0", "temp1"], ["gpsd0", "gpsd1"]]
cls = self.target("/foo/bar")
self.assertEqual(cls.endtime, TDP * 2)
self.assertEqual(cls.starttime, TDP)
@@ -196,49 +168,62 @@ class TestNTPStats(unittest.TestCase):
self.assertEqual(cls.sitename, "bar")
self.assertEqual(logjig.data, [])
self.assertEqual(cls.clockstats,
- [[604810000, '604810.0'],
- [604811000, '604811.0'],
- [604820000, '604820.0'],
- [604821000, '604821.0']])
+ ["clockstats clock0", "clockstats clock1"])
self.assertEqual(cls.peerstats,
- [[604830000, '604830.0'],
- [604831000, '604831.0'],
- [604840000, '604840.0'],
- [604841000, '604841.0']])
+ ["peerstats peer0", "peerstats peer1"])
self.assertEqual(cls.loopstats,
- [[604850000, '604850.0'],
- [604851000, '604851.0'],
- [604860000, '604860.0'],
- [604861000, '604861.0']])
+ ["loopstats loop0", "loopstats loop1"])
self.assertEqual(cls.rawstats,
- [[604870000, '604870.0'],
- [604871000, '604871.0'],
- [604880000, '604880.0'],
- [604881000, '604881.0']])
+ ["rawstats raw0", "rawstats raw1"])
self.assertEqual(cls.temps,
- [[604801250, '604801.25', '40594', '90'],
- [604802250, '604802.25', '40594', '91'],
- [604803250, '604803.25', '40594', '100'],
- [604804250, '604804.25', '40594', '101']])
+ ["temps temp0", "temps temp1"])
self.assertEqual(cls.gpsd,
- [[604805250, '604805.25', '40594', '110'],
- [604806250, '604806.25', '40594', '111']])
+ ["gpsd gpsd0", "gpsd gpsd1"])
# Test all arguments
faketimemod.time_returns = [TDP * 2]
fakesockmod.fqdn_returns = ["jabber"]
fakeosmod.path.isdir_returns = [True]
- self.open_returns = [None]
- fakeglobmod.glob_returns = [(["/foo/bar/clockstats.0"]),
- ([]), ([]), ([]), ([]), ([])]
- fakeosmod.path.getmtime_returns = [101, 102, 103, 104, 105, 106]
- cls = self.target("/foo/bar", "Sitename", 100, 50, 150)
+ load_returns = [[], [], [], [], [], []]
+ cls = self.target("/foo/bar", "Sitename", 200, 50, 150)
+ self.assertEqual(cls.endtime, 150)
+ self.assertEqual(cls.starttime, 50)
+ self.assertEqual(cls.period, 200)
+ self.assertEqual(cls.sitename, "Sitename")
+ self.assertEqual(logjig.data, [])
+ self.assertEqual(cls.clockstats, [])
+ self.assertEqual(cls.peerstats, [])
+ self.assertEqual(cls.loopstats, [])
+ self.assertEqual(cls.rawstats, [])
+ self.assertEqual(cls.temps, [])
+ self.assertEqual(cls.gpsd, [])
+ # Test endtime, but no starttime
+ faketimemod.time_returns = [TDP * 2]
+ fakesockmod.fqdn_returns = ["jabber"]
+ fakeosmod.path.isdir_returns = [True]
+ load_returns = [[], [], [], [], [], []]
+ cls = self.target("/foo/bar", "Sitename", 100, endtime=150)
self.assertEqual(cls.endtime, 150)
self.assertEqual(cls.starttime, 50)
self.assertEqual(cls.period, 100)
self.assertEqual(cls.sitename, "Sitename")
- self.assertEqual(logjig.data,
- ["ntpviz: WARNING: could not read "
- "/foo/bar/clockstats.0\n"])
+ self.assertEqual(logjig.data, [])
+ self.assertEqual(cls.clockstats, [])
+ self.assertEqual(cls.peerstats, [])
+ self.assertEqual(cls.loopstats, [])
+ self.assertEqual(cls.rawstats, [])
+ self.assertEqual(cls.temps, [])
+ self.assertEqual(cls.gpsd, [])
+ # Test endtime, but no starttime
+ faketimemod.time_returns = [TDP * 2]
+ fakesockmod.fqdn_returns = ["jabber"]
+ fakeosmod.path.isdir_returns = [True]
+ load_returns = [[], [], [], [], [], []]
+ cls = self.target("/foo/bar", "Sitename", 100, starttime=150)
+ self.assertEqual(cls.endtime, 250)
+ self.assertEqual(cls.starttime, 150)
+ self.assertEqual(cls.period, 100)
+ self.assertEqual(cls.sitename, "Sitename")
+ self.assertEqual(logjig.data, [])
self.assertEqual(cls.clockstats, [])
self.assertEqual(cls.peerstats, [])
self.assertEqual(cls.loopstats, [])
@@ -250,9 +235,7 @@ class TestNTPStats(unittest.TestCase):
faketimemod.time_returns = [TDP * 2]
fakesockmod.getfqdn_returns = ["jabber"]
fakeosmod.path.isdir_returns = [True]
- self.open_returns = [None]
- fakeglobmod.glob_returns = [([]), ([]), ([]), ([]), ([]), ([])]
- fakeosmod.path.getmtime_returns = []
+ load_returns = [[], [], [], [], [], []]
cls = self.target("/foo/bar", "ntpstats", 100, 50, 150)
self.assertEqual(cls.endtime, 150)
self.assertEqual(cls.starttime, 50)
@@ -268,6 +251,8 @@ class TestNTPStats(unittest.TestCase):
finally:
ntp.statfiles.os = ostemp
ntp.statfiles.time = timetemp
+ self.target._NTPStats__load_stem = stemtemp
+ self.target._NTPStats__process_stem = processtemp
sys.stderr = errtemp
def test___load_statfiles(self):
@@ -407,8 +392,9 @@ class TestNTPStats(unittest.TestCase):
self.target._NTPStats__load_stem = self.load_stem_jig
dataOne = ["#clockstats0\n", "40587 5 foo\n", "40587 10 bar\n",
"40587 15 baz\n", "#clockstats1\n", "40587 20 quux"]
- dataTwo = ["#gpsdstats0\n", "20.5 5 foo\n", "21.5 10 bar\n",
- "22.5 15 baz\n", "#gpsdstats1\n", "23.5 20 quux"]
+ dataTwo = ["#gpsdstats0 foo bar quux\n",
+ "20.5 5 foo\n", "21.5 10 bar\n", "22.5 15 baz\n",
+ "#gpsdstats1\n", "23.5 20 quux"]
TestNTPStats.load_stem_returns = [[]] * 6
fakeosmod.path.isdir_returns = [True] * 10
cls = self.target("/foo/bar", "sitename", starttime=0,
=====================================
tests/pylib/test_util.py
=====================================
--- a/tests/pylib/test_util.py
+++ b/tests/pylib/test_util.py
@@ -145,6 +145,12 @@ class TestPylibUtilMethods(unittest.TestCase):
"[0000:1111:2222:3333:4444:5555:6666:7777]:123"),
("0000:1111:2222:3333:4444:5555:6666:7777", ":123"))
+ def test_stringfilt(self):
+ f = ntp.util.stringfilt
+
+ self.assertEqual(f("1.2345 2.3456 3.4567 4.5678 5.6789 6.789"),
+ " 1.2345 2.3456 3.4567 4.5678 5.6789 6.789")
+
def test_oomsbetweenunits(self):
f = ntp.util.oomsbetweenunits
@@ -270,6 +276,8 @@ class TestPylibUtilMethods(unittest.TestCase):
self.assertEqual(f("1.995", 4), "2.00")
# Attempt to catch bug
self.assertEqual(f("15937.5", None), "15937.5")
+ # Test to bid and no decimals
+ self.assertEqual(f("123456", 4), "123456")
def test_cropprecision(self):
f = ntp.util.cropprecision
@@ -394,6 +402,8 @@ class TestPylibUtilMethods(unittest.TestCase):
self.assertEqual(f("1.234", "offset"), " 1.234ms")
# PPM var
self.assertEqual(f("1.234", "frequency"), "1.234ppm")
+ # No unit type
+ self.assertEqual(f("1.234", "blahblahblah"), "1.234")
def test_f8dot4(self):
f = ntp.util.f8dot4
@@ -866,6 +876,9 @@ class TestPylibUtilMethods(unittest.TestCase):
# Test PPM_VARS, with units
self.assertEqual(f(data, showunits=True),
"frequency=0ppm, clk_wander=1ppm\n")
+ # Test unrecognized variable
+ data = od((("yeahyeah", (1, "1")),))
+ self.assertEqual(f(data), "yeahyeah=1\n")
finally:
ntp.util.termsize = termtemp
@@ -1128,22 +1141,22 @@ class TestPeerSummary(unittest.TestCase):
# Test, no units, hmode=BCLIENTX, peers
cdns_jig_returns = ["clock_canon"]
faketimemod.time_returns = [0xA0000000]
- self.assertEqual(cls.summary(5, data, 12345),
- " clock_canon .FAIL. 8 b 6926"
+ self.assertEqual(cls.summary(0x500, data, 12345),
+ "#clock_canon .FAIL. 8 b 6926"
" 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 = [0xA0000000]
- self.assertEqual(cls.summary(5, data, 12345),
- " clock_canon .FAIL. 8 B 6926"
+ self.assertEqual(cls.summary(0x500, data, 12345),
+ "#clock_canon .FAIL. 8 B 6926"
" 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 = [0xA0000000]
- self.assertEqual(cls.summary(5, data, 12345),
- " clock_canon .FAIL. 8 M 6926"
+ self.assertEqual(cls.summary(0x500, data, 12345),
+ "#clock_canon .FAIL. 8 M 6926"
" 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")
@@ -1151,82 +1164,107 @@ class TestPeerSummary(unittest.TestCase):
data["srchost"] = ("(blah)", "(blah)")
cdns_jig_returns = ["clock_canon"]
faketimemod.time_returns = [0xA0000000]
- self.assertEqual(cls.summary(5, data, 12345),
- " clock_canon .FAIL. 8 l 6926"
+ self.assertEqual(cls.summary(0x500, data, 12345),
+ "#clock_canon .FAIL. 8 l 6926"
" 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 = [0xA0000000]
- self.assertEqual(cls.summary(5, data, 12345),
- " clock_canon .POOL. 8 p 6926"
+ self.assertEqual(cls.summary(0x500, data, 12345),
+ "#clock_canon .POOL. 8 p 6926"
" 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 = [0xA0000000]
- self.assertEqual(cls.summary(5, data, 12345),
- " clock_canon .FAIL. 8 a 6926"
+ self.assertEqual(cls.summary(0x500, data, 12345),
+ "#clock_canon .FAIL. 8 a 6926"
" 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 = [0xA0000000]
- self.assertEqual(cls.summary(5, data, 12345),
- " clock_canon .FAIL. 8 u 6926"
+ self.assertEqual(cls.summary(0x500, data, 12345),
+ "#clock_canon .FAIL. 8 u 6926"
" 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 = [0xA0000000]
- self.assertEqual(cls.summary(5, data, 12345),
- " clock_canon .FAIL. 8 s 6926"
+ self.assertEqual(cls.summary(0x500, data, 12345),
+ "#clock_canon .FAIL. 8 s 6926"
" 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 = [0xA0000000]
- self.assertEqual(cls.summary(5, data, 12345),
- " clock_canon .FAIL. 8 S 6926"
+ self.assertEqual(cls.summary(0x500, data, 12345),
+ "#clock_canon .FAIL. 8 S 6926"
" 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 = [0xA0000000]
- self.assertEqual(cls.summary(5, data, 12345),
- " 10.20.30.40 .FAIL. 8 S 6926"
+ self.assertEqual(cls.summary(0x500, data, 12345),
+ "#10.20.30.40 .FAIL. 8 S 6926"
" 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 = [0xA0000000]
- self.assertEqual(cls.summary(5, data, 12345),
- " clock_canon_bla .FAIL. 8 S 6926"
+ self.assertEqual(cls.summary(0x500, data, 12345),
+ "#clock_canon_bla .FAIL. 8 S 6926"
" 32 764 1.2346 2.7183 3.1416\n")
- # Test, no units, name crop
+ # Test, no units, no name crop
cls.wideremote = True
cdns_jig_returns = ["clock_canon_blah_jabber_quantum"]
faketimemod.time_returns = [0xA0000000]
- self.assertEqual(cls.summary(5, data, 12345),
- " clock_canon_blah_jabber_quantum\n"
+ self.assertEqual(cls.summary(0x500, data, 12345),
+ "#clock_canon_blah_jabber_quantum\n"
" .FAIL. 8 S"
" 6926 32 764 1.2346 2.7183 3.1416\n")
# Test, with units
cls.showunits = True
cdns_jig_returns = ["clock_canon"]
faketimemod.time_returns = [0xA0000000]
- self.assertEqual(cls.summary(5, data, 12345),
- " clock_canon .FAIL. 8 S 6926"
+ self.assertEqual(cls.summary(0x500, data, 12345),
+ "#clock_canon .FAIL. 8 S 6926"
" 32 764 1.2346ms 2.7183ms 3.1416ms\n")
+ # Test, low precision formatting with units
+ lowpdata = data.copy()
+ lowpdata["delay"] = (1.234, "1.234")
+ cls.showunits = True
+ cdns_jig_returns = ["clock_canon"]
+ faketimemod.time_returns = [0xA0000000]
+ self.assertEqual(cls.summary(0x500, lowpdata, 12345),
+ "#clock_canon .FAIL. 8 S 6926"
+ " 32 764 1.234ms 2.7183ms 3.1416ms\n")
+ # Test, low precision formatting with units
+ cls.showunits = False
+ cdns_jig_returns = ["clock_canon"]
+ faketimemod.time_returns = [0xA0000000]
+ self.assertEqual(cls.summary(0x500, lowpdata, 12345),
+ "#clock_canon .FAIL. 8 S 6926"
+ " 32 764 1.234 2.718 3.142\n")
# Test, apeers
cls.showunits = True
cls.displaymode = "apeers"
cdns_jig_returns = ["clock_canon"]
faketimemod.time_returns = [0xA0000000]
- self.assertEqual(cls.summary(5, data, 12345),
- " clock_canon .FAIL. 12345 8 S 6926"
+ self.assertEqual(cls.summary(0x500, data, 12345),
+ "#clock_canon .FAIL. 12345 8 S 6926"
+ " 32 764 1.2346ms 2.7183ms 3.1416ms\n")
+ # Test rstatus, previous version
+ cls.showunits = True
+ cls.displaymode = "apeers"
+ cdns_jig_returns = ["clock_canon"]
+ faketimemod.time_returns = [0xA0000000]
+ cls.pktversion = 0
+ self.assertEqual(cls.summary(0x300, data, 12345),
+ "*clock_canon .FAIL. 12345 8 S 6926"
" 32 764 1.2346ms 2.7183ms 3.1416ms\n")
finally:
ntp.util.time = timetemp
View it on GitLab: https://gitlab.com/NTPsec/ntpsec/compare/52c4a02c90bdf085f0c6f84f12060b46c58f9749...79f957635bb6d0745b9784ccbf71303d7b2acad7
---
View it on GitLab: https://gitlab.com/NTPsec/ntpsec/compare/52c4a02c90bdf085f0c6f84f12060b46c58f9749...79f957635bb6d0745b9784ccbf71303d7b2acad7
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/20180112/741f8874/attachment.html>
More information about the vc
mailing list