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

How to use get_auth_string method in localstack

Best Python code snippet using localstack_python

mngmt.py

Source:mngmt.py

...64            }65            cluster_req.append(fpgaNode)66        r1 = requests.post(67            "http://" + __cf_manager_url__ + "/clusters?{0}&dont_verify_memory=0".format(68                user.get_auth_string(with_project=True)),69            json=cluster_req, timeout=__POST_CLUSTER_TIMEOUT__)70        elapsed = time.time() - start71        if r1.status_code != 200:72            # something went wrong73            return errorReqExit("POST cluster", r1.status_code)74        cluster_data = json.loads(r1.text)75        print("Id of new cluster: {}".format(cluster_data['cluster_id']))76        print("Time for POST cluster: \t{0}s\n".format(elapsed))77        new_cluster = cFcluster(user, cluster_data)78        return new_cluster79    except requests.exceptions.Timeout as e:80        # Maybe set up for a retry81        print(e)82        print("ERROR: Something went wrong with post_cluster request and it reached timeout="+str(__POST_CLUSTER_TIMEOUT__)+". Maybe retry or increase timeout value.\n")83        sys.exit(1)84def get_cluster_data(cluster: cFcluster):85    print("Requesting cluster data for cluster_id={0} ...".format(cluster.get_id())) # FIXME: .get_id()86    try:87        start = time.time()88        r1 = requests.get(89            "http://" + __cf_manager_url__ + "/clusters/" + str(cluster.get_id()) + "?{0}".format(90                cluster.user.get_auth_string()), timeout=__GET_CLUSTER_TIMEOUT__)91        elapsed = time.time() - start92        print("Time for GET cluster: \t{0}s\n".format(elapsed))93        if r1.status_code != 200:94            # something went horrible wrong95            return errorReqExit("GET cluster", r1.status_code)96        cluster_data = json.loads(r1.text)97        # update, if necessary98        cluster.cluster_data = cluster_data99        return cluster_data100    except requests.exceptions.Timeout as e:101        # Maybe set up for a retry102        print(e)103        print("ERROR: Something went wrong with get_cluster_data request and it reached timeout="+str(__GET_CLUSTER_TIMEOUT__)+". Maybe retry or increase timeout value.\n")104def get_clusters_data(user: cFuser, limit=100):105    print("Requesting clusters data (limit="+str(limit)+")...")106    try:107        start = time.time()108        r1 = requests.get(109            "http://" + __cf_manager_url__ + "/clusters" + "?{0}&limit={1}".format(110                user.get_auth_string(), limit), timeout=__GET_CLUSTER_TIMEOUT__)111        elapsed = time.time() - start112        print(r1.request.url)113        print("Time for GET clusters: \t{0}s\n".format(elapsed))114        if r1.status_code != 200:115            # something went horrible wrong116            return errorReqExit("GET clusters", r1.status_code)117        clusters_data = json.loads(r1.text)118        return clusters_data119    except requests.exceptions.Timeout as e:120        # Maybe set up for a retry121        print(e)122        print("ERROR: Something went wrong with get_cluster request and it reached timeout="+str(__GET_CLUSTER_TIMEOUT__)+". Maybe retry or increase timeout value.\n")123def delete_cluster_data(cluster: cFcluster):124    print("Requesting delete cluster_id={0} ...".format(cluster.get_id()))125    try:126        start = time.time()127        r1 = requests.delete(128            "http://" + __cf_manager_url__ + "/clusters/" + str(cluster.get_id()) + "?{0}".format(129                cluster.user.get_auth_string()), timeout=__DELETE_CLUSTER_TIMEOUT__)130        elapsed = time.time() - start131        print("Time for DELETE cluster: \t{0}s\n".format(elapsed))132        if r1.status_code != 204:133            # something went horrible wrong134            return errorReqExit("DELETE cluster", r1.status_code)135        return 0136    except requests.exceptions.Timeout as e:137        # Maybe set up for a retry138        print(e)139        print("ERROR: Something went wrong with delete_cluster_data request and it reached timeout="+str(__DELETE_CLUSTER_TIMEOUT__)+". Maybe retry or increase timeout value.\n")140def restart_cluster_apps(cluster: cFcluster):141    print("Requesting restart for (all) FPGA(s) of cluster_id={0} ...".format(cluster.get_id()))142    try:143        start = time.time()        144        r1 = requests.patch(145            "http://" + __cf_manager_url__ + "/clusters/" + str(cluster.get_id()) + "/restart?{0}".format(146            cluster.user.get_auth_string()))147        elapsed = time.time() - start148        print("Time for RESTART cluster: \t{0}s\n".format(elapsed))149        if r1.status_code != 200:150            # something went horrible wrong151            return errorReqExit("PATCH cluster restart", r1.status_code)152        print(r1.content.decode())153    except Exception as e:154        print("ERROR: Failed to reset the FPGA(s) role(s)")155        print(str(e))156        exit(1)157####################################################################################################158# Instances functions159####################################################################################################160class cFinstance:161    def __init__(self, user: cFuser, instance_data):162        self.instance_data = instance_data163        self.user = user164        self.id = instance_data['instance_id']165    def get_id(self):166        return self.id167    168def get_instances_data(user: cFuser, limit=100):169    print("Requesting instances data (limit="+str(limit)+")...")170    try:171        start = time.time()172        r1 = requests.get(173            "http://" + __cf_manager_url__ + "/instances" + "?{0}&limit={1}".format(174                user.get_auth_string(), limit), timeout=__GET_INSTANCE_TIMEOUT__)175        elapsed = time.time() - start176        print(r1.request.url)177        print("Time for GET instances: \t{0}s\n".format(elapsed))178        if r1.status_code != 200:179            # something went horrible wrong180            return errorReqExit("GET instances", r1.status_code)181        instances_data = json.loads(r1.text)182        return instances_data183    except requests.exceptions.Timeout as e:184        # Maybe set up for a retry185        print(e)186        print("ERROR: Something went wrong with get_instances request and it reached timeout="+str(__GET_INSTANCE_TIMEOUT__)+". Maybe retry or increase timeout value.\n")187def create_instance():188    print("TODO \n")189def get_instance_data(instance: cFinstance):190    print("Requesting instance data for instance_id={0} ...".format(instance.get_id()))191    try:192        start = time.time()193        r1 = requests.get(194            "http://" + __cf_manager_url__ + "/instances/" + str(instance.get_id()) + "?{0}".format(195                instance.user.get_auth_string()), timeout=__GET_INSTANCE_TIMEOUT__)196        elapsed = time.time() - start197        print("Time for GET instance: \t{0}s\n".format(elapsed))198        if r1.status_code != 200:199            # something went horrible wrong200            return errorReqExit("GET instance", r1.status_code)201        instance_data = json.loads(r1.text)202        # update, if necessary203        instance.instance_data = instance_data204        return instance_data205    except requests.exceptions.Timeout as e:206        # Maybe set up for a retry207        print(e)208        print("ERROR: Something went wrong with get_instance_data request and it reached timeout="+str(__GET_INSTANCE_TIMEOUT__)+". Maybe retry or increase timeout value.\n")209def reprogram_instance():210    print("TODO \n")211def api_request_instance():212    print("TODO \n")213def restart_instance_app(instance: cFinstance):214    print("Requesting restart for instance_id={0} ...".format(instance.get_id()))215    try:216        start = time.time()       217        r1 = requests.patch(218            "http://" + __cf_manager_url__ + "/instances/" + str(instance.get_id()) + "/app_restart?{0}".format(219            instance.user.get_auth_string()))220        elapsed = time.time() - start221        print("Time for RESTART instance: \t{0}s\n".format(elapsed))222        if r1.status_code != 200:223            # something went horrible wrong224            return errorReqExit("PATCH instance restart", r1.status_code)225        print(r1.content.decode())226    except Exception as e:227        print("ERROR: Failed to reset the FPGA role")228        print(str(e))229        exit(1)230def delete_instance(instance: cFinstance):231    print("deleting instance {}".format(instance.id))232    r1 = requests.delete(233        "http://" + __cf_manager_url__ + "/instances/{0}?{1}".format(instance.id, instance.user.get_auth_string()))234    if r1.status_code > 204:235        # error codes236        # 204 Instance was deleted237        # 401 Unauthenticated, bad login238        # 403 Unauthorized239        # 404 Instance does not exist240        return r1.status_code241    else:242        print("Instance {} removed".format(instance.id))243    instance_data = r1.status_code244    return instance_data245####################################################################################################246# Images functions247####################################################################################################248class cFimage:249    def __init__(self, user: cFuser, image_data):250        self.user = user251        self.image_data = image_data252        self.id = image_data['id']253        self.comment = image_data['comment']254        self.required_shell = image_data['shell_type']255def get_images(user: cFuser):256    print("TODO \n")257def get_image(image: cFimage):258    print("TODO \n")259def post_image(image: cFimage):260    print("TODO \n")261def delete_image(image: cFimage):262    print("TODO \n")263####################################################################################################264# Resources functions (admin only)265####################################################################################################266# TODO: no resource class so far, because the resource data should reside inside the CFRM, not the cFSP267def get_resource_status(resource_id, admin: cFuser):268    print("Requesting resource status...")269    r1 = requests.get(270        "http://" + __cf_manager_url__ + "/resources/" + str(271            resource_id) + "/status/" + "?{0}".format(admin.get_auth_string()))272    if r1.status_code != 200:273        # something went wrong274        return errorReqExit("GET resource status", r1.status_code)275    resource_status = json.loads(r1.text)276    return resource_status277def set_resource_status(resource_id, new_status, admin: cFuser):278    # print("set resource status")279    # possible status: "AVAILABLE", "USED", "MAINTENANCE"280    r1 = requests.put(281        "http://" + __cf_manager_url__ + "/resources/{0}/status/?{1}&new_status={2}".format(282            resource_id, admin.get_auth_string(), new_status))283    if r1.status_code != 204:284        # something went wrong285        return errorReqExit("PUT /resources/{resource_id}/status/", r1.status_code)286    else:287        print("Resource {} set to {}".format(resource_id, new_status))288    resource_data = r1.status_code...

test_unc_credentials.py

Source:test_unc_credentials.py

...17        self.assertEqual(creds.get_password(), 'pass')18    def test_invalid(self):19        self.assertRaises(InvalidUsernameError, UncCredentials, '"user"')20        self.assertRaises(InvalidUsernameError, UncCredentials, '>user')21    def test_get_auth_string(self):22        self.assertEqual(UncCredentials(None, None).get_auth_string(), '')23        self.assertEqual(UncCredentials(None, '').get_auth_string(), ':')24        self.assertEqual(UncCredentials('user', None).get_auth_string(), 'user')25        self.assertEqual(UncCredentials('user', '').get_auth_string(), 'user:')26        self.assertEqual(UncCredentials(None, 'pass').get_auth_string(), ':pass')27        self.assertEqual(UncCredentials('user', ':').get_auth_string(), 'user::')28        self.assertEqual(UncCredentials('user', 'pass').get_auth_string(), 'user:pass')29    def test_eq(self):30        self.assertEqual(UncCredentials(), UncCredentials())31        self.assertEqual(UncCredentials('user', None), UncCredentials('user', None))32        self.assertEqual(UncCredentials(None, 'pass'), UncCredentials(None, 'pass'))33        self.assertEqual(UncCredentials('user', 'pass'), UncCredentials('user', 'pass'))34    def test_ne(self):35        self.assertNotEqual(UncCredentials(), UncCredentials('user'))36        self.assertNotEqual(UncCredentials('user'), UncCredentials('USER'))37        self.assertNotEqual(UncCredentials(None, 'pass'), UncCredentials(None, 'USER'))38        self.assertNotEqual(UncCredentials(), 'somestring')39        self.assertNotEqual(UncCredentials(), 10)40class TestUncParsingFunctions(TestCase):41    def test_get_creds_from_string(self):42        self.assertEqual(get_creds_from_string(''), UncCredentials(None, None))...

test_edge.py

Source:test_edge.py

1import unittest2from werkzeug.datastructures import Headers3from localstack.services.edge import get_auth_string4class EdgeServiceTest(unittest.TestCase):5    def test_get_auth_string(self):6        # Typical Header with Authorization7        headers_with_auth = Headers(8            [9                ("X-Amz-Date", "20210313T160953Z"),10                (11                    "Authorization",12                    (13                        "AWS4-HMAC-SHA256 Credential="14                        "test/20210313/us-east-1/sqs/aws4_request, "15                        "SignedHeaders=content-type;host;x-amz-date, "16                        "Signature="17                        "3cba88ae6cbb8036126d2ba18ba8ded5"18                        "eea9e5484d70822affce9dad03be5993"19                    ),20                ),21            ]22        )23        body_with_auth = (24            b"X-Amz-Algorithm=AWS4-HMAC-SHA256&"25            + b"X-Amz-Credential="26            + b"test%2F20210313%2Fus-east-1%2Fsqs%2Faws4_request&"27            + b"X-Amz-Date=20210313T011059Z&"28            + b"X-Amz-Expires=86400000&"29            + b"X-Amz-SignedHeaders=content-type%3Bhost%3Bx-amz-date&"30            + b"X-Amz-Signature="31            + b"3cba88ae6cbb8036126d2ba18ba8ded5eea9e5484d70822affce9dad03be5993"32        )33        # check getting auth string from header with Authorization header34        self.assertEqual(35            headers_with_auth.get("authorization"),36            get_auth_string("POST", "/", headers_with_auth, b""),37        )38        # check getting auth string from body with authorization params39        self.assertEqual(40            headers_with_auth.get("authorization"),41            get_auth_string("POST", "/", Headers(), body_with_auth),...

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.