site_utils/lxc/container_pool/client_unittest.py

#!/usr/bin/python
# Copyright 2017 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import os
import shutil
import socket
import tempfile
import unittest

import common
from autotest_lib.site_utils import lxc
from autotest_lib.site_utils.lxc import unittest_setup
from autotest_lib.site_utils.lxc.container_pool import async_listener
from autotest_lib.site_utils.lxc.container_pool import client


# Timeout for tests.
TIMEOUT = 30


class ClientTests(unittest.TestCase):
    """Unit tests for the Client class."""

    @classmethod
    def setUpClass(cls):
        """Creates a directory for running the unit tests."""
        # Explicitly use /tmp as the tmpdir.  Board specific TMPDIRs inside of
        # the chroot are set to a path that causes the socket address to exceed
        # the maximum allowable length.
        cls.test_dir = tempfile.mkdtemp(prefix='client_unittest_', dir='/tmp')


    @classmethod
    def tearDownClass(cls):
        """Deletes the test directory."""
        shutil.rmtree(cls.test_dir)


    def setUp(self):
        """Per-test setup."""
        # Put each test in its own test dir, so it's hermetic.
        self.test_dir = tempfile.mkdtemp(dir=ClientTests.test_dir)
        self.address = os.path.join(self.test_dir,
                                    lxc.DEFAULT_CONTAINER_POOL_SOCKET)
        self.listener = async_listener.AsyncListener(self.address)
        self.listener.start()


    def tearDown(self):
        self.listener.close()


    def testConnection(self):
        """Tests a basic client connection."""
        # Verify that no connections are pending.
        self.assertIsNone(self.listener.get_connection())

        # Connect a client, then verify that the host connection is established.
        host = None
        with client.Client.connect(self.address, TIMEOUT):
            host = self.listener.get_connection(TIMEOUT)
            self.assertIsNotNone(host)

        # Client closed - check that the host connection also closed.
        self.assertTrue(host.poll(TIMEOUT))
        with self.assertRaises(EOFError):
            host.recv()


    def testConnection_badAddress(self):
        """Tests that connecting to a bad address fails."""
        # Make a bogus address, then assert that the client fails.
        address = '%s.foobar' % self.address
        with self.assertRaises(socket.error):
            client.Client(address, 0)


    def testConnection_timeout(self):
        """Tests that connection attempts time out properly."""
        with tempfile.NamedTemporaryFile(dir=self.test_dir) as tmp:
            with self.assertRaises(socket.timeout):
                client.Client(tmp.name, 0)


    def testConnection_deadLine(self):
        """Tests that the connection times out if no action is ever taken."""
        id = 3
        short_timeout = TIMEOUT/2
        with client.Client.connect(self.address, TIMEOUT) as c:
            self.assertIsNone(c.get_container(id, short_timeout))

if __name__ == '__main__':
    unittest_setup.setup(require_sudo=False)
    unittest.main()