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