test_driver.py 文件源码

def test_issue_etcd_request_recursive_watch(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", timeout=10,
                                        wait_index=11, recursive=True)
        self.assertEqual(
            m_pool.request.mock_calls,
            [call("GET",
                  "http://localhost:4001/v2/keys/calico/v1",
                  fields={"recursive": "true",
                          "wait": "true",
                          "waitIndex": 11},
                  timeout=10,
                  preload_content=False)]
        )