Release 0.4.6

Bugfix:
* Use persist-queue instead of queuelib
* [GH-87]Add VNXLunSyncCompletedError for migration

README.rst

@@ -3,17 +3,17 @@ StorOps: The Python Library for VNX & Unity
 
 .. image:: https://img.shields.io/travis/emc-openstack/storops.svg
     :target: https://travis-ci.org/emc-openstack/storops
-    
+
 .. image:: https://img.shields.io/coveralls/emc-openstack/storops.svg
     :target: https://coveralls.io/github/emc-openstack/storops
-    
+
 .. image:: https://img.shields.io/pypi/v/storops.svg
     :target: https://pypi.python.org/pypi/storops
 
 .. image:: https://landscape.io/github/emc-openstack/storops/master/landscape.svg?style=flat
     :target: https://landscape.io/github/emc-openstack/storops/
 
-VERSION: 0.4.5
+VERSION: 0.4.6
 
 A minimalist Python library to manage VNX/Unity systems.
 This document lies in the source code and go with the release.

requirements.txt

@@ -8,4 +8,4 @@ python-dateutil>=2.4.2 # BSD
 retryz>=0.1.8 # Apache-2.0
 cachez>=0.1.0 # Apache-2.0
 bitmath>=1.3.0 # MIT
-queuelib>=1.4.2 # BSD
+persist-queue>=0.1.4 # Apache-2.0

storops/exception.py

@@ -711,6 +711,11 @@ class VNXLunNotMigratingError(VNXMigrationError):
     error_message = 'The specified source LUN is not currently migrating.'
 
 
+@cli_exception
+class VNXLunSyncCompletedError(VNXMigrationError):
+    error_code = 0x714a8021
+
+
 @cli_exception
 class VNXTargetNotReadyError(VNXMigrationError):
     error_message = 'The destination LUN is not available for migration'

storops/lib/tasks.py

@@ -15,14 +15,16 @@
 #    under the License.
 
 import logging
-import queuelib
-import pickle
+import persistqueue
+
+from persistqueue import Empty
 import threading
 import time
 from storops.exception import StoropsException
 from storops.exception import VNXObjectNotFoundError
 
 __author__ = 'Peter Wang'
+
 log = logging.getLogger(__name__)
 
 
@@ -32,19 +34,21 @@ class PQueue(object):
 
     def __init__(self, path, interval=None):
         self.path = path
-        self._q = queuelib.FifoDiskQueue(self.path)
+        self._q = persistqueue.Queue(self.path)
         self._interval = (
             self.DEFAULT_INTERVAL if interval is None else interval)
         self.started = False
 
     def put(self, func, **kwargs):
         item = {'object': func.__self__, 'method': func.__name__,
                 'params': kwargs}
-        self._q.push(self._dumps(item))
+        self._q.put_nowait(item)
+
+    def get(self, block=False):
+        return self._q.get(block=block)
 
-    def get(self):
-        item = self._q.pop()
-        return self._loads(item) if item else None
+    def task_done(self):
+            self._q.task_done()
 
     def start(self):
         if not self.started:
@@ -57,12 +61,6 @@ def stop(self):
         self._interval = 0
         self.started = False
 
-    def _dumps(self, obj):
-        return pickle.dumps(obj)
-
-    def _loads(self, pickle_bytes):
-        return pickle.loads(pickle_bytes)
-
     def set_interval(self, interval):
         self._interval = interval
 
@@ -81,16 +79,17 @@ def re_enqueue(self, item):
             else:
                 retries += 1
                 item['retries'] = retries
-                self._q.push(self._dumps(item))
+                self._q.put_nowait(item)
         else:
             item['retries'] = 1
-            self._q.push(self._dumps(item))
+            self._q.put_nowait(item)
 
     def _run_tasks(self):
         while self._interval > 0:
             log.debug("Running periodical check.")
-            data = self.get()
-            if not data:
+            try:
+                data = self.get()
+            except Empty:
                 log.debug("Queue is empty now.")
             else:
                 method = getattr(data['object'], data['method'], None)
@@ -110,6 +109,7 @@ def _run_tasks(self):
                         log.error("Unexpected error occurs when executing {}:"
                                   " {}, this job will not be executed"
                                   " again".format(method.__name__, ex))
+                self.task_done()
             time.sleep(self._interval)
         log.info("{} with path {} has been "
                  "stopped.".format(self.__class__.__name__, self._q.path))

storops/vnx/resource/migration.py

@@ -60,6 +60,10 @@ def get(cls, cli, source=None):
     @property
     @instance_cache
     def source_lun(self):
+        if self._source is not None:
+            return storops.vnx.resource.lun.VNXLun(
+                lun_id=self._source,
+                cli=self._cli)
         return storops.vnx.resource.lun.VNXLun.get(
             cli=self._cli, lun_id=self.source_lu_id, name=self.source_lu_name)
 

test/lib/test_tasks.py

@@ -19,7 +19,8 @@
 import shutil
 from unittest import TestCase
 import tempfile
-from hamcrest import assert_that, equal_to
+from hamcrest import assert_that, equal_to, raises
+from persistqueue import Empty
 
 from storops.lib import tasks
 from test.vnx.cli_mock import patch_cli, t_vnx
@@ -52,6 +53,10 @@ def test_get(self):
         assert_that(pickled_item['object']._ip, equal_to(fake_vnx._ip))
         assert_that(pickled_item['method'], equal_to('delete_lun'))
         assert_that(pickled_item['params']['name'], equal_to('l1'))
+        self.q.task_done()
+        self.q = None
+        self.q = tasks.PQueue(self.path)
+        assert_that(self.q.get, raises(Empty))
 
     def test_run_empty_queue(self):
         self.q.set_interval(0.01)
@@ -91,8 +96,7 @@ def test_enqueue_expected_error(self):
         self.q.put(fake_vnx.delete_hba, hba_uid=uid)
         self.q.start()
         time.sleep(0.2)
-        reenqueued_item = self.q.get()
-        assert_that(None, equal_to(reenqueued_item))
+        assert_that(self.q.get, raises(Empty))
 
     @patch_cli
     def test_enqueue_storops_error(self):

test/vnx/resource/test_migration.py

@@ -19,7 +19,7 @@
 
 from hamcrest import assert_that, equal_to, instance_of, raises
 
-from storops.exception import VNXLunNotMigratingError
+from storops.exception import VNXLunNotMigratingError, VNXLunSyncCompletedError
 from storops.vnx.resource.lun import VNXLun
 from test.vnx.cli_mock import t_cli, patch_cli
 from storops.vnx.enums import VNXMigrationRate
@@ -88,3 +88,12 @@ def f():
 
         assert_that(f, raises(VNXLunNotMigratingError,
                               'not currently migrating'))
+
+    @patch_cli
+    def test_cancel_migrate_sync_completed(self):
+        def f():
+            ms = VNXMigrationSession(1, t_cli())
+            ms.cancel()
+
+        assert_that(f, raises(VNXLunSyncCompletedError,
+                              'because data sychronization is completed'))

test/vnx/testdata/block_output/migrate_-cancel_-source_1_-o.txt

@@ -0,0 +1,2 @@
+Error: migrate -cancel command failed.
+Migration cannot be cancelled because data sychronization is completed (0x714a8021)