master: reject drop/tweak ctl commands that could lead to unwanted status

The following 2 operations can be onerous and they should not be
directly usable without some kind of confirmation by the user:
- Dropping a node now requires to first stop it.
- Tweaking does not exclude anymore automatically DOWN nodes,
  because a node could go DOWN between the moment the user sends
  the command to tweak and the actual tweak by the master.

neo/master/handlers/administration.py

@@ -120,6 +120,9 @@ class AdministrationHandler(MasterHandler):
                    'node already in %s state' % state)
         if node.isStorage():
             keep = state == NodeStates.DOWN
+            if node.isRunning() and not keep:
+                raise AnswerDenied(
+                    "a running node must be stopped before removal")
             try:
                 cell_list = app.pt.dropNodeList([node], keep)
             except PartitionTableException, e:
@@ -197,8 +200,14 @@ class AdministrationHandler(MasterHandler):
     @__change_pt_rpc
     def tweakPartitionTable(self, conn, dry_run, uuid_list):
         app = self.app
-        drop_list = [node for node in app.nm.getStorageList()
-            if node.getUUID() in uuid_list or not node.isRunning()]
+        drop_list = []
+        for node in app.nm.getStorageList():
+            if node.getUUID() in uuid_list or node.isPending():
+                drop_list.append(node)
+            elif not node.isRunning():
+                drop_list.append(node)
+                raise AnswerDenied(
+                    'tweak: down nodes must be listed explicitly')
         if dry_run:
             pt = object.__new__(app.pt.__class__)
             new_nodes = pt.load(app.pt.getID(), app.pt.getReplicas(),

neo/tests/threaded/test.py

@@ -472,6 +472,7 @@ class Test(NEOThreadedTest):
             self.assertFalse(conn.isClosed())
             getCellSortKey = cluster.client.getCellSortKey
             self.assertEqual(getCellSortKey(s0, good), 0)
+            cluster.neoctl.killNode(s0.getUUID())
             cluster.neoctl.dropNode(s0.getUUID())
             self.assertEqual([s1], cluster.client.nm.getStorageList())
             self.assertTrue(conn.isClosed())
@@ -777,6 +778,7 @@ class Test(NEOThreadedTest):
             checkNodeState(NodeStates.RUNNING)
             self.assertEqual([], cluster.getOutdatedCells())
             # drop one
+            cluster.neoctl.killNode(s1.uuid)
             cluster.neoctl.dropNode(s1.uuid)
             checkNodeState(None)
             self.tic() # Let node state update reach remaining storage

neo/tests/threaded/testReplication.py

@@ -601,8 +601,9 @@ class ReplicationTests(NEOThreadedTest):
                     tweak()
                 t.commit()
                 t2.join()
-            cluster.neoctl.dropNode(S[2].uuid)
-            cluster.neoctl.dropNode(S[3].uuid)
+            for s in S[2:]:
+                cluster.neoctl.killNode(s.uuid)
+                cluster.neoctl.dropNode(s.uuid)
             cluster.neoctl.tweakPartitionTable()
             if done:
                 f.remove(delay)
@@ -988,6 +989,19 @@ class ReplicationTests(NEOThreadedTest):
         expected = '|'.join(['U.U.|.U.U'] * 3)
         self.assertPartitionTable(cluster, expected)
 
+    @with_cluster(partitions=3, replicas=1, storage_count=3)
+    def testAdminOnerousOperationCondition(self, cluster):
+        s = cluster.storage_list[2]
+        cluster.neoctl.killNode(s.uuid)
+        tweak = cluster.neoctl.tweakPartitionTable
+        self.assertRaises(SystemExit, tweak)
+        self.assertRaises(SystemExit, tweak, dry_run=True)
+        self.assertTrue(tweak((s.uuid,))[0])
+        self.tic()
+        cluster.neoctl.dropNode(s.uuid)
+        s = cluster.storage_list[1]
+        self.assertRaises(SystemExit, cluster.neoctl.dropNode, s.uuid)
+
     @with_cluster(partitions=5, replicas=2, storage_count=3)
     def testCheckReplicas(self, cluster):
         from neo.storage import checker