git-lfs: fixed tests for upload, and added new for file verification.

vcsserver/git_lfs/lib.py

@@ -1,172 +1,175 b''
 # RhodeCode VCSServer provides access to different vcs backends via network.
 # Copyright (C) 2014-2017 RodeCode 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 shutil
 import logging
 from collections import OrderedDict
 
 log = logging.getLogger(__name__)
 
 
 class OidHandler(object):
 
     def __init__(self, store, repo_name, auth, oid, obj_size, obj_data, obj_href,
                  obj_verify_href=None):
         self.current_store = store
         self.repo_name = repo_name
         self.auth = auth
         self.oid = oid
         self.obj_size = obj_size
         self.obj_data = obj_data
         self.obj_href = obj_href
         self.obj_verify_href = obj_verify_href
 
     def get_store(self, mode=None):
         return self.current_store
 
     def get_auth(self):
         """returns auth header for re-use in upload/download"""
         return " ".join(self.auth)
 
     def download(self):
 
         store = self.get_store()
         response = None
         has_errors = None
 
         if not store.has_oid():
             # error reply back to client that something is wrong with dl
             err_msg = 'object: {} does not exist in store'.format(store.oid)
             has_errors = OrderedDict(
                 error=OrderedDict(
                     code=404,
                     message=err_msg
                 )
             )
 
         download_action = OrderedDict(
             href=self.obj_href,
             header=OrderedDict([("Authorization", self.get_auth())])
         )
         if not has_errors:
             response = OrderedDict(download=download_action)
         return response, has_errors
 
     def upload(self, skip_existing=True):
         """
         Write upload action for git-lfs server
         """
 
         store = self.get_store()
         response = None
         has_errors = None
 
         # verify if we have the OID before, if we do, reply with empty
         if store.has_oid():
             log.debug('LFS: store already has oid %s', store.oid)
 
             # validate size
-            size_match = store.size_oid() == self.obj_size
+            store_size = store.size_oid()
+            size_match = store_size == self.obj_size
             if not size_match:
-                log.warning('LFS: size mismatch for oid:%s', self.oid)
+                log.warning(
+                    'LFS: size mismatch for oid:%s, in store:%s expected: %s',
+                    self.oid, store_size, self.obj_size)
             elif skip_existing:
                 log.debug('LFS: skipping further action as oid is existing')
                 return response, has_errors
 
         chunked = ("Transfer-Encoding", "chunked")
         upload_action = OrderedDict(
             href=self.obj_href,
             header=OrderedDict([("Authorization", self.get_auth()), chunked])
         )
         if not has_errors:
             response = OrderedDict(upload=upload_action)
             # if specified in handler, return the verification endpoint
             if self.obj_verify_href:
                 verify_action = OrderedDict(
                     href=self.obj_verify_href,
                     header=OrderedDict([("Authorization", self.get_auth())])
                 )
                 response['verify'] = verify_action
         return response, has_errors
 
     def exec_operation(self, operation, *args, **kwargs):
         handler = getattr(self, operation)
         log.debug('LFS: handling request using %s handler', handler)
         return handler(*args, **kwargs)
 
 
 class LFSOidStore(object):
 
     def __init__(self, oid, repo, store_location=None):
         self.oid = oid
         self.repo = repo
         self.store_path = store_location or self.get_default_store()
         self.tmp_oid_path = os.path.join(self.store_path, oid + '.tmp')
         self.oid_path = os.path.join(self.store_path, oid)
         self.fd = None
 
     def get_engine(self, mode):
         """
         engine = .get_engine(mode='wb')
         with engine as f:
             f.write('...')
         """
 
         class StoreEngine(object):
             def __init__(self, mode, store_path, oid_path, tmp_oid_path):
                 self.mode = mode
                 self.store_path = store_path
                 self.oid_path = oid_path
                 self.tmp_oid_path = tmp_oid_path
 
             def __enter__(self):
                 if not os.path.isdir(self.store_path):
                     os.makedirs(self.store_path)
 
                 # TODO(marcink): maybe write metadata here with size/oid ?
                 fd = open(self.tmp_oid_path, self.mode)
                 self.fd = fd
                 return fd
 
             def __exit__(self, exc_type, exc_value, traceback):
                 # close tmp file, and rename to final destination
                 self.fd.close()
                 shutil.move(self.tmp_oid_path, self.oid_path)
 
         return StoreEngine(
             mode, self.store_path, self.oid_path, self.tmp_oid_path)
 
     def get_default_store(self):
         """
         Default store, consistent with defaults of Mercurial large files store
         which is /home/username/.cache/largefiles
         """
         user_home = os.path.expanduser("~")
         return os.path.join(user_home, '.cache', 'lfs-store')
 
     def has_oid(self):
         return os.path.exists(os.path.join(self.store_path, self.oid))
 
     def size_oid(self):
         size = -1
 
         if self.has_oid():
             oid = os.path.join(self.store_path, self.oid)
             size = os.stat(oid).st_size
 
         return size

vcsserver/git_lfs/tests/test_lib.py

@@ -1,123 +1,141 b''
 # RhodeCode VCSServer provides access to different vcs backends via network.
 # Copyright (C) 2014-2017 RodeCode 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 vcsserver.git_lfs.lib import OidHandler, LFSOidStore
 
 
 @pytest.fixture()
 def lfs_store(tmpdir):
     repo = 'test'
     oid = '123456789'
     store = LFSOidStore(oid=oid, repo=repo, store_location=str(tmpdir))
     return store
 
 
 @pytest.fixture()
 def oid_handler(lfs_store):
     store = lfs_store
     repo = store.repo
     oid = store.oid
 
     oid_handler = OidHandler(
         store=store, repo_name=repo, auth=('basic', 'xxxx'),
         oid=oid,
         obj_size='1024', obj_data={}, obj_href='http://localhost/handle_oid',
         obj_verify_href='http://localhost/verify')
     return oid_handler
 
 
 class TestOidHandler(object):
 
     @pytest.mark.parametrize('exec_action', [
         'download',
         'upload',
     ])
     def test_exec_action(self, exec_action, oid_handler):
         handler = oid_handler.exec_operation(exec_action)
         assert handler
 
     def test_exec_action_undefined(self, oid_handler):
         with pytest.raises(AttributeError):
             oid_handler.exec_operation('wrong')
 
     def test_download_oid_not_existing(self, oid_handler):
         response, has_errors = oid_handler.exec_operation('download')
 
         assert response is None
         assert has_errors['error'] == {
             'code': 404,
             'message': 'object: 123456789 does not exist in store'}
 
     def test_download_oid(self, oid_handler):
         store = oid_handler.get_store()
         if not os.path.isdir(os.path.dirname(store.oid_path)):
             os.makedirs(os.path.dirname(store.oid_path))
 
         with open(store.oid_path, 'wb') as f:
             f.write('CONTENT')
 
         response, has_errors = oid_handler.exec_operation('download')
 
         assert has_errors is None
         assert response['download'] == {
             'header': {'Authorization': 'basic xxxx'},
             'href': 'http://localhost/handle_oid'
         }
 
     def test_upload_oid_that_exists(self, oid_handler):
         store = oid_handler.get_store()
         if not os.path.isdir(os.path.dirname(store.oid_path)):
             os.makedirs(os.path.dirname(store.oid_path))
 
         with open(store.oid_path, 'wb') as f:
             f.write('CONTENT')
-
+        oid_handler.obj_size = 7
         response, has_errors = oid_handler.exec_operation('upload')
         assert has_errors is None
         assert response is None
 
+    def test_upload_oid_that_exists_but_has_wrong_size(self, oid_handler):
+        store = oid_handler.get_store()
+        if not os.path.isdir(os.path.dirname(store.oid_path)):
+            os.makedirs(os.path.dirname(store.oid_path))
+
+        with open(store.oid_path, 'wb') as f:
+            f.write('CONTENT')
+
+        oid_handler.obj_size = 10240
+        response, has_errors = oid_handler.exec_operation('upload')
+        assert has_errors is None
+        assert response['upload'] == {
+            'header': {'Authorization': 'basic xxxx',
+                       'Transfer-Encoding': 'chunked'},
+            'href': 'http://localhost/handle_oid',
+        }
+
     def test_upload_oid(self, oid_handler):
         response, has_errors = oid_handler.exec_operation('upload')
         assert has_errors is None
         assert response['upload'] == {
-            'header': {'Authorization': 'basic xxxx'},
+            'header': {'Authorization': 'basic xxxx',
+                       'Transfer-Encoding': 'chunked'},
             'href': 'http://localhost/handle_oid'
         }
 
 
 class TestLFSStore(object):
     def test_write_oid(self, lfs_store):
         oid_location = lfs_store.oid_path
 
         assert not os.path.isfile(oid_location)
 
         engine = lfs_store.get_engine(mode='wb')
         with engine as f:
             f.write('CONTENT')
 
         assert os.path.isfile(oid_location)
 
     def test_detect_has_oid(self, lfs_store):
 
         assert lfs_store.has_oid() is False
         engine = lfs_store.get_engine(mode='wb')
         with engine as f:
             f.write('CONTENT')
 
         assert lfs_store.has_oid() is True
  No newline at end of file