diff --git a/dulwich/lfs.py b/dulwich/lfs.py new file mode 100644 index 00000000..8d14dfea --- /dev/null +++ b/dulwich/lfs.py @@ -0,0 +1,75 @@ +# lfs.py -- Implementation of the LFS +# Copyright (C) 2020 Jelmer Vernooij +# +# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU +# General Public License as public by the Free Software Foundation; version 2.0 +# or (at your option) any later version. You can redistribute it and/or +# modify it under the terms of either of these two licenses. +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# You should have received a copy of the licenses; if not, see +# for a copy of the GNU General Public License +# and for a copy of the Apache +# License, Version 2.0. +# + +import hashlib +import os +import tempfile + + +class LFSStore(object): + """Stores objects on disk, indexed by SHA256.""" + + def __init__(self, path): + self.path = path + + @classmethod + def create(cls, lfs_dir): + if not os.path.isdir(lfs_dir): + os.mkdir(lfs_dir) + os.mkdir(os.path.join(lfs_dir, 'tmp')) + os.mkdir(os.path.join(lfs_dir, 'objects')) + return cls(lfs_dir) + + @classmethod + def from_repo(cls, repo, create=False): + lfs_dir = os.path.join(repo.controldir, 'lfs') + if create: + return cls.create(lfs_dir) + return cls(lfs_dir) + + def _sha_path(self, sha): + return os.path.join(self.path, 'objects', sha[0:2], sha[2:4], sha) + + def open_object(self, sha): + """Open an object by sha.""" + try: + return open(self._sha_path(sha), 'rb') + except FileNotFoundError: + raise KeyError(sha) + + def write_object(self, chunks): + """Write an object. + + Returns: object SHA + """ + sha = hashlib.sha256() + tmpdir = os.path.join(self.path, 'tmp') + with tempfile.NamedTemporaryFile( + dir=tmpdir, mode='wb', delete=False) as f: + for chunk in chunks: + sha.update(chunk) + f.write(chunk) + f.flush() + tmppath = f.name + path = self._sha_path(sha.hexdigest()) + if not os.path.exists(os.path.dirname(path)): + os.makedirs(os.path.dirname(path)) + os.rename(tmppath, path) + return sha.hexdigest() diff --git a/dulwich/tests/__init__.py b/dulwich/tests/__init__.py index eea22fa3..f46cff60 100644 --- a/dulwich/tests/__init__.py +++ b/dulwich/tests/__init__.py @@ -1,197 +1,198 @@ # __init__.py -- The tests for dulwich # Copyright (C) 2007 James Westby # # Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU # General Public License as public by the Free Software Foundation; version 2.0 # or (at your option) any later version. You can redistribute it and/or # modify it under the terms of either of these two licenses. # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # # You should have received a copy of the licenses; if not, see # for a copy of the GNU General Public License # and for a copy of the Apache # License, Version 2.0. # """Tests for Dulwich.""" import doctest import os import shutil import subprocess import sys import tempfile # If Python itself provides an exception, use that import unittest from unittest import ( # noqa: F401 SkipTest, TestCase as _TestCase, skipIf, expectedFailure, ) class TestCase(_TestCase): def setUp(self): super(TestCase, self).setUp() self._old_home = os.environ.get("HOME") os.environ["HOME"] = "/nonexistant" def tearDown(self): super(TestCase, self).tearDown() if self._old_home: os.environ["HOME"] = self._old_home else: del os.environ["HOME"] class BlackboxTestCase(TestCase): """Blackbox testing.""" # TODO(jelmer): Include more possible binary paths. bin_directories = [os.path.abspath(os.path.join( os.path.dirname(__file__), "..", "..", "bin")), '/usr/bin', '/usr/local/bin'] def bin_path(self, name): """Determine the full path of a binary. Args: name: Name of the script Returns: Full path """ for d in self.bin_directories: p = os.path.join(d, name) if os.path.isfile(p): return p else: raise SkipTest("Unable to find binary %s" % name) def run_command(self, name, args): """Run a Dulwich command. Args: name: Name of the command, as it exists in bin/ args: Arguments to the command """ env = dict(os.environ) env["PYTHONPATH"] = os.pathsep.join(sys.path) # Since they don't have any extensions, Windows can't recognize # executablility of the Python files in /bin. Even then, we'd have to # expect the user to set up file associations for .py files. # # Save us from all that headache and call python with the bin script. argv = [sys.executable, self.bin_path(name)] + args return subprocess.Popen( argv, stdout=subprocess.PIPE, stdin=subprocess.PIPE, stderr=subprocess.PIPE, env=env) def self_test_suite(): names = [ 'archive', 'blackbox', 'client', 'config', 'diff_tree', 'fastexport', 'file', 'grafts', 'greenthreads', 'hooks', 'ignore', 'index', + 'lfs', 'line_ending', 'lru_cache', 'mailmap', 'objects', 'objectspec', 'object_store', 'missing_obj_finder', 'pack', 'patch', 'porcelain', 'protocol', 'reflog', 'refs', 'repository', 'server', 'stash', 'utils', 'walk', 'web', ] module_names = ['dulwich.tests.test_' + name for name in names] loader = unittest.TestLoader() return loader.loadTestsFromNames(module_names) def tutorial_test_suite(): import dulwich.client # noqa: F401 import dulwich.config # noqa: F401 import dulwich.index # noqa: F401 import dulwich.reflog # noqa: F401 import dulwich.repo # noqa: F401 import dulwich.server # noqa: F401 import dulwich.patch # noqa: F401 tutorial = [ 'introduction', 'file-format', 'repo', 'object-store', 'remote', 'conclusion', ] tutorial_files = ["../../docs/tutorial/%s.txt" % name for name in tutorial] def setup(test): test.__old_cwd = os.getcwd() test.tempdir = tempfile.mkdtemp() test.globs.update({'tempdir': test.tempdir}) os.chdir(test.tempdir) def teardown(test): os.chdir(test.__old_cwd) shutil.rmtree(test.tempdir) return doctest.DocFileSuite( module_relative=True, package='dulwich.tests', setUp=setup, tearDown=teardown, *tutorial_files) def nocompat_test_suite(): result = unittest.TestSuite() result.addTests(self_test_suite()) result.addTests(tutorial_test_suite()) from dulwich.contrib import test_suite as contrib_test_suite result.addTests(contrib_test_suite()) return result def compat_test_suite(): result = unittest.TestSuite() from dulwich.tests.compat import test_suite as compat_test_suite result.addTests(compat_test_suite()) return result def test_suite(): result = unittest.TestSuite() result.addTests(self_test_suite()) if sys.platform != 'win32': result.addTests(tutorial_test_suite()) from dulwich.tests.compat import test_suite as compat_test_suite result.addTests(compat_test_suite()) from dulwich.contrib import test_suite as contrib_test_suite result.addTests(contrib_test_suite()) return result diff --git a/dulwich/tests/test_lfs.py b/dulwich/tests/test_lfs.py new file mode 100644 index 00000000..c4ec5897 --- /dev/null +++ b/dulwich/tests/test_lfs.py @@ -0,0 +1,44 @@ +# test_lfs.py -- tests for LFS +# Copyright (C) 2020 Jelmer Vernooij +# +# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU +# General Public License as public by the Free Software Foundation; version 2.0 +# or (at your option) any later version. You can redistribute it and/or +# modify it under the terms of either of these two licenses. +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# You should have received a copy of the licenses; if not, see +# for a copy of the GNU General Public License +# and for a copy of the Apache +# License, Version 2.0. +# + +"""Tests for LFS support.""" + +from . import TestCase +from ..lfs import LFSStore +import shutil +import tempfile + + +class LFSTests(TestCase): + + def setUp(self): + super(LFSTests, self).setUp() + self.test_dir = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, self.test_dir) + self.lfs = LFSStore.create(self.test_dir) + + def test_create(self): + sha = self.lfs.write_object([b'a', b'b']) + with self.lfs.open_object(sha) as f: + self.assertEqual(b'ab', f.read()) + + def test_missing(self): + self.assertRaises( + KeyError, self.lfs.open_object, 'abcdeabcdeabcdeabcde')