1#!/usr/bin/env python3 2# 3# Copyright 2016 - The Android Open Source Project 4# 5# Licensed under the Apache License, Version 2.0 (the "License"); 6# you may not use this file except in compliance with the License. 7# You may obtain a copy of the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an "AS IS" BASIS, 13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14# See the License for the specific language governing permissions and 15# limitations under the License. 16 17import copy 18import shutil 19import tempfile 20import unittest 21 22import acts.controllers.relay_lib.fugu_remote as fugu_remote 23from acts.controllers.relay_lib.errors import RelayConfigError 24from acts.controllers.relay_lib.errors import RelayDeviceConnectionError 25from acts.controllers.relay_lib.generic_relay_device import GenericRelayDevice 26from acts.controllers.relay_lib.relay import Relay 27from acts.controllers.relay_lib.relay import RelayDict 28from acts.controllers.relay_lib.relay import RelayState 29from acts.controllers.relay_lib.relay import SynchronizeRelays 30from acts.controllers.relay_lib.relay_board import RelayBoard 31from acts.controllers.relay_lib.relay_device import RelayDevice 32from acts.controllers.relay_lib.relay_rig import RelayRig 33from acts.controllers.relay_lib.sain_smart_board import SainSmartBoard 34from mock import patch 35 36 37class MockBoard(RelayBoard): 38 def __init__(self, config): 39 self.relay_states = dict() 40 self.relay_previous_states = dict() 41 RelayBoard.__init__(self, config) 42 43 def get_relay_position_list(self): 44 return [0, 1] 45 46 def get_relay_status(self, relay_position): 47 if relay_position not in self.relay_states: 48 self.relay_states[relay_position] = RelayState.NO 49 self.relay_previous_states[relay_position] = RelayState.NO 50 return self.relay_states[relay_position] 51 52 def set(self, relay_position, state): 53 self.relay_previous_states[relay_position] = self.get_relay_status( 54 relay_position) 55 self.relay_states[relay_position] = state 56 return state 57 58 59class ActsRelayTest(unittest.TestCase): 60 def setUp(self): 61 Relay.transition_wait_time = 0 62 Relay.button_press_time = 0 63 self.config = { 64 'name': 'MockBoard', 65 'relays': [{ 66 'name': 'Relay', 67 'relay_pos': 0 68 }] 69 } 70 self.board = MockBoard(self.config) 71 self.relay = Relay(self.board, 'Relay') 72 self.board.set(self.relay.position, RelayState.NO) 73 74 def tearDown(self): 75 Relay.transition_wait_time = .2 76 Relay.button_press_time = .25 77 78 def test_turn_on_from_off(self): 79 self.board.set(self.relay.position, RelayState.NO) 80 self.relay.set_nc() 81 self.assertEqual( 82 self.board.get_relay_status(self.relay.position), RelayState.NC) 83 84 def test_turn_on_from_on(self): 85 self.board.set(self.relay.position, RelayState.NC) 86 self.relay.set_nc() 87 self.assertEqual( 88 self.board.get_relay_status(self.relay.position), RelayState.NC) 89 90 def test_turn_off_from_on(self): 91 self.board.set(self.relay.position, RelayState.NC) 92 self.relay.set_no() 93 self.assertEqual( 94 self.board.get_relay_status(self.relay.position), RelayState.NO) 95 96 def test_turn_off_from_off(self): 97 self.board.set(self.relay.position, RelayState.NO) 98 self.relay.set_no() 99 self.assertEqual( 100 self.board.get_relay_status(self.relay.position), RelayState.NO) 101 102 def test_toggle_off_to_on(self): 103 self.board.set(self.relay.position, RelayState.NO) 104 self.relay.toggle() 105 self.assertEqual( 106 self.board.get_relay_status(self.relay.position), RelayState.NC) 107 108 def test_toggle_on_to_off(self): 109 self.board.set(self.relay.position, RelayState.NC) 110 self.relay.toggle() 111 self.assertEqual( 112 self.board.get_relay_status(self.relay.position), RelayState.NO) 113 114 def test_set_on(self): 115 self.board.set(self.relay.position, RelayState.NO) 116 self.relay.set(RelayState.NC) 117 self.assertEqual( 118 self.board.get_relay_status(self.relay.position), RelayState.NC) 119 120 def test_set_off(self): 121 self.board.set(self.relay.position, RelayState.NC) 122 self.relay.set(RelayState.NO) 123 self.assertEqual( 124 self.board.get_relay_status(self.relay.position), RelayState.NO) 125 126 def test_set_foo(self): 127 with self.assertRaises(ValueError): 128 self.relay.set('FOO') 129 130 def test_set_nc_for(self): 131 # Here we set twice so relay_previous_state will also be OFF 132 self.board.set(self.relay.position, RelayState.NO) 133 self.board.set(self.relay.position, RelayState.NO) 134 135 self.relay.set_nc_for(0) 136 137 self.assertEqual( 138 self.board.get_relay_status(self.relay.position), RelayState.NO) 139 self.assertEqual(self.board.relay_previous_states[self.relay.position], 140 RelayState.NC) 141 142 def test_set_no_for(self): 143 # Here we set twice so relay_previous_state will also be OFF 144 self.board.set(self.relay.position, RelayState.NC) 145 self.board.set(self.relay.position, RelayState.NC) 146 147 self.relay.set_no_for(0) 148 149 self.assertEqual( 150 self.board.get_relay_status(self.relay.position), RelayState.NC) 151 self.assertEqual(self.board.relay_previous_states[self.relay.position], 152 RelayState.NO) 153 154 def test_get_status_on(self): 155 self.board.set(self.relay.position, RelayState.NC) 156 self.assertEqual(self.relay.get_status(), RelayState.NC) 157 158 def test_get_status_off(self): 159 self.board.set(self.relay.position, RelayState.NO) 160 self.assertEqual(self.relay.get_status(), RelayState.NO) 161 162 def test_clean_up_default_on(self): 163 new_relay = Relay(self.board, 0) 164 new_relay._original_state = RelayState.NO 165 self.board.set(new_relay.position, RelayState.NO) 166 new_relay.clean_up() 167 168 self.assertEqual( 169 self.board.get_relay_status(new_relay.position), RelayState.NO) 170 171 def test_clean_up_default_off(self): 172 new_relay = Relay(self.board, 0) 173 new_relay._original_state = RelayState.NO 174 self.board.set(new_relay.position, RelayState.NC) 175 new_relay.clean_up() 176 177 self.assertEqual( 178 self.board.get_relay_status(new_relay.position), RelayState.NO) 179 180 def test_clean_up_original_state_none(self): 181 val = 'STAYS_THE_SAME' 182 new_relay = Relay(self.board, 0) 183 # _original_state is none by default 184 # The line below sets the dict to an impossible value. 185 self.board.set(new_relay.position, val) 186 new_relay.clean_up() 187 # If the impossible value is cleared, then the test should fail. 188 self.assertEqual(self.board.get_relay_status(new_relay.position), val) 189 190 191class ActsSainSmartBoardTest(unittest.TestCase): 192 STATUS_MSG = ('<small><a href="{}"></a>' 193 '</small><a href="{}/{}TUX">{}TUX</a><p>') 194 195 RELAY_ON_PAGE_CONTENTS = 'relay_on page' 196 RELAY_OFF_PAGE_CONTENTS = 'relay_off page' 197 198 def setUp(self): 199 Relay.transition_wait_time = 0 200 Relay.button_press_time = 0 201 self.test_dir = 'file://' + tempfile.mkdtemp() + '/' 202 203 # Creates the files used for testing 204 self._set_status_page('0000000000000000') 205 with open(self.test_dir[7:] + '00', 'w+') as file: 206 file.write(self.RELAY_OFF_PAGE_CONTENTS) 207 with open(self.test_dir[7:] + '01', 'w+') as file: 208 file.write(self.RELAY_ON_PAGE_CONTENTS) 209 210 self.config = ({ 211 'name': 'SSBoard', 212 'base_url': self.test_dir, 213 'relays': [{ 214 'name': '0', 215 'relay_pos': 0 216 }, { 217 'name': '1', 218 'relay_pos': 1 219 }, { 220 'name': '2', 221 'relay_pos': 7 222 }] 223 }) 224 self.ss_board = SainSmartBoard(self.config) 225 self.r0 = Relay(self.ss_board, 0) 226 self.r1 = Relay(self.ss_board, 1) 227 self.r7 = Relay(self.ss_board, 7) 228 229 def tearDown(self): 230 shutil.rmtree(self.test_dir[7:]) 231 Relay.transition_wait_time = .2 232 Relay.button_press_time = .25 233 234 def test_get_url_code(self): 235 result = self.ss_board._get_relay_url_code(self.r0.position, 236 RelayState.NO) 237 self.assertEqual(result, '00') 238 239 result = self.ss_board._get_relay_url_code(self.r0.position, 240 RelayState.NC) 241 self.assertEqual(result, '01') 242 243 result = self.ss_board._get_relay_url_code(self.r7.position, 244 RelayState.NO) 245 self.assertEqual(result, '14') 246 247 result = self.ss_board._get_relay_url_code(self.r7.position, 248 RelayState.NC) 249 self.assertEqual(result, '15') 250 251 def test_load_page_status(self): 252 self._set_status_page('0000111100001111') 253 result = self.ss_board._load_page(SainSmartBoard.HIDDEN_STATUS_PAGE) 254 self.assertTrue( 255 result.endswith( 256 '0000111100001111TUX">0000111100001111TUX</a><p>')) 257 258 def test_load_page_relay(self): 259 result = self.ss_board._load_page('00') 260 self.assertEqual(result, self.RELAY_OFF_PAGE_CONTENTS) 261 262 result = self.ss_board._load_page('01') 263 self.assertEqual(result, self.RELAY_ON_PAGE_CONTENTS) 264 265 def test_load_page_no_connection(self): 266 with self.assertRaises(RelayDeviceConnectionError): 267 self.ss_board._load_page('**') 268 269 def _set_status_page(self, status_16_chars): 270 with open(self.test_dir[7:] + '99', 'w+') as status_file: 271 status_file.write( 272 self.STATUS_MSG.format(self.test_dir[:-1], self.test_dir[:-1], 273 status_16_chars, status_16_chars)) 274 275 def _test_sync_status_dict(self, status_16_chars): 276 self._set_status_page(status_16_chars) 277 expected_dict = dict() 278 279 for index, char in enumerate(status_16_chars): 280 expected_dict[ 281 index] = RelayState.NC if char == '1' else RelayState.NO 282 283 self.ss_board._sync_status_dict() 284 self.assertDictEqual(expected_dict, self.ss_board.status_dict) 285 286 def test_sync_status_dict(self): 287 self._test_sync_status_dict('0000111100001111') 288 self._test_sync_status_dict('0000000000000000') 289 self._test_sync_status_dict('0101010101010101') 290 self._test_sync_status_dict('1010101010101010') 291 self._test_sync_status_dict('1111111111111111') 292 293 def test_get_relay_status_status_dict_none(self): 294 self._set_status_page('1111111111111111') 295 self.ss_board.status_dict = None 296 self.assertEqual( 297 self.ss_board.get_relay_status(self.r0.position), RelayState.NC) 298 299 def test_get_relay_status_status_dict_on(self): 300 self.r0.set(RelayState.NC) 301 self.assertEqual( 302 self.ss_board.get_relay_status(self.r0.position), RelayState.NC) 303 304 def test_get_relay_status_status_dict_off(self): 305 self.r0.set(RelayState.NO) 306 self.assertEqual( 307 self.ss_board.get_relay_status(self.r0.position), RelayState.NO) 308 309 def test_set_on(self): 310 patch_path = 'acts.controllers.relay_lib.sain_smart_board.urlopen' 311 with patch(patch_path) as urlopen: 312 board = SainSmartBoard(self.config) 313 board.status_dict = {} 314 board.set(self.r0.position, RelayState.NC) 315 urlopen.assert_called_once_with('%s%s' % (self.ss_board.base_url, '01')) 316 317 def test_set_off(self): 318 patch_path = 'acts.controllers.relay_lib.sain_smart_board.urlopen' 319 with patch(patch_path) as urlopen: 320 board = SainSmartBoard(self.config) 321 board.status_dict = {} 322 board.set(self.r0.position, RelayState.NO) 323 urlopen.assert_called_once_with('%s%s' % (self.ss_board.base_url, '00')) 324 325 def test_connection_error_no_tux(self): 326 default_status_msg = self.STATUS_MSG 327 self.STATUS_MSG = self.STATUS_MSG.replace('TUX', '') 328 try: 329 self._set_status_page('1111111111111111') 330 self.ss_board.get_relay_status(0) 331 except RelayDeviceConnectionError: 332 self.STATUS_MSG = default_status_msg 333 return 334 335 self.fail('Should have thrown an error without TUX appearing.') 336 337 338class ActsRelayRigTest(unittest.TestCase): 339 def setUp(self): 340 Relay.transition_wait_time = 0 341 Relay.button_press_time = 0 342 self.config = { 343 'boards': [{ 344 'type': 'SainSmartBoard', 345 'name': 'ss_control', 346 'base_url': 'http://192.168.1.4/30000/' 347 }, { 348 'type': 'SainSmartBoard', 349 'name': 'ss_control_2', 350 'base_url': 'http://192.168.1.4/30000/' 351 }], 352 'devices': [{ 353 'type': 'GenericRelayDevice', 354 'name': 'device', 355 'relays': { 356 'Relay00': 'ss_control/0', 357 'Relay10': 'ss_control/1' 358 } 359 }] 360 } 361 362 def tearDown(self): 363 Relay.transition_wait_time = .2 364 Relay.button_press_time = .25 365 366 def test_init_relay_rig_missing_boards(self): 367 flawed_config = copy.deepcopy(self.config) 368 del flawed_config['boards'] 369 with self.assertRaises(RelayConfigError): 370 RelayRig(flawed_config) 371 372 def test_init_relay_rig_is_not_list(self): 373 flawed_config = copy.deepcopy(self.config) 374 flawed_config['boards'] = self.config['boards'][0] 375 with self.assertRaises(RelayConfigError): 376 RelayRig(flawed_config) 377 378 def test_init_relay_rig_duplicate_board_names(self): 379 flawed_config = copy.deepcopy(self.config) 380 flawed_config['boards'][1]['name'] = (self.config['boards'][0]['name']) 381 with self.assertRaises(RelayConfigError): 382 RelayRigMock(flawed_config) 383 384 def test_init_relay_rig_device_gets_relays(self): 385 modded_config = copy.deepcopy(self.config) 386 del modded_config['devices'][0]['relays']['Relay00'] 387 rig = RelayRigMock(modded_config) 388 self.assertEqual(len(rig.relays), 4) 389 self.assertEqual(len(rig.devices['device'].relays), 1) 390 391 rig = RelayRigMock(self.config) 392 self.assertEqual(len(rig.devices['device'].relays), 2) 393 394 def test_init_relay_rig_correct_device_type(self): 395 rig = RelayRigMock(self.config) 396 self.assertEqual(len(rig.devices), 1) 397 self.assertIsInstance(rig.devices['device'], GenericRelayDevice) 398 399 def test_init_relay_rig_missing_devices_creates_generic_device(self): 400 modded_config = copy.deepcopy(self.config) 401 del modded_config['devices'] 402 rig = RelayRigMock(modded_config) 403 self.assertEqual(len(rig.devices), 1) 404 self.assertIsInstance(rig.devices['device'], GenericRelayDevice) 405 self.assertDictEqual(rig.devices['device'].relays, rig.relays) 406 407 408class RelayRigMock(RelayRig): 409 """A RelayRig that substitutes the MockBoard for any board.""" 410 411 _board_constructors = { 412 'SainSmartBoard': lambda x: MockBoard(x), 413 'FuguMockBoard': lambda x: FuguMockBoard(x) 414 } 415 416 def __init__(self, config=None): 417 if not config: 418 config = { 419 "boards": [{ 420 'name': 'MockBoard', 421 'type': 'SainSmartBoard' 422 }] 423 } 424 425 RelayRig.__init__(self, config) 426 427 428class ActsGenericRelayDeviceTest(unittest.TestCase): 429 def setUp(self): 430 Relay.transition_wait_time = 0 431 Relay.button_press_time = 0 432 self.board_config = {'name': 'MockBoard', 'type': 'SainSmartBoard'} 433 434 self.board = MockBoard(self.board_config) 435 self.r0 = self.board.relays[0] 436 self.r1 = self.board.relays[1] 437 438 self.device_config = { 439 'name': 'MockDevice', 440 'relays': { 441 'r0': 'MockBoard/0', 442 'r1': 'MockBoard/1' 443 } 444 } 445 config = { 446 'boards': [self.board_config], 447 'devices': [self.device_config] 448 } 449 self.rig = RelayRigMock(config) 450 self.rig.boards['MockBoard'] = self.board 451 self.rig.relays[self.r0.relay_id] = self.r0 452 self.rig.relays[self.r1.relay_id] = self.r1 453 454 def tearDown(self): 455 Relay.transition_wait_time = .2 456 Relay.button_press_time = .25 457 458 def test_setup_single_relay(self): 459 self.r0.set(RelayState.NC) 460 self.r1.set(RelayState.NC) 461 462 modified_config = copy.deepcopy(self.device_config) 463 del modified_config['relays']['r1'] 464 465 grd = GenericRelayDevice(modified_config, self.rig) 466 grd.setup() 467 468 self.assertEqual(self.r0.get_status(), RelayState.NO) 469 self.assertEqual(self.r1.get_status(), RelayState.NC) 470 471 def test_setup_multiple_relays(self): 472 self.board.set(self.r0.position, RelayState.NC) 473 self.board.set(self.r1.position, RelayState.NC) 474 475 grd = GenericRelayDevice(self.device_config, self.rig) 476 grd.setup() 477 478 self.assertEqual(self.r0.get_status(), RelayState.NO) 479 self.assertEqual(self.r1.get_status(), RelayState.NO) 480 481 def test_cleanup_single_relay(self): 482 self.test_setup_single_relay() 483 484 def test_cleanup_multiple_relays(self): 485 self.test_setup_multiple_relays() 486 487 def change_state(self, begin_state, call, end_state, previous_state=None): 488 self.board.set(self.r0.position, begin_state) 489 grd = GenericRelayDevice(self.device_config, self.rig) 490 call(grd) 491 self.assertEqual(self.r0.get_status(), end_state) 492 if previous_state: 493 self.assertEqual( 494 self.board.relay_previous_states[self.r0.position], 495 previous_state) 496 497 def test_press_while_no(self): 498 self.change_state(RelayState.NO, lambda x: x.press('r0'), 499 RelayState.NO, RelayState.NC) 500 501 def test_press_while_nc(self): 502 self.change_state(RelayState.NC, lambda x: x.press('r0'), 503 RelayState.NO, RelayState.NC) 504 505 def test_hold_down_while_no(self): 506 self.change_state(RelayState.NO, lambda x: x.hold_down('r0'), 507 RelayState.NC) 508 509 def test_hold_down_while_nc(self): 510 self.change_state(RelayState.NC, lambda x: x.hold_down('r0'), 511 RelayState.NC) 512 513 def test_release_while_nc(self): 514 self.change_state(RelayState.NC, lambda x: x.release('r0'), 515 RelayState.NO) 516 517 518class ActsRelayDeviceTest(unittest.TestCase): 519 def setUp(self): 520 Relay.transition_wait_time = 0 521 Relay.button_press_time = 0 522 523 self.board_config = { 524 'name': 'MockBoard', 525 'relays': [{ 526 'id': 0, 527 'relay_pos': 0 528 }, { 529 'id': 1, 530 'relay_pos': 1 531 }] 532 } 533 534 self.board = MockBoard(self.board_config) 535 self.r0 = Relay(self.board, 0) 536 self.r1 = Relay(self.board, 1) 537 self.board.set(self.r0.position, RelayState.NO) 538 self.board.set(self.r1.position, RelayState.NO) 539 540 self.rig = RelayRigMock() 541 self.rig.boards['MockBoard'] = self.board 542 self.rig.relays[self.r0.relay_id] = self.r0 543 self.rig.relays[self.r1.relay_id] = self.r1 544 545 self.device_config = { 546 "type": "GenericRelayDevice", 547 "name": "device", 548 "relays": { 549 'r0': 'MockBoard/0', 550 'r1': 'MockBoard/1' 551 } 552 } 553 554 def tearDown(self): 555 Relay.transition_wait_time = .2 556 Relay.button_press_time = .25 557 558 def test_init_raise_on_name_missing(self): 559 flawed_config = copy.deepcopy(self.device_config) 560 del flawed_config['name'] 561 with self.assertRaises(RelayConfigError): 562 RelayDevice(flawed_config, self.rig) 563 564 def test_init_raise_on_name_wrong_type(self): 565 flawed_config = copy.deepcopy(self.device_config) 566 flawed_config['name'] = {} 567 with self.assertRaises(RelayConfigError): 568 RelayDevice(flawed_config, self.rig) 569 570 def test_init_raise_on_relays_missing(self): 571 flawed_config = copy.deepcopy(self.device_config) 572 del flawed_config['relays'] 573 with self.assertRaises(RelayConfigError): 574 RelayDevice(flawed_config, self.rig) 575 576 def test_init_raise_on_relays_wrong_type(self): 577 flawed_config = copy.deepcopy(self.device_config) 578 flawed_config['relays'] = str 579 with self.assertRaises(RelayConfigError): 580 RelayDevice(flawed_config, self.rig) 581 582 def test_init_raise_on_relays_is_empty(self): 583 flawed_config = copy.deepcopy(self.device_config) 584 flawed_config['relays'] = [] 585 with self.assertRaises(RelayConfigError): 586 RelayDevice(flawed_config, self.rig) 587 588 def test_init_raise_on_relays_are_dicts_without_names(self): 589 flawed_config = copy.deepcopy(self.device_config) 590 flawed_config['relays'] = [{'id': 0}, {'id': 1}] 591 with self.assertRaises(RelayConfigError): 592 RelayDevice(flawed_config, self.rig) 593 594 def test_init_raise_on_relays_are_dicts_without_ids(self): 595 flawed_config = copy.deepcopy(self.device_config) 596 flawed_config['relays'] = [{'name': 'r0'}, {'name': 'r1'}] 597 with self.assertRaises(RelayConfigError): 598 RelayDevice(flawed_config, self.rig) 599 600 def test_init_pass_relays_have_ids_and_names(self): 601 RelayDevice(self.device_config, self.rig) 602 603 604class TestRelayRigParser(unittest.TestCase): 605 def setUp(self): 606 Relay.transition_wait_time = 0 607 Relay.button_press_time = 0 608 self.board_config = { 609 'name': 'MockBoard', 610 'relays': [{ 611 'id': 'r0', 612 'relay_pos': 0 613 }, { 614 'id': 'r1', 615 'relay_pos': 1 616 }] 617 } 618 self.r0 = self.board_config['relays'][0] 619 self.r1 = self.board_config['relays'][1] 620 self.board = MockBoard(self.board_config) 621 622 def tearDown(self): 623 Relay.transition_wait_time = .2 624 Relay.button_press_time = .25 625 626 def test_create_relay_board_raise_on_missing_type(self): 627 with self.assertRaises(RelayConfigError): 628 RelayRigMock().create_relay_board(self.board_config) 629 630 def test_create_relay_board_valid_config(self): 631 config = copy.deepcopy(self.board_config) 632 config['type'] = 'SainSmartBoard' 633 RelayRigMock().create_relay_board(config) 634 635 def test_create_relay_board_raise_on_type_not_found(self): 636 flawed_config = copy.deepcopy(self.board_config) 637 flawed_config['type'] = 'NonExistentBoard' 638 with self.assertRaises(RelayConfigError): 639 RelayRigMock().create_relay_board(flawed_config) 640 641 def test_create_relay_device_create_generic_on_missing_type(self): 642 rig = RelayRigMock() 643 rig.relays['r0'] = self.r0 644 rig.relays['r1'] = self.r1 645 config = { 646 'name': 'name', 647 'relays': { 648 'r0': 'MockBoard/0', 649 'r1': 'MockBoard/1' 650 } 651 } 652 device = rig.create_relay_device(config) 653 self.assertIsInstance(device, GenericRelayDevice) 654 655 def test_create_relay_device_config_with_type(self): 656 rig = RelayRigMock() 657 rig.relays['r0'] = self.r0 658 rig.relays['r1'] = self.r1 659 config = { 660 'type': 'GenericRelayDevice', 661 'name': '.', 662 'relays': { 663 'r0': 'MockBoard/0', 664 'r1': 'MockBoard/1' 665 } 666 } 667 device = rig.create_relay_device(config) 668 self.assertIsInstance(device, GenericRelayDevice) 669 670 def test_create_relay_device_raise_on_type_not_found(self): 671 rig = RelayRigMock() 672 rig.relays['r0'] = self.r0 673 rig.relays['r1'] = self.r1 674 config = { 675 'type': 'SomeInvalidType', 676 'name': '.', 677 'relays': [{ 678 'name': 'r0', 679 'pos': 'MockBoard/0' 680 }, { 681 'name': 'r1', 682 'pos': 'MockBoard/1' 683 }] 684 } 685 with self.assertRaises(RelayConfigError): 686 rig.create_relay_device(config) 687 688 689class TestSynchronizeRelays(unittest.TestCase): 690 def test_synchronize_relays(self): 691 Relay.transition_wait_time = .1 692 with SynchronizeRelays(): 693 self.assertEqual(Relay.transition_wait_time, 0) 694 self.assertEqual(Relay.transition_wait_time, .1) 695 696 697class FuguMockBoard(MockBoard): 698 def get_relay_position_list(self): 699 return range(4) 700 701 702class TestFuguRemote(unittest.TestCase): 703 def setUp(self): 704 Relay.transition_wait_time = 0 705 self.mock_rig = RelayRigMock({ 706 "boards": [{ 707 'name': 'MockBoard', 708 'type': 'FuguMockBoard' 709 }] 710 }) 711 self.mock_board = self.mock_rig.boards['MockBoard'] 712 self.fugu_config = { 713 'type': 'FuguRemote', 714 'name': 'UniqueDeviceName', 715 'mac_address': '00:00:00:00:00:00', 716 'relays': { 717 'Power': 'MockBoard/0', 718 fugu_remote.Buttons.BACK.value: 'MockBoard/1', 719 fugu_remote.Buttons.HOME.value: 'MockBoard/2', 720 fugu_remote.Buttons.PLAY_PAUSE.value: 'MockBoard/3' 721 } 722 } 723 Relay.button_press_time = 0 724 725 def tearDown(self): 726 Relay.button_press_time = .25 727 Relay.transition_wait_time = .2 728 729 def test_config_missing_button(self): 730 """FuguRemote __init__ should throw an error if a relay is missing.""" 731 flawed_config = copy.deepcopy(self.fugu_config) 732 del flawed_config['relays']['Power'] 733 del flawed_config['relays'][fugu_remote.Buttons.BACK.value] 734 with self.assertRaises(RelayConfigError): 735 fugu_remote.FuguRemote(flawed_config, self.mock_rig) 736 737 def test_config_missing_mac_address(self): 738 """FuguRemote __init__ should throw an error without a mac address.""" 739 flawed_config = copy.deepcopy(self.fugu_config) 740 del flawed_config['mac_address'] 741 with self.assertRaises(RelayConfigError): 742 fugu_remote.FuguRemote(flawed_config, self.mock_rig) 743 744 def test_config_no_issues(self): 745 """FuguRemote __init__ should not throw errors for a correct config.""" 746 fugu_remote.FuguRemote(self.fugu_config, self.mock_rig) 747 748 def test_power_nc_after_setup(self): 749 """Power should be NORMALLY_CLOSED after calling setup if it exists.""" 750 fugu = fugu_remote.FuguRemote(self.fugu_config, self.mock_rig) 751 fugu.setup() 752 self.assertEqual(self.mock_board.get_relay_status(0), RelayState.NC) 753 pass 754 755 def press_button_success(self, relay_position): 756 self.assertEqual(self.mock_board.relay_states[relay_position], 757 RelayState.NO) 758 self.assertEqual(self.mock_board.relay_previous_states[relay_position], 759 RelayState.NC) 760 761 def test_press_play_pause(self): 762 fugu = fugu_remote.FuguRemote(self.fugu_config, self.mock_rig) 763 fugu.press_play_pause() 764 self.press_button_success(3) 765 766 def test_press_back(self): 767 fugu = fugu_remote.FuguRemote(self.fugu_config, self.mock_rig) 768 fugu.press_back() 769 self.press_button_success(1) 770 771 def test_press_home(self): 772 fugu = fugu_remote.FuguRemote(self.fugu_config, self.mock_rig) 773 fugu.press_home() 774 self.press_button_success(2) 775 776 def test_enter_pairing_mode(self): 777 fugu = fugu_remote.FuguRemote(self.fugu_config, self.mock_rig) 778 fugu_remote.PAIRING_MODE_WAIT_TIME = 0 779 fugu.enter_pairing_mode() 780 self.press_button_success(2) 781 self.press_button_success(1) 782 783 784class TestRelayDict(unittest.TestCase): 785 def test_init(self): 786 mock_device = object() 787 blank_dict = dict() 788 relay_dict = RelayDict(mock_device, blank_dict) 789 self.assertEqual(relay_dict._store, blank_dict) 790 self.assertEqual(relay_dict.relay_device, mock_device) 791 792 def test_get_item_valid_key(self): 793 mock_device = object() 794 blank_dict = {'key': 'value'} 795 relay_dict = RelayDict(mock_device, blank_dict) 796 self.assertEqual(relay_dict['key'], 'value') 797 798 def test_get_item_invalid_key(self): 799 # Create an object with a single attribute 'name' 800 mock_device = type('', (object, ), {'name': 'name'})() 801 blank_dict = {'key': 'value'} 802 relay_dict = RelayDict(mock_device, blank_dict) 803 with self.assertRaises(RelayConfigError): 804 value = relay_dict['not_key'] 805 806 def test_iter(self): 807 mock_device = type('', (object, ), {'name': 'name'})() 808 data_dict = {'a': '1', 'b': '2', 'c': '3'} 809 relay_dict = RelayDict(mock_device, data_dict) 810 811 rd_set = set() 812 for key in relay_dict: 813 rd_set.add(key) 814 dd_set = set() 815 for key in data_dict: 816 dd_set.add(key) 817 818 self.assertSetEqual(rd_set, dd_set) 819 820 821if __name__ == "__main__": 822 unittest.main() 823