openstack-cyborg-7.0.0.tar.gz: .../test_fpga_ext_arq.py

"Fossies" - the Fresh Open Source Software Archive

Member "openstack-cyborg-7.0.0/cyborg/tests/unit/objects/test_fpga_ext_arq.py" (6 Oct 2021, 17337 Bytes) of package /linux/misc/openstack/openstack-cyborg-7.0.0.tar.gz:

As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file.

1 # Copyright 2019 Intel. 2 # All Rights Reserved. 3 # 4 # Licensed under the Apache License, Version 2.0 (the "License"); you may 5 # not use this file except in compliance with the License. You may obtain 6 # a copy of the License at 7 # 8 # http://www.apache.org/licenses/LICENSE-2.0 9 # 10 # Unless required by applicable law or agreed to in writing, software 11 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 12 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 13 # License for the specific language governing permissions and limitations 14 # under the License. 15 16 from io import BytesIO 17 import json 18 import requests 19 from requests import structures 20 from requests import utils 21 from unittest import mock 22 23 from testtools.matchers import HasLength 24 25 from cyborg.common import constants 26 from cyborg.common import exception 27 from cyborg import objects 28 from cyborg.tests.unit.db import base 29 from cyborg.tests.unit import fake_deployable 30 from cyborg.tests.unit import fake_extarq 31 32 33 class TestFPGAExtARQObject(base.DbTestCase): 34 35 def setUp(self): 36 super(TestFPGAExtARQObject, self).setUp() 37 self.fake_fpga_db_extarqs = fake_extarq.get_fake_fpga_db_extarqs() 38 self.fake_obj_fpga_extarqs = fake_extarq.get_fake_fpga_extarq_objs() 39 classes = ["no_program", "bitstream_program", 40 "function_program", "bad_program"] 41 self.class_fgpa_objects = dict( 42 zip(classes, self.fake_obj_fpga_extarqs)) 43 self.class_fgpa_dbs = dict( 44 zip(classes, self.fake_fpga_db_extarqs)) 45 self.bitstream_id = self.class_fgpa_objects["bitstream_program"][ 46 "device_profile_group"][constants.ACCEL_BITSTREAM_ID] 47 self.function_id = self.class_fgpa_objects["function_program"][ 48 "device_profile_group"][constants.ACCEL_FUNCTION_ID] 49 self.images_md = { 50 "/images": 51 [ 52 {"id": self.bitstream_id, 53 "tags": ["trait:CUSTOM_FPGA_INTEL"], 54 constants.ACCEL_FUNCTION_ID: self.function_id}, 55 ] 56 } 57 self.deployable_uuids = ['0acbf8d6-e02a-4394-aae3-57557d209498'] 58 self.bdf = {"domain": "0000", "bus": "00", 59 "device": "01", "function": "1"} 60 self.cpid = { 61 "id": 0, 62 "uuid": "e4a66b0d-b377-40d6-9cdc-6bf7e720e596", 63 "device_id": "1", 64 "cpid_type": "PCI", 65 "cpid_info": json.dumps(self.bdf).encode('utf-8') 66 } 67 68 def response(self, status_code=200, content='', headers=None, 69 reason=None, elapsed=0, request=None, stream=False): 70 res = requests.Response() 71 res.status_code = status_code 72 if isinstance(content, (dict, list)): 73 content = json.dumps(content).encode('utf-8') 74 if isinstance(content, str): 75 content = content.encode('utf-8') 76 res._content = content 77 res._content_consumed = content 78 res.headers = structures.CaseInsensitiveDict(headers or {}) 79 res.encoding = utils.get_encoding_from_headers(res.headers) 80 res.reason = reason 81 res.request = request 82 if hasattr(request, 'url'): 83 res.url = request.url 84 if isinstance(request.url, bytes): 85 res.url = request.url.decode('utf-8') 86 87 if stream: 88 res.raw = BytesIO(content) 89 else: 90 res.raw = BytesIO(b'') 91 # normally this closes the underlying connection, 92 # but we have nothing to free. 93 res.close = lambda *args, **kwargs: None 94 return res 95 96 def images_get(self, url, *args, **kw): 97 images = {"images": []} 98 params = kw.get("params", {}) 99 bit_id = None 100 res = None 101 if url != "/images": 102 bit_id = url.rsplit("/", 1)[-1] 103 params["id"] = bit_id 104 tags = kw.pop("tags", []) 105 for image in self.images_md["/images"]: 106 if tags <= image["tags"] and params.items() <= image.items(): 107 images["images"].append(image) 108 res = self.response(content=image) 109 return res if bit_id else self.response(content=images) 110 111 @mock.patch('cyborg.objects.ExtARQ._from_db_object') 112 def test_get_bitstream_id_return_None(self, mock_from_db_obj): 113 db_extarq = self.fake_fpga_db_extarqs[0] 114 uuid = db_extarq['uuid'] 115 mock_from_db_obj.return_value = self.fake_obj_fpga_extarqs[0] 116 with mock.patch.object( 117 self.dbapi, 'extarq_get') as mock_extarq_get: 118 mock_extarq_get.return_value = db_extarq 119 obj_extarq = objects.FPGAExtARQ.get(self.context, uuid) 120 mock_extarq_get.assert_called_once_with(self.context, uuid) 121 self.assertEqual(obj_extarq.arq.uuid, uuid) 122 self.assertIsNone(obj_extarq._get_bitstream_id()) 123 124 @mock.patch('cyborg.objects.ExtARQ._from_db_object') 125 def test_get_bitstream_id_return_UUID(self, mock_from_db_obj): 126 db_extarq = self.fake_fpga_db_extarqs[1] 127 bit_id = db_extarq['device_profile_group'][ 128 constants.ACCEL_BITSTREAM_ID] 129 uuid = db_extarq['uuid'] 130 mock_from_db_obj.return_value = self.fake_obj_fpga_extarqs[1] 131 with mock.patch.object( 132 self.dbapi, 'extarq_get') as mock_extarq_get: 133 mock_extarq_get.return_value = db_extarq 134 obj_extarq = objects.FPGAExtARQ.get(self.context, uuid) 135 mock_extarq_get.assert_called_once_with(self.context, uuid) 136 self.assertEqual(obj_extarq.arq.uuid, uuid) 137 self.assertEqual(obj_extarq._get_bitstream_id(), bit_id) 138 139 @mock.patch('cyborg.objects.ExtARQ._from_db_object') 140 def test_get_function_id_return_UUID(self, mock_from_db_obj): 141 db_extarq = self.fake_fpga_db_extarqs[2] 142 fun_id = db_extarq['device_profile_group'][constants.ACCEL_FUNCTION_ID] 143 uuid = db_extarq['uuid'] 144 mock_from_db_obj.return_value = self.fake_obj_fpga_extarqs[2] 145 with mock.patch.object( 146 self.dbapi, 'extarq_get') as mock_extarq_get: 147 mock_extarq_get.return_value = db_extarq 148 obj_extarq = objects.FPGAExtARQ.get(self.context, uuid) 149 mock_extarq_get.assert_called_once_with(self.context, uuid) 150 self.assertEqual(obj_extarq.arq.uuid, uuid) 151 self.assertEqual(obj_extarq._get_function_id(), fun_id) 152 153 @mock.patch('cyborg.objects.ExtARQ._from_db_object_list') 154 def test_list(self, mock_from_db_obj_list): 155 db_extarqs = self.fake_fpga_db_extarqs 156 mock_from_db_obj_list.return_value = self.fake_obj_fpga_extarqs 157 with mock.patch.object(self.dbapi, 'extarq_list', 158 autospec=True) as mock_get_list: 159 mock_get_list.return_value = db_extarqs 160 obj_extarqs = objects.FPGAExtARQ.list(self.context) 161 self.assertEqual(1, mock_get_list.call_count) 162 self.assertThat(obj_extarqs, HasLength(4)) 163 self.assertIsInstance(obj_extarqs[0], objects.FPGAExtARQ) 164 for i, obj_extarq in enumerate(obj_extarqs): 165 self.assertEqual(obj_extarq.arq.uuid, db_extarqs[i]['uuid']) 166 167 @mock.patch('openstack.connection.Connection') 168 def test_get_bitstream_md_from_bitstream_id(self, mock_conn): 169 mock_conn.return_value = type( 170 "Connection", (object,), 171 {"image": type("image", (object,), {"get": self.images_get})}) 172 obj_extarq = self.class_fgpa_objects["bitstream_program"] 173 md = obj_extarq._get_bitstream_md_from_bitstream_id(self.bitstream_id) 174 self.assertDictEqual(self.images_md["/images"][0], md) 175 176 @mock.patch('openstack.connection.Connection') 177 def test_get_bitstream_md_from_function_id(self, mock_conn): 178 mock_conn.return_value = type( 179 "Connection", (object,), 180 {"image": type("image", (object,), {"get": self.images_get})}) 181 obj_extarq = self.class_fgpa_objects["function_program"] 182 md = obj_extarq._get_bitstream_md_from_function_id(self.function_id) 183 self.assertDictEqual(self.images_md["/images"][0], md) 184 185 @mock.patch('openstack.connection.Connection') 186 @mock.patch('cyborg.objects.ExtARQ.update_check_state') 187 def test_needs_programming(self, mock_check_state, mock_conn): 188 mock_conn.return_value = type( 189 "Connection", (object,), 190 {"image": type("image", (object,), {"get": self.images_get})}) 191 dep_uuid = self.deployable_uuids[0] 192 fake_dep = fake_deployable.fake_deployable_obj( 193 self.context, uuid=dep_uuid) 194 195 # function_id is require 196 obj_extarq = self.class_fgpa_objects["function_program"] 197 need = obj_extarq._needs_programming(self.context, fake_dep) 198 self.assertTrue(need) 199 200 # bitstream_id is require 201 obj_extarq = self.class_fgpa_objects["bitstream_program"] 202 need = obj_extarq._needs_programming(self.context, fake_dep) 203 self.assertTrue(need) 204 205 # Both bitstream_id and function_id are require 206 obj_extarq = self.class_fgpa_objects["bad_program"] 207 self.assertRaises( 208 exception.InvalidParameterValue, obj_extarq._needs_programming, 209 self.context, fake_dep) 210 211 # None of bitstream_id or function_id is require 212 obj_extarq = self.class_fgpa_objects["no_program"] 213 need = obj_extarq._needs_programming(self.context, fake_dep) 214 self.assertFalse(need) 215 216 @mock.patch('openstack.connection.Connection') 217 @mock.patch('cyborg.objects.ExtARQ.update_check_state') 218 def test_get_bitstream_md(self, mock_check_state, mock_conn): 219 mock_conn.return_value = type( 220 "Connection", (object,), 221 {"image": type("image", (object,), {"get": self.images_get})}) 222 dep_uuid = self.deployable_uuids[0] 223 fake_dep = fake_deployable.fake_deployable_obj( 224 self.context, uuid=dep_uuid) 225 226 # function_id is require 227 obj_extarq = self.class_fgpa_objects["function_program"] 228 bs_id = obj_extarq._get_bitstream_id() 229 fun_id = obj_extarq._get_function_id() 230 md = obj_extarq.get_bitstream_md( 231 self.context, fake_dep, fun_id, bs_id) 232 self.assertDictEqual(self.images_md["/images"][0], md) 233 234 # bitstream_id is require 235 obj_extarq = self.class_fgpa_objects["bitstream_program"] 236 bs_id = obj_extarq._get_bitstream_id() 237 fun_id = obj_extarq._get_function_id() 238 md = obj_extarq.get_bitstream_md( 239 self.context, fake_dep, fun_id, bs_id) 240 self.assertDictEqual(self.images_md["/images"][0], md) 241 242 # Both bitstream_id and function_id are require 243 obj_extarq = self.class_fgpa_objects["bad_program"] 244 bs_id = obj_extarq._get_bitstream_id() 245 fun_id = obj_extarq._get_function_id() 246 self.assertRaises( 247 exception.InvalidParameterValue, obj_extarq.get_bitstream_md, 248 self.context, fake_dep, fun_id, bs_id) 249 250 # None of bitstream_id or function_id is require 251 obj_extarq = self.class_fgpa_objects["no_program"] 252 bs_id = obj_extarq._get_bitstream_id() 253 fun_id = obj_extarq._get_function_id() 254 md = obj_extarq.get_bitstream_md( 255 self.context, fake_dep, fun_id, bs_id) 256 self.assertIsNone(md) 257 258 @mock.patch('openstack.connection.Connection') 259 @mock.patch('cyborg.objects.ExtARQ.update_check_state') 260 def test_need_extra_bind_job(self, mock_check_state, mock_conn): 261 mock_conn.return_value = type( 262 "Connection", (object,), 263 {"image": type("image", (object,), {"get": self.images_get})}) 264 dep_uuid = self.deployable_uuids[0] 265 fake_dep = fake_deployable.fake_deployable_obj( 266 self.context, uuid=dep_uuid) 267 268 # function_id is require 269 obj_extarq = self.class_fgpa_objects["function_program"] 270 need = obj_extarq._need_extra_bind_job(self.context, fake_dep) 271 self.assertTrue(need) 272 273 # bitstream_id is require 274 obj_extarq = self.class_fgpa_objects["bitstream_program"] 275 need = obj_extarq._need_extra_bind_job(self.context, fake_dep) 276 self.assertTrue(need) 277 278 # Both bitstream_id and function_id are require 279 obj_extarq = self.class_fgpa_objects["bad_program"] 280 self.assertRaises( 281 exception.InvalidParameterValue, obj_extarq._need_extra_bind_job, 282 self.context, fake_dep) 283 284 # None of bitstream_id or function_id is require 285 obj_extarq = self.class_fgpa_objects["no_program"] 286 need = obj_extarq._need_extra_bind_job(self.context, fake_dep) 287 self.assertFalse(need) 288 289 @mock.patch('cyborg.agent.rpcapi.AgentAPI.fpga_program') 290 @mock.patch('openstack.connection.Connection') 291 @mock.patch('cyborg.objects.ExtARQ.update_check_state') 292 @mock.patch('cyborg.objects.Deployable.get_cpid_list') 293 def test_do_programming(self, mock_cpid_list, mock_check_state, mock_conn, 294 mock_program): 295 dep_uuid = self.deployable_uuids[0] 296 fake_dep = fake_deployable.fake_deployable_obj( 297 self.context, uuid=dep_uuid) 298 299 # function_id is require 300 obj_extarq = self.class_fgpa_objects["function_program"] 301 obj_extarq.arq.hostname = 'newtestnode1' 302 fake_dep.driver_name = "intel_fpga" 303 304 mock_cpid_list.return_value = [self.cpid] 305 bs_id = obj_extarq._get_bitstream_id() 306 obj_extarq._do_programming(self.context, fake_dep, bs_id) 307 mock_program.assert_called_once_with( 308 self.context, 'newtestnode1', self.cpid, bs_id, "intel_fpga") 309 310 @mock.patch('cyborg.common.placement_client.PlacementClient.' 311 '__init__') 312 @mock.patch('cyborg.common.placement_client.PlacementClient.' 313 'add_traits_to_rp') 314 @mock.patch('cyborg.common.placement_client.PlacementClient.' 315 'delete_traits_with_prefixes') 316 def test_update_placement( 317 self, mock_delete_traits, mock_add_traits, mock_placement_init): 318 mock_placement_init.return_value = None 319 dep_uuid = self.deployable_uuids[0] 320 fake_dep = fake_deployable.fake_deployable_obj( 321 self.context, uuid=dep_uuid) 322 323 # function_id is require 324 obj_extarq = self.class_fgpa_objects["function_program"] 325 obj_extarq.arq.hostname = 'newtestnode1' 326 # fake_dep.driver_name = "intel_fpga" 327 fun_id = obj_extarq._get_function_id() 328 rp_uuid = obj_extarq.arq.device_rp_uuid 329 obj_extarq._update_placement( 330 self.context, fun_id, fake_dep.driver_name) 331 332 function_id = fun_id.upper().replace('-', '_-') 333 vendor = fake_dep.driver_name.upper() 334 trait_names = ["_".join(( 335 constants.FPGA_FUNCTION_ID, vendor, function_id))] 336 mock_add_traits.assert_called_once_with(rp_uuid, trait_names) 337 mock_delete_traits.assert_called_once_with( 338 self.context, rp_uuid, [constants.FPGA_FUNCTION_ID]) 339 340 @mock.patch('cyborg.agent.rpcapi.AgentAPI.fpga_program') 341 @mock.patch('cyborg.objects.Deployable.get_cpid_list') 342 @mock.patch('cyborg.objects.Deployable.update') 343 @mock.patch('cyborg.common.placement_client.PlacementClient.' 344 '__init__') 345 @mock.patch('cyborg.common.placement_client.PlacementClient.' 346 'add_traits_to_rp') 347 @mock.patch('cyborg.common.placement_client.PlacementClient.' 348 'delete_traits_with_prefixes') 349 @mock.patch('openstack.connection.Connection') 350 @mock.patch('cyborg.objects.ExtARQ.update_check_state') 351 @mock.patch('cyborg.objects.ExtARQ.bind') 352 def test_bind( 353 self, mock_bind, mock_check_state, mock_conn, mock_delete_traits, 354 mock_add_traits, mock_placement_init, mock_dp_update, 355 mock_cpid_list, mock_program): 356 357 mock_placement_init.return_value = None 358 mock_conn.return_value = type( 359 "Connection", (object,), 360 {"image": type("image", (object,), {"get": self.images_get})}) 361 362 dep_uuid = self.deployable_uuids[0] 363 fake_dep = fake_deployable.fake_deployable_obj( 364 self.context, uuid=dep_uuid) 365 366 mock_cpid_list.return_value = [self.cpid] 367 368 # None of bitstream_id or function_id is require 369 obj_extarq = self.class_fgpa_objects["no_program"] 370 obj_extarq.bind(self.context, fake_dep) 371 self.assertFalse(mock_program.called) 372 self.assertFalse(mock_placement_init.called) 373 self.assertFalse(mock_dp_update.called) 374 mock_bind.assert_called_once_with(self.context, fake_dep) 375 376 self.cpid["cpid_info"] = json.dumps(self.bdf).encode('utf-8') 377 # function_id is require 378 obj_extarq = self.class_fgpa_objects["function_program"] 379 obj_extarq.bind(self.context, fake_dep) 380 mock_bind.assert_called_with(self.context, fake_dep) 381 self.assertEqual(mock_bind.call_count, 2) 382 383 self.cpid["cpid_info"] = json.dumps(self.bdf).encode('utf-8') 384 # bitstream_id is require 385 obj_extarq = self.class_fgpa_objects["bitstream_program"] 386 obj_extarq.bind(self.context, fake_dep) 387 mock_bind.assert_called_with(self.context, fake_dep) 388 self.assertEqual(mock_bind.call_count, 3)