KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use response_409 method in Django Test Plus

Best Python code snippet using django-test-plus_python
views.py
Source:views.py
1# Copyright (c) 2012-2016 Seafile Ltd.2# encoding: utf-83import os4import json5import logging6import urllib27import requests8import hashlib9import urlparse10import posixpath11from rest_framework.views import APIView12from django.http import HttpResponse13from django.core.cache import cache14from pysearpc import SearpcError15from seaserv import seafile_api16from seahub.base.accounts import User17from seahub.utils import gen_inner_file_get_url, \18    gen_inner_file_upload_url, get_file_type_and_ext, is_pro_version19from seahub.utils.file_op import check_file_lock20from seahub.base.templatetags.seahub_tags import email2nickname21from seahub.settings import SITE_ROOT22from .utils import get_file_info_by_token23from .settings import ENABLE_OFFICE_WEB_APP_EDIT, \24        OFFICE_WEB_APP_EDIT_FILE_EXTENSION25logger = logging.getLogger(__name__)26json_content_type = 'application/json; charset=utf-8'27WOPI_LOCK_EXPIRATION = 30 * 6028def generate_file_lock_key_value(request):29    token = request.GET.get('access_token', None)30    request_user, repo_id, file_path = get_file_info_by_token(token)31    repo = seafile_api.get_repo(repo_id)32    if repo.is_virtual:33        origin_repo_id = repo.origin_repo_id34        origin_file_path = posixpath.join(repo.origin_path, file_path.strip('/'))35        file_path_hash = hashlib.sha256(origin_file_path.encode('utf8')).hexdigest()36        lock_cache_key = '_'.join(['HTTP_X_WOPI_LOCK', origin_repo_id, file_path_hash])37    else:38        file_path_hash = hashlib.sha256(file_path.encode('utf8')).hexdigest()39        lock_cache_key = '_'.join(['HTTP_X_WOPI_LOCK', repo_id, file_path_hash])40    x_wopi_lock = request.META.get('HTTP_X_WOPI_LOCK', None)41    return lock_cache_key, x_wopi_lock42def lock_file(request):43    key, value = generate_file_lock_key_value(request)44    cache.set(key, value, WOPI_LOCK_EXPIRATION)45def unlock_file(request):46    key, value = generate_file_lock_key_value(request)47    cache.delete(key)48def refresh_file_lock(request):49    lock_file(request)50def file_is_locked(request):51    key, value = generate_file_lock_key_value(request)52    return True if cache.get(key, '') else False53def get_current_lock_id(request):54    key, value = generate_file_lock_key_value(request)55    return cache.get(key, '')56def access_token_check(func):57    def _decorated(view, request, file_id, *args, **kwargs):58        token = request.GET.get('access_token', None)59        if not token:60            logger.error('access_token invalid.')61            return HttpResponse(json.dumps({}), status=401,62                                content_type=json_content_type)63        request_user, repo_id, file_path = get_file_info_by_token(token)64        if not request_user or not repo_id or not file_path:65            logger.error('File info invalid, user: %s, repo_id: %s, path: %s.' \66                    % (request_user, repo_id, file_path))67            return HttpResponse(json.dumps({}), status=404,68                                content_type=json_content_type)69        try:70            User.objects.get(email=request_user)71        except User.DoesNotExist:72            logger.error('User %s not found.' % request_user)73            return HttpResponse(json.dumps({}), status=404,74                                content_type=json_content_type)75        repo = seafile_api.get_repo(repo_id)76        if not repo:77            logger.error('Library %s not found.' % repo_id)78            return HttpResponse(json.dumps({}), status=404,79                                content_type=json_content_type)80        obj_id = seafile_api.get_file_id_by_path(repo_id, file_path)81        if not obj_id:82            logger.error('File %s not found.' % file_path)83            return HttpResponse(json.dumps({}), status=404,84                                content_type=json_content_type)85        return func(view, request, file_id, *args, **kwargs)86    return _decorated87class WOPIFilesView(APIView):88    @access_token_check89    def get(self, request, file_id, format=None):90        """ WOPI endpoint for check file info91        """92        token = request.GET.get('access_token', None)93        request_user, repo_id, file_path = get_file_info_by_token(token)94        repo = seafile_api.get_repo(repo_id)95        obj_id = seafile_api.get_file_id_by_path(repo_id, file_path)96        try:97            file_size = seafile_api.get_file_size(repo.store_id,98                                                  repo.version,99                                                  obj_id)100        except SearpcError as e:101            logger.error(e)102            return HttpResponse(json.dumps({}), status=500,103                                content_type=json_content_type)104        if file_size == -1:105            logger.error('File %s not found.' % file_path)106            return HttpResponse(json.dumps({}), status=401,107                                content_type=json_content_type)108        result = {}109        # necessary110        result['BaseFileName'] = os.path.basename(file_path)111        result['Size'] = file_size112        result['UserId'] = request_user113        result['Version'] = obj_id114        try:115            if is_pro_version():116                result['OwnerId'] = seafile_api.get_repo_owner(repo_id) or \117                        seafile_api.get_org_repo_owner(repo_id)118            else:119                result['OwnerId'] = seafile_api.get_repo_owner(repo_id)120        except Exception as e:121            logger.error(e)122            return HttpResponse(json.dumps({}), status=500,123                                content_type=json_content_type)124        # optional125        result['UserFriendlyName'] = email2nickname(request_user)126        absolute_uri = request.build_absolute_uri('/')127        result['PostMessageOrigin'] = urlparse.urljoin(absolute_uri, SITE_ROOT).strip('/')128        result['HidePrintOption'] = False129        result['HideSaveOption'] = False130        result['HideExportOption'] = False131        result['EnableOwnerTermination'] = True132        result['SupportsLocks'] = True133        result['SupportsGetLock'] = True134        result['SupportsUpdate'] = True135        filename = os.path.basename(file_path)136        filetype, fileext = get_file_type_and_ext(filename)137        try:138            is_locked, locked_by_me = check_file_lock(repo_id,139                    file_path, request_user)140        except Exception as e:141            logger.error(e)142            return HttpResponse(json.dumps({}),143                    status=500, content_type=json_content_type)144        perm = seafile_api.check_permission_by_path(repo_id,145                file_path, request_user)146        if ENABLE_OFFICE_WEB_APP_EDIT and not repo.encrypted and \147            perm == 'rw' and ((not is_locked) or (is_locked and locked_by_me)) and \148                fileext in OFFICE_WEB_APP_EDIT_FILE_EXTENSION:149            result['UserCanWrite'] = True150        return HttpResponse(json.dumps(result), status=200,151                            content_type=json_content_type)152    @access_token_check153    def post(self, request, file_id, format=None):154        token = request.GET.get('access_token', None)155        request_user, repo_id, file_path = get_file_info_by_token(token)156        response_409 = HttpResponse(json.dumps({}),157                status=409, content_type=json_content_type)158        x_wopi_override = request.META.get('HTTP_X_WOPI_OVERRIDE', None)159        x_wopi_lock = request.META.get('HTTP_X_WOPI_LOCK', None)160        x_wopi_oldlock = request.META.get('HTTP_X_WOPI_OLDLOCK', None)161        current_lock_id = get_current_lock_id(request)162        if x_wopi_override == 'LOCK':163            if x_wopi_oldlock:164                # UnlockAndRelock endpoint165                if file_is_locked(request):166                    if x_wopi_oldlock != current_lock_id:167                        # If the file is currently locked168                        # and the X-WOPI-OldLock value does NOT match the lock currently on the file169                        # the host must return a âlock mismatchâ response (409 Conflict)170                        # and include an X-WOPI-Lock response header containing the value of the current lock on the file171                        response_409['X-WOPI-Lock'] = current_lock_id172                        return response_409173                    else:174                        unlock_file(request)175                        lock_file(request)176                        return HttpResponse()177                else:178                    response_409['X-WOPI-Lock'] = ''179                    return response_409180            else:181                # Lock endpoint182                if not file_is_locked(request):183                    # If the file is currently unlocked184                    # the host should lock the file and return 200 OK.185                    lock_file(request)186                    return HttpResponse()187                else:188                    # If the file is currently locked189                    # and the X-WOPI-Lock value matches the lock currently on the file190                    # the host should treat the request as if it is a RefreshLock request191                    # That is, the host should refresh the lock timer and return 200 OK.192                    if current_lock_id == x_wopi_lock:193                        refresh_file_lock(request)194                        return HttpResponse()195                    else:196                        # In all other cases197                        # the host must return a âlock mismatchâ response (409 Conflict)198                        # and include an X-WOPI-Lock response header199                        # containing the value of the current lock on the file.200                        response_409['X-WOPI-Lock'] = current_lock_id201                        return response_409202        elif x_wopi_override == 'GET_LOCK':203            response = HttpResponse()204            if not file_is_locked(request):205                # If the file is currently NOT locked206                # the host must return a 200 OK207                # and include an X-WOPI-Lock response header set to the empty string.208                response['X-WOPI-Lock'] = ''209                return response210            else:211                # If the file is currently locked212                # the host should return a 200 OK213                # and include an X-WOPI-Lock response header214                # containing the value of the current lock on the file.215                response['X-WOPI-Lock'] = current_lock_id216                return response217        elif x_wopi_override in ('REFRESH_LOCK', 'UNLOCK'):218            if file_is_locked(request):219                # If the file is currently locked220                # and the X-WOPI-Lock value does NOT match the lock currently on the file221                # the host must return a âlock mismatchâ response (409 Conflict)222                # and include an X-WOPI-Lock response header containing the value of the current lock on the file223                if x_wopi_lock != current_lock_id:224                    response_409['X-WOPI-Lock'] = current_lock_id225                    return response_409226                else:227                    if x_wopi_override == 'REFRESH_LOCK':228                        refresh_file_lock(request)229                    else:230                        unlock_file(request)231                    return HttpResponse()232            else:233                # or if the file is unlocked234                # the host must return a âlock mismatchâ response (409 Conflict)235                # and include an X-WOPI-Lock response header containing the value of the current lock on the file236                # In the case where the file is unlocked237                # the host must set X-WOPI-Lock to the empty string238                response_409['X-WOPI-Lock'] = ''239                return response_409240        else:241            logger.info('HTTP_X_WOPI_OVERRIDE: %s' % x_wopi_override)242            logger.info('HTTP_X_WOPI_LOCK: %s' % x_wopi_lock)243            logger.info('HTTP_X_WOPI_OLDLOCK: %s' % x_wopi_oldlock)244            return HttpResponse(json.dumps({'error_msg': 'HTTP_X_WOPI_OVERRIDE invalid'}),245                    status=401, content_type=json_content_type)246class WOPIFilesContentsView(APIView):247    @access_token_check248    def get(self, request, file_id, format=None):249        """ WOPI endpoint for get file content250        """251        token = request.GET.get('access_token', None)252        request_user, repo_id, file_path = get_file_info_by_token(token=token)253        obj_id = seafile_api.get_file_id_by_path(repo_id, file_path)254        file_name = os.path.basename(file_path)255        try:256            fileserver_token = seafile_api.get_fileserver_access_token(repo_id,257                    obj_id, 'view', '', use_onetime = False)258        except SearpcError as e:259            logger.error(e)260            return HttpResponse(json.dumps({}), status=500,261                    content_type=json_content_type)262        if not fileserver_token:263            return HttpResponse(json.dumps({}), status=500,264                    content_type=json_content_type)265        inner_path = gen_inner_file_get_url(fileserver_token, file_name)266        try:267            file_content = urllib2.urlopen(inner_path).read()268        except urllib2.URLError as e:269            logger.error(e)270            return HttpResponse(json.dumps({}), status=500,271                                content_type=json_content_type)272        return HttpResponse(file_content, content_type="application/octet-stream")273    @access_token_check274    def post(self, request, file_id, format=None):275        token = request.GET.get('access_token', None)276        request_user, repo_id, file_path = get_file_info_by_token(token=token)277        try:278            file_obj = request.read()279            # get file update url280            token = seafile_api.get_fileserver_access_token(repo_id,281                    'dummy', 'update', request_user)282            if not token:283                return HttpResponse(json.dumps({}), status=500,284                        content_type=json_content_type)285            update_url = gen_inner_file_upload_url('update-api', token)286            # update file287            files = {288                'file': file_obj,289                'file_name': os.path.basename(file_path),290                'target_file': file_path,291            }292            requests.post(update_url, files=files)293        except Exception as e:294            logger.error(e)295            return HttpResponse(json.dumps({}), status=500,296                                content_type=json_content_type)297        return HttpResponse(json.dumps({}), status=200,...
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.