blob: 93c90c165ec915b4779daf54b5b8f7fc3904175e [file] [log] [blame]
"""Unit Tests for diagui.py implementation."""
__author__ = 'anandkhare@google.com (Anand Khare)'
import ast
import json
import os
import google3
import diagui.main
import tornado.httpclient
import tr.mainloop
import tr.helpers
import dm_root
import dm.fakewifi
import dm.host
from tr.wvtest import unittest
class AsynchFetch(object):
"""Creates instance of client object, makes asynchronous calls to server."""
def __init__(self, url_temp):
self.http_client = tornado.httpclient.AsyncHTTPClient()
self.resp = None
self.http_client.fetch(url_temp, method='GET',
callback=self.HandleRequest)
def HandleRequest(self, response):
self.resp = response
def Wait(self, loop):
while not self.resp:
loop.RunOnce()
class FakeHostsList(dm.host.CATA181HOSTS):
def __init__(self, count=1):
self._hosts = {}
for idx in range(1, count+1):
host = tr.core.Extensible(dm.host.CATA181HOST)()
host.X_CATAWAMPUS_ORG_ClientIdentification = (
dm.host.ClientIdentification())
self._hosts[str(idx)] = host
@property
def HostList(self):
return self._hosts
class DiaguiTest(unittest.TestCase):
"""Tests whether 2 clients receive the same data from the server.
Also checks if both receive updates.
"""
def setUp(self):
self.save_activewan = diagui.main.ACTIVEWAN
diagui.main.ACTIVEWAN = 'testdata/activewan'
self.checksum = '0'
self.url_string = 'http://localhost:8880/content.json?checksum='
def tearDown(self):
diagui.main.ACTIVEWAN = self.save_activewan
def testUpdateDict(self):
test_data = """acs OK (May 21 2013 18:58:41+700)
softversion 1.16a
uptime 76:28:39
serialnumber 123456789
temperature 54 C
fiberjack Up
wanmac 1a:2b:3c:4d:5e:6f
wanip 63.28.214.97
lanip 192.168.1.1
subnetmask 255.255.255.0
dhcpstart 192.158.1.100
dhcpend 192.168.1.254
wiredlan 6a:5b:4c:3d:2e:1f Up
wireddevices Living Room (TV box, 6a:5b:4c:3d:2e:1f)
ssid24 AllenFamilyNetwork
ssid5 (same)
wpa2 (configured)
wirelesslan 3a:1b:4c:1d:5e:9f Up
wirelessdevices Dad\'s Phone (6a:5b:4c:3d:2e:1f)
upnp O
portforwarding 80-80: Dad\'s Computer (6a:5b:4c:3d:2e:1f)
dmzdevice Wireless Device (1) (6a:5b:4c:3d:2e:1f)
dyndns DynDNS
username allenfamily
domain home.allenfamily.com"""
url_temp = self.url_string + self.checksum
app = diagui.main.MainApplication(None, None, run_diagui=True)
app.listen(8880)
app.diagui.data = dict(line.decode('utf-8').strip().split(None, 1)
for line in test_data.split('\n'))
app.diagui.UpdateCheckSum()
response1 = AsynchFetch(url_temp)
response2 = AsynchFetch(url_temp)
main_loop = tr.mainloop.MainLoop()
response1.Wait(main_loop)
response2.Wait(main_loop)
self.assertEqual(response1.resp.body,
response2.resp.body)
self.assertNotEqual(response1.resp.body, None)
self.checksum = ast.literal_eval(response1.resp.body).get(
'checksum')
test_data = """acs OK (May 21 2013 18:58:41+700)
softversion 2.16a
uptime 76:28:39
serialnumber 987654321
temperature 54 C
fiberjack Up
wanmac 1a:2b:3c:4d:5e:6f
wanip 63.28.214.97
lanip 192.168.1.1
subnetmask 255.255.255.0
dhcpstart 192.158.1.100
dhcpend 192.168.1.254
wiredlan 6a:5b:4c:3d:2e:1f Up
wireddevices Living Room (TV box, 6a:5b:4c:3d:2e:1f)
ssid24 AllenFamilyNetwork
ssid5 (same)
wpa2 (configured)
wirelesslan 3a:1b:4c:1d:5e:9f Up
wirelessdevices Dad\'s Phone (6a:5b:4c:3d:2e:1f)
upnp O
portforwarding 80-80: Dad\'s Computer (6a:5b:4c:3d:2e:1f)
dmzdevice Wireless Device (1) (6a:5b:4c:3d:2e:1f)
dyndns DynDNS
username allenfamily
domain home.allenfamily.com"""
app.diagui.data = dict(line.decode('utf-8').strip().split(None, 1)
for line in test_data.split('\n'))
app.diagui.UpdateCheckSum()
url_temp = self.url_string + self.checksum
response1_new = AsynchFetch(url_temp)
response2_new = AsynchFetch(url_temp)
response1_new.Wait(main_loop)
response2_new.Wait(main_loop)
self.assertEqual(response1_new.resp.body,
response2_new.resp.body)
self.assertNotEqual(response1_new.resp.body, None)
self.assertNotEqual(response1.resp.body,
response1_new.resp.body)
def testOnuStats(self):
app = diagui.main.MainApplication(None, None, run_diagui=True)
app.listen(8880)
main_loop = tr.mainloop.MainLoop()
diagui.main.ONU_STAT_FILE = 'testdata/onu_stats1.json'
app.diagui.UpdateOnuStats()
self.assertTrue('onu_wan_connected' in app.diagui.data)
self.assertFalse('onu_serial' in app.diagui.data)
self.checksum = '0'
url_temp = self.url_string + self.checksum
response = AsynchFetch(url_temp)
response.Wait(main_loop)
self.assertNotEqual(response.resp.body, None)
jsdata = json.loads(response.resp.body)
self.assertTrue(jsdata['onu_wan_connected'])
diagui.main.ONU_STAT_FILE = 'testdata/onu_stats2.json'
app.diagui.UpdateOnuStats()
response = AsynchFetch(url_temp)
response.Wait(main_loop)
jsdata = json.loads(response.resp.body)
self.assertTrue(jsdata['onu_wan_connected'])
self.assertTrue(jsdata['onu_acs_contacted'])
self.assertEqual(jsdata['onu_acs_contact_time'], 100000)
self.assertEqual(jsdata['onu_serial'], '12345')
def testNoOnuStats(self):
app = diagui.main.MainApplication(None, None, run_diagui=True)
diagui.main.ONU_STAT_FILE = '/no/such/file'
app.diagui.UpdateOnuStats()
# just checking whether there is an exception
class TechuiTest(unittest.TestCase):
"""Tests the data gathering functions for the TechUI."""
def testMainApp(self):
url = 'http://localhost:8880/techui.json?checksum=0'
app = diagui.main.MainApplication(None, None, run_diagui=True,
run_techui=True)
fake_data = {'moca_bitloading': {},
'ip_addr': {'ec:88:92:91:3d:67': '111.111.11.1',
'aa:aa:aa:aa:aa:aa': '123.456.78.90'},
'wifi_signal_strength': {},
'softversion': 'gfrg200-46-pre0-39-g056a912-th',
'serialnumber': 'G0123456789',
'other_aps': {'f4:f5:e8:80:58:d7': -67.0},
'host_names': {'ec:88:92:91:3d:67': 'android',
'aa:aa:aa:aa:aa:aa': 'GFiberTV'},
'moca_corrected_codewords': {},
'moca_uncorrected_codewords': {},
'moca_signal_strength': {},
'self_signals': {'f4:f5:e8:83:01:94': -25},
'moca_nbas': {},
'checksum': 0}
app.techui.data = fake_data
app.listen(8880)
main_loop = tr.mainloop.MainLoop()
response1 = AsynchFetch(url)
response1.Wait(main_loop)
result1 = json.loads(response1.resp.body)
self.assertNotEqual(result1, None)
self.assertEqual(result1, fake_data)
# Send another request, update the data, and call callbacks.
# Should update the checksum.
result1_checksum = result1['checksum']
response2 = AsynchFetch(url)
app.techui.data['other_aps'] = {'f4:f5:e8:80:58:d7': -50.0}
app.techui.NotifyUpdatedDict()
response2.Wait(main_loop)
result2 = json.loads(response2.resp.body)
# Set fake data to expected output and compare.
fake_data['other_aps'] = {'f4:f5:e8:80:58:d7': -50.0}
fake_data['checksum'] = app.techui.data['checksum']
result2_checksum = result2['checksum']
self.assertNotEqual(result2, None)
self.assertEqual(result2, fake_data)
self.assertNotEqual(result1_checksum, result2_checksum)
self.assertEqual(app.techui.FindTVBoxes(), ['123.456.78.90'])
# Update the url to have the new checksum, update data, and check for
# correct response.
url = 'http://localhost:8880/techui.json?checksum=' + result2_checksum
response3 = AsynchFetch(url)
app.techui.data['other_aps'] = {'f4:f5:e8:80:58:d7': -40.0}
app.techui.NotifyUpdatedDict()
response3.Wait(main_loop)
result3 = json.loads(response3.resp.body)
# Set fake data to expected output and compare.
fake_data['other_aps'] = {'f4:f5:e8:80:58:d7': -40.0}
fake_data['checksum'] = app.techui.data['checksum']
result3_checksum = result3['checksum']
self.assertNotEqual(result3, None)
self.assertEqual(result3, fake_data)
self.assertNotEqual(result2_checksum, result3_checksum)
def testSetTechUIDict(self):
techui = diagui.main.TechUI(None)
techui.SetTechUIDict('fake', {})
self.assertEqual(techui.data['fake'], {})
test_dict = {'11:22:33:44:55:66': 1, '11:22:33:44:55:67': 2}
techui.SetTechUIDict('fake', test_dict)
self.assertEqual(techui.data['fake'], test_dict)
def testLoadJson(self):
dne = '/tmp/does_not_exist'
try:
os.remove(dne)
except OSError:
pass
result = diagui.main.LoadJson(dne)
self.assertEqual(result, {})
jsonfile = '/tmp/json'
test_dict = {'11:22:33:44:55:66': 1, '11:22:33:44:55:67': 2}
tr.helpers.WriteFileAtomic(jsonfile, json.dumps(test_dict))
result = diagui.main.LoadJson(jsonfile)
self.assertEqual(result, test_dict)
try:
os.remove(jsonfile)
except OSError:
pass
def testUpdateMocaDict(self):
techui = diagui.main.TechUI(None)
techui.root = dm_root.DeviceModelRoot(None, 'fakecpe', None)
interface_list = techui.root.Device.MoCA.InterfaceList
snr = {}
bitloading = {}
corrected_cw = {}
uncorrected_cw = {}
nbas = {}
for unused_i, inter in interface_list.iteritems():
for unused_j, dev in inter.AssociatedDeviceList.iteritems():
snr[dev.MACAddress] = dev.X_CATAWAMPUS_ORG_RxSNR_dB
bitloading[dev.MACAddress] = dev.X_CATAWAMPUS_ORG_RxBitloading
nbas[dev.MACAddress] = dev.X_CATAWAMPUS_ORG_RxNBAS
corrected = (dev.X_CATAWAMPUS_ORG_RxPrimaryCwCorrected +
dev.X_CATAWAMPUS_ORG_RxSecondaryCwCorrected)
uncorrected = (dev.X_CATAWAMPUS_ORG_RxPrimaryCwUncorrected +
dev.X_CATAWAMPUS_ORG_RxSecondaryCwUncorrected)
no_errors = (dev.X_CATAWAMPUS_ORG_RxPrimaryCwNoErrors +
dev.X_CATAWAMPUS_ORG_RxSecondaryCwNoErrors)
total = corrected + uncorrected + no_errors
if total > 0:
corrected_cw[dev.MACAddress] = corrected/total
uncorrected_cw[dev.MACAddress] = uncorrected/total
else:
corrected_cw[dev.MACAddress] = 0
uncorrected_cw[dev.MACAddress] = 0
techui.UpdateMocaDict()
self.assertEqual(snr, techui.data['moca_signal_strength'])
self.assertEqual(bitloading, techui.data['moca_bitloading'])
self.assertEqual(corrected_cw,
techui.data['moca_corrected_codewords'])
self.assertEqual(uncorrected_cw,
techui.data['moca_uncorrected_codewords'])
self.assertEqual(nbas, techui.data['moca_nbas'])
def testUpdateWifiDict(self):
techui = diagui.main.TechUI(None)
wlan0 = dm.fakewifi.FakeWifiWlanConfiguration()
wlan1 = dm.fakewifi.FakeWifiWlanConfiguration()
techui.root = dm_root.DeviceModelRoot(None, 'fakecpe', None)
lans = techui.root.InternetGatewayDevice.LANDeviceList
lans['1'].WLANConfigurationList = {
'1': wlan0,
'2': wlan1,
}
wlan0.signals = {'11:22:33:44:55:66': -66}
wlan1.signals = {'66:55:44:33:22:11': -11}
techui.UpdateWifiDict()
self.assertEquals(
techui.data['wifi_signal_strength'],
{'66:55:44:33:22:11': -11, '11:22:33:44:55:66': -66})
def testNoSignals(self):
techui = diagui.main.TechUI(None)
wlan0 = dm.fakewifi.FakeWifiWlanConfiguration()
wlan1 = object()
techui.root = dm_root.DeviceModelRoot(None, 'fakecpe', None)
lans = techui.root.InternetGatewayDevice.LANDeviceList
lans['1'].WLANConfigurationList = {
'1': wlan0,
'2': wlan1,
}
wlan0.signals = {'11:22:33:44:55:66': -66}
techui.UpdateWifiDict()
self.assertEquals(
techui.data['wifi_signal_strength'],
{'11:22:33:44:55:66': -66})
class LicenseuiTest(unittest.TestCase):
"""Make sure server can retrieve encrypted license file."""
def testLicenseExists(self):
app = diagui.main.MainApplication(None, None, run_licenseui=True)
app.listen(8880)
main_loop = tr.mainloop.MainLoop()
response = AsynchFetch('http://localhost:8880/license/LICENSES.zip')
response.Wait(main_loop)
self.assertNotEqual(response.resp.body, None)
if __name__ == '__main__':
unittest.main()