| #!/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. |
| This file includes the following test cases: |
| - testID: 11843140 7.6.11 Verify firmware upgrade/downgrade by ACS |
| - testID: 14165036 7.6.18 Negative test: Bruno Image download interruption |
| - testID: 14334001 7.6.25.3 File Transfer (File upload is optional, this is |
| then basically the same test case as download.) |
| """ |
| |
| __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.""" |
| if not self.p_ssh: |
| self.p_ssh = ssh.SSH(user=self.params['user'], |
| addr=self.params['addr'], |
| bruno_prompt=self.params['bruno_prompt'], |
| addr_ipv6=self.params['addr_ipv6']) |
| 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 |
| log_list = self.p_ssh.GetCmdOutput() |
| |
| 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 |
| info = self.log.createErrorInfo( |
| 'Critical', 'Exit! Can not find time information from ' |
| 'device log ...') |
| self.log.sendLine(self.test_info, info) |
| return os.sys.exit(1) |
| |
| def ConfigACS(self, version_id): |
| """Setup the ACS device profile with expected image. |
| |
| Then restart the cwmpd process on device to |
| initiate the inform/download process |
| Args: |
| version_id: the expected version id that needs to be configured |
| Returns: |
| return the time stamp string if succeed, return -1 otherwise. |
| """ |
| #configure ACS: |
| acs_instance = acs.ACS(acs_path=self.params['acs_path']) |
| acs_instance.SetLogging(self.log) |
| acs_instance.SaveConfiguration( |
| self.params['profile_id'], version_id, self.params['acs_app_id'], |
| self.params['acs_host_name']) |
| |
| 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 |
| """ |
| time.sleep(5) |
| s = self.GetTimeStamp() |
| print 'previous time stamp: ' + time_stamp |
| print 'current time stamp: ' + s |
| if float(s) < float(time_stamp) and s is not -1: |
| return True |
| return False |
| |
| def CheckImageInstallation(self, time_stamp, version): |
| """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 |
| version: the image version that needs to be installed |
| Returns: |
| return True if installation succeeded |
| return False otherwise |
| """ |
| self.p_ssh.Close() |
| time.sleep(6*self.__short_delay) |
| self.p_ssh.SshRetry() |
| count = 0 |
| ts = time_stamp |
| while not self.IsRebooted(ts): |
| count +=1 |
| if count*self.__short_delay*6 > 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 float(t) > float(ts): |
| ts = t |
| self.p_ssh.Close() |
| info = self.log.CreateProgressInfo( |
| '---', 'Wait ' + str(6*self.__short_delay) |
| + ' seconds for image installation and device reboot...') |
| self.log.SendLine(self.test_info, info) |
| time.sleep(6*self.__short_delay) |
| self.p_ssh.SshRetry() |
| |
| 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 + '########' + version |
| if crt_version == version: |
| return True |
| else: |
| return False |
| |
| def VerifyImageDownload(self): |
| """Verify image download and upgrade. |
| |
| This test checks the following: |
| - check current image version |
| - if current_version is not expected_version as defined in config file: |
| - Upgrade to expected_version (verify expected version can boot up) |
| - Downgrade from expected_version to downgrade_version |
| (verify downgrade handling capability of expected_version image) |
| - Finally upgrade to expected_version, and keep device in this state. |
| - if the current version is already the expected_version, |
| - Downgrade from expected_version to downgrade_version |
| (verify downgrade handling capability of expected_version image) |
| - Finally upgrade to expected_version, and keep device in this state. |
| |
| At the same time, it also verifies: |
| - configure ACS with these software versions |
| """ |
| max_test_idx = 3 |
| current_test_idx = 1 |
| upgrade = True |
| |
| while current_test_idx <= max_test_idx: |
| info = self.log.CreateResultInfo( |
| '---', 'Test Sequence Number: ' + str(current_test_idx)) |
| 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) |
| |
| if self.params['current_version'] == self.params['expected_version']: |
| if current_test_idx == 1: |
| current_test_idx += 1 |
| upgrade = False |
| continue |
| |
| version = None |
| if upgrade: |
| self.time_stamp = self.ConfigACS(self.params['expected_version_id']) |
| info = self.log.CreateProgressInfo('40%', 'Expected Version: ' |
| + self.params['expected_version']) |
| self.log.SendLine(self.test_info, info) |
| version = self.params['expected_version'] |
| else: |
| self.time_stamp = self.ConfigACS(self.params['downgrade_version_id']) |
| info = self.log.CreateProgressInfo('40%', 'Downgrade Version: ' |
| + self.params['downgrade_version']) |
| self.log.SendLine(self.test_info, info) |
| version = self.params['downgrade_version'] |
| |
| info = self.log.CreateProgressInfo( |
| '50%', 'Successfully configured ACS for image version: ' + version) |
| self.log.SendLine(self.test_info, info) |
| |
| time.sleep(self.__short_delay) # time delay for log update |
| self.time_stamp = self.GetTimeStamp() |
| if self.CheckImageInstallation(self.time_stamp, version): |
| info = self.log.CreateProgressInfo( |
| '100%', 'Image successfully upgraded to expected version: ' |
| 'from: ' + self.params['current_version'] |
| + ' to: ' + 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: ' + 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) |
| info = self.log.CreateResultInfo( |
| 'Failed', 'Image failed to upgraded to expected version: ' |
| 'from: ' + self.params['current_version'] |
| + ' to: ' + version) |
| self.log.SendLine(self.test_info, info) |
| |
| current_test_idx += 1 |
| if upgrade: |
| upgrade = False |
| else: |
| upgrade = True |
| |
| def ImageUpgrade(self, expected_version, expected_version_id): |
| """This function upgrade device image to expected version. |
| |
| Args: |
| expected_version: the version to be upgraded to |
| expected_version_id: the id of the expected version |
| Returns: |
| True if upgrade succeeds |
| False if upgrade fails |
| """ |
| version = self.GetVersion() |
| if version == expected_version: |
| info = self.log.CreateProgressInfo( |
| '---', 'Image is already expected versioin: ' |
| + expected_version) |
| self.log.SendLine(self.test_info, info) |
| return True |
| self.time_stamp = self.ConfigACS(expected_version_id) |
| info = self.log.CreateProgressInfo( |
| '---', 'Successfully configured ACS for image version: ' |
| + expected_version) |
| self.log.SendLine(self.test_info, info) |
| |
| time.sleep(self.__short_delay) # time delay for log update |
| self.time_stamp = self.GetTimeStamp() |
| if self.CheckImageInstallation(self.time_stamp, expected_version): |
| info = self.log.CreateProgressInfo( |
| '---', 'Image successfully upgraded to expected version: ' |
| 'from: ' + version + ' to: ' + expected_version) |
| self.log.SendLine(self.test_info, info) |
| return True |
| else: |
| info = self.log.CreateErrorInfo( |
| 'critical', |
| 'Error! Image version does not match expected version!') |
| self.log.SendLine(self.test_info, info) |
| return False |
| |
| def VerifyDownloadInterrupt(self): |
| """Negative test case: verify image download interruption. |
| |
| This case tests device's capability of image upgrade in case of |
| power loss, power cycle, or multiple download requests from ACS. |
| The device should finally be able to complete the download and |
| image upgrade process from expected_version to downgrade_version. |
| |
| Returns: |
| return: False if the verification fails |
| return: True otherwise |
| """ |
| self.CheckVersion() |
| if self.params['current_version'] != self.params['expected_version']: |
| info = self.log.CreateProgressInfo( |
| '5%', 'Current version is not expected version to test, upgrade to: ' |
| + self.params['expected_version']) |
| self.log.SendLine(self.test_info, info) |
| if not self.ImageUpgrade(self.params['expected_version'], |
| self.params['expected_version_id']): |
| info = self.log.CreateErrorInfo( |
| 'critical', |
| 'Failed to upgrade version to: ' |
| + self.params['expected_version']) |
| self.log.SendLine(self.test_info, info) |
| return False |
| self.ConfigACS(self.params['downgrade_version_id']) |
| info = self.log.CreateProgressInfo( |
| '10%', 'Successfully configured ACS for image version: ' |
| + self.params['downgrade_version']) |
| self.log.SendLine(self.test_info, info) |
| |
| time.sleep(6*self.__short_delay) |
| |
| info = self.log.CreateProgressInfo( |
| '40%', 'Download started, reboot the device.') |
| self.log.SendLine(self.test_info, info) |
| time_stamp = self.GetTimeStamp() |
| self.p_ssh.SendCmd('reboot') |
| self.p_ssh.GetCmdOutput() |
| while not self.IsRebooted(time_stamp): |
| self.p_ssh.Close() |
| time.sleep(self.__short_delay*3) |
| self.p_ssh.SshRetry() |
| info = self.log.CreateProgressInfo( |
| '60%', 'Downloading interrupted, waiting for device boot up.') |
| self.log.SendLine(self.test_info, info) |
| |
| time_stamp = self.GetTimeStamp() |
| version = self.GetVersion() |
| if version != self.params['downgrade_version']: |
| count = 0 |
| while not self.IsRebooted(time_stamp): |
| count += 1 |
| if count*self.__short_delay > self.__long_delay: |
| info = self.log.CreateErrorInfo( |
| 'critical', |
| 'Error! Image downloading is not successful after interruption!') |
| self.log.SendLine(self.test_info, info) |
| info = self.log.CreateResultInfo( |
| 'Failed', |
| 'Image failed to upgraded to version: ' |
| + self.params['downgrade_version'] + ' (' |
| + self.__long_delay + ' seconds) after interrupted.') |
| self.log.SendLine(self.test_info, info) |
| return False |
| time_stamp = self.GetTimeStamp() |
| info = self.log.CreateProgressInfo( |
| '80%', 'Waiting for downloading complete after interruption.') |
| self.log.SendLine(self.test_info, info) |
| time.sleep(self.__short_delay*5) |
| |
| info = self.log.CreateProgressInfo( |
| '100%', 'Image successfully upgraded to version: ' |
| + self.params['downgrade_version'] + ' after interrupted.') |
| self.log.SendLine(self.test_info, info) |
| info = self.log.CreateResultInfo( |
| 'Pass', 'Image successfully upgraded to version: ' |
| + self.params['downgrade_version'] + ' after interrupted.') |
| self.log.SendLine(self.test_info, info) |
| return True |
| |
| def Run(self): |
| """Run the test case.""" |
| ####### Add your code here -- BEGIN ####### |
| print 'Test Started...' |
| self.Init() |
| self.p_ssh.SshRetry() |
| info = self.log.CreateProgressInfo( |
| '5%', 'SSH session to device successfully established!') |
| self.log.SendLine(self.test_info, info) |
| |
| if self.test_info['testID'] == '11843140': |
| self.VerifyImageDownload() |
| info = self.log.CreateResultInfo( |
| '----', 'For file transfer test, refer to image download test result:' |
| ' testID: 11843140.') |
| self.log.SendLine(self.test_info, info) |
| elif self.test_info['testID'] == '14165036': |
| self.VerifyDownloadInterrupt() |
| |
| 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__': |
| # Test cases defined in this file: |
| test_defined = ['11843140', '14165036', '14334001'] |
| # To execute test cases that defined in config file: |
| test_to_execute = testCase.TestCase.FindTestCaseToExecute('config.cfg') |
| for test_id in test_to_execute: |
| if test_id in test_defined: |
| test = DownloadTest(testID=test_id) |
