Merge "Add disable_root test scenario"

trove_tempest_plugin/services/client.py

@@ -58,10 +58,11 @@ class TroveClient(rest_client.RestClient):
         retry=tenacity.retry_if_exception_type(),
         reraise=True
     )
-    def delete_resource(self, obj, id, ignore_notfound=False):
+    def delete_resource(self, obj, id, ignore_notfound=False,
+                        expected_status_code=202):
         try:
             resp, _ = self.delete('/{obj}/{id}'.format(obj=obj, id=id))
-            self.expected_success(202, resp.status)
+            self.expected_success(expected_status_code, resp.status)
             return resp
         except exceptions.NotFound:
             if ignore_notfound:

trove_tempest_plugin/tests/base.py

@@ -652,6 +652,22 @@ class BaseTroveTest(test.BaseTestCase):
         resp = cls.client.create_resource(f"instances/{instance_id}/root", {})
         return resp['user']['password']
 
+    @classmethod
+    def disable_root_access(cls, instance_id):
+        cls.client.delete_resource(f"instances/{instance_id}", 'root',
+                                   expected_status_code=204)
+
+    @classmethod
+    def enable_root_access(cls, instance_id, password={}):
+        if password:
+            password = {"password": password}
+        cls.client.create_resource(f"instances/{instance_id}/root", password)
+
+    @classmethod
+    def is_root_enabled(cls, instance_id):
+        resp = cls.client.get_resource(f"instances/{instance_id}", 'root')
+        return resp['rootEnabled']
+
     @classmethod
     def rebuild_instance(cls, instance_id, image_id):
         rebuild_req = {

trove_tempest_plugin/tests/scenario/base_basic.py

@@ -112,6 +112,32 @@ class TestInstanceBasicBase(trove_base.BaseTroveTest):
         self.assertNotEqual(value, cur_value)
         self.assertNotEqual(new_value, cur_value)
 
+    @decorators.idempotent_id('bb262e9f-10c1-4271-8ad1-4b8b585e0ee1')
+    def test_root_disable(self):
+        # get the default root status
+        root_enabled = self.is_root_enabled(self.instance_id)
+        self.assertEqual(self.enable_root, root_enabled)
+        # enable the root
+        if self.enable_root:
+            password = self.get_root_pass(self.instance_id)
+            self.enable_root_access(self.instance_id, password)
+        else:
+            self.enable_root_access(self.instance_id)
+        root_enabled = self.is_root_enabled(self.instance_id)
+        self.assertTrue(root_enabled)
+        # disable the root
+        self.disable_root_access(self.instance_id)
+        root_enabled = self.is_root_enabled(self.instance_id)
+        self.assertFalse(root_enabled)
+        # enable again
+        if self.enable_root:
+            password = self.get_root_pass(self.instance_id)
+            self.enable_root_access(self.instance_id, password)
+        else:
+            self.enable_root_access(self.instance_id)
+        root_enabled = self.is_root_enabled(self.instance_id)
+        self.assertTrue(root_enabled)
+
 
 class TestInstanceBasicMySQLBase(TestInstanceBasicBase):
     def _access_db(self, ip, username=constants.DB_USER,