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

How to use validate_api_key method in localstack

Best Python code snippet using localstack_python

sparepartitem.py

Source:sparepartitem.py

...26        """27        Returns a list of all spare part item childs.28        """29        api_key = request.headers.get('apikey')30        if validate_api_key(api_key).status_code != 200:31            return validate_api_key(api_key)32        sparepartitem_all_entries = SparepartitemModel.query.order_by(SparepartitemModel.datetime_display.desc()).all()33        sparepartitem_entry_list = []34        for sparepartitem_entry in sparepartitem_all_entries:35            sparepartitem_data = sparepartitem_schema.dump(sparepartitem_entry)36            sparepartitem_data['sparepart'] = sparepart_schema.dump(sparepartitem_entry.sparepart)37            sparepartitem_entry_list.append(sparepartitem_data)38        response = jsonify(sparepartitem_entry_list)39        response.status_code = 20040        return response41    @api.doc(security='apikey')42    @api.expect(sparepartitem_input_parameters)43    @api.response(201, 'Spare part item successfully added.')44    def post(self):45        """46        Creates a new spare part item child.47        """48        api_key = request.headers.get('apikey')49        if validate_api_key(api_key).status_code != 200:50            return validate_api_key(api_key)51        inserted_data = request.get_json()52        new_sparepartitem = SparepartitemModel(53            sparepart_id=inserted_data.get('sparepart_id'),54            condition=inserted_data.get('condition'),55            description=inserted_data.get('description'),56            stock=inserted_data.get('stock'),57        )58        db.session.add(new_sparepartitem)59        db.session.commit()60        response = jsonify(new_sparepartitem.sparepartitem_id)61        response.status_code = 20162        return response63@ns.route('/<string:id_>')64@api.response(404, 'Spare part item not found.')65class SparepartitemItem(Resource):66    @api.doc(security='apikey')67    @api.response(200, f"Spare part item with requested id successfully fetched.")68    def get(self, id_: str):69        """70        Returns a spare part item child.71        """72        api_key = request.headers.get('apikey')73        if validate_api_key(api_key).status_code != 200:74            return validate_api_key(api_key)75        sparepart_item = SparepartitemModel.query.filter(SparepartitemModel.sparepartitem_id == id_).one()76        response = jsonify(sparepartitem_schema.dump(sparepart_item))77        response.status_code = 20078        return response79    @api.doc(security='apikey')80    @api.expect(sparepartitem_input_parameters)81    @api.response(204, f"Spare part item with requested id successfully updated.")82    def put(self, id_: str):83        """84        Updates a spare part item child.85        """86        api_key = request.headers.get('apikey')87        if validate_api_key(api_key).status_code != 200:88            return validate_api_key(api_key)89        inserted_data = request.get_json()90        sparepart_item = SparepartitemModel.query.filter(SparepartitemModel.sparepartitem_id == id_).one()91        if inserted_data.get('condition', 'ParameterNotInPayload') != 'ParameterNotInPayload':92            sparepart_item.condition = inserted_data.get('condition')93        if inserted_data.get('description', 'ParameterNotInPayload') != 'ParameterNotInPayload':94            sparepart_item.description = inserted_data.get('description')95        if inserted_data.get('stock', 'ParameterNotInPayload') != 'ParameterNotInPayload':96            sparepart_item.stock = inserted_data.get('stock')97        db.session.add(sparepart_item)98        db.session.commit()99        return None, 204100    @api.doc(security='apikey')101    @api.response(204, f"Spare part item with requested id successfully deleted.")102    def delete(self, id_: str):103        """104        Deletes a spare part item child.105        """106        api_key = request.headers.get('apikey')107        if validate_api_key(api_key).status_code != 200:108            return validate_api_key(api_key)109        sparepart_item = SparepartitemModel.query.filter(SparepartitemModel.sparepartitem_id == id_).one()110        db.session.delete(sparepart_item)111        db.session.commit()...

databases.py

Source:databases.py

1from fastapi import APIRouter, Request, status, Depends2from fastapi.openapi.models import APIKey3from fastapi.responses import JSONResponse4from commands import commands5from router.key import validate_api_key6# Defining our API router7def get_router(app):8    # Create a FastAPI router9    router = APIRouter()10    # List all databases11    @router.get("/databases/{plugin_name}", response_description="List all databases")12    async def list_databases(request: Request, plugin_name: str, api_key: APIKey = Depends(validate_api_key)):13        success, message = commands.list_databases(plugin_name)14        content = {"success": success, "message": message}15        return JSONResponse(status_code=status.HTTP_200_OK, content=content)16    # Check if a database exists17    @router.get("/databases/{plugin_name}/{database_name}", response_description="Check if a database exists")18    async def database_exists(request: Request, plugin_name: str, database_name: str,19                              api_key: APIKey = Depends(validate_api_key)):20        success, message = commands.database_exists(plugin_name, database_name)21        content = {"success": success, "message": message}22        return JSONResponse(status_code=status.HTTP_200_OK, content=content)23    # List linked apps24    @router.get("/databases/links/{plugin_name}/{database_name}", response_description="List linked apps")25    async def database_linked_apps(request: Request, plugin_name: str, database_name: str,26                                   api_key: APIKey = Depends(validate_api_key)):27        success, message = commands.database_linked_apps(plugin_name, database_name)28        content = {"success": success, "message": message}29        return JSONResponse(status_code=status.HTTP_200_OK, content=content)30    # Link a database to an app31    @router.post("/databases/links/{plugin_name}/{database_name}/{app_name}",32                 response_description="Link a database to an app")33    async def link_database(request: Request, plugin_name: str, database_name: str, app_name: str,34                            api_key: APIKey = Depends(validate_api_key)):35        success, message = commands.link_database(plugin_name, database_name, app_name)36        content = {"success": success, "message": message}37        return JSONResponse(status_code=status.HTTP_200_OK, content=content)38    # Unlink a database from an app39    @router.delete("/databases/links/{plugin_name}/{database_name}/{app_name}",40                   response_description="Unlink a database from an app")41    async def unlink_database(request: Request, plugin_name: str, database_name: str, app_name: str,42                              api_key: APIKey = Depends(validate_api_key)):43        success, message = commands.unlink_database(plugin_name, database_name, app_name)44        content = {"success": success, "message": message}45        return JSONResponse(status_code=status.HTTP_200_OK, content=content)46    # Create a database47    @router.post("/databases/{plugin_name}/{database_name}", response_description="Create a database")48    async def create_database(request: Request, plugin_name: str, database_name: str,  api_key: APIKey = Depends(validate_api_key)):49        success, message = commands.create_database(plugin_name, database_name)50        content = {"success": success, "message": message}51        return JSONResponse(status_code=status.HTTP_200_OK, content=content)52    # Delete a database53    @router.delete("/databases/{plugin_name}/{database_name}", response_description="Delete a database")54    async def delete_database(request: Request, plugin_name: str, database_name: str,55                              api_key: APIKey = Depends(validate_api_key)):56        success, message = commands.delete_database(plugin_name, database_name)57        content = {"success": success, "message": message}58        return JSONResponse(status_code=status.HTTP_200_OK, content=content)59    # We return our router...

test_config.py

Source:test_config.py

...20        """21        _config._checked_api_keys = dict()22    def test_err_empty_string(self, mocker):23        p = mocker.patch("codepost.util.config._logger.warning")24        assert not _config.validate_api_key(self.BAD_KEY_EMPTY, log_outcome=False)25        p.assert_not_called()26    def test_err_empty_string_logged(self, mocker):27        p = mocker.patch("codepost.util.config._logger.warning")28        assert not _config.validate_api_key(self.BAD_KEY_EMPTY, log_outcome=True)29        p.assert_called()30    def test_err_unstringable(self, stringable_err):31        assert not _config.validate_api_key(stringable_err)32    def test_err_tooshort_string(self):33        assert not _config.validate_api_key(self.BAD_KEY_TOOSHORT)34    def test_err_toolong_string(self):35        assert not _config.validate_api_key(self.BAD_KEY_TOOLONG)36    def test_err_bad_request(self, requests_mock):37        api_key = self.FAKE_KEY_RIGHTLENGTH38        requests_mock.get(39            "{}/courses/".format(_config.BASE_URL),40            request_headers={"Authorization": "Token {}".format(api_key)},41            status_code=401)42        assert not _config.validate_api_key(api_key)43        assert requests_mock.called44    def test_success(self, requests_mock):45        api_key = self.FAKE_KEY_RIGHTLENGTH46        requests_mock.get(47            "{}/courses/".format(_config.BASE_URL),48            request_headers={"Authorization": "Token {}".format(api_key)},49            status_code=200)50        assert _config.validate_api_key(api_key)51        assert api_key in _config._checked_api_keys52        assert requests_mock.called53    def test_err_request_fail(self, requests_mock):54        api_key = self.FAKE_KEY_RIGHTLENGTH55        requests_mock.get(56            "{}/courses/".format(_config.BASE_URL),57            exc=Exception)58        assert not _config.validate_api_key(api_key)59        assert requests_mock.called60    def test_cached_true(self, mocker):61        api_key = self.FAKE_KEY_RIGHTLENGTH62        _config._checked_api_keys[api_key] = True63        p = mocker.patch("codepost.util.config._logger.debug")64        assert _config.validate_api_key(api_key, log_outcome=True)65        p.assert_called()66    def test_cached_false(self, mocker):67        api_key = self.FAKE_KEY_RIGHTLENGTH68        _config._checked_api_keys[api_key] = False69        assert not _config.validate_api_key(api_key)70    def test_cached_refresh_true(self, mocker):71        api_key = self.FAKE_KEY_RIGHTLENGTH72        _config._checked_api_keys[api_key] = True73        p = mocker.patch("codepost.util.config._logger.debug")74        assert not _config.validate_api_key(api_key, refresh=True, log_outcome=True)...

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.