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

How to use no_body method in Splinter

Best Python code snippet using splinter

common.py

Source:common.py

1from __future__ import annotations2from typing import Any, Dict, Iterable, Iterator, Mapping, Optional, Union3from typing_extensions import TypeGuard45import attr67from .io import FlvReader8from .io_protocols import RandomIO9from . import scriptdata10from .avc import extract_resolution11from .utils import OffsetRepositor12from .models import (13    AVCPacketType, FlvTag, AudioTag, ScriptTag, TagType, VideoTag14)151617def read_tags(18    reader: FlvReader, count: int, *, no_body: bool = False19) -> Iterator[FlvTag]:20    assert count > 0, 'count must greater than 0'21    for c, tag in enumerate(reader.read_tags(no_body=no_body), start=1):22        yield tag23        if c >= count:24            break252627def rread_tags(28    reader: FlvReader, count: int, *, no_body: bool = False29) -> Iterator[FlvTag]:30    assert count > 0, 'count must greater than 0'31    for c, tag in enumerate(reader.rread_tags(no_body=no_body), start=1):32        yield tag33        if c >= count:34            break353637def read_tags_in_duration(38    reader: FlvReader, duration: int, *, no_body: bool = False39) -> Iterator[FlvTag]:40    for tag in reader.read_tags(no_body=no_body):41        yield tag42        if tag.timestamp > 0:43            break4445    start = tag.timestamp46    end = start + duration4748    for tag in reader.read_tags(no_body=no_body):49        yield tag50        if tag.timestamp >= end:51            break525354def peek_tags(55    file: RandomIO, reader: FlvReader, count: int, *, no_body: bool = False,56) -> Iterator[FlvTag]:57    with OffsetRepositor(file):58        yield from read_tags(reader, count, no_body=no_body)596061def rpeek_tags(62    file: RandomIO, reader: FlvReader, count: int, *, no_body: bool = False,63) -> Iterator[FlvTag]:64    with OffsetRepositor(file):65        yield from rread_tags(reader, count, no_body=no_body)666768def find_metadata_tag(tags: Iterable[FlvTag]) -> Optional[ScriptTag]:69    for tag in tags:70        if is_script_tag(tag):71            return tag72    return None737475def find_header_tag(76    tags: Iterable[FlvTag]77) -> Optional[Union[AudioTag, VideoTag]]:78    for tag in tags:79        if is_sequence_header(tag):80            return tag81    return None828384def find_avc_header_tag(tags: Iterable[FlvTag]) -> Optional[VideoTag]:85    for tag in tags:86        if is_video_sequence_header(tag):87            return tag88    return None899091def find_aac_header_tag(tags: Iterable[FlvTag]) -> Optional[AudioTag]:92    for tag in tags:93        if is_audio_sequence_header(tag):94            return tag95    return None969798def find_nalu_keyframe_tag(tags: Iterable[FlvTag]) -> Optional[VideoTag]:99    for tag in tags:100        if is_video_nalu_keyframe(tag):101            return tag102    return None103104105def find_aac_raw_tag(tags: Iterable[FlvTag]) -> Optional[AudioTag]:106    for tag in tags:107        if is_audio_tag(tag) and tag.is_aac_raw():108            return tag109    return None110111112def is_audio_tag(tag: FlvTag) -> TypeGuard[AudioTag]:113    return tag.tag_type == TagType.AUDIO114115116def is_video_tag(tag: FlvTag) -> TypeGuard[VideoTag]:117    return tag.tag_type == TagType.VIDEO118119120def is_script_tag(tag: FlvTag) -> TypeGuard[ScriptTag]:121    return tag.tag_type == TagType.SCRIPT122123124def is_metadata_tag(tag: FlvTag) -> TypeGuard[ScriptTag]:125    if is_script_tag(tag):126        script_data = parse_scriptdata(tag)127        return script_data['name'] == 'onMetaData'128    return False129130131def is_data_tag(tag: FlvTag) -> TypeGuard[Union[AudioTag, VideoTag]]:132    return is_audio_data_tag(tag) or is_video_data_tag(tag)133134135def is_audio_data_tag(tag: FlvTag) -> TypeGuard[AudioTag]:136    return is_audio_tag(tag) and tag.is_aac_raw()137138139def is_video_data_tag(tag: FlvTag) -> TypeGuard[VideoTag]:140    return is_video_tag(tag) and tag.is_avc_nalu()141142143def is_sequence_header(tag: FlvTag) -> TypeGuard[Union[AudioTag, VideoTag]]:144    return is_audio_sequence_header(tag) or is_video_sequence_header(tag)145146147def is_audio_sequence_header(tag: FlvTag) -> TypeGuard[AudioTag]:148    return is_audio_tag(tag) and tag.is_aac_header()149150151def is_video_sequence_header(tag: FlvTag) -> TypeGuard[VideoTag]:152    return is_video_tag(tag) and tag.is_avc_header()153154155def is_video_nalu_keyframe(tag: FlvTag) -> TypeGuard[VideoTag]:156    return is_video_tag(tag) and tag.is_keyframe() and tag.is_avc_nalu()157158159def parse_scriptdata(script_tag: ScriptTag) -> scriptdata.ScriptData:160    assert script_tag.body is not None161    return scriptdata.load(script_tag.body)162163164def unparse_scriptdata(165    script_tag: ScriptTag, script_data: scriptdata.ScriptData, **changes: Any166) -> ScriptTag:167    body = scriptdata.dump(script_data)168    return script_tag.evolve(body=body, **changes)169170171def create_script_tag(172    script_data: scriptdata.ScriptData, offset: int = 0, timestamp: int = 0173) -> ScriptTag:174    body = scriptdata.dump(script_data)175    return ScriptTag(176        filtered=False,177        tag_type=TagType.SCRIPT,178        data_size=len(body),179        timestamp=timestamp,180        stream_id=0,181        offset=offset,182        body=body,183    )184185186def create_metadata_tag(187    metadata: Dict[str, Any], offset: int = 0, timestamp: int = 0188) -> ScriptTag:189    script_data = scriptdata.ScriptData(name='onMetaData', value=metadata)190    return create_script_tag(script_data, offset, timestamp)191192193def parse_metadata(metadata_tag: ScriptTag) -> Dict[str, Any]:194    script_data = parse_scriptdata(metadata_tag)195    assert script_data['name'] == 'onMetaData'196    return script_data['value']197198199def unparse_metadata(200    metadata_tag: ScriptTag, metadata: Dict[str, Any], **changes: Any201) -> ScriptTag:202    script_data = scriptdata.ScriptData(name='onMetaData', value=metadata)203    return unparse_scriptdata(metadata_tag, script_data, **changes)204205206def enrich_metadata(207    metadata_tag: ScriptTag,208    extra_metadata: Mapping[str, Any],209    offset: Optional[int] = None,210) -> ScriptTag:211    metadata = parse_metadata(metadata_tag)212    metadata.update(extra_metadata)213    metadata = ensure_order(metadata)214215    if offset is None:216        return unparse_metadata(metadata_tag, metadata)217    else:218        return unparse_metadata(metadata_tag, metadata, offset=offset)219220221def update_metadata(222    metadata_tag: ScriptTag,223    metadata: Mapping[str, Any],224    offset: Optional[int] = None,225) -> ScriptTag:226    original_tag_size = metadata_tag.tag_size227    new_tag = enrich_metadata(metadata_tag, metadata, offset)228    assert new_tag.tag_size == original_tag_size, 'tag size must unchanged'229    return new_tag230231232def ensure_order(metadata: Dict[str, Any]) -> Dict[str, Any]:233    # the order of metadata property matters!!!234    # some typical properties such as 'keyframes' must be before some custom235    # properties such as 'Comment' otherwise, it won't take effect in some236    # players!237    from .data_analyser import MetaData238    typical_props = attr.fields_dict(MetaData).keys()239    return {240        **{k: v for k, v in metadata.items() if k in typical_props},241        **{k: v for k, v in metadata.items() if k not in typical_props},242    }243244245@attr.s(auto_attribs=True, slots=True, frozen=True)246class Resolution:247    width: int248    height: int249250    @classmethod251    def from_metadata(cls, metadata: Dict[str, Any]) -> Resolution:252        return cls(253            width=metadata.get('width') or metadata['displayWidth'],254            height=metadata.get('height') or metadata['displayHeight'],255        )256257    @classmethod258    def from_aac_sequence_header(cls, tag: VideoTag) -> Resolution:259        assert tag.avc_packet_type == AVCPacketType.AVC_SEQUENCE_HEADER260        assert tag.body is not None261        width, height = extract_resolution(tag.body)
...

test_http.py

Source:test_http.py

...30        def index(self, *args, **kwargs):31            return "Hello world!"32        index.exposed = True33        34        def no_body(self, *args, **kwargs):35            return "Hello world!"36        no_body.exposed = True37        no_body._cp_config = {'request.process_request_body': False}38        39        def post_multipart(self, file):40            """Return a summary ("a * 65536\nb * 65536") of the uploaded file."""41            contents = file.file.read()42            summary = []43            curchar = ""44            count = 045            for c in contents:46                if c == curchar:47                    count += 148                else:...

vnf_package_views.py

Source:vnf_package_views.py

1# Copyright 2018 ZTE Corporation.2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7#         http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14import logging15from django.http import StreamingHttpResponse16from drf_yasg.utils import swagger_auto_schema, no_body17from rest_framework import status18from rest_framework.decorators import api_view19from rest_framework.response import Response20from genericparser.packages.serializers.upload_vnf_pkg_from_uri_req import UploadVnfPackageFromUriRequestSerializer21from genericparser.packages.serializers.create_vnf_pkg_info_req import CreateVnfPkgInfoRequestSerializer22from genericparser.packages.serializers.vnf_pkg_info import VnfPkgInfoSerializer23from genericparser.packages.serializers.vnf_pkg_infos import VnfPkgInfosSerializer24from genericparser.packages.biz.vnf_package import VnfPackage25from genericparser.packages.biz.vnf_package import VnfPkgUploadThread26from genericparser.packages.biz.vnf_package import parse_vnfd_and_save27from genericparser.packages.biz.vnf_package import handle_upload_failed28from .common import validate_data29from .common import view_safe_call_with_log30logger = logging.getLogger(__name__)31@swagger_auto_schema(32    method="GET",33    operation_description="Query multiple VNF package resource",34    request_body=no_body,35    responses={36        status.HTTP_200_OK: VnfPkgInfosSerializer(),37        status.HTTP_500_INTERNAL_SERVER_ERROR: "Internal error"38    }39)40@swagger_auto_schema(41    method="POST",42    operation_description="Create an individual VNF package resource",43    request_body=CreateVnfPkgInfoRequestSerializer,44    responses={45        status.HTTP_201_CREATED: VnfPkgInfoSerializer(),46        status.HTTP_500_INTERNAL_SERVER_ERROR: "Internal error"47    }48)49@api_view(http_method_names=["GET", "POST"])50@view_safe_call_with_log(logger=logger)51def vnf_packages_rc(request):52    if request.method == 'GET':53        logger.debug("Query VNF packages> %s" % request.data)54        data = VnfPackage().query_multiple()55        vnf_pkg_infos = validate_data(data, VnfPkgInfosSerializer)56        return Response(data=vnf_pkg_infos.data, status=status.HTTP_200_OK)57    if request.method == 'POST':58        logger.debug("Create VNF package> %s" % request.data)59        create_vnf_pkg_info_request = validate_data(request.data,60                                                    CreateVnfPkgInfoRequestSerializer)61        data = VnfPackage().create_vnf_pkg(create_vnf_pkg_info_request.data)62        vnf_pkg_info = validate_data(data, VnfPkgInfoSerializer)63        return Response(data=vnf_pkg_info.data, status=status.HTTP_201_CREATED)64@swagger_auto_schema(65    method='PUT',66    operation_description="Upload VNF package content",67    request_body=no_body,68    responses={69        status.HTTP_202_ACCEPTED: "Successfully",70        status.HTTP_500_INTERNAL_SERVER_ERROR: "Internal error"71    }72)73@swagger_auto_schema(74    method="GET",75    operation_description="Fetch VNF package content",76    request_body=no_body,77    responses={78        status.HTTP_200_OK: VnfPkgInfosSerializer(),79        status.HTTP_404_NOT_FOUND: "VNF package does not exist",80        status.HTTP_500_INTERNAL_SERVER_ERROR: "Internal error"81    }82)83@api_view(http_method_names=["PUT", "GET"])84@view_safe_call_with_log(logger=logger)85def package_content_ru(request, **kwargs):86    vnf_pkg_id = kwargs.get("vnfPkgId")87    if request.method == "PUT":88        logger.debug("Upload VNF package %s" % vnf_pkg_id)89        files = request.FILES.getlist('file')90        try:91            local_file_name = VnfPackage().upload(vnf_pkg_id, files[0])92            parse_vnfd_and_save(vnf_pkg_id, local_file_name)93            return Response(None, status=status.HTTP_202_ACCEPTED)94        except Exception as e:95            handle_upload_failed(vnf_pkg_id)96            raise e97    if request.method == "GET":98        file_range = request.META.get('HTTP_RANGE')99        file_iterator = VnfPackage().download(vnf_pkg_id, file_range)100        return StreamingHttpResponse(file_iterator, status=status.HTTP_200_OK)101@swagger_auto_schema(102    method='POST',103    operation_description="Upload VNF package content from uri",104    request_body=UploadVnfPackageFromUriRequestSerializer,105    responses={106        status.HTTP_202_ACCEPTED: "Successfully",107        status.HTTP_500_INTERNAL_SERVER_ERROR: "Internal error"108    }109)110@api_view(http_method_names=['POST'])111@view_safe_call_with_log(logger=logger)112def upload_from_uri_c(request, **kwargs):113    vnf_pkg_id = kwargs.get("vnfPkgId")114    try:115        upload_vnf_from_uri_request = validate_data(request.data,116                                                    UploadVnfPackageFromUriRequestSerializer)117        VnfPkgUploadThread(upload_vnf_from_uri_request.data, vnf_pkg_id).start()118        return Response(None, status=status.HTTP_202_ACCEPTED)119    except Exception as e:120        handle_upload_failed(vnf_pkg_id)121        raise e122@swagger_auto_schema(123    method='GET',124    operation_description="Query an individual VNF package resource",125    request_body=no_body,126    responses={127        status.HTTP_200_OK: VnfPkgInfoSerializer(),128        status.HTTP_404_NOT_FOUND: "VNF package does not exist",129        status.HTTP_500_INTERNAL_SERVER_ERROR: "Internal error"130    }131)132@swagger_auto_schema(133    method='DELETE',134    operation_description="Delete an individual VNF package resource",135    request_body=no_body,136    responses={137        status.HTTP_204_NO_CONTENT: "No content",138        status.HTTP_500_INTERNAL_SERVER_ERROR: "Internal error"139    }140)141@api_view(http_method_names=['GET', 'DELETE'])142@view_safe_call_with_log(logger=logger)143def vnf_package_rd(request, **kwargs):144    vnf_pkg_id = kwargs.get("vnfPkgId")145    if request.method == 'GET':146        logger.debug("Query an individual VNF package> %s" % request.data)147        data = VnfPackage().query_single(vnf_pkg_id)148        vnf_pkg_info = validate_data(data, VnfPkgInfoSerializer)149        return Response(data=vnf_pkg_info.data, status=status.HTTP_200_OK)150    if request.method == 'DELETE':151        logger.debug("Delete an individual VNF package> %s" % request.data)152        VnfPackage().delete_vnf_pkg(vnf_pkg_id)...

pnf_descriptor_views.py

Source:pnf_descriptor_views.py

1# Copyright 2018 ZTE Corporation.2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7#         http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14import logging15from django.http import StreamingHttpResponse16from drf_yasg.utils import no_body, swagger_auto_schema17from rest_framework import status18from rest_framework.decorators import api_view19from rest_framework.response import Response20from genericparser.packages.biz.pnf_descriptor import PnfDescriptor21from genericparser.packages.serializers.create_pnfd_info_request import CreatePnfdInfoRequestSerializer22from genericparser.packages.serializers.pnfd_info import PnfdInfoSerializer23from genericparser.packages.serializers.pnfd_infos import PnfdInfosSerializer24from genericparser.packages.views.common import validate_data25from genericparser.packages.serializers.genericparser_serializers import ParseModelRequestSerializer26from genericparser.packages.serializers.genericparser_serializers import ParseModelResponseSerializer27from genericparser.packages.serializers.genericparser_serializers import InternalErrorRequestSerializer28from genericparser.packages.serializers.response import ProblemDetailsSerializer29from genericparser.pub.utils.syscomm import fun_name30from genericparser.pub.utils.values import ignore_case_get31from .common import view_safe_call_with_log32logger = logging.getLogger(__name__)33@swagger_auto_schema(34    method='GET',35    operation_description="Query a PNFD",36    request_body=no_body,37    responses={38        status.HTTP_200_OK: PnfdInfoSerializer(),39        status.HTTP_404_NOT_FOUND: ProblemDetailsSerializer(),40        status.HTTP_500_INTERNAL_SERVER_ERROR: ProblemDetailsSerializer()41    }42)43@swagger_auto_schema(44    method='DELETE',45    operation_description="Delete a PNFD",46    request_body=no_body,47    responses={48        status.HTTP_204_NO_CONTENT: "No content",49        status.HTTP_500_INTERNAL_SERVER_ERROR: ProblemDetailsSerializer()50    }51)52@api_view(http_method_names=['GET', 'DELETE'])53@view_safe_call_with_log(logger=logger)54def pnfd_info_rd(request, **kwargs):  # TODO55    pnfd_info_id = kwargs.get('pnfdInfoId')56    if request.method == 'GET':57        logger.debug("Query an individual PNF descriptor> %s" % request.data)58        data = PnfDescriptor().query_single(pnfd_info_id)59        pnfd_info = validate_data(data, PnfdInfoSerializer)60        return Response(data=pnfd_info.data, status=status.HTTP_200_OK)61    if request.method == 'DELETE':62        logger.debug("Delete an individual PNFD resource> %s" % request.data)63        PnfDescriptor().delete_single(pnfd_info_id)64        return Response(data=None, status=status.HTTP_204_NO_CONTENT)65@swagger_auto_schema(66    method='POST',67    operation_description="Create a  PNFD",68    request_body=CreatePnfdInfoRequestSerializer(),69    responses={70        status.HTTP_201_CREATED: PnfdInfoSerializer(),71        status.HTTP_500_INTERNAL_SERVER_ERROR: ProblemDetailsSerializer()72    }73)74@swagger_auto_schema(75    method='GET',76    operation_description="Query multiple PNFDs",77    request_body=no_body,78    responses={79        status.HTTP_200_OK: PnfdInfosSerializer(),80        status.HTTP_500_INTERNAL_SERVER_ERROR: ProblemDetailsSerializer()81    }82)83@api_view(http_method_names=['POST', 'GET'])84@view_safe_call_with_log(logger=logger)85def pnf_descriptors_rc(request):86    if request.method == 'POST':87        create_pnfd_info_request = validate_data(request.data, CreatePnfdInfoRequestSerializer)88        data = PnfDescriptor().create(create_pnfd_info_request.data)89        pnfd_info = validate_data(data, PnfdInfoSerializer)90        return Response(data=pnfd_info.data, status=status.HTTP_201_CREATED)91    if request.method == 'GET':92        data = PnfDescriptor().query_multiple(request)93        pnfd_infos = validate_data(data, PnfdInfosSerializer)94        return Response(data=pnfd_infos.data, status=status.HTTP_200_OK)95@swagger_auto_schema(96    method='PUT',97    operation_description="Upload PNFD content",98    request_body=no_body,99    responses={100        status.HTTP_204_NO_CONTENT: "No content",101        status.HTTP_500_INTERNAL_SERVER_ERROR: ProblemDetailsSerializer()102    }103)104@swagger_auto_schema(105    method='GET',106    operation_description="Fetch PNFD content",107    request_body=no_body,108    responses={109        status.HTTP_204_NO_CONTENT: 'PNFD file',110        status.HTTP_404_NOT_FOUND: ProblemDetailsSerializer(),111        status.HTTP_500_INTERNAL_SERVER_ERROR: ProblemDetailsSerializer()112    }113)114@api_view(http_method_names=['PUT', 'GET'])115@view_safe_call_with_log(logger=logger)116def pnfd_content_ru(request, **kwargs):117    pnfd_info_id = kwargs.get("pnfdInfoId")118    if request.method == 'PUT':119        files = request.FILES.getlist('file')120        try:121            local_file_name = PnfDescriptor().upload(files[0], pnfd_info_id)122            PnfDescriptor().parse_pnfd_and_save(pnfd_info_id, local_file_name)123            return Response(data=None, status=status.HTTP_204_NO_CONTENT)124        except Exception as e:125            PnfDescriptor().handle_upload_failed(pnfd_info_id)126            raise e127    if request.method == 'GET':128        file_iterator = PnfDescriptor().download(pnfd_info_id)129        return StreamingHttpResponse(file_iterator, status=status.HTTP_200_OK)130@swagger_auto_schema(131    method='POST',132    operation_description="Parse PNF model",133    request_body=ParseModelRequestSerializer,134    responses={135        status.HTTP_202_ACCEPTED: ParseModelResponseSerializer,136        status.HTTP_500_INTERNAL_SERVER_ERROR: InternalErrorRequestSerializer})137@api_view(http_method_names=['POST'])138def pnf_model_parser(request, *args, **kwargs):139    csar_id = ignore_case_get(request.data, "csarId")140    inputs = ignore_case_get(request.data, "inputs")141    logger.debug(142        "Enter %s, csar_id=%s, inputs=%s",143        fun_name(),144        csar_id,145        inputs)146    ret = PnfDescriptor().parse_pnfd(csar_id, inputs)147    logger.info("Leave %s, Return value is %s", fun_name(), ret)148    if ret[0] != 0:149        return Response(data={'error': ret[1]}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)150    response = validate_data(ret[1], ParseModelResponseSerializer)...

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.