Merge "Add compute microversion 2.78"

openstack/compute/v2/server.py

@@ -555,6 +555,22 @@ class Server(resource.Resource, metadata.MetadataMixin, resource.TagMixin):
         self._action(
             session, {'os-migrateLive': body}, microversion=microversion)
 
+    def fetch_topology(self, session):
+        utils.require_microversion(session, 2.78)
+
+        url = utils.urljoin(Server.base_path, self.id, 'topology')
+
+        response = session.get(url)
+
+        exceptions.raise_from_response(response)
+
+        try:
+            data = response.json()
+            if 'topology' in data:
+                return data['topology']
+        except ValueError:
+            pass
+
     def fetch_security_groups(self, session):
         """Fetch security groups of a server.
 

openstack/tests/unit/compute/v2/test_server.py

@@ -1010,6 +1010,73 @@ class TestServer(base.TestCase):
         self.sess.post.assert_called_with(
             url, json=body, headers=headers, microversion='2.30')
 
+    def test_get_topology(self):
+        sot = server.Server(**EXAMPLE)
+
+        class FakeEndpointData:
+            min_microversion = '2.1'
+            max_microversion = '2.78'
+        self.sess.get_endpoint_data.return_value = FakeEndpointData()
+        self.sess.default_microversion = None
+
+        response = mock.Mock()
+
+        topology = {
+            "nodes": [
+                {
+                    "cpu_pinning": {
+                        "0": 0,
+                        "1": 5
+                    },
+                    "host_node": 0,
+                    "memory_mb": 1024,
+                    "siblings": [
+                        [
+                            0,
+                            1
+                        ]
+                    ],
+                    "vcpu_set": [
+                        0,
+                        1
+                    ]
+                },
+                {
+                    "cpu_pinning": {
+                        "2": 1,
+                        "3": 8
+                    },
+                    "host_node": 1,
+                    "memory_mb": 2048,
+                    "siblings": [
+                        [
+                            2,
+                            3
+                        ]
+                    ],
+                    "vcpu_set": [
+                        2,
+                        3
+                    ]
+                }
+            ],
+            "pagesize_kb": 4
+        }
+
+        response.status_code = 200
+        response.json.return_value = {
+            'topology': topology
+        }
+
+        self.sess.get.return_value = response
+
+        fetched_topology = sot.fetch_topology(self.sess)
+
+        url = 'servers/IDENTIFIER/topology'
+        self.sess.get.assert_called_with(url)
+
+        self.assertEqual(fetched_topology, topology)
+
     def test_get_security_groups(self):
         sot = server.Server(**EXAMPLE)