How to use check_entry method in lisa

Best Python code snippet using lisa_python

security.py

Source:security.py

1from nil_lib import format_switch_list, switch_list_send_command, file_loader, file_create2def rogue_bpdu_switches(switch_list, user, pwd=None) -> None:3    '''4    '''5    switch_list = format_switch_list(6        switch_list, user, pwd=pwd)7    switch_list = switch_list_send_command(8        switch_list, ['sh spanning-tree detail', 'sh run'])9    fsm_bpdu = file_loader(10        'nil_lib/templates/spanning_tree_int_detail.fsm')11    fsm_interfaces = file_loader(12        'nil_lib/templates/interfaces.fsm')13    audit_list, debug_list = [], []14    for switch in switch_list:15        if not switch['name']:16            continue    # Could not connect to switch17        fsm_bpdu._result, fsm_interfaces._result = [], []18        switch['output'][0] = fsm_bpdu.ParseTextToDicts(19            switch['output'][0])20        switch['output'][1] = fsm_interfaces.ParseTextToDicts(21            switch['output'][1])22        interface_check = []23        for spanning_entry in switch['output'][0]:24            if ('Port-channel' not in spanning_entry['INTERFACE']25                and spanning_entry['INTERFACE'] not in interface_check):26                if spanning_entry['BPDU_RECIEVE'] != '0':27                    for int_entry in switch['output'][1]:28                        # Find interface configs29                        if spanning_entry['INTERFACE'] == int_entry['INTERFACE_NAME']:30                            found = False31                            for int_details in int_entry['INTERFACE_DETAILS']:32                                if 'portfast' in int_details:33                                    interface_check.append(spanning_entry['INTERFACE'])34                                    found = True35                                    break36                            if not found:37                                # Entry does not have portfast38                                int_entry['SWITCH_NAME'] = switch['name']39                                if int_entry not in debug_list:40                                    debug_list.append(int_entry)41                            else:42                                break43        if interface_check:44            audit_list.append({45                'name': switch['name'], 'interface': interface_check})46    file_create(47        'rogue_bpdu_switches',48        'logs/audit/',49        audit_list,50        file_extension='yml',51        override=True)52def rogue_mac_ports(switch_list, user, pwd=None) -> None:53    '''54    '''55    switch_list = format_switch_list(56        switch_list, user, pwd=pwd)57    switch_list = switch_list_send_command(58        switch_list, 'show mac address-table',59        fsm=True, fsm_template='nil_lib/templates/mac_address_table')60    audit_list = []61    for switch in switch_list:62        mac_list, processed_list = [], []63        for check_entry in switch['output'][:]:64            if check_entry['ports'] in processed_list:65                continue66            if ('CPU' in check_entry['ports']67                or 'Po' in check_entry['ports']):68                switch['output'].remove(check_entry)69                continue70            found = []71            for entry in switch['output']:72                if (check_entry == entry73                    or entry['ports'] in processed_list):74                    continue75                if check_entry['ports'] == entry['ports']:76                    found.append(entry)77            list_check = [78                {'key': 'vlan', 'value': check_entry['vlan']},79                {'key': 'mac', 'value': check_entry['mac']},80                {'key': 'type', 'value': check_entry['type']}]81            for check in list_check:82                if not isinstance(check['value'], list):83                    check_entry[check['key']] = [check['value']]84            if found:85                for entry in found:86                    list_check = [87                        {'key': 'vlan', 'value': entry['vlan']},88                        {'key': 'mac', 'value': entry['mac']},89                        {'key': 'type', 'value': entry['type']}]90                    for check in list_check:91                        if entry[check['key']] not in check_entry[check['key']]:92                            check_entry[check['key']].append(check['value'])93                    switch['output'].remove(entry)94            processed_list.append(check_entry['ports'])95            if len(check_entry['mac']) > 2:96                mac_list.append(check_entry['ports'])97        if mac_list:98            audit_list.append({99                'name': switch['name'], 'ports': sorted(mac_list)})100    file_create(101        'rogue_mac_ports',102        'logs/audit/',103        audit_list,104        file_extension='yml',105        override=True)106if __name__ == "__main__":...

test_init.py

Source:test_init.py

1"""Test Subaru component setup and updates."""2from unittest.mock import patch3from subarulink import InvalidCredentials, SubaruException4from homeassistant.components.homeassistant import (5    DOMAIN as HA_DOMAIN,6    SERVICE_UPDATE_ENTITY,7)8from homeassistant.components.subaru.const import DOMAIN9from homeassistant.config_entries import ConfigEntryState10from homeassistant.const import ATTR_ENTITY_ID11from homeassistant.setup import async_setup_component12from .api_responses import (13    TEST_VIN_1_G1,14    TEST_VIN_2_EV,15    TEST_VIN_3_G2,16    VEHICLE_DATA,17    VEHICLE_STATUS_EV,18    VEHICLE_STATUS_G2,19)20from .conftest import (21    MOCK_API_FETCH,22    MOCK_API_UPDATE,23    TEST_ENTITY_ID,24    setup_subaru_integration,25)26async def test_setup_with_no_config(hass):27    """Test DOMAIN is empty if there is no config."""28    assert await async_setup_component(hass, DOMAIN, {})29    await hass.async_block_till_done()30    assert DOMAIN not in hass.config_entries.async_domains()31async def test_setup_ev(hass, ev_entry):32    """Test setup with an EV vehicle."""33    check_entry = hass.config_entries.async_get_entry(ev_entry.entry_id)34    assert check_entry35    assert check_entry.state is ConfigEntryState.LOADED36async def test_setup_g2(hass):37    """Test setup with a G2 vehcile ."""38    entry = await setup_subaru_integration(39        hass,40        vehicle_list=[TEST_VIN_3_G2],41        vehicle_data=VEHICLE_DATA[TEST_VIN_3_G2],42        vehicle_status=VEHICLE_STATUS_G2,43    )44    check_entry = hass.config_entries.async_get_entry(entry.entry_id)45    assert check_entry46    assert check_entry.state is ConfigEntryState.LOADED47async def test_setup_g1(hass):48    """Test setup with a G1 vehicle."""49    entry = await setup_subaru_integration(50        hass, vehicle_list=[TEST_VIN_1_G1], vehicle_data=VEHICLE_DATA[TEST_VIN_1_G1]51    )52    check_entry = hass.config_entries.async_get_entry(entry.entry_id)53    assert check_entry54    assert check_entry.state is ConfigEntryState.LOADED55async def test_unsuccessful_connect(hass):56    """Test unsuccessful connect due to connectivity."""57    entry = await setup_subaru_integration(58        hass,59        connect_effect=SubaruException("Service Unavailable"),60        vehicle_list=[TEST_VIN_2_EV],61        vehicle_data=VEHICLE_DATA[TEST_VIN_2_EV],62        vehicle_status=VEHICLE_STATUS_EV,63    )64    check_entry = hass.config_entries.async_get_entry(entry.entry_id)65    assert check_entry66    assert check_entry.state is ConfigEntryState.SETUP_RETRY67async def test_invalid_credentials(hass):68    """Test invalid credentials."""69    entry = await setup_subaru_integration(70        hass,71        connect_effect=InvalidCredentials("Invalid Credentials"),72        vehicle_list=[TEST_VIN_2_EV],73        vehicle_data=VEHICLE_DATA[TEST_VIN_2_EV],74        vehicle_status=VEHICLE_STATUS_EV,75    )76    check_entry = hass.config_entries.async_get_entry(entry.entry_id)77    assert check_entry78    assert check_entry.state is ConfigEntryState.SETUP_ERROR79async def test_update_skip_unsubscribed(hass):80    """Test update function skips vehicles without subscription."""81    await setup_subaru_integration(82        hass, vehicle_list=[TEST_VIN_1_G1], vehicle_data=VEHICLE_DATA[TEST_VIN_1_G1]83    )84    with patch(MOCK_API_FETCH) as mock_fetch:85        await hass.services.async_call(86            HA_DOMAIN,87            SERVICE_UPDATE_ENTITY,88            {ATTR_ENTITY_ID: TEST_ENTITY_ID},89            blocking=True,90        )91        await hass.async_block_till_done()92        mock_fetch.assert_not_called()93async def test_update_disabled(hass, ev_entry):94    """Test update function disable option."""95    with patch(MOCK_API_FETCH, side_effect=SubaruException("403 Error"),), patch(96        MOCK_API_UPDATE,97    ) as mock_update:98        await hass.services.async_call(99            HA_DOMAIN,100            SERVICE_UPDATE_ENTITY,101            {ATTR_ENTITY_ID: TEST_ENTITY_ID},102            blocking=True,103        )104        await hass.async_block_till_done()105        mock_update.assert_not_called()106async def test_fetch_failed(hass):107    """Tests when fetch fails."""108    await setup_subaru_integration(109        hass,110        vehicle_list=[TEST_VIN_2_EV],111        vehicle_data=VEHICLE_DATA[TEST_VIN_2_EV],112        vehicle_status=VEHICLE_STATUS_EV,113        fetch_effect=SubaruException("403 Error"),114    )115    test_entity = hass.states.get(TEST_ENTITY_ID)116    assert test_entity.state == "unavailable"117async def test_unload_entry(hass, ev_entry):118    """Test that entry is unloaded."""119    assert ev_entry.state is ConfigEntryState.LOADED120    assert await hass.config_entries.async_unload(ev_entry.entry_id)121    await hass.async_block_till_done()...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.