v 1.0.10

- fixed iteration bug in qman
- added missing unit test

work related to [https://bugzilla.ipr.univ-rennes.fr/show_bug.cgi?id=3093]

cocluto/quman.py

@@ -91,7 +91,7 @@ class QueueManager:
     def request_queue_deactivation(self, queue_name: QueueMachineId, requester_id: RequesterId, reason: str):
 
         disable_reasons = self.get_disable_reasons(queue_name)
-        for dr in disable_reasons:
+        for dr in disable_reasons.values():
             assert dr.requester_id != requester_id, f"Requester {requester_id} has already requested deactivation of queue {queue_name} for reason '{dr.reason}' at {dr.timestamp.isoformat()}. Cannot request deactivation again without reactivating first."
 
         if len(disable_reasons) == 0:

cocluto/version.py

@@ -1,4 +1,4 @@
-__version__ = '1.0.9'
+__version__ = '1.0.10'
 
 
 class Version(object):

test/test_quman.py

@@ -0,0 +1,31 @@
+from pathlib import Path
+import unittest
+import logging
+# from cocluto import ClusterController
+from cocluto.SimpaDbUtil import SqliteDb
+from cocluto.quman import QueueManager, init_db
+
+
+class QumanTestCase(unittest.TestCase):
+
+    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+
+    def setUp(self) -> None:
+        return super().setUp()
+
+    def test_quman(self):
+        logging.info('test_quman')
+        db_path = Path('./quman_test/quman.sqlite')
+        if db_path.exists():
+            db_path.unlink()
+        db_backend = SqliteDb(db_path)
+        init_db(db_backend)
+        quman = QueueManager(db_backend)
+        quman.request_queue_deactivation('main.q@alambix42', 'sysadmin.graffy', 'because I want to test quman')
+        quman.request_queue_activation('main.q@alambix42', 'sysadmin.graffy', 'because I want to test quman')
+        # self.assertIsInstance(job_state, JobsState)
+        db_backend.dump(Path('./quman_test/quman_dump.sql'))
+
+
+if __name__ == '__main__':
+    unittest.main()