Add QoS policy object and CRUD commands.

Closes-Bug: 1609037
Depends-On: Idf319cd182304952071bc976a2e56c42fbcb8468

Change-Id: I33bafeca979410d329ae10a82772ccdb48c10daa

doc/source/users/resources/network/index.rst

@@ -19,6 +19,7 @@ Network Resources
    v2/pool
    v2/pool_member
    v2/port
+   v2/qos_policy
    v2/quota
    v2/rbac_policy
    v2/router

doc/source/users/resources/network/v2/qos_policy.rst

@@ -0,0 +1,12 @@
+openstack.network.v2.qos_policy
+===============================
+
+.. automodule:: openstack.network.v2.qos_policy
+
+The QoSPolicy Class
+-------------------
+
+The ``QoSPolicy`` class inherits from :class:`~openstack.resource.Resource`.
+
+.. autoclass:: openstack.network.v2.qos_policy.QoSPolicy
+   :members:

openstack/network/v2/_proxy.py

@@ -25,6 +25,7 @@ from openstack.network.v2 import network_ip_availability
 from openstack.network.v2 import pool as _pool
 from openstack.network.v2 import pool_member as _pool_member
 from openstack.network.v2 import port as _port
+from openstack.network.v2 import qos_policy as _qos_policy
 from openstack.network.v2 import quota as _quota
 from openstack.network.v2 import rbac_policy as _rbac_policy
 from openstack.network.v2 import router as _router
@@ -1890,3 +1891,86 @@ class Proxy(proxy.BaseProxy):
         :rtype: :class:`~openstack.network.v2.vpn_service.VPNService`
         """
         return self._update(_vpn_service.VPNService, vpn_service, **attrs)
+
+    def create_qos_policy(self, **attrs):
+        """Create a new QoS policy from attributes
+
+        :param dict attrs: Keyword arguments which will be used to create
+                           a :class:`~openstack.network.v2.qos_policy.
+                           QoSPolicy`, comprised of the properties on the
+                           QoSPolicy class.
+
+        :returns: The results of QoS policy creation
+        :rtype: :class:`~openstack.network.v2.qos_policy.QoSPolicy`
+        """
+        return self._create(_qos_policy.QoSPolicy, **attrs)
+
+    def delete_qos_policy(self, qos_policy, ignore_missing=True):
+        """Delete a QoS policy
+
+        :param qos_policy: The value can be either the ID of a QoS policy or a
+                           :class:`~openstack.network.v2.qos_policy.QoSPolicy`
+                           instance.
+        :param bool ignore_missing: When set to ``False``
+                    :class:`~openstack.exceptions.ResourceNotFound` will be
+                    raised when the QoS policy does not exist.
+                    When set to ``True``, no exception will be set when
+                    attempting to delete a nonexistent QoS policy.
+
+        :returns: ``None``
+        """
+        self._delete(_qos_policy.QoSPolicy, qos_policy,
+                     ignore_missing=ignore_missing)
+
+    def find_qos_policy(self, name_or_id, ignore_missing=True):
+        """Find a single QoS policy
+
+        :param name_or_id: The name or ID of a QoS policy.
+        :param bool ignore_missing: When set to ``False``
+                    :class:`~openstack.exceptions.ResourceNotFound` will be
+                    raised when the resource does not exist.
+                    When set to ``True``, None will be returned when
+                    attempting to find a nonexistent resource.
+        :returns: One :class:`~openstack.network.v2.qos_policy.QoSPolicy` or
+                  None
+        """
+        return self._find(_qos_policy.QoSPolicy, name_or_id,
+                          ignore_missing=ignore_missing)
+
+    def get_qos_policy(self, qos_policy):
+        """Get a single QoS policy
+
+        :param qos_policy: The value can be the ID of a QoS policy or a
+                           :class:`~openstack.network.v2.qos_policy.QoSPolicy`
+                           instance.
+
+        :returns: One :class:`~openstack.network.v2.qos_policy.QoSPolicy`
+        :raises: :class:`~openstack.exceptions.ResourceNotFound`
+                 when no resource can be found.
+        """
+        return self._get(_qos_policy.QoSPolicy, qos_policy)
+
+    def qos_policies(self, **query):
+        """Return a generator of QoS policies
+
+        :param kwargs \*\*query: Optional query parameters to be sent to limit
+                                 the resources being returned.
+
+        :returns: A generator of QoS policy objects
+        :rtype: :class:`~openstack.network.v2.qos_policy.QoSPolicy`
+        """
+        return self._list(_qos_policy.QoSPolicy, paginated=False, **query)
+
+    def update_qos_policy(self, qos_policy, **attrs):
+        """Update a QoS policy
+
+        :param qos_policy: Either the id of a QoS policy or a
+                           :class:`~openstack.network.v2.qos_policy.QoSPolicy`
+                           instance.
+        :attrs kwargs: The attributes to update on the QoS policy represented
+                       by ``value``.
+
+        :returns: The updated QoS policy
+        :rtype: :class:`~openstack.network.v2.qos_policy.QoSPolicy`
+        """
+        return self._update(_qos_policy.QoSPolicy, qos_policy, **attrs)

openstack/network/v2/qos_policy.py

@@ -0,0 +1,42 @@
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+from openstack.network import network_service
+from openstack import resource
+
+
+class QoSPolicy(resource.Resource):
+    resource_key = 'policy'
+    resources_key = 'policies'
+    base_path = '/qos/policies'
+    service = network_service.NetworkService()
+
+    # capabilities
+    allow_create = True
+    allow_retrieve = True
+    allow_update = True
+    allow_delete = True
+    allow_list = True
+
+    # Properties
+    #: QoS policy name.
+    name = resource.prop('name')
+    #: The ID of the project who owns the network. Only administrative
+    #: users can specify a project ID other than their own.
+    project_id = resource.prop('tenant_id')
+    #: The QoS policy description.
+    description = resource.prop('description')
+    #: Indicates whether this QoS policy is shared across all projects.
+    #: *Type: bool*
+    is_shared = resource.prop('shared', type=bool)
+    #: List of QoS rules applied to this QoS policy.
+    rules = resource.prop('rules')

openstack/tests/functional/network/v2/test_qos_policy.py

@@ -0,0 +1,64 @@
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+import uuid
+
+from openstack.network.v2 import qos_policy as _qos_policy
+from openstack.tests.functional import base
+
+
+class TestQoSPolicy(base.BaseFunctionalTest):
+
+    QOS_POLICY_ID = None
+    QOS_POLICY_NAME = uuid.uuid4().hex
+    QOS_POLICY_NAME_UPDATED = uuid.uuid4().hex
+    IS_SHARED = False
+    RULES = []
+    QOS_POLICY_DESCRIPTION = "QoS policy description"
+
+    @classmethod
+    def setUpClass(cls):
+        super(TestQoSPolicy, cls).setUpClass()
+        qos = cls.conn.network.create_qos_policy(
+            description=cls.QOS_POLICY_DESCRIPTION,
+            name=cls.QOS_POLICY_NAME,
+            shared=cls.IS_SHARED,
+        )
+        assert isinstance(qos, _qos_policy.QoSPolicy)
+        cls.assertIs(cls.QOS_POLICY_NAME, qos.name)
+        cls.QOS_POLICY_ID = qos.id
+
+    @classmethod
+    def tearDownClass(cls):
+        sot = cls.conn.network.delete_qos_policy(cls.QOS_POLICY_ID)
+        cls.assertIs(None, sot)
+
+    def test_find(self):
+        sot = self.conn.network.find_qos_policy(self.QOS_POLICY_NAME)
+        self.assertEqual(self.QOS_POLICY_ID, sot.id)
+
+    def test_get(self):
+        sot = self.conn.network.get_qos_policy(self.QOS_POLICY_ID)
+        self.assertEqual(self.QOS_POLICY_NAME, sot.name)
+        self.assertEqual(self.IS_SHARED, sot.is_shared)
+        self.assertEqual(self.RULES, sot.rules)
+        self.assertEqual(self.QOS_POLICY_DESCRIPTION, sot.description)
+
+    def test_list(self):
+        names = [o.name for o in self.conn.network.qos_policies()]
+        self.assertIn(self.QOS_POLICY_NAME, names)
+
+    def test_update(self):
+        sot = self.conn.network.update_qos_policy(
+            self.QOS_POLICY_ID,
+            name=self.QOS_POLICY_NAME_UPDATED)
+        self.assertEqual(self.QOS_POLICY_NAME_UPDATED, sot.name)

openstack/tests/unit/network/v2/test_proxy.py

@@ -28,6 +28,7 @@ from openstack.network.v2 import network_ip_availability
 from openstack.network.v2 import pool
 from openstack.network.v2 import pool_member
 from openstack.network.v2 import port
+from openstack.network.v2 import qos_policy
 from openstack.network.v2 import quota
 from openstack.network.v2 import rbac_policy
 from openstack.network.v2 import router
@@ -382,6 +383,30 @@ class TestNetworkProxy(test_proxy_base.TestProxyBase):
     def test_port_update(self):
         self.verify_update(self.proxy.update_port, port.Port)
 
+    def test_qos_policy_create_attrs(self):
+        self.verify_create(self.proxy.create_qos_policy, qos_policy.QoSPolicy)
+
+    def test_qos_policy_delete(self):
+        self.verify_delete(self.proxy.delete_qos_policy, qos_policy.QoSPolicy,
+                           False)
+
+    def test_qos_policy_delete_ignore(self):
+        self.verify_delete(self.proxy.delete_qos_policy, qos_policy.QoSPolicy,
+                           True)
+
+    def test_qos_policy_find(self):
+        self.verify_find(self.proxy.find_qos_policy, qos_policy.QoSPolicy)
+
+    def test_qos_policy_get(self):
+        self.verify_get(self.proxy.get_qos_policy, qos_policy.QoSPolicy)
+
+    def test_qos_policies(self):
+        self.verify_list(self.proxy.qos_policies, qos_policy.QoSPolicy,
+                         paginated=False)
+
+    def test_qos_policy_update(self):
+        self.verify_update(self.proxy.update_qos_policy, qos_policy.QoSPolicy)
+
     def test_quota_delete(self):
         self.verify_delete(self.proxy.delete_quota, quota.Quota, False)
 

openstack/tests/unit/network/v2/test_qos_policy.py

@@ -0,0 +1,49 @@
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+import testtools
+import uuid
+
+from openstack.network.v2 import qos_policy
+
+EXAMPLE = {
+    'id': 'IDENTIFIER',
+    'description': 'QoS policy description',
+    'name': 'qos-policy-name',
+    'shared': True,
+    'tenant_id': '2',
+    'rules': [uuid.uuid4().hex]
+}
+
+
+class TestQoSPolicy(testtools.TestCase):
+
+    def test_basic(self):
+        sot = qos_policy.QoSPolicy()
+        self.assertEqual('policy', sot.resource_key)
+        self.assertEqual('policies', sot.resources_key)
+        self.assertEqual('/qos/policies', sot.base_path)
+        self.assertEqual('network', sot.service.service_type)
+        self.assertTrue(sot.allow_create)
+        self.assertTrue(sot.allow_retrieve)
+        self.assertTrue(sot.allow_update)
+        self.assertTrue(sot.allow_delete)
+        self.assertTrue(sot.allow_list)
+
+    def test_make_it(self):
+        sot = qos_policy.QoSPolicy(EXAMPLE)
+        self.assertEqual(EXAMPLE['id'], sot.id)
+        self.assertEqual(EXAMPLE['description'], sot.description)
+        self.assertEqual(EXAMPLE['name'], sot.name)
+        self.assertTrue(sot.is_shared)
+        self.assertEqual(EXAMPLE['tenant_id'], sot.project_id)
+        self.assertEqual(EXAMPLE['rules'], sot.rules)