Best Python code snippet using avocado_python
NIOSSPT-6623.py
Source:NIOSSPT-6623.py  
1#!/usr/bin/env python2# -*- coding: utf-8 -*-3__author__ = "Shashikala R S"4__email__  = "srs@infoblox.com"5#############################################################################6#  Grid Set up required:                                                    #7#  1. Licenses : DNS(enabled), DHCP, Grid                                   #8#############################################################################9import os10import re11import config12import pytest13import unittest14import logging15import json16from time import sleep17import ib_utils.ib_NIOS as ib_NIOS18import pexpect19logging.basicConfig(format='%(asctime)s - %(name)s(%(process)d) - %(levelname)s - %(message)s',filename="rfe_10324.log" ,level=logging.DEBUG,filemode='w')20class NIOSSPT_6623(unittest.TestCase):21    @pytest.mark.run(order=1)22    def test_000_start_dns_services(self):23        print("Start DNS service")24        get_ref = ib_NIOS.wapi_request('GET', object_type='member:dns')25        print(get_ref)26        for ref in json.loads(get_ref):27            response = ib_NIOS.wapi_request('PUT', ref=ref['_ref'], fields=json.dumps({"enable_dns":True}))28            if type(response) == tuple:29                if response[0]==400 or response[0]==401:30                    assert False31    32    33    34    @pytest.mark.run(order=2)35    def test_001_create_nsg_services(self):36        print("create 'default' NSG using the GM as Grid Primary and set as default NSG")37        data = {"name": "NSG",38               "comment": "Created a new NSG server",39               "grid_primary":[{"name": config.grid_fqdn}]40               }41        response = ib_NIOS.wapi_request('POST',object_type="nsgroup",fields=json.dumps(data))42        print(response)43        if type(response) == tuple:44            if response[0]==400 or response[0]==401:45                print("Failure: Create default NSG")46                assert False47                  48                   49    @pytest.mark.run(order=3)50    def test_002_create_zone_default_NSG(self):     51        print("Create auth zone localdomain using the 'default' NSG")52        data = {"fqdn": "nsg_testing.com",53               "view": "default",54               "ns_group":"NSG"}55        response = ib_NIOS.wapi_request('POST',object_type="zone_auth",fields=json.dumps(data))56        print(response)57        if type(response) == tuple:58            if response[0]==400 or response[0]==401:59                print("Failure: Create auth zone localdomain using the 'default' NSG")60                assert False61                  62        63    @pytest.mark.run(order=4)64    def test_003_enable_ddns_update(self):      65        print("Enable DDNS updates from anywhere")66        get_ref = ib_NIOS.wapi_request('GET', object_type='grid:dns')67        data = {"allow_update":[{"_struct": "addressac","address":"Any", "permission":"ALLOW"}]}68        for ref in json.loads(get_ref):69            response = ib_NIOS.wapi_request('PUT', ref=json.loads(get_ref)[0]['_ref'], fields=json.dumps(data))70            print(response)71            if type(response) == tuple:72                if response[0]==400 or response[0]==401:73                    print("Failure: Enable DDNS updates from anywhere")74                    assert False75                76        print("Restart DHCP Services")77        grid =  ib_NIOS.wapi_request('GET', object_type="grid", grid_vip=config.grid_vip)78        ref = json.loads(grid)[0]['_ref']79        data= {"member_order" : "SIMULTANEOUSLY","restart_option":"FORCE_RESTART","service_option": "ALL"}80        request_restart = ib_NIOS.wapi_request('POST', object_type = ref + "?_function=restartservices",fields=json.dumps(data),grid_vip=config.grid_vip)81        sleep(20)82        83    @pytest.mark.run(order=5)84    def test_004_create_CAA_Record(self):     85        print("Add a CAA record")86        data = {"name": "c_rec.nsg_testing.com",87                "ca_flag": 0,88                "ca_tag": "issue",89                "ca_value": "no",90               "view": "default"}91        response = ib_NIOS.wapi_request('POST',object_type="record:caa",fields=json.dumps(data))92        print(response)93        if type(response) == tuple:94            if response[0]==400 or response[0]==401:95                print("Failure: Add a CAA record")96                assert False97     98    @pytest.mark.run(order=6)99    def test_005_create_CAA_Record_nsupdate(self):     100        print("Add a CAA record through nsupdate")101        child = pexpect.spawn("nsupdate")102        try:103            child.expect('> ')104            child.send('server '+config.grid_vip+'\n')105            print(child.before)106            child.expect('>')107            child.send('update add caa_rec.nsg_testing.com. 100 caa 0 issue "nsg_testing.com"\n')108            print(child.before)109            child.expect('>')110            child.send('send\n')111            print(child.before)112            child.expect('>')113            child.send('quit\n')114        except Exception as e:115            print(e)116            assert False117        finally:118            child.close()119            120    @pytest.mark.run(order=7)121    def test_006_validate_caa_record_updation(self):        122        print("\nValidating caa record got updated or not")     123        get_ref = ib_NIOS.wapi_request('GET', object_type='record:caa?name=caa_rec.nsg_testing.com')124        for ref in json.loads(get_ref):125            if 'caa_rec' in ref['_ref']:126                print("Success: Validating caa record got updated")127                assert True128            else:129                print("Failure: Validating caa record got updated")130                assert False131                        132    @pytest.mark.run(order=8)133    def test_007_sign_zone_notice_message(self):        134        print("Try to sign the zone, and notice the error message received")     135        get_ref = ib_NIOS.wapi_request('GET', object_type='zone_auth')136        for ref in json.loads(get_ref):137            if 'nsg_testing' in ref['_ref']:138                response = ib_NIOS.wapi_request('POST', ref=ref['_ref'], params='?_function=dnssec_operation', fields=json.dumps({"operation":"SIGN"}))139                print(response)140                if type(response) == tuple:141                    if response[0]==400 or response[0]==401:142                        print("Failure: Sign Zone signs without error")143                        assert False144                    else:145                        print("Success: Sign Zone ssigns without error")146                        147    148    @pytest.mark.run(order=9)149    def test_008_unsign_zone_(self):        150        print("Try to unsign the zone")     151        get_ref = ib_NIOS.wapi_request('GET', object_type='zone_auth')152        for ref in json.loads(get_ref):153            if 'nsg_testing' in ref['_ref']:154                response = ib_NIOS.wapi_request('POST', ref=ref['_ref'], params='?_function=dnssec_operation', fields=json.dumps({"operation":"UNSIGN"}))155                print(response)156                if type(response) == tuple:157                    if response[0]==400 or response[0]==401:158                        print("Failure: Unsign Zone signs without error")159                        assert False160                    else:161                        print("Success: Unsign Zone ssigns without error")162    @pytest.mark.run(order=10)   163    def test_009_change_nsg_to_grid_primary(self):164        print("Edit the zone and move from the 'default' NSG to a manually defined GM as Grid Primary")165    166        get_ref = ib_NIOS.wapi_request('GET', object_type='zone_auth')167        data = {"view":"default","grid_primary":[{"name": config.grid_fqdn}],"comment":"Move from NSG to grid"}168        for ref in json.loads(get_ref):169            if 'nsg_testing' in ref['_ref']:170                response = ib_NIOS.wapi_request('PUT', ref=ref['_ref'], fields=json.dumps(data))171                print(response)172                if type(response) == tuple:173                    if response[0]==400 or response[0]==401:174                        print("Failure: move from NSG to Grid")175                        assert False176        print("Restart DHCP Services")177        grid =  ib_NIOS.wapi_request('GET', object_type="grid", grid_vip=config.grid_vip)178        ref = json.loads(grid)[0]['_ref']179        data= {"member_order" : "SIMULTANEOUSLY","restart_option":"FORCE_RESTART","service_option": "ALL"}180        request_restart = ib_NIOS.wapi_request('POST', object_type = ref + "?_function=restartservices",fields=json.dumps(data),grid_vip=config.grid_vip)181        sleep(30)182    183    184    @pytest.mark.run(order=11)   185    def test_010_sign_zone_without_error(self):186        print("Try to sign the zone, and confirm it signs without error")187        get_ref = ib_NIOS.wapi_request('GET', object_type='zone_auth')188        for ref in json.loads(get_ref):189            if 'nsg_testing' in ref['_ref']:190                response = ib_NIOS.wapi_request('POST', ref=ref['_ref'], params='?_function=dnssec_operation', fields=json.dumps({"operation":"SIGN"}),grid_vip=config.grid_vip)191                print(response)192                if type(response) == tuple:193                    if response[0]==400 or response[0]==401:194                        print("Failure: Sign Zone signs without error")195                        assert False196                    else:197                        print("Success: Sign Zone ssigns without error")198    199    200        201                   202    @pytest.mark.run(order=12)203    def test_011_cleanup(self):204        print("Clean up all created objects")205        get_ref = ib_NIOS.wapi_request('GET', object_type="nsgroup", grid_vip=config.grid_vip)206        for ref in json.loads(get_ref):207            response = ib_NIOS.wapi_request('DELETE',ref=ref['_ref'])208             209            print(response)210                    211            if type(response) == tuple:212                        213                if response[0]==400 or response[0]==401:214                            215                    assert False216                else:217                    assert True218        get_ref = ib_NIOS.wapi_request('GET', object_type="zone_auth", grid_vip=config.grid_vip)219        for ref in json.loads(get_ref):220            response = ib_NIOS.wapi_request('DELETE',ref=ref['_ref'])221             222            print(response)223                    224            if type(response) == tuple:225                        226                if response[0]==400 or response[0]==401:227                            228                    assert False229                else:230                    assert True231        get_ref = ib_NIOS.wapi_request('GET', object_type="record:caa", grid_vip=config.grid_vip)232        for ref in json.loads(get_ref):233            response = ib_NIOS.wapi_request('DELETE',ref=ref['_ref'])234             235            print(response)236                    237            if type(response) == tuple:238                        239                if response[0]==400 or response[0]==401:240                            241                    assert False242                else:243                    assert True244        print("Restart DHCP Services")245        grid =  ib_NIOS.wapi_request('GET', object_type="grid", grid_vip=config.grid_vip)246        ref = json.loads(grid)[0]['_ref']247        data= {"member_order" : "SIMULTANEOUSLY","restart_option":"FORCE_RESTART","service_option": "ALL"}248        request_restart = ib_NIOS.wapi_request('POST', object_type = ref + "?_function=restartservices",fields=json.dumps(data),grid_vip=config.grid_vip)...test_data_manager.py
Source:test_data_manager.py  
...54    return empty_testdset, other_dset55def test_create_ref(datamanager, empty_testref):56    dm = datamanager57    # check that ref dataset opens and has correct type58    assert dm.get_ref(*empty_testref)[0].shape == (0, 2)59    assert dm.get_ref(*empty_testref)[-1] == (0, 1)60    assert dm.get_ref_region(*empty_testref).dtype == ref_region_dtype61    # check that ref dataset is still empty62    assert len(dm.get_ref(*empty_testref)[0]) == 063    assert len(dm.get_ref_region(*empty_testref)) == 064@pytest.fixture65def full_testdset(datamanager, empty_testdset, empty_testref):66    dm = datamanager67    sl = dm.reserve_data(empty_testdset, 100)68    dm.write_data(empty_testdset, sl, rank)69    return empty_testdset, sl70def test_write_dset(datamanager, full_testdset):71    dm = datamanager72    # check that we have access to the *full* dataset after writing73    assert len(dm.get_dset(full_testdset[0])) == size * 10074    # check that processes wrote to correct region75    assert all(dm.get_dset(full_testdset[0])[full_testdset[1]] == rank)76@pytest.fixture77def full_testref(datamanager, empty_testref, full_testdset):78    dm = datamanager79    sl = dm.reserve_data(empty_testref[-1], full_testdset[-1])80    dm.write_data(empty_testref[-1], sl, rank)81    ref_idcs = np.r_[full_testdset[1]].reshape(1, -1, 1)82    idcs = np.r_[full_testdset[1]].reshape(-1, 1, 1)83    idcs, ref_idcs = np.broadcast_arrays(idcs, ref_idcs)84    ref = np.concatenate((idcs, ref_idcs), axis=-1).reshape(-1, 2)85    dm.write_ref(*empty_testref, ref)86    return empty_testref, full_testdset[-1]87def test_write_ref(datamanager, full_testdset, full_testref):88    dm = datamanager89    n = len(dm.get_dset(full_testdset[0]))90    # check that we have access to the *full* ref dataset after writing91    assert len(dm.get_ref(*full_testref[0])[0]) == size * 100**292    # check that child attribute is accessible and correct93    assert dm.fh[dm.get_ref(*full_testref[0])[0].attrs['dset0']] == dm.get_dset(full_testdset[0])94    assert dm.fh[dm.get_ref(*full_testref[0])[0].attrs['dset1']] == dm.get_dset(full_testref[0][-1])95    ref, ref_dir = dm.get_ref(*full_testref[0])96    # check that first of process' refs point to the correct chunk of the dataset97    sel = full_testref[1]98    ref_region = dm.get_ref_region(*full_testref[0])99    assert all(ref_region[sel]['start'] != ref_region[sel]['stop'])100    data = dereference(sel, ref, dm.get_dset(full_testdset[0]), ref_region, ref_direction=ref_dir)101    assert all([np.all(dm.get_dset(full_testdset[0])[sel] == d) for d in data])102def test_getitem(datamanager, full_testdset, full_testref):103    dm = datamanager104    assert dm.get_dset(full_testdset[0]) == dm[full_testdset[0] + '/data']105    assert len(dm[full_testdset[0], :10]) == 10106    assert np.all(dm[full_testdset[0], :10] == dm.get_dset(full_testdset[0])[:10])107    assert dm[tuple(full_testref[0])].dtype == dm.get_dset(full_testref[0][-1]).dtype108    assert len(dm[tuple(full_testref[0])]) == len(dm.get_dset(full_testref[0][0]))109    assert len(dm[full_testref[0][0], full_testref[0][1], :10]) == 10110    assert len(dm[full_testref[0][0], full_testref[0][1], 0]) == 1111def test_context(testfile):112    with H5FlowDataManager(testfile, 'a', mpi=H5FLOW_MPI) as dm:113        dm.create_dset('test', int)114        assert dm['test']115def test_repr(datamanager):116    print(datamanager)117@pytest.mark.skipif(size != 1, reason='test designed for single process only')118def test_write_ref(datamanager):119    dm = datamanager120    dm.create_dset('A', int)121    dm.create_dset('B', int)122    dm.create_ref('A', 'B')123    dm.reserve_data('A', 100)124    dm.reserve_data('B', 100)125    assert len(dm.get_dset('A')) == 100126    assert len(dm.get_dset('B')) == 100127    assert len(dm.get_ref_region('A', 'B')) == 100128    assert len(dm.get_ref_region('B', 'A')) == 100129    assert len(dm.get_ref('B', 'A')[0]) == 0130    # test writing A -> B references131    ref = np.c_[np.arange(0, 10), np.arange(20, 30)]132    dm.write_ref('A', 'B', ref)133    assert len(dm.get_dset('A')) == 100134    assert len(dm.get_dset('B')) == 100135    assert len(dm.get_ref_region('A', 'B')) == 100136    assert len(dm.get_ref_region('B', 'A')) == 100137    assert len(dm.get_ref('A', 'B')[0]) == 10138    assert len(dm.get_ref('B', 'A')[0]) == 10139    assert np.all(dm.get_ref('A', 'B')[0][:10, 0] == ref[:, 0])140    assert np.all(dm.get_ref('A', 'B')[0][:10, 1] == ref[:, 1])141    assert dm.get_ref('A', 'B')[1] == (0, 1)142    assert dm.get_ref('B', 'A')[1] == (1, 0)143    assert np.all(dm.get_ref_region('A', 'B')[0:10]['start'] == np.arange(10))144    assert np.all(dm.get_ref_region('A', 'B')[0:10]['stop'] == np.arange(10) + 1)145    assert np.all(dm.get_ref_region('B', 'A')[20:30]['start'] == np.arange(10))146    assert np.all(dm.get_ref_region('B', 'A')[20:30]['stop'] == np.arange(10) + 1)147    # test writing B -> A references148    ref = np.c_[np.arange(0, 10), np.arange(20, 30)]149    dm.write_ref('B', 'A', ref)150    assert len(dm.get_dset('A')) == 100151    assert len(dm.get_dset('B')) == 100152    assert len(dm.get_ref_region('A', 'B')) == 100153    assert len(dm.get_ref_region('B', 'A')) == 100154    assert len(dm.get_ref('A', 'B')[0]) == 20155    assert len(dm.get_ref('B', 'A')[0]) == 20156    assert np.all(dm.get_ref('B', 'A')[0][10:20, 1] == ref[:, 0])157    assert np.all(dm.get_ref('B', 'A')[0][10:20, 0] == ref[:, 1])158    assert dm.get_ref('A', 'B')[1] == (0, 1)159    assert dm.get_ref('B', 'A')[1] == (1, 0)160    assert np.all(dm.get_ref_region('A', 'B')[20:30]['start'] == np.arange(10, 20))161    assert np.all(dm.get_ref_region('A', 'B')[20:30]['stop'] == np.arange(10, 20) + 1)162    assert np.all(dm.get_ref_region('B', 'A')[0:10]['start'] == np.arange(10, 20))...field_reference.py
Source:field_reference.py  
...32def change_lazy_computation():33  """Overriding lazily computed values."""34  config = config_dict.ConfigDict()35  config.reference = 136  config.reference_0 = config.get_ref('reference') + 1037  config.reference_1 = config.get_ref('reference') + 2038  config.reference_1_0 = config.get_ref('reference_1') + 10039  print(config.reference)  # Prints 1.40  print(config.reference_0)  # Prints 11.41  print(config.reference_1)  # Prints 21.42  print(config.reference_1_0)  # Prints 121.43  config.reference_1 = 3044  print(config.reference)  # Prints 1 (unchanged).45  print(config.reference_0)  # Prints 11 (unchanged).46  print(config.reference_1)  # Prints 30.47  print(config.reference_1_0)  # Prints 130.48def create_cycle():49  """Creates a cycle within a ConfigDict."""50  config = config_dict.ConfigDict()51  config.integer_field = 152  config.bigger_integer_field = config.get_ref('integer_field') + 1053  try:54    # Raises a MutabilityError because setting config.integer_field would55    # cause a cycle.56    config.integer_field = config.get_ref('bigger_integer_field') + 257  except config_dict.MutabilityError as e:58    print(e)59def lazy_configdict():60  """Example usage of lazy computation with ConfigDict."""61  config = config_dict.ConfigDict()62  config.reference_field = config_dict.FieldReference(1)63  config.integer_field = 264  config.float_field = 2.565  # No lazy evaluatuations because we didn't use get_ref()66  config.no_lazy = config.integer_field * config.float_field67  # This will lazily evaluate ONLY config.integer_field68  config.lazy_integer = config.get_ref('integer_field') * config.float_field69  # This will lazily evaluate ONLY config.float_field70  config.lazy_float = config.integer_field * config.get_ref('float_field')71  # This will lazily evaluate BOTH config.integer_field and config.float_Field72  config.lazy_both = (config.get_ref('integer_field') *73                      config.get_ref('float_field'))74  config.integer_field = 375  print(config.no_lazy)  # Prints 5.0 - It uses integer_field's original value76  print(config.lazy_integer)  # Prints 7.577  config.float_field = 3.578  print(config.lazy_float)  # Prints 7.079  print(config.lazy_both)  # Prints 10.580def lazy_configdict_advanced():81  """Advanced lazy computation with ConfigDict."""82  # FieldReferences can be used with ConfigDict as well83  config = config_dict.ConfigDict()84  config.float_field = 12.685  config.integer_field = 12386  config.list_field = [0, 1, 2]87  config.float_multiply_field = config.get_ref('float_field') * 388  print(config.float_multiply_field)  # Prints 37.889  config.float_field = 10.090  print(config.float_multiply_field)  # Prints 30.091  config.longer_list_field = config.get_ref('list_field') + [3, 4, 5]92  print(config.longer_list_field)  # Prints [0, 1, 2, 3, 4, 5]93  config.list_field = [-1]94  print(config.longer_list_field)  # Prints [-1, 3, 4, 5]95  # Both operands can be references96  config.ref_subtraction = (97      config.get_ref('float_field') - config.get_ref('integer_field'))98  print(config.ref_subtraction)  # Prints -113.099  config.integer_field = 10100  print(config.ref_subtraction)  # Prints 0.0101def main(argv=()):102  del argv  # Unused.103  lazy_computation()104  lazy_configdict()105  change_lazy_computation()106  create_cycle()107  lazy_configdict_advanced()108if __name__ == '__main__':...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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
