vcsserver/git_lfs/tests/test_lfs_app.py

# RhodeCode VCSServer provides access to different vcs backends via network.
# Copyright (C) 2014-2024 RhodeCode GmbH
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA

import os
import pytest
from webtest.app import TestApp as WebObTestApp

from vcsserver.lib.ext_json import json
from vcsserver.lib.str_utils import safe_bytes
from vcsserver.git_lfs.app import create_app
from vcsserver.git_lfs.lib import LFSOidStore


@pytest.fixture(scope='function')
def git_lfs_app(tmpdir):
    custom_app = WebObTestApp(create_app(
        git_lfs_enabled=True, git_lfs_store_path=str(tmpdir),
        git_lfs_http_scheme='http'))
    custom_app._store = str(tmpdir)
    return custom_app


@pytest.fixture(scope='function')
def git_lfs_https_app(tmpdir):
    custom_app = WebObTestApp(create_app(
        git_lfs_enabled=True, git_lfs_store_path=str(tmpdir),
        git_lfs_http_scheme='https'))
    custom_app._store = str(tmpdir)
    return custom_app


@pytest.fixture()
def http_auth():
    return {'HTTP_AUTHORIZATION': "Basic XXXXX"}


class TestLFSApplication:

    def test_app_wrong_path(self, git_lfs_app):
        git_lfs_app.get('/repo/info/lfs/xxx', status=404)

    def test_app_deprecated_endpoint(self, git_lfs_app):
        response = git_lfs_app.post('/repo/info/lfs/objects', status=501)
        assert response.status_code == 501
        assert json.loads(response.text) == {'message': 'LFS: v1 api not supported'}

    def test_app_lock_verify_api_not_available(self, git_lfs_app):
        response = git_lfs_app.post('/repo/info/lfs/locks/verify', status=501)
        assert response.status_code == 501
        assert json.loads(response.text) == {
            'message': 'GIT LFS locking api not supported'}

    def test_app_lock_api_not_available(self, git_lfs_app):
        response = git_lfs_app.post('/repo/info/lfs/locks', status=501)
        assert response.status_code == 501
        assert json.loads(response.text) == {
            'message': 'GIT LFS locking api not supported'}

    def test_app_batch_api_missing_auth(self, git_lfs_app):
        git_lfs_app.post_json(
            '/repo/info/lfs/objects/batch', params={}, status=403)

    def test_app_batch_api_unsupported_operation(self, git_lfs_app, http_auth):
        response = git_lfs_app.post_json(
            '/repo/info/lfs/objects/batch', params={}, status=400,
            extra_environ=http_auth)
        assert json.loads(response.text) == {
            'message': 'unsupported operation mode: `None`'}

    def test_app_batch_api_missing_objects(self, git_lfs_app, http_auth):
        response = git_lfs_app.post_json(
            '/repo/info/lfs/objects/batch', params={'operation': 'download'},
            status=400, extra_environ=http_auth)
        assert json.loads(response.text) == {
            'message': 'missing objects data'}

    def test_app_batch_api_unsupported_data_in_objects(
            self, git_lfs_app, http_auth):
        params = {'operation': 'download',
                  'objects': [{}]}
        response = git_lfs_app.post_json(
            '/repo/info/lfs/objects/batch', params=params, status=400,
            extra_environ=http_auth)
        assert json.loads(response.text) == {
            'message': 'unsupported data in objects'}

    def test_app_batch_api_download_missing_object(
            self, git_lfs_app, http_auth):
        params = {
            'operation': 'download',
            'objects': [{'oid': '123', 'size': '1024'}]
        }
        response = git_lfs_app.post_json(
            '/repo/info/lfs/objects/batch', params=params,
            extra_environ=http_auth)

        expected_objects = [
            {
                'oid': '123',
                'size': '1024',
                'authenticated': True,
                'errors': {'error': {'code': 404, 'message': 'object: 123 does not exist in store'}},
            }
        ]

        assert json.loads(response.text) == {
            'objects': expected_objects,
            'transfer': 'basic'
        }

    def test_app_batch_api_download(self, git_lfs_app, http_auth):
        oid = '456'
        oid_path = LFSOidStore(oid=oid, repo=None, store_location=git_lfs_app._store).oid_path
        if not os.path.isdir(os.path.dirname(oid_path)):
            os.makedirs(os.path.dirname(oid_path))
        with open(oid_path, 'wb') as f:
            f.write(safe_bytes('OID_CONTENT'))

        params = {'operation': 'download',
                  'objects': [{'oid': oid, 'size': '1024'}]}
        response = git_lfs_app.post_json(
            '/repo/info/lfs/objects/batch', params=params,
            extra_environ=http_auth)

        expected_objects = [
            {'authenticated': True,
             'actions': {
                 'download': {
                     'header': {'Authorization': 'Basic XXXXX'},
                     'href': 'http://localhost/repo/info/lfs/objects/456'},
             },
             'oid': '456',
             'size': '1024'}
        ]
        assert json.loads(response.text) == {
            'objects': expected_objects,
            'transfer': 'basic'
        }

    def test_app_batch_api_upload(self, git_lfs_app, http_auth):
        params = {'operation': 'upload',
                  'objects': [{'oid': '123', 'size': '1024'}]}
        response = git_lfs_app.post_json(
            '/repo/info/lfs/objects/batch', params=params,
            extra_environ=http_auth)
        expected_objects = [
            {
                'authenticated': True,
                'actions': {
                    'upload': {
                        'header': {
                            'Authorization': 'Basic XXXXX',
                            'Transfer-Encoding': 'chunked'
                        },
                        'href': 'http://localhost/repo/info/lfs/objects/123'
                    },
                    'verify': {
                        'header': {
                            'Authorization': 'Basic XXXXX'
                        },
                        'href': 'http://localhost/repo/info/lfs/verify'
                    }
                },
                'oid': '123',
                'size': '1024'
            }
        ]
        assert json.loads(response.text) == {
            'objects': expected_objects,
            'transfer': 'basic'
        }

    def test_app_batch_api_upload_for_https(self, git_lfs_https_app, http_auth):
        params = {'operation': 'upload',
                  'objects': [{'oid': '123', 'size': '1024'}]}
        response = git_lfs_https_app.post_json(
            '/repo/info/lfs/objects/batch', params=params,
            extra_environ=http_auth)
        expected_objects = [
            {'authenticated': True,
             'actions': {
                 'upload': {
                     'header': {'Authorization': 'Basic XXXXX',
                                'Transfer-Encoding': 'chunked'},
                     'href': 'https://localhost/repo/info/lfs/objects/123'},
                 'verify': {
                     'header': {'Authorization': 'Basic XXXXX'},
                     'href': 'https://localhost/repo/info/lfs/verify'}
             },
             'oid': '123',
             'size': '1024'}
        ]
        assert json.loads(response.text) == {
            'objects': expected_objects, 'transfer': 'basic'}

    def test_app_verify_api_missing_data(self, git_lfs_app):
        params = {'oid': 'missing'}
        response = git_lfs_app.post_json(
            '/repo/info/lfs/verify', params=params,
            status=400)

        assert json.loads(response.text) == {
            'message': 'missing oid and size in request data'}

    def test_app_verify_api_missing_obj(self, git_lfs_app):
        params = {'oid': 'missing', 'size': '1024'}
        response = git_lfs_app.post_json(
            '/repo/info/lfs/verify', params=params,
            status=404)

        assert json.loads(response.text) == {
            'message': 'oid `missing` does not exists in store'
        }

    def test_app_verify_api_size_mismatch(self, git_lfs_app):
        oid = 'existing'
        oid_path = LFSOidStore(oid=oid, repo=None, store_location=git_lfs_app._store).oid_path
        if not os.path.isdir(os.path.dirname(oid_path)):
            os.makedirs(os.path.dirname(oid_path))
        with open(oid_path, 'wb') as f:
            f.write(safe_bytes('OID_CONTENT'))

        params = {'oid': oid, 'size': '1024'}
        response = git_lfs_app.post_json(
            '/repo/info/lfs/verify', params=params, status=422)

        assert json.loads(response.text) == {
            'message': 'requested file size mismatch store size:11 requested:1024'
        }

    def test_app_verify_api(self, git_lfs_app):
        oid = 'existing'
        oid_path = LFSOidStore(oid=oid, repo=None, store_location=git_lfs_app._store).oid_path
        if not os.path.isdir(os.path.dirname(oid_path)):
            os.makedirs(os.path.dirname(oid_path))
        with open(oid_path, 'wb') as f:
            f.write(safe_bytes('OID_CONTENT'))

        params = {'oid': oid, 'size': 11}
        response = git_lfs_app.post_json(
            '/repo/info/lfs/verify', params=params)

        assert json.loads(response.text) == {
            'message': {'size': 11, 'oid': oid}
        }

    def test_app_download_api_oid_not_existing(self, git_lfs_app):
        oid = 'missing'

        response = git_lfs_app.get(f'/repo/info/lfs/objects/{oid}', status=404)

        assert json.loads(response.text) == {
            'message': 'requested file with oid `missing` not found in store'}

    def test_app_download_api(self, git_lfs_app):
        oid = 'existing'
        oid_path = LFSOidStore(oid=oid, repo=None, store_location=git_lfs_app._store).oid_path
        if not os.path.isdir(os.path.dirname(oid_path)):
            os.makedirs(os.path.dirname(oid_path))
        with open(oid_path, 'wb') as f:
            f.write(safe_bytes('OID_CONTENT'))

        response = git_lfs_app.get(f'/repo/info/lfs/objects/{oid}')
        assert response

    def test_app_upload(self, git_lfs_app):
        oid = '65f23e22a9bfedda96929b3cfcb8b6d2fdd34a2e877ddb81f45d79ab05710e12'

        response = git_lfs_app.put(
            f'/repo/info/lfs/objects/{oid}', params='CONTENT')

        assert json.loads(response.text) == {'upload': 'ok', 'state': 'written'}

        # verify that we actually wrote that OID
        oid_path = LFSOidStore(oid=oid, repo=None, store_location=git_lfs_app._store).oid_path
        assert os.path.isfile(oid_path)
        assert 'CONTENT' == open(oid_path).read()

        response = git_lfs_app.put(
            f'/repo/info/lfs/objects/{oid}', params='CONTENT')

        assert json.loads(response.text) == {'upload': 'ok', 'state': 'in-store'}


    def test_app_upload_wrong_sha(self, git_lfs_app):
        oid = 'i-am-a-wrong-sha'

        response = git_lfs_app.put(f'/repo/info/lfs/objects/{oid}', params='CONTENT', status=400)

        assert json.loads(response.text) == {
            'message': 'oid i-am-a-wrong-sha does not match expected sha '
                       '65f23e22a9bfedda96929b3cfcb8b6d2fdd34a2e877ddb81f45d79ab05710e12'}

        # check this OID wasn't written to store
        response = git_lfs_app.get(f'/repo/info/lfs/objects/{oid}', status=404)
        assert json.loads(response.text) == {'message': 'requested file with oid `i-am-a-wrong-sha` not found in store'}