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(self.board.get_relay_status(self.relay.position), 82 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(self.board.get_relay_status(self.relay.position), 88 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(self.board.get_relay_status(self.relay.position), 94 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(self.board.get_relay_status(self.relay.position), 100 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(self.board.get_relay_status(self.relay.position), 106 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(self.board.get_relay_status(self.relay.position), 112 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(self.board.get_relay_status(self.relay.position), 118 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(self.board.get_relay_status(self.relay.position), 124 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(self.board.get_relay_status(self.relay.position), 138 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(self.board.get_relay_status(self.relay.position), 150 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(self.board.get_relay_status(new_relay.position), 169 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(self.board.get_relay_status(new_relay.position), 178 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': 212 'SSBoard', 213 'base_url': 214 self.test_dir, 215 'relays': [{ 216 'name': '0', 217 'relay_pos': 0 218 }, { 219 'name': '1', 220 'relay_pos': 1 221 }, { 222 'name': '2', 223 'relay_pos': 7 224 }] 225 }) 226 self.ss_board = SainSmartBoard(self.config) 227 self.r0 = Relay(self.ss_board, 0) 228 self.r1 = Relay(self.ss_board, 1) 229 self.r7 = Relay(self.ss_board, 7) 230 231 def tearDown(self): 232 shutil.rmtree(self.test_dir[7:]) 233 Relay.transition_wait_time = .2 234 Relay.button_press_time = .25 235 236 def test_get_url_code(self): 237 result = self.ss_board._get_relay_url_code(self.r0.position, 238 RelayState.NO) 239 self.assertEqual(result, '00') 240 241 result = self.ss_board._get_relay_url_code(self.r0.position, 242 RelayState.NC) 243 self.assertEqual(result, '01') 244 245 result = self.ss_board._get_relay_url_code(self.r7.position, 246 RelayState.NO) 247 self.assertEqual(result, '14') 248 249 result = self.ss_board._get_relay_url_code(self.r7.position, 250 RelayState.NC) 251 self.assertEqual(result, '15') 252 253 def test_load_page_status(self): 254 self._set_status_page('0000111100001111') 255 result = self.ss_board._load_page(SainSmartBoard.HIDDEN_STATUS_PAGE) 256 self.assertTrue( 257 result.endswith( 258 '0000111100001111TUX">0000111100001111TUX</a><p>')) 259 260 def test_load_page_relay(self): 261 result = self.ss_board._load_page('00') 262 self.assertEqual(result, self.RELAY_OFF_PAGE_CONTENTS) 263 264 result = self.ss_board._load_page('01') 265 self.assertEqual(result, self.RELAY_ON_PAGE_CONTENTS) 266 267 def test_load_page_no_connection(self): 268 with self.assertRaises(RelayDeviceConnectionError): 269 self.ss_board._load_page('**') 270 271 def _set_status_page(self, status_16_chars): 272 with open(self.test_dir[7:] + '99', 'w+') as status_file: 273 status_file.write( 274 self.STATUS_MSG.format(self.test_dir[:-1], self.test_dir[:-1], 275 status_16_chars, status_16_chars)) 276 277 def _test_sync_status_dict(self, status_16_chars): 278 self._set_status_page(status_16_chars) 279 expected_dict = dict() 280 281 for index, char in enumerate(status_16_chars): 282 expected_dict[ 283 index] = RelayState.NC if char == '1' else RelayState.NO 284 285 self.ss_board._sync_status_dict() 286 self.assertDictEqual(expected_dict, self.ss_board.status_dict) 287 288 def test_sync_status_dict(self): 289 self._test_sync_status_dict('0000111100001111') 290 self._test_sync_status_dict('0000000000000000') 291 self._test_sync_status_dict('0101010101010101') 292 self._test_sync_status_dict('1010101010101010') 293 self._test_sync_status_dict('1111111111111111') 294 295 def test_get_relay_status_status_dict_none(self): 296 self._set_status_page('1111111111111111') 297 self.ss_board.status_dict = None 298 self.assertEqual(self.ss_board.get_relay_status(self.r0.position), 299 RelayState.NC) 300 301 def test_get_relay_status_status_dict_on(self): 302 self.r0.set(RelayState.NC) 303 self.assertEqual(self.ss_board.get_relay_status(self.r0.position), 304 RelayState.NC) 305 306 def test_get_relay_status_status_dict_off(self): 307 self.r0.set(RelayState.NO) 308 self.assertEqual(self.ss_board.get_relay_status(self.r0.position), 309 RelayState.NO) 310 311 def test_set_on(self): 312 patch_path = 'acts.controllers.relay_lib.sain_smart_board.urlopen' 313 with patch(patch_path) as urlopen: 314 board = SainSmartBoard(self.config) 315 board.status_dict = {} 316 board.set(self.r0.position, RelayState.NC) 317 urlopen.assert_called_once_with('%s%s' % 318 (self.ss_board.base_url, '01')) 319 320 def test_set_off(self): 321 patch_path = 'acts.controllers.relay_lib.sain_smart_board.urlopen' 322 with patch(patch_path) as urlopen: 323 board = SainSmartBoard(self.config) 324 board.status_dict = {} 325 board.set(self.r0.position, RelayState.NO) 326 urlopen.assert_called_once_with('%s%s' % 327 (self.ss_board.base_url, '00')) 328 329 def test_connection_error_no_tux(self): 330 default_status_msg = self.STATUS_MSG 331 self.STATUS_MSG = self.STATUS_MSG.replace('TUX', '') 332 try: 333 self._set_status_page('1111111111111111') 334 self.ss_board.get_relay_status(0) 335 except RelayDeviceConnectionError: 336 self.STATUS_MSG = default_status_msg 337 return 338 339 self.fail('Should have thrown an error without TUX appearing.') 340 341 342class ActsRelayRigTest(unittest.TestCase): 343 def setUp(self): 344 Relay.transition_wait_time = 0 345 Relay.button_press_time = 0 346 self.config = { 347 'boards': [{ 348 'type': 'SainSmartBoard', 349 'name': 'ss_control', 350 'base_url': 'http://192.168.1.4/30000/' 351 }, { 352 'type': 'SainSmartBoard', 353 'name': 'ss_control_2', 354 'base_url': 'http://192.168.1.4/30000/' 355 }], 356 'devices': [{ 357 'type': 'GenericRelayDevice', 358 'name': 'device', 359 'relays': { 360 'Relay00': 'ss_control/0', 361 'Relay10': 'ss_control/1' 362 } 363 }] 364 } 365 366 def tearDown(self): 367 Relay.transition_wait_time = .2 368 Relay.button_press_time = .25 369 370 def test_init_relay_rig_missing_boards(self): 371 flawed_config = copy.deepcopy(self.config) 372 del flawed_config['boards'] 373 with self.assertRaises(RelayConfigError): 374 RelayRig(flawed_config) 375 376 def test_init_relay_rig_is_not_list(self): 377 flawed_config = copy.deepcopy(self.config) 378 flawed_config['boards'] = self.config['boards'][0] 379 with self.assertRaises(RelayConfigError): 380 RelayRig(flawed_config) 381 382 def test_init_relay_rig_duplicate_board_names(self): 383 flawed_config = copy.deepcopy(self.config) 384 flawed_config['boards'][1]['name'] = (self.config['boards'][0]['name']) 385 with self.assertRaises(RelayConfigError): 386 RelayRigMock(flawed_config) 387 388 def test_init_relay_rig_device_gets_relays(self): 389 modded_config = copy.deepcopy(self.config) 390 del modded_config['devices'][0]['relays']['Relay00'] 391 rig = RelayRigMock(modded_config) 392 self.assertEqual(len(rig.relays), 4) 393 self.assertEqual(len(rig.devices['device'].relays), 1) 394 395 rig = RelayRigMock(self.config) 396 self.assertEqual(len(rig.devices['device'].relays), 2) 397 398 def test_init_relay_rig_correct_device_type(self): 399 rig = RelayRigMock(self.config) 400 self.assertEqual(len(rig.devices), 1) 401 self.assertIsInstance(rig.devices['device'], GenericRelayDevice) 402 403 def test_init_relay_rig_missing_devices_creates_generic_device(self): 404 modded_config = copy.deepcopy(self.config) 405 del modded_config['devices'] 406 rig = RelayRigMock(modded_config) 407 self.assertEqual(len(rig.devices), 1) 408 self.assertIsInstance(rig.devices['device'], GenericRelayDevice) 409 self.assertDictEqual(rig.devices['device'].relays, rig.relays) 410 411 412class RelayRigMock(RelayRig): 413 """A RelayRig that substitutes the MockBoard for any board.""" 414 415 _board_constructors = { 416 'SainSmartBoard': lambda x: MockBoard(x), 417 'FuguMockBoard': lambda x: FuguMockBoard(x) 418 } 419 420 def __init__(self, config=None): 421 if not config: 422 config = { 423 "boards": [{ 424 'name': 'MockBoard', 425 'type': 'SainSmartBoard' 426 }] 427 } 428 429 RelayRig.__init__(self, config) 430 431 432class ActsGenericRelayDeviceTest(unittest.TestCase): 433 def setUp(self): 434 Relay.transition_wait_time = 0 435 Relay.button_press_time = 0 436 self.board_config = {'name': 'MockBoard', 'type': 'SainSmartBoard'} 437 438 self.board = MockBoard(self.board_config) 439 self.r0 = self.board.relays[0] 440 self.r1 = self.board.relays[1] 441 442 self.device_config = { 443 'name': 'MockDevice', 444 'relays': { 445 'r0': 'MockBoard/0', 446 'r1': 'MockBoard/1' 447 } 448 } 449 config = { 450 'boards': [self.board_config], 451 'devices': [self.device_config] 452 } 453 self.rig = RelayRigMock(config) 454 self.rig.boards['MockBoard'] = self.board 455 self.rig.relays[self.r0.relay_id] = self.r0 456 self.rig.relays[self.r1.relay_id] = self.r1 457 458 def tearDown(self): 459 Relay.transition_wait_time = .2 460 Relay.button_press_time = .25 461 462 def test_setup_single_relay(self): 463 self.r0.set(RelayState.NC) 464 self.r1.set(RelayState.NC) 465 466 modified_config = copy.deepcopy(self.device_config) 467 del modified_config['relays']['r1'] 468 469 grd = GenericRelayDevice(modified_config, self.rig) 470 grd.setup() 471 472 self.assertEqual(self.r0.get_status(), RelayState.NO) 473 self.assertEqual(self.r1.get_status(), RelayState.NC) 474 475 def test_setup_multiple_relays(self): 476 self.board.set(self.r0.position, RelayState.NC) 477 self.board.set(self.r1.position, RelayState.NC) 478 479 grd = GenericRelayDevice(self.device_config, self.rig) 480 grd.setup() 481 482 self.assertEqual(self.r0.get_status(), RelayState.NO) 483 self.assertEqual(self.r1.get_status(), RelayState.NO) 484 485 def test_cleanup_single_relay(self): 486 self.test_setup_single_relay() 487 488 def test_cleanup_multiple_relays(self): 489 self.test_setup_multiple_relays() 490 491 def change_state(self, begin_state, call, end_state, previous_state=None): 492 self.board.set(self.r0.position, begin_state) 493 grd = GenericRelayDevice(self.device_config, self.rig) 494 call(grd) 495 self.assertEqual(self.r0.get_status(), end_state) 496 if previous_state: 497 self.assertEqual( 498 self.board.relay_previous_states[self.r0.position], 499 previous_state) 500 501 def test_press_while_no(self): 502 self.change_state(RelayState.NO, lambda x: x.press('r0'), 503 RelayState.NO, RelayState.NC) 504 505 def test_press_while_nc(self): 506 self.change_state(RelayState.NC, lambda x: x.press('r0'), 507 RelayState.NO, RelayState.NC) 508 509 def test_hold_down_while_no(self): 510 self.change_state(RelayState.NO, lambda x: x.hold_down('r0'), 511 RelayState.NC) 512 513 def test_hold_down_while_nc(self): 514 self.change_state(RelayState.NC, lambda x: x.hold_down('r0'), 515 RelayState.NC) 516 517 def test_release_while_nc(self): 518 self.change_state(RelayState.NC, lambda x: x.release('r0'), 519 RelayState.NO) 520 521 522class ActsRelayDeviceTest(unittest.TestCase): 523 def setUp(self): 524 Relay.transition_wait_time = 0 525 Relay.button_press_time = 0 526 527 self.board_config = { 528 'name': 'MockBoard', 529 'relays': [{ 530 'id': 0, 531 'relay_pos': 0 532 }, { 533 'id': 1, 534 'relay_pos': 1 535 }] 536 } 537 538 self.board = MockBoard(self.board_config) 539 self.r0 = Relay(self.board, 0) 540 self.r1 = Relay(self.board, 1) 541 self.board.set(self.r0.position, RelayState.NO) 542 self.board.set(self.r1.position, RelayState.NO) 543 544 self.rig = RelayRigMock() 545 self.rig.boards['MockBoard'] = self.board 546 self.rig.relays[self.r0.relay_id] = self.r0 547 self.rig.relays[self.r1.relay_id] = self.r1 548 549 self.device_config = { 550 "type": "GenericRelayDevice", 551 "name": "device", 552 "relays": { 553 'r0': 'MockBoard/0', 554 'r1': 'MockBoard/1' 555 } 556 } 557 558 def tearDown(self): 559 Relay.transition_wait_time = .2 560 Relay.button_press_time = .25 561 562 def test_init_raise_on_name_missing(self): 563 flawed_config = copy.deepcopy(self.device_config) 564 del flawed_config['name'] 565 with self.assertRaises(RelayConfigError): 566 RelayDevice(flawed_config, self.rig) 567 568 def test_init_raise_on_name_wrong_type(self): 569 flawed_config = copy.deepcopy(self.device_config) 570 flawed_config['name'] = {} 571 with self.assertRaises(RelayConfigError): 572 RelayDevice(flawed_config, self.rig) 573 574 def test_init_raise_on_relays_missing(self): 575 flawed_config = copy.deepcopy(self.device_config) 576 del flawed_config['relays'] 577 with self.assertRaises(RelayConfigError): 578 RelayDevice(flawed_config, self.rig) 579 580 def test_init_raise_on_relays_wrong_type(self): 581 flawed_config = copy.deepcopy(self.device_config) 582 flawed_config['relays'] = str 583 with self.assertRaises(RelayConfigError): 584 RelayDevice(flawed_config, self.rig) 585 586 def test_init_raise_on_relays_is_empty(self): 587 flawed_config = copy.deepcopy(self.device_config) 588 flawed_config['relays'] = [] 589 with self.assertRaises(RelayConfigError): 590 RelayDevice(flawed_config, self.rig) 591 592 def test_init_raise_on_relays_are_dicts_without_names(self): 593 flawed_config = copy.deepcopy(self.device_config) 594 flawed_config['relays'] = [{'id': 0}, {'id': 1}] 595 with self.assertRaises(RelayConfigError): 596 RelayDevice(flawed_config, self.rig) 597 598 def test_init_raise_on_relays_are_dicts_without_ids(self): 599 flawed_config = copy.deepcopy(self.device_config) 600 flawed_config['relays'] = [{'name': 'r0'}, {'name': 'r1'}] 601 with self.assertRaises(RelayConfigError): 602 RelayDevice(flawed_config, self.rig) 603 604 def test_init_pass_relays_have_ids_and_names(self): 605 RelayDevice(self.device_config, self.rig) 606 607 608class TestRelayRigParser(unittest.TestCase): 609 def setUp(self): 610 Relay.transition_wait_time = 0 611 Relay.button_press_time = 0 612 self.board_config = { 613 'name': 'MockBoard', 614 'relays': [{ 615 'id': 'r0', 616 'relay_pos': 0 617 }, { 618 'id': 'r1', 619 'relay_pos': 1 620 }] 621 } 622 self.r0 = self.board_config['relays'][0] 623 self.r1 = self.board_config['relays'][1] 624 self.board = MockBoard(self.board_config) 625 626 def tearDown(self): 627 Relay.transition_wait_time = .2 628 Relay.button_press_time = .25 629 630 def test_create_relay_board_raise_on_missing_type(self): 631 with self.assertRaises(RelayConfigError): 632 RelayRigMock().create_relay_board(self.board_config) 633 634 def test_create_relay_board_valid_config(self): 635 config = copy.deepcopy(self.board_config) 636 config['type'] = 'SainSmartBoard' 637 RelayRigMock().create_relay_board(config) 638 639 def test_create_relay_board_raise_on_type_not_found(self): 640 flawed_config = copy.deepcopy(self.board_config) 641 flawed_config['type'] = 'NonExistentBoard' 642 with self.assertRaises(RelayConfigError): 643 RelayRigMock().create_relay_board(flawed_config) 644 645 def test_create_relay_device_create_generic_on_missing_type(self): 646 rig = RelayRigMock() 647 rig.relays['r0'] = self.r0 648 rig.relays['r1'] = self.r1 649 config = { 650 'name': 'name', 651 'relays': { 652 'r0': 'MockBoard/0', 653 'r1': 'MockBoard/1' 654 } 655 } 656 device = rig.create_relay_device(config) 657 self.assertIsInstance(device, GenericRelayDevice) 658 659 def test_create_relay_device_config_with_type(self): 660 rig = RelayRigMock() 661 rig.relays['r0'] = self.r0 662 rig.relays['r1'] = self.r1 663 config = { 664 'type': 'GenericRelayDevice', 665 'name': '.', 666 'relays': { 667 'r0': 'MockBoard/0', 668 'r1': 'MockBoard/1' 669 } 670 } 671 device = rig.create_relay_device(config) 672 self.assertIsInstance(device, GenericRelayDevice) 673 674 def test_create_relay_device_raise_on_type_not_found(self): 675 rig = RelayRigMock() 676 rig.relays['r0'] = self.r0 677 rig.relays['r1'] = self.r1 678 config = { 679 'type': 680 'SomeInvalidType', 681 'name': 682 '.', 683 'relays': [{ 684 'name': 'r0', 685 'pos': 'MockBoard/0' 686 }, { 687 'name': 'r1', 688 'pos': 'MockBoard/1' 689 }] 690 } 691 with self.assertRaises(RelayConfigError): 692 rig.create_relay_device(config) 693 694 695class TestSynchronizeRelays(unittest.TestCase): 696 def test_synchronize_relays(self): 697 Relay.transition_wait_time = .1 698 with SynchronizeRelays(): 699 self.assertEqual(Relay.transition_wait_time, 0) 700 self.assertEqual(Relay.transition_wait_time, .1) 701 702 703class FuguMockBoard(MockBoard): 704 def get_relay_position_list(self): 705 return range(4) 706 707 708class TestFuguRemote(unittest.TestCase): 709 def setUp(self): 710 Relay.transition_wait_time = 0 711 self.mock_rig = RelayRigMock( 712 {"boards": [{ 713 'name': 'MockBoard', 714 'type': 'FuguMockBoard' 715 }]}) 716 self.mock_board = self.mock_rig.boards['MockBoard'] 717 self.fugu_config = { 718 'type': 'FuguRemote', 719 'name': 'UniqueDeviceName', 720 'mac_address': '00:00:00:00:00:00', 721 'relays': { 722 'Power': 'MockBoard/0', 723 fugu_remote.Buttons.BACK.value: 'MockBoard/1', 724 fugu_remote.Buttons.HOME.value: 'MockBoard/2', 725 fugu_remote.Buttons.PLAY_PAUSE.value: 'MockBoard/3' 726 } 727 } 728 Relay.button_press_time = 0 729 730 def tearDown(self): 731 Relay.button_press_time = .25 732 Relay.transition_wait_time = .2 733 734 def test_config_missing_button(self): 735 """FuguRemote __init__ should throw an error if a relay is missing.""" 736 flawed_config = copy.deepcopy(self.fugu_config) 737 del flawed_config['relays']['Power'] 738 del flawed_config['relays'][fugu_remote.Buttons.BACK.value] 739 with self.assertRaises(RelayConfigError): 740 fugu_remote.FuguRemote(flawed_config, self.mock_rig) 741 742 def test_config_missing_mac_address(self): 743 """FuguRemote __init__ should throw an error without a mac address.""" 744 flawed_config = copy.deepcopy(self.fugu_config) 745 del flawed_config['mac_address'] 746 with self.assertRaises(RelayConfigError): 747 fugu_remote.FuguRemote(flawed_config, self.mock_rig) 748 749 def test_config_no_issues(self): 750 """FuguRemote __init__ should not throw errors for a correct config.""" 751 fugu_remote.FuguRemote(self.fugu_config, self.mock_rig) 752 753 def test_power_nc_after_setup(self): 754 """Power should be NORMALLY_CLOSED after calling setup if it exists.""" 755 fugu = fugu_remote.FuguRemote(self.fugu_config, self.mock_rig) 756 fugu.setup() 757 self.assertEqual(self.mock_board.get_relay_status(0), RelayState.NC) 758 759 def press_button_success(self, relay_position): 760 self.assertEqual(self.mock_board.relay_states[relay_position], 761 RelayState.NO) 762 self.assertEqual(self.mock_board.relay_previous_states[relay_position], 763 RelayState.NC) 764 765 def test_press_play_pause(self): 766 fugu = fugu_remote.FuguRemote(self.fugu_config, self.mock_rig) 767 fugu.press_play_pause() 768 self.press_button_success(3) 769 770 def test_press_back(self): 771 fugu = fugu_remote.FuguRemote(self.fugu_config, self.mock_rig) 772 fugu.press_back() 773 self.press_button_success(1) 774 775 def test_press_home(self): 776 fugu = fugu_remote.FuguRemote(self.fugu_config, self.mock_rig) 777 fugu.press_home() 778 self.press_button_success(2) 779 780 def test_enter_pairing_mode(self): 781 fugu = fugu_remote.FuguRemote(self.fugu_config, self.mock_rig) 782 fugu_remote.PAIRING_MODE_WAIT_TIME = 0 783 fugu.enter_pairing_mode() 784 self.press_button_success(2) 785 self.press_button_success(1) 786 787 788class TestRelayDict(unittest.TestCase): 789 def test_init(self): 790 mock_device = object() 791 blank_dict = dict() 792 relay_dict = RelayDict(mock_device, blank_dict) 793 self.assertEqual(relay_dict._store, blank_dict) 794 self.assertEqual(relay_dict.relay_device, mock_device) 795 796 def test_get_item_valid_key(self): 797 mock_device = object() 798 blank_dict = {'key': 'value'} 799 relay_dict = RelayDict(mock_device, blank_dict) 800 self.assertEqual(relay_dict['key'], 'value') 801 802 def test_get_item_invalid_key(self): 803 # Create an object with a single attribute 'name' 804 mock_device = type('', (object, ), {'name': 'name'})() 805 blank_dict = {'key': 'value'} 806 relay_dict = RelayDict(mock_device, blank_dict) 807 with self.assertRaises(RelayConfigError): 808 value = relay_dict['not_key'] 809 810 def test_iter(self): 811 mock_device = type('', (object, ), {'name': 'name'})() 812 data_dict = {'a': '1', 'b': '2', 'c': '3'} 813 relay_dict = RelayDict(mock_device, data_dict) 814 815 rd_set = set() 816 for key in relay_dict: 817 rd_set.add(key) 818 dd_set = set() 819 for key in data_dict: 820 dd_set.add(key) 821 822 self.assertSetEqual(rd_set, dd_set) 823 824 825if __name__ == "__main__": 826 unittest.main() 827