# See the License for the specific language governing permissions and
# limitations under the License.
+import glob
import os
import re
PCI_PATH = "/sys/class/net/%s/device/virtfn%s/net"
VIRTFN_FORMAT = "^virtfn(?P<vf_index>\d+)"
VIRTFN_REG_EX = re.compile(VIRTFN_FORMAT)
+ MAC_VTAP_PREFIX = "upper_macvtap*"
@classmethod
def scan_vf_devices(cls, dev_name):
"""Check if VF is assigned.
Checks if a given vf index of a given device name is assigned
- by checking the relevant path in the system
+ by checking the relevant path in the system:
+ VF is assigned if:
+ Direct VF: PCI_PATH does not exist.
+ Macvtap VF: upper_macvtap path exists.
@param dev_name: pf network device name
@param vf_index: vf index
"""
path = cls.PCI_PATH % (dev_name, vf_index)
- return not (os.path.isdir(path))
+ if not os.path.isdir(path):
+ return True
+ upper_macvtap_path = os.path.join(path, "*", cls.MAC_VTAP_PREFIX)
+ return bool(glob.glob(upper_macvtap_path))
class EmbSwitch(object):
def test_is_assigned_vf_false(self):
self._mock_assign_vf(False)
+
+ def _mock_assign_vf_macvtap(self, macvtap_exists):
+ def _glob(file_path):
+ return ["upper_macvtap0"] if macvtap_exists else []
+
+ with contextlib.nested(
+ mock.patch("os.path.isdir",
+ return_value=True),
+ mock.patch("glob.glob",
+ side_effect=_glob)):
+ result = esm.PciOsWrapper.is_assigned_vf(self.DEV_NAME,
+ self.VF_INDEX)
+ self.assertEqual(macvtap_exists, result)
+
+ def test_is_assigned_vf_macvtap_true(self):
+ self._mock_assign_vf_macvtap(True)
+
+ def test_is_assigned_vf_macvtap_false(self):
+ self._mock_assign_vf_macvtap(False)