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

How to use get_created method in locust

Best Python code snippet using locust

app.py

Source:app.py

...60    Returns:61        int: The number of players on the server62    """63    # If the server is <3 minutes old, just return 0. Chances are its still being setup and checking a server thats still starting causes a slowdown64    if (datetime.now() - get_created(service)).total_seconds() < min_age:65        return -166    try:67        service.reload()68        ip = service.name + ":25565"69        print(ip)70        server = MinecraftServer.lookup(ip)71        return server.status().players.online72    except Exception as e:73        print(e)74        return -275def get_created(service: Service) -> datetime:76    """Gets the datetime of when a service was created77    Args:78        service (Service): The service to find the time created of79    Returns:80        datetime: The datetime object of when the service was created81    """82    # Might be something like ISO 8601 to reduce the headache with this formatting83    createdAt = service.attrs['CreatedAt']84    pi = createdAt.index(".")85    return datetime.strptime(createdAt[0:pi], "%Y-%m-%dT%H:%M:%S")86# def get_created_volume(volume) -> datetime:87#     # Might be something like ISO 8601 to reduce the headache with this formatting88#     return datetime.strptime(volume.attrs['CreatedAt'], "%Y-%m-%dT%H:%M:%SZ")89def get_services():90    """Returns list of services running MC Servers91    Returns:92        List: List of Services93    """94    return client.services.list(filters={"label": [stack_name]})95    # return sorted(client.services.list(filters={"label": [check_label]}), key=lambda x: get_created(x), reverse=True)96def get_nodes():97    """Returns a list of nodes in the swarm98    Returns:99        List: List of Nodes100    """101    return client.nodes.list()102resources = Resources(mem_limit='1.5g')103def create_service(username):104    """Created a new server service105    Args:106        username (string): The username for the OP user107    Returns:108        Service: The mc server service109    """110    # global port_last111    # vol = {check_label+"_user_"+username: {'bind': '/server', 'mode': 'rw'}} if username != None else False112    env = [f"OP_USERNAME={username}"] if username != None else [f"OP_USERNAME="]113    # port_last = port_last + 1114    # if port_last > port_max:115    #     port_last = port_min116    port = randint(port_min, port_max)117    return client.services.create('emwjacobson/mcserver:latest', endpoint_spec=EndpointSpec(ports={port: (25565, "tcp")}),118                                 labels={stack_name: '', 'username': username}, env=env, networks=[stack_name+"_default"])119    # return client.containers.run('emwjacobson/mcserver:latest', mem_limit='1.5g', cpu_quota=100000, cpu_period=100000,120    #                              remove=True, detach=True, ports={'25565/tcp': port_range, '25565/udp': port_range},121    #                              labels={check_label: '', 'username': username}, network=check_label,122    #                              volumes=vol, environment=env)123def stop_service(service: Service):124    """Stops service `service`125    Args:126        service (Service): The service you want to stop127    """128    # service.exec_run("/bin/sh -c 'kill $(pidof java)'", detach=True)129    service.remove()130###################################131#132# END Helper Functions133#134###################################135###################################136#137# REST Endpoints138#139###################################140@app.route('/stats/')141def stats():142    try:143        services = get_services()144        nodes = get_nodes()145        rtn = {146            **response_success,147            "message": "",148            "num_nodes": len(nodes),149            "num_running": len(services),150            "servers": []151        }152        for service in services:153            created = get_created(service)154            rtn['servers'].append({155                "id": service.id,156                "port": get_port(service),157                "created": created,158                "alive_for": (datetime.now() - created).total_seconds(),159                "num_players": get_online(service),160                "username": service.attrs['Spec']['Labels']['username']161            })162        return rtn163    except docker.errors.APIError as e:164        return {165            **response_error,166            "message": e.explanation167        }168    except Exception as e:169        print(e)170        return {171            **response_error,172            "message": "an error has occured"173        }174@app.route('/stats/<cid>') # TODO: This was never implemented in v1, so no need to reimplement yet.175def stats_service(cid):176    return177    # cid = escape(cid)178    # if len(cid) < 64:179    #     return {180    #         **error,181    #         "message": "Invalid server id"182    #     }183    # try:184    #     container = client.containers.get(cid)185    #     return {186    #         **success,187    #         "message": "",188    #         "id": cid,189    #         "port": get_port(container),190    #         "created": get_created(container),191    #         "alive_for": (datetime.now() - get_created(container)).total_seconds(),192    #         "num_players": get_online(container),193    #         "username": container.labels['username']194    #     }195    # except docker.errors.NotFound as e:196    #     return {197    #         **error,198    #         "message": e.explanation199    #     }200    # except Exception as e:201    #     print(e)202    #     return {203    #         **error,204    #         "message": "An error has occured"205    #     }206@app.route('/start/')207@app.route('/start/<username>')208def start_server(username=None):209    services = sorted(get_services(), key=lambda x : get_created(x), reverse=True)210    if len(services) > 0:211        newest = services[0]212        diff = datetime.now() - get_created(newest)213        # If newest server was made less than 60 seconds ago, wait a bit before starting a new one214        if diff.total_seconds() < 60:215            return {216                **response_error,217                "message": f"Another server was created too recently. Please wait {60-diff.total_seconds():.0f} more seconds and try again."218            }219    if username != None:220        username = escape(username)221        if len(username) < 3:222            return {223                **response_error,224                "message": "Username too short"225            }226        if username.startswith(stack_name):227            return {228                **response_error,229                "message": f"Username cannot start with {stack_name}"230            }231    try:232        services = get_services()233        # if len(services) >= server_per_node * len(get_nodes()):234        #     return {235        #         **response_error,236        #         "message": "Maximum amount of servers reached"237        #     }238        for service in services:239            if service.attrs['Spec']['Labels']['username'] == username:240                return {241                    **response_error,242                    "message": "A server with that username has already been started"243                }244    except docker.errors.APIError as e:245        return {246            **response_error,247            "message": e.explanation248        }249    try:250        service = create_service(username)251    except docker.errors.ImageNotFound as e:252        return {253            **response_error,254            "message": e.explanation255        }256    except docker.errors.APIError as e:257        return {258            **response_error,259            "message": "Error, please try again. Port might already be in use."260        }261    try:262        service.reload()263        return {264            **response_success,265            "message": "Server started",266            "id": service.id,267            "port": get_port(service)268        }269    except Exception as e:270        print(e)271        service.remove()272        return {273            **response_error,274            "message": "An error has occured"275        }276@app.route('/stop/<service_id>')277def stop_server(service_id):278    service_id = escape(service_id)279    if len(service_id) < 25:280        return {281            **response_error,282            "message": "Invalid server ID"283        }284    try:285        service = client.services.get(service_id)286        if stack_name in service.attrs['Spec']['Labels']:287            if (datetime.now() - get_created(service)).total_seconds() < min_age:288                return {289                    **response_error,290                    "message": "Server is too new to be stopped, please wait at least 3 minutes after starting"291                }292            stop_service(service)293            return {294                **response_success,295                "message": "Server stopped"296            }297        else:298            raise docker.errors.NotFound("", explanation=f"No such server: {service_id}")299    except docker.errors.NotFound as e:300        return {301            **response_error,302            "message": e.explanation303        }304@app.route('/reset/<username>') # TODO: This was never implemented in v1, so no need to reimplement yet.305def reset_server(username):306    return307    # if username != None:308    #     username = escape(username)309    #     if len(username) < 5:310    #         return {311    #             **error,312    #             "message": "Username too short"313    #         }314    # # Stop any servers with the username if they are running315    # for c in get_containers():316    #     if c.labels['username'] == username:317    #         stop_container(c)318    #         break319    # # Make sure the volume for `username` exists320    # try:321    #     vol = client.volumes.get(username)322    # except docker.errors.NotFound as e:323    #     return {324    #         **error,325    #         "message": "Username not found"326    #     }327    # except Exception as e:328    #     print(e)329    #     return {330    #         **error,331    #         "message": "There was an error"332    #     }333    # # Finally remove the volume334    # try:335    #     vol.reload()336    #     vol.remove(force=True)337    # except Exception as e:338    #     print(e)339    #     return {340    #         **error,341    #         "message": "There was an error"342    #     }343    # return {344    #     **success,345    #     "message": "Server has been reset"346    # }347@app.route('/')348def index():349    return "", 404350###################################351#352# END REST Endpoints353#354###################################355# This code is ran when the program is run as pure python.356# Intended to be used as a cron-job to clean up servers357if __name__ == "__main__":358    for service in get_services():359        diff = datetime.now() - get_created(service)360        total_seconds = diff.total_seconds()361        if total_seconds > 600:362            try:363                num_online = get_online(service)364                if num_online == 0:365                    print("Stopping server for being alive too long without active players")366                    stop_service(service)367            except ConnectionRefusedError as e:...

admin.py

Source:admin.py

...37    def get_author(self, obj):38        return obj.comment.author39    get_author.short_description = _('Author')40    get_author.admin_order_field = 'comment__author'41    def get_created(self, obj):42        return obj.comment.created43    get_created.short_description = _('Created')44    get_created.admin_order_field = 'comment__created'45    @method_decorator(staff_member_required)46    def approve(self, request, obj):47        obj.approve()48        form = AdminCommentForm(49            data={50                'author': obj.comment.author.id,51                'message': obj.comment.message,52                'post': obj.comment.post.id53            },54            instance=obj.comment55        )56        if form.is_valid():57            form.save()58        return redirect(self.return_url)59    approve.short_description = _('Approve')60    @method_decorator(staff_member_required)61    def disapprove(self, request, obj):62        obj.disapprove()63        NotificationDataView.approval_notification(request.user, obj.comment.author, NotificationDataView.get_model().DISSAPROVED_COMMENT)64        return redirect(self.return_url)65    disapprove.short_description = _('Disapprove')66    def approve_show(self, obj):67        return '<a class="addlink" href="{0}/approve">Approve</a>'.format(obj.id)68    approve_show.allow_tags = True69    approve_show.short_description = _("Approve")70    def disapprove_show(self, obj):71        return '<a class="deletelink" href="{0}/disapprove">Disapprove</a>'.format(obj.id)72    disapprove_show.allow_tags = True73    disapprove_show.short_description = _("Disapprove")74    def image_thumbnail(self, obj):75        return render_to_string('approval/_admin_thumbnail.html', context={"image": obj.get_image()})76    image_thumbnail.allow_tags = True77    image_thumbnail.short_description = _("Image")78    """79    Declare your admin buttons here!80    """81    buttons = [(approve.func_name, approve.short_description), (disapprove.func_name, disapprove.short_description), ]82class PostApprovalAdmin(ButtonableAdmin, ImprovedRawIdFields):83    return_url = reverse_lazy("admin:approval_postapproval_changelist")84    search_fields = ['post__title', 'post__author__username']85    list_display = ['reason', 'image_thumbnail', 'get_link', 'get_author', 'approve_show', 'disapprove_show', 'get_created']86    raw_id_fields = ['post']87    readonly_fields = ['reason']88    fields = ['post', 'reason']89    raw_id_managers = {'post': 'all_objects'}90    def get_link(self, obj):91        return '<a href="{0}">{1}</a>'.format(reverse_lazy("admin:post_post_change", args=[obj.post_id]), obj.post)92    get_link.short_description = _('Post')93    get_link.allow_tags = True94    def get_author(self, obj):95        return obj.post.author96    get_author.short_description = _('Author')97    get_author.admin_order_field = 'post__author'98    def get_created(self, obj):99        return obj.post.created100    get_created.short_description = _('Created')101    get_created.admin_order_field = 'post__created'102    @method_decorator(staff_member_required)103    def approve(self, request, obj):104        obj.approve()105        data = {106            'author': obj.post.author.id,107            'message': obj.post.message,108            'title': obj.post.title,109            'community': obj.post.community_id110        }111        if obj.post.community:112            data['wall-selector'] = 'community'...

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.