1 # Copyright (c) 2018 Intel Corporation
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
15 import multiprocessing
21 from yardstick.common import utils
24 class ImportModulesFromPackageTestCase(unittest.TestCase):
26 def test_import_package(self):
27 module_name = 'yardstick.tests.functional.common.fake_module'
28 library_name = 'fake_library'
29 class_name = 'FakeClassToBeImported'
30 self.assertNotIn(module_name, sys.modules)
32 utils.import_modules_from_package(module_name)
33 self.assertIn(module_name, sys.modules)
34 module_obj = sys.modules[module_name]
35 library_obj = getattr(module_obj, library_name)
36 class_obj = getattr(library_obj, class_name)
37 self.assertEqual(class_name, class_obj().__class__.__name__)
40 class SendSocketCommandTestCase(unittest.TestCase):
43 def _run_socket_server(port):
44 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
45 sock.bind(('localhost', port))
49 conn, _ = sock.accept()
53 def _terminate_server(socket_server):
54 # Wait until the socket server closes the open port.
56 if socket_server and socket_server.is_alive():
57 socket_server.terminate()
59 def test_send_command(self):
62 socket_server = multiprocessing.Process(
63 name='run_socket_server',
64 target=SendSocketCommandTestCase._run_socket_server,
65 args=(port, )).start()
67 self.addCleanup(self._terminate_server, socket_server)
69 # Wait until the socket is open.
72 0, utils.send_socket_command('localhost', port, 'test_command'))