How to use response_content method in Httmock

Best Python code snippet using httmock_python

test_views.py

Source:test_views.py

1from django.test import TestCase, Client2from rest_framework.test import APIClient3from django.urls import reverse4from django.contrib.auth.models import User5from api.models import Company, Review6import json7class TestViews(TestCase):8    def setUp(self):9        self.api_client = APIClient()10        u = User.objects.create_user('MaddieKan', 'dr.almanovamadina@gmail.com', 'MaddieKan')11        a = User.objects.create_user('admin', 'admin@gmail.com', 'admin')12        a.is_staff = 113        a.save()14        c = Company.objects.create(name='1Fit', description='Gym subscriptions',15                               address='Kazakhstan, Almaty, Makataev st. 117')16        Review.objects.create(rating=4, title='First Review', summary='Summary', ip=2130706433, company=c, reviewer=u)17        self.login('MaddieKan', 'MaddieKan')18    def login(self, login, password):19        response = self.api_client.post(reverse('login'), {"username": login, "password": password})20        response_content = json.loads(response.content.decode('utf-8'))21        self.token = response_content["token"]22        self.api_client.credentials(HTTP_AUTHORIZATION='JWT ' + self.token)23    def get_request(self, url):24        response = self.api_client.get(url, format='json')25        response_content = json.loads(response.content.decode('utf-8'))26        return response, response_content27    def post_request(self, url, data):28        response = self.api_client.post(url, data, format='json')29        response_content = json.loads(response.content.decode('utf-8'))30        return response, response_content31    def put_request(self, url, data):32        response = self.api_client.put(url, data, format='json')33        response_content = json.loads(response.content.decode('utf-8'))34        return response, response_content35    def delete_request(self, url):36        response = self.api_client.delete(url, format='json')37        response_content = json.loads(response.content.decode('utf-8'))38        return response, response_content39    def test_sign_up_POST(self):40        url = reverse('sign_up')41        data = {"username": "test", "email": "test@gmail.com", "password": "test", "first_name": "test",42                "last_name": "test", "is_staff": 0}43        response = self.api_client.post(url, data, format='json')44        self.assertEquals(response.status_code, 201)45    def test_companies_list_GET(self):46        response, response_content = self.get_request(reverse('companies'))47        data = [{'id': 1, 'name': '1Fit', 'description': 'Gym subscriptions',48                 'address': 'Kazakhstan, Almaty, Makataev st. 117'}]49        self.assertEquals(response.status_code, 200)50        self.assertEquals(response_content, data)51    def test_companies_list_POST(self):52        data = {"name": "Codebusters", "description": "desc", "address": "Kazakhstan, Almaty, Makataev st. 117"}53        response, response_content = self.post_request(reverse('companies'), data)54        res_data = {"id": 2, "name": "Codebusters", "description": "desc",55                    "address": "Kazakhstan, Almaty, Makataev st. 117"}56        self.assertEquals(response.status_code, 201)57        self.assertEquals(res_data, response_content)58    def test_company_details_GET(self):59        response, response_content = self.get_request(reverse('company_details', args=[1]))60        data = {'id': 1, 'name': '1Fit', 'description': 'Gym subscriptions',61                 'address': 'Kazakhstan, Almaty, Makataev st. 117'}62        self.assertEquals(response.status_code, 200)63        self.assertEquals(response_content, data)64    def test_company_details_PUT(self):65        data = {"name": "1fit", "description": "desc", "address": "Kazakhstan, Almaty, Makataev st. 117"}66        response, response_content = self.put_request(reverse('company_details', args=[1]), data)67        res_data = {"id": 1, "name": "1fit", "description": "desc", "address": "Kazakhstan, Almaty, Makataev st. 117"}68        self.assertEquals(response.status_code, 200)69        self.assertEquals(response_content, res_data)70    def test_company_details_DELETE(self):71        response, response_content = self.delete_request(reverse('company_details', args=[1]))72        data = {'deleted': True}73        self.assertEquals(response.status_code, 200)74        self.assertEquals(response_content, data)75    def test_reviews_list_not_admin_GET(self):76        response, response_content = self.get_request(reverse('reviews_full_list'))77        data = {'details': 'permission denied'}78        self.assertEquals(response.status_code, 403)79        self.assertEquals(response_content, data)80    def test_reviews_list_admin_GET(self):81        self.login('admin', 'admin')82        response, response_content = self.get_request(reverse('reviews_full_list'))83        self.assertEquals(response.status_code, 200)84        self.assertEquals(response_content[0]['title'], 'First Review')85    def test_reviews_list_POST(self):86        data = {"rating": 5, "title": "Second Review", "summary": "Summary", "ip": "192.168.0.1", "company": 1,87                "reviewer": 1}88        response, response_content = self.post_request(reverse('reviews_full_list'), data)89        self.assertEquals(response.status_code, 201)90        self.assertEquals(response_content['title'], 'Second Review')91    def test_review_details_GET(self):92        response, response_content = self.get_request(reverse('review_details', args=[1]))93        self.assertEquals(response.status_code, 200)94        self.assertEquals(response_content['title'], 'First Review')95    def test_review_details_PUT(self):96        data = {"rating": 5, "title": "Edited First Review", "summary": "Summary", "company": 1}97        response, response_content = self.put_request(reverse('review_details', args=[1]), data)98        self.assertEquals(response.status_code, 200)99        self.assertEquals(response_content['title'], 'Edited First Review')100    def test_review_details_DELETE(self):101        response, response_content = self.delete_request(reverse('review_details', args=[1]))102        data = {'deleted': True}103        self.assertEquals(response.status_code, 200)104        self.assertEquals(response_content, data)105    def test_user_reviews_GET(self):106        response, response_content = self.get_request(reverse('user_reviews'))107        self.assertEquals(response.status_code, 200)...

get.py

Source:get.py

1import json2from django.http import JsonResponse3from dsproj_app.api_functions.get_val_and_payload import val_and_payload4from os import environ5from dsproj_app.views import get_array_views6from hashlib import sha17import random8import requests9UP_TO_DATE_PAYLOAD = {}10def get_handling(request, details, key):11    store = details["store"]12    latest_timestamp = details["latest_timestamp"]13    curr_node_vc = details["clock"]14    shards = details["shards"]15    response_content = {}16    payload_json = val_and_payload(request.body)["payload_json"]17    # OPTION: NO KEY PROVIDED18    if key == None:19        response_content = {20            "result": "Error",21            "msg": "Not provided",22            "error": "Key does not exist",23        }24        return JsonResponse(response_content, status=422)25    if store.is_exists(key):26        value = store.get_item(key)["val"]27        if value == "this-is-invalid":28            response_content = {29                "result": "Error",30                "msg": "Unable to access key: " + key,31                "payload": payload_json,32            }33            return JsonResponse(response_content, status=400)34    # OPTION: SEARCH-GET35    if "/keyValue-store/search" in request.path:36        response_content["result"] = "Success"37        response_content["payload"] = payload_json38        if store.is_exists(key):39            response_content["isExists"] = True40        else:41            response_content["isExists"] = False42        return JsonResponse(response_content, status=200)43    # OPTION: EMPTY PAYLOAD (USER REQUEST), HASH KEY AND CHECK SHARD DIRECTORY44    if not payload_json:45        payload_json = {46            "vc": curr_node_vc.get_vc(),47            "pos": get_array_views().index(environ.get("IP_PORT")),48            "tstamp": latest_timestamp.get_timestamp(),49            "causal_context": details["causal_context"],50        }51        details["causal_context"] = None52        data = "payload=" + json.dumps(payload_json)53        if store.is_exists(key):54            response_content = {55                "result": "Success",56                "val": store.get_item(key)["val"],57                "payload": payload_json,58                "owner": shards.get_my_shard()59            }60            status = 20061        else:62            response_content = {63                "result": "Error",64                "msg": "Key does not exist",65                "payload": payload_json,66            }67            status = 40468        shard_location = None69        binary_key = sha1(key.encode())70        shard_location = int(binary_key.hexdigest(), 16) % shards.get_shard_size()71        # OPTION: WE'RE IN THE WRONG SHARD, REDIRECT REQUEST TO NODE WITH CORRECT SHARD72        current_node_not_in_shard = shards.get_members_in_ID(shard_location) is not None and not (73            environ.get("IP_PORT") in shards.get_members_in_ID(shard_location)74        )75        if current_node_not_in_shard:76            members = shards.get_members_in_ID(shard_location)77            if members != None and len(members) > 0:78                rand_address = random.choice(members)79                data = "payload=" + json.dumps(payload_json)80                response = requests.get(81                    "http://" + rand_address + "/keyValue-store/" + key, data=data82                )83                response_content = response.json()84                if response.status_code == 404:85                    response_content = {86                        "result": "Error",87                        "msg": "Key does not exist",88                        "payload": payload_json,89                    }90                else:91                    del response_content["owner"]92                return JsonResponse(response_content, status=response.status_code)93            else:94                response_content = {95                    "result": "Error",96                    "msg": "No nodes in shard " + shard_location,97                    "payload": payload_json,98                }99                status = 400100                return JsonResponse(response_content, status=status)101        else:102            return JsonResponse(response_content, status=status)103    # OPTION: NON-EMPTY PAYLOAD (NODES COMMUNICATING). WE'RE IN THE104    # RIGHT CONTAINER, JUST DO NORMAL GET105    else:106        views = get_array_views()107        # CONDITION: If store already has the key108        if store.is_exists(key):109            details["causal_context"] = {"key": key, "vc": curr_node_vc.get_vc()}110            response_content["result"] = "Success"111            response_content["val"] = store.get_item(key)["val"]112            # payload_json['value'] = store.get_item(key)['val']113            # CONDITION: If causal context exists and requestor's vc > this vc114            # if cc_key != None and curr_node_vc.greater_than_or_equal(payload_json["vc"]) is False:115            #     store.add(cc_key, "too_old", None)116            #     UP_TO_DATE_PAYLOAD = payload_json117            # CONDITION: if reading an old value, return Payload out of date,118            # remove val from response_content, and set status to 400119            if store.get_item(key)["val"] == "too_old":120                del response_content["val"]121                response_content = {122                    "msg": "Payload out of date",123                    "status": 400,124                    "payload": UP_TO_DATE_PAYLOAD,125                }126                status = 400127            status = 200128            payload_json = {129                "vc": curr_node_vc.get_vc(),130                "pos": views.index(environ.get("IP_PORT")),131                "tstamp": latest_timestamp.get_timestamp(),132                "causal_context": details["causal_context"],133            }134        else:135            payload_json = {136                "value": None,137                "vc": curr_node_vc.get_vc(),138                "pos": views.index(environ.get("IP_PORT")),139                "tstamp": latest_timestamp.get_timestamp(),140                "causal_context": details["causal_context"],141            }142            response_content["payload"] = payload_json143            status = 404144        response_content["owner"] = shards.get_my_shard()145        response_content["payload"] = payload_json...

put.py

Source:put.py

1from dsproj_app.api_functions.get_val_and_payload import val_and_payload2from django.http import JsonResponse3from sys import getsizeof4from os import environ5from dsproj_app.views import get_array_views6from hashlib import sha17import json8import re9from urllib.parse import parse_qs10import time11import requests12import random13def put_handling(request, details, key):14    store = details["store"]15    latest_timestamp = details["latest_timestamp"]16    curr_node_vc = details["clock"]17    shards = details["shards"]18    response_content = {}19    # OPTION: VALUE MISSING20    if len(request.body) <= 0:21        response_content["msg"] = "Error"22        response_content["result"] = "Error"23        response_content["error"] = "Value is missing"24        return JsonResponse(response_content, status=422)25    # OPTION: KEY LENGTH INVALID26    if 0 < len(key) > 200:27        response_content["msg"] = "Error"28        response_content["result"] = "Error"29        response_content["error"] = "Key not valid"30        return JsonResponse(response_content, status=422)31    body_unicode = request.body.decode("utf-8")32    body = parse_qs(body_unicode)33    payload_json = None34    if "rebalance" not in body:35        payload_json = val_and_payload(request.body)["payload_json"]36    else:37        print("HIII")38    val = val_and_payload(request.body)["val"]39    # OPTION: VALUE SIZE TOO BIG40    if getsizeof(val) > 1024:41        response_content["result"] = "Error"42        response_content["msg"] = "Object too large. Size limit is 1MB"43        response_content["error"] = "Key not valid"44        return JsonResponse(response_content, status=422)45    shard_location = None46    print("Before: ", payload_json)47    # OPTION: NON-EMPTY PAYLOAD (NODES COMMUNICATING)48    if payload_json:49        req_vc = payload_json["vc"]50        req_timestamp = payload_json["tstamp"]51        if (52            "latest_timestamp" in payload_json53            and latest_timestamp.get_timestamp() == None54        ):55            latest_timestamp.set_timestamp(payload_json["latest_timestamp"])56        else:57            lt = latest_timestamp.max_timestamp(req_timestamp)58            latest_timestamp.set_timestamp(lt)59        req_position = int(payload_json["pos"])60    # OPTION: EMPTY PAYLOAD (USER REQUEST)61    else:62        views = get_array_views()63        req_vc = curr_node_vc.get_vc()64        req_position = views.index(environ.get("IP_PORT"))65        req_timestamp = time.time()66        if latest_timestamp.get_timestamp() == None:67            latest_timestamp.set_timestamp(req_timestamp)68            payload_json["latest_timestamp"] = latest_timestamp.get_timestamp()69        payload_json = {70            "pos": req_position,71            "tstamp": req_timestamp,72            "causal_context": details["causal_context"],73            "vc": req_vc,74        }75        details["causal_context"] = None76        binary_key = sha1(key.encode())77        shard_location = int(binary_key.hexdigest(), 16) % shards.get_shard_size()78    # OPTION: KEY NEVER EXISTED79    if not store.has_key(key):80        response_content["replaced"] = False81        response_content["msg"] = "Added successfully"82        response_content["payload"] = payload_json83        status = 20184    # OPTION: KEY ALREADY EXISTS AND IS BEING REPLACED85    elif store.has_key(key):86        response_content["replaced"] = True87        response_content["msg"] = "Updated successfully"88        response_content["payload"] = payload_json89        status = 20090    members = shards.get_members_in_ID(shard_location)91    # if in right shard92    if shard_location != None:93        if members != None:94            rand_address = random.choice(members)95            if "rebalance" in body:96                store.add(key, val, {})97            data = "val=" + val + "&&payload=" + json.dumps(payload_json)98            response = requests.put(99                "http://" + rand_address + "/keyValue-store/" + key, data=data100            )101            return JsonResponse(response.json(), status=response.status_code)102        else:103            response_content = {104                "result": "Error",105                "msg": "No nodes in shard " + str(shard_location),106                "payload": payload_json,107            }108            status = 400109            return JsonResponse(response_content, status=status)110    else:111        if "rebalance" not in body:112            curr_node_vc.increment_self()113            payload_json["vc"] = curr_node_vc.get_vc()114        store.add(key, val, payload_json["causal_context"])115        print("I AM ADDING KEY")116        response_content["owner"] = shards.get_my_shard()117        return JsonResponse(response_content, status=status)...

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.