| #!/usr/bin/python |
| # Copyright 2012 Google Inc. All Rights Reserved. |
| |
| """This class defines the test case "image download" from ACS. |
| |
| This class extends the TestCase Class. |
| testID: 11843140 |
| This test does the following tasks: |
| - check current image version |
| - if the current version is the same as expected version, |
| this test will first to try to downgrade it to the downgrade_version. |
| - Then, it will Run image download test |
| - Fianlly, it will downgrade to downgrade_version if this has not been |
| done. This is to test the download handling capabilities of the |
| expected_version |
| - configure ACS instance with these software versions |
| - verify image download and installation to be complete and successful, by |
| observing a successfully reboot after installation |
| """ |
| |
| __author__ = 'Lehan Meng (lmeng@google.com)' |
| |
| |
| import os |
| import re |
| import time |
| |
| import acs |
| import ssh |
| import testCase |
| |
| |
| class DownloadTest(testCase.TestCase): |
| """Verify Bruno image download from ACS (TestID: 11843140). |
| |
| Configuration Parameters: |
| testID: 11843140 |
| user = root |
| addr = 10.0.28.191 |
| pwd = google |
| bruno_prompt = (none)# |
| file=cmdOutput.txt |
| title =Image_Download |
| current_version=None |
| expected_version=bruno-iguana-2-dg |
| ACS_URL = https://gfiber-acs-staging.appspot.com/cwmp |
| """ |
| |
| def Init(self): |
| """Initialte the DownloadTest instance.""" |
| self.p_ssh = ssh.SSH(user=self.params['user'], |
| addr=self.params['addr'], |
| pwd=self.params['pwd'], |
| bruno_prompt=self.params['bruno_prompt'], |
| addr_ipv6=self.params['addr_ipv6'], |
| athena_user=self.params['athena_user'], |
| athena_pwd=self.params['athena_pwd'], |
| jump_server=self.params['jump_server']) |
| self.p_ssh.Setlogging(self.log) |
| self.__debug = False |
| self.__short_delay = 5 #short delay: 5 seconds |
| self.__delay = 180 #medium delay: 180 seconds |
| self.__long_delay = 600 #long delay: 600 seconds |
| |
| def CheckACSUrl(self): |
| """Check current ACS URL of the device. |
| |
| Set it to a default URL if the current URL is not recognized |
| """ |
| self.p_ssh.SendCmd('cat /tmp/cwmp/acs_url') |
| line = self.p_ssh.GetCmdOutput()[1] |
| m_1 = re.match(self.params['ACS_URL'], line) |
| if m_1 is None: |
| info = self.log.CreateErrorInfo( |
| 'intermediate', |
| 'ACS URL unknown! Update ACS URL manually') |
| self.log.SendLine(self.test_info, info) |
| |
| self.p_ssh.SendCmd('echo ' + self.params['ACS_URL'] |
| + ' > /tmp/cwmp/acs_url') |
| self.AddConfigParams('current_ACS_url', self.params['ACS_URL']) |
| else: |
| self.AddConfigParams('current_ACS_url', self.params['ACS_URL']) |
| |
| def SetACSUrl(self, url='http://acs-sandbox.appspot.com/cwmp'): |
| """Set ACS URL on the device.""" |
| self.p_ssh.SendCmd('echo ' + url + ' > /tmp/cwmp/acs_url') |
| self.p_ssh.SendCmd('cat /tmp/cwmp/acs_url') |
| info = self.log.CreateProgressInfo('---', 'Set ACS url to: ' + url) |
| self.log.SendLine(self.test_info, info) |
| |
| def CheckVersion(self): |
| """Check current software version.""" |
| self.p_ssh.SendCmd('more /etc/version') |
| current_version = self.p_ssh.GetCmdOutput(2)[1].strip() |
| self.params['current_version'] = current_version |
| |
| def GetVersion(self): |
| """Get current software version.""" |
| self.p_ssh.SendCmd('more /etc/version') |
| return self.p_ssh.GetCmdOutput(2)[1].strip() |
| |
| def GetTimeStamp(self, line=None): |
| """Extract the time information of most current event from Device log. |
| |
| i.e., the time that the last event happens since reboot (in seconds) |
| Args: |
| line: a line of device log which contains event time information. |
| If it has value of 'None', then retrieve the last a few lines from |
| device log to extract time information. |
| Returns: |
| return the time stamp string if succeed, return -1 otherwise. |
| """ |
| if line is None: |
| # get time stamp from Device log file |
| self.p_ssh.SendCmd('dmesg | grep cwmpd:') |
| # search the last 30 lines in Device log for time information |
| latest_log_entry = 30 |
| log_list = self.p_ssh.GetCmdOutput(latest_log_entry) |
| |
| for line in log_list: |
| m = re.match('\\[[\s]*(\d*\\.\d{3})\\]', line) |
| if m is not None: |
| # Match, return the time stamp |
| return m.group(1) |
| else: |
| # get time stamp from a Device log line |
| m = re.match('\\[[\s]*(\d*\\.\d{3})\\]', line) |
| if m is not None: |
| # Match, return the time stamp |
| return m.group(1) |
| |
| # No time stamp found |
| return -1 |
| |
| def ConfigACS(self): |
| """Setup the ACS device profile with expected image. |
| |
| Then restart the cwmpd process on device to |
| initiate the inform/download process |
| Returns: |
| return the time stamp string if succeed, return -1 otherwise. |
| """ |
| #configure ACS: |
| acs_instance = acs.ACS(self.params['expected_version_id']) |
| acs_instance.SetLogging(self.log) |
| acs_instance.SaveConfiguration() |
| |
| #self.p_ssh.SendCmd('/etc/Init.d/S85catawampus restart') |
| # To setup any device service parameter using nbi_client will trigger |
| # ACS to make a remote connection request to the device. |
| # However, currently ACS does not provide separate stubby call to |
| # only request a remote connection. |
| # As walk around: |
| # 1. call remote management URL locally to initiate Inform immediately |
| # 2. wait for the next periodic inform to initiate download |
| # Here go for the 2nd option. |
| time_stamp = self.GetTimeStamp() |
| return time_stamp |
| |
| def GetRemoteConnectionURL(self): |
| """Get remote connection URL from the cwmp log of the device.""" |
| self.p_ssh.SendCmd('dmesg | grep cwmpd:') |
| log_list = self.p_ssh.GetCmdOutput() |
| for line in log_list: |
| if 'InternetGatewayDevice.ManagementServer.ConnectionRequestURL' in line: |
| index = log_list.index(line) |
| index -= 1 |
| if index < 0: |
| # End of long while download not started, wait 281 |
| info = self.log.createErrorInfo( |
| 'low', 'Can not find ConnectionRequestURL from device log ...') |
| self.log.sendLine(self.test_info, info) |
| return None |
| |
| log_str = log_list[index] |
| m = re.search('(http://[\d\l:]+)</Value>', log_str) |
| if m is not None: |
| # match successful: |
| return m.group(1) |
| # No ConnectionRequestURL find: |
| info = self.log.createErrorInfo( |
| 'low', 'Can not find ConnectionRequestURL from device log ...') |
| self.log.sendLine(self.test_info, info) |
| return None |
| |
| def CheckDownloadFileExists(self): |
| """check if a downloaded file is still in the folder. |
| |
| A file in the folder '/rw/tr69/dnld/' indicates either of the |
| following: |
| 1. download is in progress |
| 2. download completed, installation is in progress |
| 3. download completed, installation completed, waiting for reboot |
| |
| the upgrade is not considered as complete until a reboot is done |
| |
| Returns: |
| return True if file exists, return False otherwise. |
| """ |
| self.p_ssh.SendCmd('ls -l /rw/tr69/dnld/') |
| latest_log_entry = 4 |
| log_list = self.p_ssh.GetCmdOutput(latest_log_entry) |
| for line in log_list: |
| m = re.search(r'[\s]+([\w,\d]{9})', line) |
| if m is not None: |
| # download file found |
| return True |
| return False |
| |
| def IsRebooted(self, time_stamp): |
| """check if the device rebooted after the software installation. |
| |
| This method checks device log file, if any event found with smaller time |
| stamp value than the last known device event, it indicates that the device |
| has been rebooted since then |
| Args: |
| time_stamp: time stamp that the last event recorded on device log |
| Returns: |
| return True if rebooted |
| return False otherwise |
| """ |
| s = self.GetTimeStamp() |
| if float(s) < float(time_stamp): |
| return True |
| return False |
| |
| def CheckImageInstallation(self, time_stamp): |
| """Check if the Image installation and Device reboot is successful. |
| |
| An upgrade is considered complete only after the device is rebooted |
| after image installation |
| Args: |
| time_stamp: check if a reboot is done after this time |
| Returns: |
| return True if installation succeeded |
| return False otherwise |
| """ |
| self.p_ssh.ExitToAthena() |
| time.sleep(3*self.__short_delay) |
| self.p_ssh.SshRetryFromAthena() |
| count = 0 |
| ts = time_stamp |
| while not self.IsRebooted(ts): |
| count +=1 |
| if count*self.__short_delay*3 > self.__long_delay: |
| info = self.log.CreateErrorInfo( |
| 'critical', |
| 'Error! File download longer than ' + str(self.__long_delay) |
| + ' seconds, check network. Exit!') |
| self.log.SendLine(self.test_info, info) |
| os.sys.exit(1) |
| |
| t = self.GetTimeStamp() |
| if t > ts: |
| ts = t |
| self.p_ssh.ExitToAthena() |
| info = self.log.CreateProgressInfo( |
| '90%', 'Wait ' + str(3*self.__short_delay) |
| + ' seconds for image installation and device reboot...') |
| self.log.SendLine(self.test_info, info) |
| time.sleep(3*self.__short_delay) |
| self.p_ssh.SshRetryFromAthena() |
| |
| time.sleep(2*self.__short_delay) |
| self.p_ssh.SendCmd('more /etc/version') |
| line = self.p_ssh.GetCmdOutput()[1] |
| crt_version = line.strip() |
| print crt_version + '########' + self.params['expected_version'] |
| if crt_version == self.params['expected_version']: |
| return True |
| else: |
| return False |
| |
| def Run(self): |
| """Run the test case.""" |
| ####### Add your code here -- BEGIN ####### |
| print 'Test Started...' |
| self.Init() |
| self.p_ssh.SshToAthena() |
| info = self.log.CreateProgressInfo( |
| '5%', 'SSH session to Athena server successfully established!') |
| self.log.SendLine(self.test_info, info) |
| |
| max_test_idx = 2 |
| current_test_idx = 1 |
| |
| while current_test_idx < max_test_idx: |
| self.p_ssh.SshRetryFromAthena() |
| info = self.log.CreateResultInfo( |
| '---', 'Test Sequence Number: ' + str(current_test_idx)) |
| self.log.SendLine(self.test_info, info) |
| while self.CheckDownloadFileExists(): |
| # current in download |
| info = self.log.CreateProgressInfo( |
| '10%', 'Another downloading in progress, retry after ' |
| + str(self.__short_delay) + 'seconds...') |
| self.log.SendLine(self.test_info, info) |
| self.p_ssh.ExitToAthena() |
| time.sleep(self.__short_delay) |
| self.p_ssh.SshRetryFromAthena() |
| |
| info = self.log.CreateProgressInfo( |
| '15%', 'SSH session to device successfully established!') |
| self.log.SendLine(self.test_info, info) |
| |
| self.CheckACSUrl() |
| info = self.log.CreateProgressInfo('20%', 'Current ACS url: ' |
| + self.params['ACS_URL']) |
| self.log.SendLine(self.test_info, info) |
| |
| self.CheckVersion() |
| info = self.log.CreateProgressInfo('30%', 'Current Version: ' |
| + self.params['current_version']) |
| self.log.SendLine(self.test_info, info) |
| |
| # Need to do: |
| # check if the currently version is already expected version, |
| # if so, downgrade first, then Run this test |
| #if self.params['current_version'] == self.params['expected_version']: |
| # info = self.log.createErrorInfo( |
| # 'low', 'Currently image is expected version, ' |
| # + 'please try other images.') |
| # self.log.sendLine(self.test_info, info) |
| |
| self.time_stamp = self.ConfigACS() |
| info = self.log.CreateProgressInfo('40%', 'Expected Version: ' |
| + self.params['expected_version']) |
| self.log.SendLine(self.test_info, info) |
| info = self.log.CreateProgressInfo( |
| '40%', 'Successfully configured ACS for expected image version') |
| self.log.SendLine(self.test_info, info) |
| |
| time.sleep(self.__short_delay) # time delay for log update |
| if self.CheckImageInstallation(self.time_stamp): |
| info = self.log.CreateProgressInfo( |
| '100%', 'Image successfully upgraded to expected version: ' |
| 'from: ' + self.params['current_version'] |
| + ' to: ' + self.params['expected_version']) |
| self.log.SendLine(self.test_info, info) |
| info = self.log.CreateResultInfo( |
| 'Pass', 'Image successfully upgraded to expected version: ' |
| 'from: ' + self.params['current_version'] |
| + ' to: ' + self.params['expected_version']) |
| self.log.SendLine(self.test_info, info) |
| else: |
| info = self.log.CreateErrorInfo( |
| 'critical', |
| 'Error! Image version does not match expected version!') |
| self.log.SendLine(self.test_info, info) |
| |
| current_test_idx += 1 |
| self.p_ssh.ExitToAthena() |
| # Need to do: |
| # downgrade from expected_version |
| # To verify the capability of the download handling on expected version |
| # image |
| print 'Test Completed...' |
| ####### Add your code here -- END ####### |
| |
| def Destructor(self): |
| """This is the Destructor function to call for the user extened class.""" |
| # exit the ssh tunnel |
| self.p_ssh.Close() |
| |
| |
| if __name__ == '__main__': |
| DownloadTest(testID='11843140', key_word='Download') |
