test_vif_pool.py 文件源码

def test__get_in_use_ports(self):
        cls = vif_pool.BaseVIFPool
        m_driver = mock.MagicMock(spec=cls)

        kubernetes = self.useFixture(k_fix.MockK8sClient()).client
        pod = get_pod_obj()
        port_id = 'f2c1b73a-6a0c-4dca-b986-0d07d09e0c02'
        versioned_object = jsonutils.dumps({
            'versioned_object.data': {
                'active': True,
                'address': 'fa:16:3e:ef:e6:9f',
                'id': port_id
            }})

        pod['metadata']['annotations'][constants.K8S_ANNOTATION_VIF] = (
            versioned_object)
        items = [pod]
        kubernetes.get.return_value = {'items': items}

        resp = cls._get_in_use_ports(m_driver)

        self.assertEqual(resp, [port_id])