You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
111 lines
3.3 KiB
111 lines
3.3 KiB
# bsv - Backup, Synchronization, Versioning
|
|
# Copyright (C) 2023 Simon Boyé
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero 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 Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
|
from __future__ import annotations
|
|
from datetime import UTC, datetime
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
from random import randbytes
|
|
from typing import Iterator
|
|
|
|
import pytest
|
|
from tempfile import TemporaryDirectory
|
|
|
|
from bsv.repository import Repository, Snapshot, Tree, TreeItem, create_repository, timestamp_from_time
|
|
from bsv.simple_cas.cas import Digest
|
|
|
|
|
|
@pytest.fixture
|
|
def tmp_dir():
|
|
with TemporaryDirectory(prefix="simple_cas_") as tmp_dir:
|
|
yield Path(tmp_dir)
|
|
|
|
|
|
@pytest.fixture
|
|
def repo(tmp_dir):
|
|
return create_repository(
|
|
tmp_dir / "bsv",
|
|
"test_repo",
|
|
)
|
|
|
|
|
|
def test_read_write_blob(tmp_dir: Path, repo: Repository):
|
|
path = tmp_dir / "test.dat"
|
|
make_random_file(path, 1 << 20)
|
|
|
|
with path.open("rb") as stream:
|
|
digest = repo.add_blob(stream)
|
|
|
|
blob = repo.get_blob(digest)
|
|
data = blob.reader().read()
|
|
|
|
with path.open("rb") as stream:
|
|
assert data == stream.read()
|
|
|
|
|
|
def test_read_write_tree(repo: Repository):
|
|
now = datetime.now(UTC)
|
|
tree = Tree(
|
|
repo,
|
|
[
|
|
TreeItem(
|
|
"xyz",
|
|
Digest(bytes([42]) * repo._cas._digest_size),
|
|
0o744,
|
|
creation_timestamp = timestamp_from_time(now),
|
|
modification_timestamp = timestamp_from_time(now),
|
|
),
|
|
TreeItem(
|
|
"foobar",
|
|
Digest(bytes([123]) * repo._cas._digest_size),
|
|
0o777,
|
|
creation_timestamp = timestamp_from_time(now),
|
|
modification_timestamp = timestamp_from_time(now),
|
|
),
|
|
]
|
|
)
|
|
|
|
assert Tree.from_bytes(repo, tree.to_bytes()) == tree
|
|
|
|
digest = repo.add_tree(tree)
|
|
assert repo.get_tree(digest) == tree
|
|
|
|
|
|
def test_read_write_snapshot(repo: Repository):
|
|
snapshot = Snapshot(
|
|
repo = repo,
|
|
tree_digest = Digest(bytes([42]) * repo._cas._digest_size),
|
|
repo_name = "test_repo",
|
|
timestamp = timestamp_from_time(datetime.now()),
|
|
)
|
|
|
|
assert Snapshot.from_bytes(repo, snapshot.to_bytes()) == snapshot
|
|
|
|
digest = repo.add_snapshot(snapshot)
|
|
assert repo.get_snapshot(digest) == snapshot
|
|
|
|
|
|
def make_random_file(path: Path, size: int):
|
|
with path.open("wb") as stream:
|
|
for chunk_size in iter_chunks(size):
|
|
stream.write(randbytes(chunk_size))
|
|
|
|
def iter_chunks(size: int, chunk_size: int=1 << 16) -> Iterator[int]:
|
|
num_full_chunks = (size - 1) // chunk_size
|
|
for _ in range(num_full_chunks):
|
|
yield chunk_size
|
|
offset = num_full_chunks * chunk_size
|
|
if offset != size:
|
|
yield size - offset
|
|
|