KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use describe_sessions method in localstack

Best Python code snippet using localstack_python

index.py

Source:index.py

...45    api_instance = swagger_client.SessionsApi(swagger_client.ApiClient(get_client_configuration()))46    set_request_headers(api_instance.api_client)47    return api_instance48		49def describe_sessions(session_ids=None, next_token=None, tags=None, owner=None):50    filters = list()51    if tags:52        for tag in tags:53            filter_key_value_pair = KeyValuePair(key='tag:' + tag['Key'], value=tag['Value'])54            filters.append(filter_key_value_pair)55    if owner:56        filter_key_value_pair = KeyValuePair(key='owner', value=owner)57        filters.append(filter_key_value_pair)58    request = DescribeSessionsRequestData(session_ids=session_ids, filters=filters, next_token=next_token)59    print('Describe Sessions Request:', request)60    api_instance = get_sessions_api()61    api_response = api_instance.describe_sessions(body=request)62    print('Describe Sessions Response', api_response)63    # FIX: return only the values required 64    return api_response65def create_sessions(sessions_to_create):66    create_sessions_request = list()67    for name, owner, session_type, init_file_path, max_concurrent_clients,\68            dcv_gl_enabled, permissions_file, requirements in sessions_to_create:69        a_request = CreateSessionRequestData(70            name=name, owner=owner, type=session_type,71            init_file_path=init_file_path, max_concurrent_clients=max_concurrent_clients,72            dcv_gl_enabled=dcv_gl_enabled, permissions_file=permissions_file, requirements=requirements)73        create_sessions_request.append(a_request)74    api_instance = get_sessions_api()75    print('Create Sessions Request:', create_sessions_request)76    api_response = api_instance.create_sessions(body=create_sessions_request)77    print('Create Sessions Response:', api_response)    78    # FIX: return only the values required 79    return api_response80def get_session_connection_api():81    api_instance = swagger_client.GetSessionConnectionDataApi(swagger_client.ApiClient(get_client_configuration()))82    set_request_headers(api_instance.api_client)83    return api_instance84def get_session_connection_data(session_id, user):85    api_response = get_session_connection_api().get_session_connection_data(session_id=session_id, user=user)86    print('Get Session Connection Data Response:', api_response)87    # FIX: return only the values required 88    return api_response89def delete_sessions(sessions_to_delete, force=False):90    delete_sessions_request = list()91    for session_id, owner in sessions_to_delete:92        a_request = DeleteSessionRequestData(session_id=session_id, owner=owner, force=force)93        delete_sessions_request.append(a_request)94    print('Delete Sessions Request:', delete_sessions_request)95    api_instance = get_sessions_api()96    api_response = api_instance.delete_sessions(body=delete_sessions_request)97    print('Delete Sessions Response', api_response)98    99    # FIX: return only the values required 100    return api_response101def handler(event, context):102    arguments = event["arguments"]103    print('Arguments: {} \n Event: {}'. format(arguments, event))104    action = arguments["action"] # mandatory105    sessionId = arguments["sessionId"] if arguments["sessionId"] else None106    owner = arguments["owner"] if arguments["owner"] else None 107    sessionType = arguments["sessionType"] if arguments["sessionType"] else None 108    concurrency = arguments["concurrency"] if arguments["concurrency"] else None 109    glenabled = arguments["glenabled"] if arguments["glenabled"] else None 110    tags = arguments["tags"] if arguments["tags"] else None 111    print('Action: {}, sessionId: {}, owner: {}'.format(action, sessionId, owner))112    # if there is a valid action go ahead, otherwise exit113    if ('create' not in action and 'list' not in action and 'connection' not in action and 'delete' not in action):114        return { 115            'statusCode': 200,116            'body': 'No valid action found. Exiting.'117        }118    else: 119        # Windows DCV only accepts CONSOLE sessions120        # Windows DCV does not accept InitFilePath 121        # Windows DCV does not accept DcvGlEnabled (only used with VIRTUAL sessions)122        # Name, Owner, Type, InitFilePath, MaxConcurrents, DcvGlEnabled, PermissionsFile, tags (format: "tag_key = tag_value")123        if (action == 'create'):124            InitFilePath = None 125            MaxConcurrents = concurrency 126            DcvGlEnabled = glenabled 127            PermissionsFile = None 128            tags = tags 129            if (sessionType == 'CONSOLE'):130                response = create_sessions([131                    (sessionId, owner, sessionType, InitFilePath, MaxConcurrents, DcvGlEnabled, PermissionsFile, tags)    132                ])133                return json.dumps(response)134            else:135                return {136                'statusCode': 200,137                'body': 'Windows DCV Servers only accept session type equals to CONSOLE. Exiting.'138                }139        # Describe can be used with or without a filter140        # filter is based on tags, sesionid, owner141        if (action == 'list' and sessionId is not None): 142            print('\n Describing session... \n ')143            response = describe_sessions()144            return json.dumps(response)145        # GetSessionConnectionData gets you the connection token and the sessionId146        # use that information to build your connection URL / .dcv file147        # Windows sessions always have Administrator as owner148        # format: 149        #   https://<dcv_server_IP>:<port>/<web_url_path>/?authToken=token#sessionId150        if (action == 'connection' and sessionId is not None and owner is not None):151            print('\n Session connection data... \n ')152            response = get_session_connection_data(sessionId, owner)153            return json.dumps(response)154        # DeleteSession requires sessionId and owner155        if (action == 'delete' and sessionId is not None and owner is not None):156            print('\n Deleting session... \n')157            response = delete_sessions([(sessionId, owner)])...

describe_ssm_sessions.py

Source:describe_ssm_sessions.py

...13        ssm_client = boto3.client('ssm', region_name = j )14        list_ssm = []15        next_token = None16        while True:17            response2 = ssm_client.describe_sessions(State='Active',18                                                                        NextToken= next_token19                                                                    ) if next_token else ssm_client.describe_sessions(State='Active')20            print(response2['Sessions'])                                                            21            for l in response2['Sessions']:22     23                24                status_var2 = l['Status']25                26                rank = 027                if status_var2 == "Connected":28                    rank = 129                if status_var2 == "Failed":30                    rank = 23132                list_ssm.append((l['SessionId'], "Sessions" , "No Document name",l['Status'], rank))33    
...

Close_Sessions_AppStream.py

Source:Close_Sessions_AppStream.py

...5ec2_re = session.resource(service_name='ec2', region_name='us-east-1')6cloudwatch_cli = session.client(service_name='appstream', region_name='us-east-1')7ec2_cli = session.client(service_name='ec2', region_name='us-east-1')8appstream_client = session.client(service_name='appstream', region_name='us-east-1')9#responses = appstream_client.describe_sessions(StackName='Stack-QX-NO-CROND', FleetName='Fleet-QX-NO-CROND')10responses = appstream_client.describe_sessions(StackName=StackName, FleetName=FleetName)11for respon in responses['Sessions']:12    Id_sesion = respon['Id']13    UserId = respon['UserId']14    ConnectionState = respon['ConnectionState']15    MaxExpirationTime = respon['MaxExpirationTime']16    EniPrivateIpAddress = respon['NetworkAccessConfiguration']['EniPrivateIpAddress']17    #Force_Brute_Close_Session = appstream_client.expire_session(SessionId=Id_sesion)18    # print(Id_sesion, UserId, ConnectionState, MaxExpirationTime, EniPrivateIpAddress) # Punto de control19    if 'NOT_CONNECTED' == ConnectionState:20        appstream_client.expire_session(SessionId=Id_sesion)21        # print('El ID de conexiÃ³n ',Id_sesion,' del usuario ',UserId, ConnectionState, MaxExpirationTime, EniPrivateIpAddress) #Punto de control22        print('El ID de conexiÃ³n ' +Id_sesion+' del usuario ' +UserId +' fue cerrado por completo, ya que presenta un estado ' +ConnectionState +'.')23    else:24        print('El ID de conexiÃ³n ' +Id_sesion +' del usuario ' +UserId +', presenta un estado ' +ConnectionState +', por lo cual no sera cerrada la sesion.')

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.