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

How to use apply_json method in localstack

Best Python code snippet using localstack_python

views.py

Source:views.py

1from django.http import JsonResponse2from django.shortcuts import get_object_or_4043from interviews.models import Apply4from profiles.models import Profile5from interviews.views import *6from django.contrib.auth import authenticate, login, logout7from rest_framework import generics8from .serializers import RegisterSerializer9from django.contrib.auth import get_user_model10from rest_framework import generics, status11from rest_framework.response import Response12from django.contrib.auth.models import User13from rest_framework.authtoken.models import Token14from .serializers import RegisterSerializer, LoginSerializer, ProfileSerializer15from .models import Profile16User=get_user_model()17# Create your views here.18def get_profile_one(request, id):19    if request.method == "GET":20        profile = get_object_or_404(Profile, pk=id)21        profile_json = {   22            "id": profile.id,23            "name": profile.name,24            "department": profile.department,25            "img": "/media/"+str(profile.img),26            "background": profile.background,27            "office": profile.office,28            "phone": profile.phone,29            "tag": profile.tag,30            "reply_rate": reply_rate(id),31            "reply_time": reply_time(id),32            "certification": profile.certification,33            "update_at": profile.update_at,34            "last_login": profile.user.last_login,35        }36        return JsonResponse({37            "status": 200,38            "success": True,39            "message": "ì ë¬¸ê° ëíì¼ íì´ì§ ìë¡ë ì±ê³µ",40            "data": profile_json41        })42    43    return JsonResponse({44        "status": 405,45        "success": False,46        "message": "method error",47        "data": None48    })49def get_profile_all(request, category_id):50    if request.method == "GET":51        profile_all = Profile.objects.filter(category_id=category_id)52        profile_all_json = []53        for profile in profile_all:54            profile_json = {55                "id": profile.id,56                "is_report" : profile.is_report,57                "name": profile.name,58                "department": profile.department,59                "img": "/media/"+str(profile.img),60                "tag": profile.tag,61                "reply_rate": reply_rate(profile.id),62                "reply_time": reply_time(profile.id),63                "certification": profile.certification,64                "update_at": profile.update_at,65                "last_login": profile.user.last_login,66            }67            profile_all_json.append(profile_json)68        return JsonResponse({69            "status": 200,70            "success": True,71            "message": "ì ë¬¸ê° ë¦¬ì¤í¸ íì´ì§ ìë¡ë ì±ê³µ",72            "data": profile_all_json73        })74    return JsonResponse({75        "status": 405,76        "success" : False,77        "message": "method error",78        "data": None79    })80def get_apply_list_for_expert(request):81    if request.method == "GET":82        token_line = request.META.get('HTTP_AUTHORIZATION')83        token = get_object_or_404(Token, key=token_line)84        apply_all = Apply.objects.filter(expert_user=token.user)85    86        request_json = []87        answered_json = []88        89        for apply in apply_all:90            sender_profile = get_object_or_404(Profile, user=apply.interview.reporter_user)91            apply_json = {92                    "id": apply.id,93                    "department": sender_profile.department,94                    "title": apply.interview.title,95                    "deadline": apply.interview.deadline,96                    "is_expired": apply.interview.is_expired,97                    "response": apply.response98                }99            if apply.response == 0:100                request_json.append(apply_json)101            elif apply.response > 0 or apply.interview.is_expired == 1:102                answered_json.append(apply_json)103        return JsonResponse({104            'status' : 200,105            'success' : True,106            'message' : 'ì ë¬¸ê° ë§ì´íì´ì§ ìì² ì¸í°ë·° ë¶ë¬ì¤ê¸° ì±ê³µ !',107            'data' : [ request_json, answered_json ]108        })109    return JsonResponse({110        "status": 405,111        "success" : False,112        "message": "method error",113        "data": None114    })115# def get_apply_answered_all_for_expert(request):116#     if request.method == "GET":117#         token_line = request.META.get('HTTP_AUTHORIZATION')118#         token = get_object_or_404(Token, key=token_line)119#         apply_all = Apply.objects.filter(expert_user=token.user)120    121#         apply_all_json = []122#         for apply in apply_all:123#             if apply.response > 0 or apply.interview.is_expired == 1:124#                 sender_profile = get_object_or_404(Profile, user=apply.interview.reporter_user)125#                 apply_json = {126#                     "id": apply.id,127#                     "department": sender_profile.department,128#                     "title": apply.interview.title,129#                     "response": apply.response,130#                     "is_expired": apply.interview.is_expired131#                 }132#                 apply_all_json.append(apply_json)133#         return JsonResponse({134#             'status' : 200,135#             'success' : True,136#             'message' : 'ì ë¬¸ê° ë§ì´íì´ì§ ì¸í°ë·° íí© ë¶ë¬ì¤ê¸° ì±ê³µ !',137#             'data' : apply_all_json138#         })139#     return JsonResponse({140#         "status": 405,141#         "success" : False,142#         "message": "method error",143#         "data": None144#     })145def get_apply_list_for_reporter(request):146    if request.method == "GET":147        token_line = request.META.get('HTTP_AUTHORIZATION')148        token = get_object_or_404(Token, key=token_line)149        interview_all = Interview.objects.filter(reporter_user=token.user)150        # apply_all = Apply.objects.filter(interview=token.user)151    152        request_json = []153        answered_json = []154        for interview in interview_all:155            sender_profile = get_object_or_404(Profile, user=interview.apply.expert_user)156            apply_json = {157                        "id": interview.id,158                        "name": sender_profile.name,159                        "title": interview.title,160                        "deadline": interview.deadline,161                        "is_send": interview.is_send,162                        "is_expired": interview.is_expired,163                        "response": interview.apply.response,164                    }165            if interview.apply.response == 0:166                if interview.is_expired == 0:167                    request_json.append(apply_json)168            elif interview.is_expired == 1 or interview.apply.response > 0:169                answered_json.append(apply_json)170        return JsonResponse({171            'status' : 200,172            'success' : True,173            'message' : 'ë¦¬í¬í° ë§ì´íì´ì§ ìì² ì¸í°ë·° ë¶ë¬ì¤ê¸° ì±ê³µ !',174            'data' : [ request_json, answered_json ]175        })176    return JsonResponse({177        "status": 405,178        "success" : False,179        "message": "method error",180        "data": None181    })182# def get_apply_answered_all_for_reporter(request):183#     #method_id == 1 -> ìë½ / == 2 -> ë³´ë¥ / == 3 -> ê±°ì  / == 4 -> ë§ë£184#     if request.method == "GET":185#         token_line = request.META.get('HTTP_AUTHORIZATION')186#         token = get_object_or_404(Token, key=token_line)187#         interview_all = Interview.objects.filter(reporter_user=token.user)188#         interview_all_json = []189    190#         #apply_all_json = []191#         for interview in interview_all:192#             if interview.is_expired == 1 or interview.apply.response > 0:193#                 sender_profile = get_object_or_404(Profile, user=interview.apply.expert_user)194#                 apply_json = {195#                     "id": interview.apply.id,196#                     "name": sender_profile.name,197#                     "title": interview.title,198#                     "response": interview.apply.response,199#                     "is_expired": interview.is_expired200                    201#                 }202#                 interview_all_json.append(apply_json)203#         return JsonResponse({204#             'status' : 200,205#             'success' : True,206#             'message' : 'ë¦¬í¬í° ë§ì´íì´ì§ ì¸í°ë·° íí© ë¶ë¬ì¤ê¸° ì±ê³µ !',207#             'data' : interview_all_json208#         })209#     return JsonResponse({210#         "status": 405,211#         "success" : False,212#         "message": "method error",213#         "data": None214#     })215# def get_apply_one_for_expert(request, id):216#     if request.method == "GET":217#         apply = get_object_or_404(Apply, pk=id)218#         interview = apply.interview219    220#         interview_json={221#                 "id" : interview.id,222#                 "title" : interview.title,223#                 "method" : interview.method,224#                 "body" : interview.body,225#                 "url" : interview.url,226#                 "deadline" : interview.deadline,227#                 "is_send" : interview.is_send,228#                 "is_expired" : interview.is_expired,229#             }230        231#         return JsonResponse({232#             'status' : 200,233#             'success' : True,234#             'message' : 'interview ìì  ì±ê³µ!',235#             'data' : interview_json236#         })237        238#     return JsonResponse({239#         'status' : 405,240#         'success' : False,241#         'message' : 'method error : get_interview',242#         'data' : None243#     })244#######################################################245#######################drf#############################246#######################################################247def logout(request):248    token_line = request.META.get('HTTP_AUTHORIZATION')249    token = get_object_or_404(Token, key=token_line)250    if token.user.is_authenticated:251        # logout(request)252        return JsonResponse({253            'status' : 200,254            'success' : True,255            'message' : 'ë¡ê·¸ìì ì±ê³µ!',256            257        })258    return JsonResponse({259        'status' : 400,260        'success' : False,261        'message' : 'ì´ë¯¸ ë¡ê·¸ìì ëììµëë¤!',262        263    })264class RegisterView(generics.CreateAPIView):265    queryset = User.objects.all()266    serializer_class = RegisterSerializer267class LoginView(generics.GenericAPIView):268    serializer_class = LoginSerializer269    def post(self, request):270        serializer = self.get_serializer(data=request.data)271        serializer.is_valid(raise_exception=True)272        token = serializer.validated_data273        login(request, token.user)274        return Response({"token": token.key, "profile_name":token.user.profile.name }, status=status.HTTP_200_OK)275class ProfileView(generics.GenericAPIView):276    serializer_class = ProfileSerializer277    def patch(self, request):278        profile = Profile.objects.get(user=request.user)279        serializer = self.get_serializer(data=request.data)280        serializer.is_valid(raise_exception=True)281        # data = serializer.validated_data282        # profile.nickname = data['name']283        # profile.position = data['department']284        # profile.subjects = data['background']285        # if request.data['image']:286        #     profile.image = request.data['image']287        profile.save()288        return Response({"result": "ok"},289                        status=status.HTTP_206_PARTIAL_CONTENT)290    def get(self, request):291        profile = Profile.objects.get(user=request.user)292        serializer = self.get_serializer(profile)293        return Response(serializer.data)...

test_reg.py

Source:test_reg.py

1from utils import *2event='Spring_1970'3event2='Fall_1923'4apply_json = {5    'event': event,6    'details': {7        'pizza_preference': 'pineapple'8    }9}10apply_json2 = {11    'event': event2,12    'details': {13        'pizza_preference': 'pineapple'14    }15}16# Test apply process17def test_apply(client, test_db, test_mail):18    # test that successful registration gives good response19    response = client.post('/api/accounts/create', json=create_json)20    assert response.status_code == 20021    response = client.post('/api/auth/login', json=login_json)22    assert response.status_code == 20023    token = response.json['access_token']24    with test_mail.record_messages() as outbox:25        response = client.post('/api/registrations/apply', json=apply_json, headers={'Authorization': 'Bearer ' + token})26        assert response.status_code == 20027        assert len(outbox) == 128        user = test_db.users.find_one({'username': login_json['username']})29        assert user['registrations'][0]['event'] == event30        assert user['registrations'][0]['details'] == apply_json['details']31        assert user['registrations'][0]['accept'] == False32        assert user['registrations'][0]['checkin'] == False33    # test that you can't apply again to the same event34    response = client.post('/api/registrations/apply', json=apply_json, headers={'Authorization': 'Bearer ' + token})35    assert response.status_code == 40036    # test that you can do a different one37    response = client.post('/api/registrations/apply', json=apply_json2, headers={'Authorization': 'Bearer ' + token})38    assert response.status_code == 20039def test_accept(client, test_db, test_mail):40    # Register three accounts but apply two accounts41    response = client.post('/api/accounts/create', json=create_json)42    response = client.post('/api/accounts/create', json=create_json2)43    response = client.post('/api/accounts/create', json=create_json3)44    response = client.post('/api/auth/login', json=login_json)45    token = response.json['access_token']46    response = client.post('/api/registrations/apply', json=apply_json, headers={'Authorization': 'Bearer ' + token})47    response = client.post('/api/auth/login', json=login_json2)48    token = response.json['access_token']49    response = client.post('/api/registrations/apply', json=apply_json, headers={'Authorization': 'Bearer ' + token})50    # Login as admin51    add_admin_account(client, test_db)52    response = client.post('/api/auth/login', json=admin_login_json)53    token = response.json['access_token']54    users = test_db.users.find({})55    ids = [str(user['_id']) for user in users]56    accept_json = {57        'event': event,58        'users': ids59    }60    # Accept both accounts61    with test_mail.record_messages() as outbox:62        response = client.post('/api/registrations/accept', json=accept_json, headers={'Authorization': 'Bearer ' + token})63        assert len(outbox) == 264    # check applied accounts are marked65    users = test_db.users.find({})66    for user in users:67        if (user['username'] != login_json3['username'] and user['username'] != 'admin'):68            assert user['registrations'][0]['accept']69def test_checkin(client, test_db):70    response = client.post('/api/accounts/create', json=create_json)71    response = client.post('/api/auth/login', json=login_json)72    token = response.json['access_token']73    response = client.post('/api/registrations/apply', json=apply_json, headers={'Authorization': 'Bearer ' + token})74    user = test_db.users.find_one({})75    user_id = str(user['_id'])76    add_admin_account(client, test_db)77    response = client.post('/api/auth/login', json=admin_login_json)78    token = response.json['access_token']79    checkin_json = {80        'event': event,81        'user': user_id82    }83    response = client.post('/api/registrations/check_in', json=checkin_json, headers={'Authorization': 'Bearer ' + token})84    assert response.status_code == 20085    user = test_db.users.find_one({})...

tickets.py

Source:tickets.py

1import datetime2from flask_restful import Resource3from flask_mail import Message4from flask import current_app5from enpi_srv.controllers.parsers import ApplyTicketFormParser6from enpi_srv import mail7class MailApplyFormContentMixin(object):8    def get_content(self, **kwargs):9        text = '''10        Hi All,11           æ´»å¨æ¶é´ï¼{0}12           é¢ç®ï¼   {1}å13           äººæ°:    {2}äºº14           åºå°é¢ç§¯: {3}å¹³ç±³15           åå¸åºå: {4}16           è¡æ¿åºï¼  {5}17           å§å:    {6}18           çµè¯:    {7}   19        Best Regards20        ä¸æµ·æ©æ¹å¹¿å21        '''.format(22            kwargs.get('start_dt'),23            kwargs.get('budget'),24            kwargs.get('people'),25            kwargs.get('area'),26            kwargs.get('city'),27            kwargs.get('city_area'),28            kwargs.get('cn_name'),29            kwargs.get('phone')30        )31        return text32    def get_subject(self):33        dt_str = datetime.datetime.now().strftime('%Y-%m-%d')34        return "[åå¸éæ±]æ¨ææ°çåå¸éæ±ï¼è¯·æ¥æ¶{}".format(dt_str)35class TicketFormView(36    MailApplyFormContentMixin,37    ApplyTicketFormParser,38    Resource):39    """40    POST http://127.0.0.1:5000/core/apply apply_json:='{'start_dt':'2019-3-27', 'budget':200000, 'people':20, 'area':'ä¸æµ·å¸', 'city_area':'è¹å£åº', 'cn_name':'é¡¾é²å°', 'phone':13999999999}'41    {42    'start_dt':xxx    # æ´»å¨æ¶é´43    'budget':xxx      # é¢ç®44    'people':xxx      # äººæ°45    'area':xx         # åºå°é¢ç§¯46    'city':xxx        # åå¸åºå47    'city_area':xx    # è¡æ¿åº48    'cn_name':xx      # å§å49    'phone':xxx       # çµè¯50    """51    def post(self):52        request_data = self.post_ticket_form.parse_args()53        apply_json = request_data.get('apply_json')54        msg = Message(55            body=self.get_content(**apply_json),56            recipients=current_app.config['MAIL_RECIPIENTS'],57            subject=self.get_subject()58        )59        mail.send(msg)60        return apply_json61class SMSSecurityCode(Resource):62    def get(self):...

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.