1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
|
# Copyright (c) 2018 Intel.
# All Rights Reserved.
#
# 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.
"""Cyborg agent resource_tracker test cases."""
import os
import fixtures
from cyborg.accelerator.drivers.fpga import utils
from cyborg.accelerator.drivers.fpga.intel import sysinfo
from cyborg.agent.resource_tracker import ResourceTracker
from cyborg.conductor import rpcapi as cond_api
from cyborg.conf import CONF
from cyborg.tests import base
from cyborg.tests.unit.accelerator.drivers.fpga.intel import prepare_test_data
class TestResourceTracker(base.TestCase):
"""Test Agent ResourceTracker """
def setUp(self):
super(TestResourceTracker, self).setUp()
self.syspath = sysinfo.SYS_FPGA
sysinfo.SYS_FPGA = "/sys/class/fpga"
tmp_sys_dir = self.useFixture(fixtures.TempDir())
prepare_test_data.create_fake_sysfs(tmp_sys_dir.path)
sysinfo.SYS_FPGA = os.path.join(
tmp_sys_dir.path, sysinfo.SYS_FPGA.split("/", 1)[-1])
utils.SYS_FPGA_PATH = sysinfo.SYS_FPGA
self.host = CONF.host
self.cond_api = cond_api.ConductorAPI()
self.rt = ResourceTracker(self.host, self.cond_api)
def tearDown(self):
super(TestResourceTracker, self).tearDown()
sysinfo.SYS_FPGA = self.syspath
utils.SYS_FPGA_PATH = self.syspath
def test_update_usage(self):
"""Update the resource usage and stats after a change in an
instance
"""
# FIXME(Shaohe Feng) need add testcase. How to check the fpgas
# has stored into DB by conductor correctly?
pass
def test_get_fpga_devices(self):
expect = {
'0000:5e:00.0': {
'function': 'pf', 'assignable': False, 'pr_num': '1',
'name': 'intel-fpga-dev.0', 'vendor_id': '0x8086',
'devices': '0000:5e:00.0',
'regions': [{
'function': 'vf', 'assignable': True,
'name': 'intel-fpga-dev.2', 'vendor_id': '0x8086',
'devices': '0000:5e:00.1',
'parent_devices': '0000:5e:00.0',
'path': '%s/intel-fpga-dev.2' % sysinfo.SYS_FPGA,
'product_id': '0xbcc1'}],
'parent_devices': '',
'path': '%s/intel-fpga-dev.0' % sysinfo.SYS_FPGA,
'product_id': '0xbcc0'},
'0000:5e:00.1': {
'function': 'vf', 'assignable': True,
'name': 'intel-fpga-dev.2', 'vendor_id': '0x8086',
'devices': '0000:5e:00.1',
'parent_devices': '0000:5e:00.0',
'path': '%s/intel-fpga-dev.2' % sysinfo.SYS_FPGA,
'product_id': '0xbcc1'},
'0000:be:00.0': {
'function': 'pf', 'assignable': True, 'pr_num': '0',
'name': 'intel-fpga-dev.1', 'vendor_id': '0x8086',
'devices': '0000:be:00.0', 'parent_devices': '',
'path': '%s/intel-fpga-dev.1' % sysinfo.SYS_FPGA,
'product_id': '0xbcc0'}}
fpgas = self.rt._get_fpga_devices()
self.assertDictEqual(expect, fpgas)
|
