Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use list_flavor_access method in tempest

Best Python code snippet using tempest_python

cron_update_ip2project.py

Source:cron_update_ip2project.py

...111    # process assigned projects but not VM running112    # assumption, 1 project can only have 1 GPU instance113    akl_gpu_flavor = osc_conn.search_flavors('akl.gpu*', get_extra=False)114    for f in akl_gpu_flavor:115        all_access = osc_conn.list_flavor_access(f.id)116        for a in all_access:117            # test if it already has an instance118            if a.project_id not in all_project_ids:119                # check if the project is still active120                project = fetch_project_info(a.project_id, allocation_client)121                if project is None: # or project.end_date < str(date.today()):122                    continue123                # we need to fix the project124                #for project in fetch_project_info:125                # get extra properties and project access info126                detail = osc_conn.get_flavor_by_id(f.id, get_extra=True)127                gpu_model = detail.extra_specs['pci_passthrough:alias'].split(':')[0]128                # print(f"{a.project_id} no instance, but has flavor {f.name}, GPU={gpu_model}")129                # find an available model and change the status130                for d in devices:131                    if d['label'] == gpu_model and d['status'] == 'available':132                        d['status'] = 'reserved'133                        d['project_id'] = a.project_id134                        break135    # update project_end date and contact136    for d in devices:137        # get project allocation info138        alloc = None139        if d['project_id'] is not None and len(d['project_id']) > 0:140            alloc = fetch_project_info(d['project_id'], allocation_client)141        #if alloc is not None:142        #    print alloc.__dict__143        if d['display_name'] is None and alloc is not None:144            d['display_name'] = alloc.project_name145        d['project_name'] = 'Auckland-CeR' if alloc is None else alloc.project_name146        d['start_date'] = 'NA' if alloc is None else alloc.start_date147        d['end_date'] = 'NA' if alloc is None else alloc.end_date148        d['contact'] = 'NA' if alloc is None else alloc.contact_email149        d['ip'] = find_ip(osc_conn, d['instance_uuid'])150        d['host'] = d['host'].replace("ntr-", "")151        d['host'] = d['host'].replace("akld2", "")152    # output153    x = PrettyTable()154    x.field_names = ['host', 'label', 'status', 'instance_uuid', 'display_name', 'project_name', 'project_id', 'start_date', 'end_date', 'contact', 'dev_id', 'ip', 'launched_at', 'terminated_at']155    for d in devices:156        row_data = []157        for f in x.field_names:158            row_data.append(d[f])159        x.add_row(row_data)160    print(x)161def list_user_projects(email, conn):162    """163    user id --> all projects (including closed ones) --> flavor164    :param email: email as user ID165    :param conn: openstack connection166    :return: 167    """168    allocation_client = allocationclient.Client(1, session=conn.session)169    # step 1: get the user ID from email170    user = conn.get_user(email)171    if user is None:172        print('can not find user')173        return174    # step 2: get all the projects of a user175    roles = conn.list_role_assignments(176        {177            'user': user.id178        }179    )180    all_projects = [r.project for r in roles]181    # print(all_projects)182    # step 3: get all auckland GPU flavors183    akl_gpu_flavors = conn.search_flavors('akl.gpu*', get_extra=False)184    user_allocations = []185    for f in akl_gpu_flavors:186        for access in conn.list_flavor_access(f.id):187            # print(access)188            if access.project_id in all_projects:189                alloc = fetch_project_info(access.project_id, allocation_client)190                if alloc:191                    user_allocations.append(alloc)192                    # print(f'{alloc.project_name}({alloc.project_id}): {alloc.start_date} -- {alloc.end_date}')193    # output194    x = PrettyTable()195    x.field_names = ['project_name', 'project_id', 'status', 'start_date', 'end_date']196    for a in user_allocations:197        x.add_row(198            [a.project_name, a.project_id, a.status_display, a.start_date, a.end_date]199        )200    print(x)201def fetch_gpu_nodes( gpudb_conn ):202  statement = 'SELECT hypervisor, pci_id FROM gpu_nodes WHERE active = 1'203  gpu_nodes = {}204  try:205    cursor = gpudb_conn.cursor()206    cursor.execute(statement)207    result = cursor.fetchall()208    for r in result:209      #For all the current active nodes in the gpu_nodes table, set to not seen yet210      gpu_nodes[r['hypervisor'] + ' ' + r['pci_id']] = { 'present': False, 'keys': r }211      return gpu_nodes212  except pymysql.Error as e:213    print "DB Update gpu_nodes: ", e214    return None215def update_gpu_db(osc_conn, db_conn, gpudb_conn):216  """217  save current metadata on current instances that are using a GPU218  :param osc_conn:219  :param gpudb_conn:220  :return:221  """222  allocation_client = allocationclient.Client(1, session=osc_conn.session)223  devices = fetch_pci_device_from_db(db_conn) 224  all_project_ids = [d['project_id'] for d in devices if d['project_id'] is not None]225  # process assigned projects but not VM running226  # assumption, 1 project can only have 1 GPU instance227  akl_gpu_flavor = osc_conn.search_flavors('akl.gpu*', get_extra=False)228  for f in akl_gpu_flavor:229    all_access = osc_conn.list_flavor_access(f.id)230    for a in all_access:231      # test if it already has an instance232      if a.project_id not in all_project_ids:233        # check if the project is still active234        project = fetch_project_info(a.project_id, allocation_client)235        if project is None: # or project.end_date < str(date.today()):236          continue237        # we need to fix the project238        #for project in fetch_project_info:239        # get extra properties and project access info240        detail = osc_conn.get_flavor_by_id(f.id, get_extra=True)241        gpu_model = detail.extra_specs['pci_passthrough:alias'].split(':')[0]242        # print(f"{a.project_id} no instance, but has flavor {f.name}, GPU={gpu_model}")243        # find an available model and change the status...

gpu.py

Source:gpu.py

...104    # process assigned projects but not VM running105    # assumption, 1 project can only have 1 GPU instance106    akl_gpu_flavor = osc_conn.search_flavors('akl.gpu*', get_extra=False)107    for f in akl_gpu_flavor:108        all_access = osc_conn.list_flavor_access(f.id)109        for a in all_access:110            # test if it already has an instance111            if a.project_id not in all_project_ids:112                # check if the project is still active113                project = fetch_project_info(a.project_id, allocation_client)114                if project is None or project.end_date < str(date.today()):115                    continue116                # we need to fix the project117                # get extra properties and project access info118                detail = osc_conn.get_flavor_by_id(f.id, get_extra=True)119                gpu_model = detail.extra_specs['pci_passthrough:alias'].split(':')[0]120                # print(f"{a.project_id} no instance, but has flavor {f.name}, GPU={gpu_model}")121                # find an available model and change the status122                status = 'conflict'123                for d in devices:124                    if d['label'] == gpu_model and d['status'] == 'available':125                        status = 'reserved'126                        d['status'] = 'reserved'127                        d['project_id'] = a.project_id128                        break129                if status == 'conflict' and project.end_date > str(date.today()):130                    devices.append({'host': '', 'label':gpu_model, 'status': status, 'instance_uuid': '', 'display_name': '', 'project_id': a.project_id, 'dev_id': '', 'launched_at': '', 'terminated_at': '' })131    # update project_end date and contact132    for d in devices:133        # get project allocation info134        alloc = None135        if d['project_id'] is not None and len(d['project_id']) > 0:136            alloc = fetch_project_info(d['project_id'], allocation_client)137        else:138           d['project_id'] = ''139        if d['display_name'] is None:140            if alloc is not None:141                d['display_name'] = alloc.project_name142            else:143                d['display_name'] = ''144        #CeR project shows up as None145        if alloc is None:146            if d['status'] != 'available':147                d['project_name'] = 'Auckland-CeR'148            else: 149                d['project_name'] = '' 150        else:151            d['project_name'] = alloc.project_name152        d['start_date'] = '' if alloc is None else alloc.start_date153        d['end_date'] = '' if alloc is None else alloc.end_date154        d['contact'] = '' if alloc is None else alloc.contact_email155        d['ip'] = find_ip(osc_conn, d['instance_uuid'])156        if d['instance_uuid'] is None:157            d['instance_uuid'] = ''158        if d['launched_at'] is None:159            d['launched_at'] = ''160    # output161    x = PrettyTable()162    if args.long:163        x.field_names = ['host', 'label', 'status', 'display_name', 'project_name', 'start_date', 'end_date', 'launched_at', 'contact',  'ip', 'dev_id', 'instance_uuid', 'project_id' ]164    elif args.short:165        x.field_names = ['host', 'label', 'status', 'project_name', 'project_id', 'end_date', 'contact']166    else:167        x.field_names = ['host', 'label', 'status', 'display_name', 'project_name', 'start_date', 'end_date', 'launched_at', 'contact']168    for d in devices:169        row_data = []170        for f in x.field_names:171            row_data.append(d[f])172        x.add_row(row_data)173    print(x)174def list_user_projects(args, conn):175    """176    user id --> all projects (including closed ones) --> flavor177    :param args: long, short, email as user ID178    :param conn: openstack connection179    :return: 180    """181    allocation_client = allocationclient.Client(1, session=conn.session)182    # step 1: get the user ID from email183    user = conn.get_user(args.email)184    if user is None:185        print('can not find user')186        return187    # step 2: get all the projects of a user188    roles = conn.list_role_assignments(189        {190            'user': user.id191        }192    )193    all_projects = [r.project for r in roles]194    # print(all_projects)195    # step 3: get all auckland GPU flavors196    akl_gpu_flavors = conn.search_flavors('akl.gpu*', get_extra=False)197    user_allocations = []198    for f in akl_gpu_flavors:199        for access in conn.list_flavor_access(f.id):200            # print(access)201            if access.project_id in all_projects:202                alloc = fetch_project_info(access.project_id, allocation_client)203                if alloc:204                    user_allocations.append(alloc)205                    # print(f'{alloc.project_name}({alloc.project_id}): {alloc.start_date} -- {alloc.end_date}')206    # output207    x = PrettyTable()208    x.field_names = ['project_name', 'project_id', 'status', 'start_date', 'end_date']209    for a in user_allocations:210        x.add_row(211            [a.project_name, a.project_id, a.status_display, a.start_date, a.end_date]212        )213    print(x)...

test_flavors_access.py

Source:test_flavors_access.py

...49                                               disk=self.disk,50                                               id=new_flavor_id,51                                               is_public='False')['flavor']52        self.addCleanup(self.client.delete_flavor, new_flavor['id'])53        flavor_access = (self.client.list_flavor_access(new_flavor_id)54                         ['flavor_access'])55        self.assertEqual(len(flavor_access), 0, str(flavor_access))56    @test.idempotent_id('59e622f6-bdf6-45e3-8ba8-fedad905a6b4')57    def test_flavor_access_add_remove(self):58        # Test to add and remove flavor access to a given tenant.59        flavor_name = data_utils.rand_name(self.flavor_name_prefix)60        new_flavor_id = data_utils.rand_int_id(start=1000)61        new_flavor = self.client.create_flavor(name=flavor_name,62                                               ram=self.ram, vcpus=self.vcpus,63                                               disk=self.disk,64                                               id=new_flavor_id,65                                               is_public='False')['flavor']66        self.addCleanup(self.client.delete_flavor, new_flavor['id'])67        # Add flavor access to a tenant....

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.