1# Copyright (C) 2010 Google Inc. All rights reserved. 2# Copyright (C) 2010 Gabor Rapcsanyi (rgabor@inf.u-szeged.hu), University of Szeged 3# 4# Redistribution and use in source and binary forms, with or without 5# modification, are permitted provided that the following conditions are 6# met: 7# 8# * Redistributions of source code must retain the above copyright 9# notice, this list of conditions and the following disclaimer. 10# * Redistributions in binary form must reproduce the above 11# copyright notice, this list of conditions and the following disclaimer 12# in the documentation and/or other materials provided with the 13# distribution. 14# * Neither the name of Google Inc. nor the names of its 15# contributors may be used to endorse or promote products derived from 16# this software without specific prior written permission. 17# 18# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 19# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 20# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR 21# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT 22# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, 23# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT 24# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, 25# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY 26# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT 27# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE 28# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. 29 30"""Unit tests for manager.py.""" 31 32import optparse 33import time 34import unittest 35 36from blinkpy.common.host_mock import MockHost 37from blinkpy.web_tests.controllers.manager import Manager 38from blinkpy.web_tests.models import test_expectations 39from blinkpy.web_tests.models.test_run_results import TestRunResults 40 41 42class FakePrinter(object): 43 44 def write_update(self, s): 45 pass 46 47 48class ManagerTest(unittest.TestCase): 49 50 def test_needs_servers(self): 51 def get_manager(): 52 host = MockHost() 53 port = host.port_factory.get('test-mac-mac10.10') 54 manager = Manager(port, options=optparse.Values({'http': True, 'max_locked_shards': 1}), printer=FakePrinter()) 55 return manager 56 57 manager = get_manager() 58 self.assertFalse(manager._needs_servers(['fast/html'])) 59 60 manager = get_manager() 61 self.assertTrue(manager._needs_servers(['http/tests/misc'])) 62 63 def test_servers_started(self): 64 def get_manager(port): 65 manager = Manager(port, options=optparse.Values({'http': True, 'max_locked_shards': 1}), printer=FakePrinter()) 66 return manager 67 68 def start_http_server(additional_dirs, number_of_drivers): 69 self.http_started = True 70 71 def start_websocket_server(): 72 self.websocket_started = True 73 74 def stop_http_server(): 75 self.http_stopped = True 76 77 def stop_websocket_server(): 78 self.websocket_stopped = True 79 80 host = MockHost() 81 port = host.port_factory.get('test-mac-mac10.10') 82 port.start_http_server = start_http_server 83 port.start_websocket_server = start_websocket_server 84 port.stop_http_server = stop_http_server 85 port.stop_websocket_server = stop_websocket_server 86 87 self.http_started = self.http_stopped = self.websocket_started = self.websocket_stopped = False 88 manager = get_manager(port) 89 manager._start_servers(['http/tests/foo.html']) 90 self.assertEqual(self.http_started, True) 91 self.assertEqual(self.websocket_started, False) 92 manager._stop_servers() 93 self.assertEqual(self.http_stopped, True) 94 self.assertEqual(self.websocket_stopped, False) 95 96 self.http_started = self.http_stopped = self.websocket_started = self.websocket_stopped = False 97 manager._start_servers(['http/tests/websocket/foo.html']) 98 self.assertEqual(self.http_started, True) 99 self.assertEqual(self.websocket_started, True) 100 manager._stop_servers() 101 self.assertEqual(self.http_stopped, True) 102 self.assertEqual(self.websocket_stopped, True) 103 104 self.http_started = self.http_stopped = self.websocket_started = self.websocket_stopped = False 105 manager._start_servers(['fast/html/foo.html']) 106 self.assertEqual(self.http_started, False) 107 self.assertEqual(self.websocket_started, False) 108 manager._stop_servers() 109 self.assertEqual(self.http_stopped, False) 110 self.assertEqual(self.websocket_stopped, False) 111 112 def test_look_for_new_crash_logs(self): 113 def get_manager(): 114 host = MockHost() 115 port = host.port_factory.get('test-mac-mac10.10') 116 manager = Manager( 117 port, 118 options=optparse.Values({'test_list': None, 'http': True, 'max_locked_shards': 1}), 119 printer=FakePrinter()) 120 return manager 121 host = MockHost() 122 port = host.port_factory.get('test-mac-mac10.10') 123 tests = ['failures/expected/crash.html'] 124 expectations = test_expectations.TestExpectations(port) 125 run_results = TestRunResults(expectations, len(tests)) 126 manager = get_manager() 127 manager._look_for_new_crash_logs(run_results, time.time()) 128 129 def _make_fake_test_result(self, host, results_directory): 130 host.filesystem.maybe_make_directory(results_directory) 131 host.filesystem.write_binary_file(results_directory + '/results.html', 'This is a test results file') 132 133 def test_rename_results_folder(self): 134 host = MockHost() 135 port = host.port_factory.get('test-mac-mac10.10') 136 137 def get_manager(): 138 manager = Manager(port, options=optparse.Values({'max_locked_shards': 1}), printer=FakePrinter()) 139 return manager 140 self._make_fake_test_result(port.host, '/tmp/layout-test-results') 141 self.assertTrue(port.host.filesystem.exists('/tmp/layout-test-results')) 142 timestamp = time.strftime( 143 '%Y-%m-%d-%H-%M-%S', time.localtime(port.host.filesystem.mtime('/tmp/layout-test-results/results.html'))) 144 archived_file_name = '/tmp/layout-test-results' + '_' + timestamp 145 manager = get_manager() 146 manager._rename_results_folder() 147 self.assertFalse(port.host.filesystem.exists('/tmp/layout-test-results')) 148 self.assertTrue(port.host.filesystem.exists(archived_file_name)) 149 150 def test_clobber_old_results(self): 151 host = MockHost() 152 port = host.port_factory.get('test-mac-mac10.10') 153 154 def get_manager(): 155 manager = Manager(port, options=optparse.Values({'max_locked_shards': 1}), printer=FakePrinter()) 156 return manager 157 self._make_fake_test_result(port.host, '/tmp/layout-test-results') 158 self.assertTrue(port.host.filesystem.exists('/tmp/layout-test-results')) 159 manager = get_manager() 160 manager._clobber_old_results() 161 self.assertFalse(port.host.filesystem.exists('/tmp/layout-test-results')) 162 163 def test_limit_archived_results_count(self): 164 host = MockHost() 165 port = host.port_factory.get('test-mac-mac10.10') 166 167 def get_manager(): 168 manager = Manager(port, options=optparse.Values({'max_locked_shards': 1}), printer=FakePrinter()) 169 return manager 170 for x in range(1, 31): 171 dir_name = '/tmp/layout-test-results' + '_' + str(x) 172 self._make_fake_test_result(port.host, dir_name) 173 manager = get_manager() 174 manager._limit_archived_results_count() 175 deleted_dir_count = 0 176 for x in range(1, 31): 177 dir_name = '/tmp/layout-test-results' + '_' + str(x) 178 if not port.host.filesystem.exists(dir_name): 179 deleted_dir_count = deleted_dir_count + 1 180 self.assertEqual(deleted_dir_count, 5) 181