| #!/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() |