test_driver.py 文件源码

def test_issue_etcd_request_basic_get(self):
        # Initialise the etcd URL.
        self.driver._handle_init({
            MSG_KEY_ETCD_URLS: ["http://localhost:4001/"],
            MSG_KEY_HOSTNAME: "ourhost",
            MSG_KEY_KEY_FILE: None,
            MSG_KEY_CERT_FILE: None,
            MSG_KEY_CA_FILE: None
        })
        m_pool = Mock(spec=HTTPConnectionPool)
        self.driver._issue_etcd_request(m_pool, "calico/v1/Ready")
        self.assertEqual(
            m_pool.request.mock_calls,
            [call("GET",
                  "http://localhost:4001/v2/keys/calico/v1/Ready",
                  fields=None,
                  timeout=5,
                  preload_content=True)]
        )