Merge changes I8be872f3,I13fe65bf
am: 29db248522
Change-Id: I64b90ca69e2cbb39bb809130a8dcab9ad4490d57
diff --git a/at-factory-tool/atft.py b/at-factory-tool/atft.py
index f50aeed..a6eeaba 100644
--- a/at-factory-tool/atft.py
+++ b/at-factory-tool/atft.py
@@ -3110,7 +3110,7 @@
evt = Event(self.print_event, wx.ID_ANY, msg)
wx.QueueEvent(self, evt)
- def _StartOperation(self, operation, target):
+ def _StartOperation(self, operation, target, show_alert=True):
if not target:
self.PauseRefresh()
return True
@@ -3120,9 +3120,11 @@
self.PauseRefresh()
return True
- self._SendAlertEvent(
- 'Target: ' + str(target) + ' is currently in another operation: ' +
- target.operation + '. Please try again later')
+ if show_alert:
+ self._SendAlertEvent(
+ 'Unable to start operation: ' + operation + ', ' +
+ 'Target: ' + str(target) + ' is currently in another operation: ' +
+ target.operation + '. Please try again later')
return False
def _EndOperation(self, target):
@@ -3898,10 +3900,9 @@
self.auto_prov_lock.acquire()
serial = target.serial_number
i = 0
- while not ProvisionStatus.isFailed(target.provision_status):
+ while True:
target = self.atft_manager.GetTargetDevice(serial)
- if not target:
- # The target disappear somehow.
+ if not target or ProvisionStatus.isFailed(target.provision_status):
break
if not self.auto_prov:
# Auto provision mode exited.
@@ -4069,7 +4070,7 @@
# Should not reach here.
return False
operation = 'ATFA device prepare and download ' + file_type + ' file'
- if not self._StartOperation(operation, atfa_dev):
+ if not self._StartOperation(operation, atfa_dev, show_alert):
return False
try:
filepath = filepath.encode('utf-8')
diff --git a/at-factory-tool/atft_unittest.py b/at-factory-tool/atft_unittest.py
index 06d458f..d6926ad 100644
--- a/at-factory-tool/atft_unittest.py
+++ b/at-factory-tool/atft_unittest.py
@@ -687,6 +687,106 @@
self.assertEqual(ProvisionStatus.REBOOT_FAILED, test_dev1.provision_status)
mock_atft._SendOperationSucceedEvent.assert_not_called()
+ def mockGetTargetDeviceDisappear(self, dev):
+ # If the device disappear, return None as target device.
+ if self.target_device_disapper:
+ return None
+ else:
+ return dev
+
+ def mockDeviceRebootTimeout(self, target, auto_prov):
+ # After reboot, the device disappear.
+ self.target_device_disapper = True
+
+ # Test device reboot timeout and disappear from device list.
+ def testHandleStateTransitionRebootTimeout(self):
+ mock_atft = MockAtft()
+ self.target_device_disapper = False
+ mock_atft._SendOperationSucceedEvent = MagicMock()
+ test_dev1 = TestDeviceInfo(self.TEST_SERIAL1, self.TEST_LOCATION1,
+ ProvisionStatus.WAITING)
+ mock_atft._FuseVbootKeyTarget = MagicMock()
+ mock_atft._FuseVbootKeyTarget.side_effect = self.mockDeviceRebootTimeout
+ mock_atft._FusePermAttrTarget = MagicMock()
+ mock_atft._FusePermAttrTarget.side_effect = (
+ lambda target=mock_atft, state=ProvisionStatus.FUSEATTR_SUCCESS:
+ self.MockStateChange(target, state))
+ mock_atft._LockAvbTarget = MagicMock()
+ mock_atft._LockAvbTarget.side_effect = (
+ lambda target=mock_atft, state=ProvisionStatus.LOCKAVB_SUCCESS:
+ self.MockStateChange(target, state))
+ mock_atft._ProvisionTarget = MagicMock()
+ mock_atft._ProvisionTarget.side_effect = (
+ lambda target, is_som_key, state=ProvisionStatus.PROVISION_SUCCESS:
+ self.MockStateChange(target, state))
+ mock_atft.auto_dev_serials = [self.TEST_SERIAL1]
+ mock_atft.auto_prov = True
+ mock_atft.atft_manager = MagicMock()
+ mock_atft.atft_manager.GetTargetDevice = MagicMock()
+ mock_atft.atft_manager.GetTargetDevice.side_effect = (
+ lambda serial, dev=test_dev1: self.mockGetTargetDeviceDisappear(dev))
+ mock_atft._HandleStateTransition(test_dev1)
+ self.assertEqual(
+ None, mock_atft.atft_manager.GetTargetDevice(self.TEST_SERIAL1))
+ mock_atft._SendOperationSucceedEvent.assert_not_called()
+
+ def mockGetTargetDeviceFuseVbootFailed(self, dev):
+ # If the device disappear, return None as target device.
+ if self.target_device_fuse_failed:
+ new_device = TestDeviceInfo(
+ dev.serial_number, dev.location, ProvisionStatus.FUSEVBOOT_FAILED)
+ return new_device
+ else:
+ return dev
+
+ def mockDeviceFuseVbootFailed(self, target, auto_prov):
+ # After reboot, the device disappear.
+ self.target_device_fuse_failed = True
+ target.provision_status = ProvisionStatus.REBOOT_ING
+
+ # Test fuse vboot key change target device state by creating a new target
+ # device instead of modifying the original one's state.
+ def testHandleStateTransitionTargetDeviceChange(self):
+ mock_atft = MockAtft()
+ self.target_device_fuse_failed = False
+ mock_atft._SendOperationSucceedEvent = MagicMock()
+ test_dev1 = TestDeviceInfo(self.TEST_SERIAL1, self.TEST_LOCATION1,
+ ProvisionStatus.WAITING)
+ mock_atft._FuseVbootKeyTarget = MagicMock()
+ mock_atft._FuseVbootKeyTarget.side_effect = self.mockDeviceFuseVbootFailed
+ mock_atft._FusePermAttrTarget = MagicMock()
+ mock_atft._FusePermAttrTarget.side_effect = (
+ lambda target=mock_atft, state=ProvisionStatus.FUSEATTR_SUCCESS:
+ self.MockStateChange(target, state))
+ mock_atft._LockAvbTarget = MagicMock()
+ mock_atft._LockAvbTarget.side_effect = (
+ lambda target=mock_atft, state=ProvisionStatus.LOCKAVB_SUCCESS:
+ self.MockStateChange(target, state))
+ mock_atft._ProvisionTarget = MagicMock()
+ mock_atft._ProvisionTarget.side_effect = (
+ lambda target, is_som_key, state=ProvisionStatus.PROVISION_SUCCESS:
+ self.MockStateChange(target, state))
+ mock_atft.auto_dev_serials = [self.TEST_SERIAL1]
+ mock_atft.auto_prov = True
+ mock_atft.atft_manager = MagicMock()
+ mock_atft.atft_manager.GetTargetDevice = MagicMock()
+ mock_atft.atft_manager.GetTargetDevice.side_effect = (
+ lambda serial, dev=test_dev1: self.mockGetTargetDeviceFuseVbootFailed(
+ dev))
+ mock_atft._HandleStateTransition(test_dev1)
+
+ # The old target device is still in REBOOT_ING state.
+ self.assertEqual(ProvisionStatus.REBOOT_ING, test_dev1.provision_status)
+
+ # The new device is in FUSEVBOOT_FAILED state
+ self.assertEqual(
+ ProvisionStatus.FUSEVBOOT_FAILED,
+ mock_atft.atft_manager.GetTargetDevice(
+ self.TEST_SERIAL1).provision_status)
+ mock_atft._SendOperationSucceedEvent.assert_not_called()
+ # Next operation should not execute.
+ mock_atft._FusePermAttrTarget.assert_not_called()
+
def testHandleStateTransitionFuseAttrFail(self):
mock_atft = MockAtft()
mock_atft._SendOperationSucceedEvent = MagicMock()
