test_datastore.py 文件源码

def test_start_driver(self, m_unlink, m_socket, m_timeout, m_popen,
                          m_exists):
        m_exists.return_value = True
        m_sck = Mock()
        m_socket.return_value = m_sck
        m_conn = Mock()
        m_sck.accept.return_value = m_conn, None
        reader, writer = self.watcher._start_driver()
        self.assertEqual(m_socket.mock_calls[0], call(socket.AF_UNIX,
                                                      socket.SOCK_STREAM))
        self.assertEqual(m_sck.bind.mock_calls,
                         [call("/run/felix-driver.sck")])
        self.assertEqual(m_sck.listen.mock_calls, [call(1)])
        self.assertEqual(m_popen.mock_calls[0],
                         call([ANY, "-m", "calico.etcddriver",
                               "/run/felix-driver.sck"]))
        self.assertEqual(m_unlink.mock_calls,
                         [call("/run/felix-driver.sck")] * 2)
        self.assertTrue(isinstance(reader, MessageReader))
        self.assertTrue(isinstance(writer, MessageWriter))
        m_exists.assert_called_once_with("/run")
        m_timeout.assert_called_once_with(10)