diff --git a/src/borg/archive.py b/src/borg/archive.py index 66204cc9ae..c9be78f067 100644 --- a/src/borg/archive.py +++ b/src/borg/archive.py @@ -40,7 +40,7 @@ from .helpers import safe_encode, safe_decode, make_path_safe, remove_surrogates from .helpers import StableDict from .helpers import bin_to_hex -from .helpers import safe_ns +from .helpers import safe_ns, pax_time_to_ns from .helpers import ellipsis_truncate, ProgressIndicatorPercent, log_multi from .helpers import os_open, flags_normal, flags_dir from .helpers import os_stat @@ -1636,6 +1636,28 @@ def process_file(self, *, path, parent_fd, name, st, cache, flags=flags_normal, return status +def tar_acl_to_borg(acl): + """Convert a POSIX ACL text from a tar PAX header (SCHILY.acl.*) to borg's ACL format. + + Borg separates entries by newlines and appends the numeric uid/gid as a 4th field to + named user/group entries (user:name:perms:uid), see acl_get on Linux and FreeBSD. + star appends it too, but separates entries by commas. GNU tar separates entries by + newlines and does not append the numeric id, so we look it up locally (like borg create + does), falling back to the name. + """ + entries = [] + for entry in acl.replace(',', '\n').split('\n'): + entry = entry.split('#', 1)[0].strip() # remove comments + if not entry: + continue + fields = entry.split(':') + if len(fields) == 3 and fields[1] and fields[0] in ('user', 'group'): + name = fields[1] + fields.append(str(user2uid(name, name) if fields[0] == 'user' else group2gid(name, name))) + entries.append(':'.join(fields)) + return '\n'.join(entries).encode('utf-8', errors='surrogateescape') + + class TarfileObjectProcessors: def __init__(self, *, cache, key, add_item, process_file_chunks, @@ -1659,6 +1681,32 @@ def create_helper(self, tarinfo, status=None, type=None): item = Item(path=make_path_safe(normalized_path), mode=tarinfo.mode | type, uid=tarinfo.uid, gid=tarinfo.gid, user=tarinfo.uname or None, group=tarinfo.gname or None, mtime=safe_ns(int(tarinfo.mtime * 1000**3))) + ph = tarinfo.pax_headers + if ph: + # the tarfile module only gives us float timestamps, parse the original strings for full precision. + for name in 'atime', 'ctime', 'mtime': + if name in ph: + ns = pax_time_to_ns(ph[name]) + if ns is not None: + setattr(item, name, ns) + xattrs = StableDict() + for key, value in ph.items(): + if key.startswith(SCHILY_XATTR): + key = key.removeprefix(SCHILY_XATTR) + if key.startswith('system.posix_acl_'): + # like borg create, we store the POSIX ACLs separately, not as xattrs. + continue + # the tarfile code gives us str keys and str values, + # but we need bytes keys and bytes values (or None for an empty value, like xattr.get_all). + bkey = key.encode('utf-8', errors='surrogateescape') + bvalue = value.encode('utf-8', errors='surrogateescape') + xattrs[bkey] = bvalue or None + elif key == SCHILY_ACL_ACCESS: + item.acl_access = tar_acl_to_borg(value) + elif key == SCHILY_ACL_DEFAULT: + item.acl_default = tar_acl_to_borg(value) + if xattrs: + item.xattrs = xattrs yield item, status # if we get here, "with"-block worked ok without error/exception, the item was processed ok... self.add_item(item, stats=self.stats) diff --git a/src/borg/archiver.py b/src/borg/archiver.py index ede7b5d569..e7aa5f8bdb 100644 --- a/src/borg/archiver.py +++ b/src/borg/archiver.py @@ -85,6 +85,7 @@ from .helpers import sig_int, ignore_sigint from .helpers import iter_separated from .helpers import get_tar_filter + from .helpers import ns_to_pax_time from .helpers import ignore_invalid_archive_tam from .helpers.parseformat import BorgJsonEncoder, safe_decode from .nanorst import rst_to_terminal @@ -248,6 +249,40 @@ def __call__(self, parser, namespace, values, option_string=None): setattr(namespace, self.dest, values) +def item_to_paxheaders(item): + """Transform (parts of) a Borg *item* into a pax_headers dict.""" + # PAX format + # ---------- + # When using the PAX (POSIX) format, we can support some things that aren't possible + # with classic tar formats, including GNU tar, such as: + # - atime, ctime, mtime with ns precision (DONE) + # - xattrs, POSIX ACLs (DONE) + # - various additions supported by GNU tar in POSIX mode (TODO) + # + ph = {} + # note: for mtime this is a bit redundant as it is already done by tarfile module, + # but it only has a float, so we do it in our way to have exact ns precision. + for name in 'atime', 'ctime', 'mtime': + if hasattr(item, name): + ns = getattr(item, name) + ph[name] = ns_to_pax_time(ns) + if hasattr(item, 'xattrs'): + for bkey, bvalue in item.xattrs.items(): + # we have bytes key and bytes value (or None for an empty value), but the tarfile code + # expects str key and str value. + key = SCHILY_XATTR + bkey.decode('utf-8', errors='surrogateescape') + value = (bvalue or b'').decode('utf-8', errors='surrogateescape') + ph[key] = value + # Add POSIX access and default ACL if present + acl_access = item.get('acl_access') + if acl_access is not None: + ph[SCHILY_ACL_ACCESS] = acl_access.decode('utf-8', errors='surrogateescape') + acl_default = item.get('acl_default') + if acl_default is not None: + ph[SCHILY_ACL_DEFAULT] = acl_default.decode('utf-8', errors='surrogateescape') + return ph + + class Archiver: def __init__(self, lock_wait=None, prog=None): @@ -1054,7 +1089,8 @@ def peek_and_store_hardlink_masters(item, matched): # The | (pipe) symbol instructs tarfile to use a streaming mode of operation # where it never seeks on the passed fileobj. - tar = tarfile.open(fileobj=tarstream, mode='w|', format=tarfile.GNU_FORMAT) + tar_format = dict(GNU=tarfile.GNU_FORMAT, PAX=tarfile.PAX_FORMAT)[args.tar_format] + tar = tarfile.open(fileobj=tarstream, mode='w|', format=tar_format) if progress: pi = ProgressIndicatorPercent(msg='%5.1f%% Processing: %s', step=0.1, msgid='extract') @@ -1085,13 +1121,6 @@ def item_to_tarinfo(item, original_path): the file contents, if any, and is None otherwise. When *tarinfo* is None, the *item* cannot be represented as a TarInfo object and should be skipped. """ - - # If we would use the PAX (POSIX) format (which we currently don't), - # we can support most things that aren't possible with classic tar - # formats, including GNU tar, such as: - # atime, ctime, possibly Linux capabilities (security.* xattrs) - # and various additions supported by GNU tar in POSIX mode. - stream = None tarinfo = tarfile.TarInfo() tarinfo.name = item.path @@ -1159,6 +1188,8 @@ def item_to_tarinfo(item, original_path): item.path = os.sep.join(orig_path.split(os.sep)[strip_components:]) tarinfo, stream = item_to_tarinfo(item, orig_path) if tarinfo: + if args.tar_format == 'PAX': + tarinfo.pax_headers = item_to_paxheaders(item) if output_list: logging.getLogger('borg.output.list').info(remove_surrogates(orig_path)) tar.addfile(tarinfo, stream) @@ -4448,12 +4479,17 @@ def diff_sort_spec_validator(s): read the uncompressed tar stream from stdin and write a compressed/filtered tar stream to stdout. - The generated tarball uses the GNU tar format. + Depending on the ``--tar-format`` option, these formats are created: - export-tar is a lossy conversion: - BSD flags, ACLs, extended attributes (xattrs), atime and ctime are not exported. - Timestamp resolution is limited to whole seconds, not the nanosecond resolution - otherwise supported by Borg. + +--------------+---------------------------+----------------------------+ + | --tar-format | Specification | Metadata | + +--------------+---------------------------+----------------------------+ + | PAX | POSIX.1-2001 (pax) format | GNU + atime/ctime/mtime ns | + | | | + xattrs, POSIX ACLs | + +--------------+---------------------------+----------------------------+ + | GNU | GNU tar format | mtime s, no atime/ctime, | + | | | no ACLs/xattrs/bsdflags | + +--------------+---------------------------+----------------------------+ A ``--sparse`` option (as found in ``borg extract``) is not supported. @@ -4476,6 +4512,9 @@ def diff_sort_spec_validator(s): help='filter program to pipe data through') subparser.add_argument('--list', dest='output_list', action='store_true', help='output verbose list of items (files, dirs, ...)') + subparser.add_argument('--tar-format', metavar='FMT', dest='tar_format', default='GNU', + choices=('PAX', 'GNU'), action=Highlander, + help='select tar format: PAX or GNU (default: GNU)') subparser.add_argument('location', metavar='ARCHIVE', type=location_validator(archive=True), help='archive to export') @@ -5619,15 +5658,17 @@ def diff_sort_spec_validator(s): Most documentation of ``borg create`` applies. Note that this command does not support excluding files. - import-tar is a lossy conversion: - BSD flags, ACLs, extended attributes (xattrs), atime and ctime are not exported. - Timestamp resolution is limited to whole seconds, not the nanosecond resolution - otherwise supported by Borg. - A ``--sparse`` option (as found in borg create) is not supported. - import-tar reads POSIX.1-1988 (ustar), POSIX.1-2001 (pax), GNU tar, UNIX V7 tar - and SunOS tar with extended attributes. + About tar formats and metadata conservation or loss, please see ``borg export-tar``. + + import-tar reads these tar formats: + + - PAX: POSIX.1-2001 + - GNU: GNU tar + - POSIX.1-1988 (ustar) + - UNIX V7 tar + - SunOS tar with extended attributes To import multiple tarballs into a single archive, they can be simply concatenated (e.g. using "cat") into a single file, and imported with an diff --git a/src/borg/constants.py b/src/borg/constants.py index 9fc572fee6..2c71a19a5c 100644 --- a/src/borg/constants.py +++ b/src/borg/constants.py @@ -92,6 +92,11 @@ FILES_CACHE_MODE_UI_DEFAULT = 'ctime,size,inode' # default for "borg create" command (CLI UI) FILES_CACHE_MODE_DISABLED = 'd' # Most Borg commands do not use the files cache at all (disable). +# tar related +SCHILY_XATTR = 'SCHILY.xattr.' # xattr key prefix in tar PAX headers +SCHILY_ACL_ACCESS = 'SCHILY.acl.access' # POSIX access ACL in tar PAX headers +SCHILY_ACL_DEFAULT = 'SCHILY.acl.default' # POSIX default ACL in tar PAX headers + # return codes returned by borg command EXIT_SUCCESS = 0 # everything done, no problems EXIT_WARNING = 1 # reached normal end of operation, but there were issues (generic warning) diff --git a/src/borg/helpers/time.py b/src/borg/helpers/time.py index 86f6583652..f459bf4170 100644 --- a/src/borg/helpers/time.py +++ b/src/borg/helpers/time.py @@ -1,6 +1,7 @@ import os import time from datetime import datetime, timezone +from decimal import Decimal, InvalidOperation from ..constants import ISO_FORMAT, ISO_FORMAT_NO_USECS @@ -89,6 +90,21 @@ def safe_ns(ts): return MAX_NS +def ns_to_pax_time(ns): + """Format a nanoseconds timestamp as an exact decimal seconds string for a tar PAX header.""" + sign = '-' if ns < 0 else '' + s, ns = divmod(abs(ns), 1000000000) + return f'{sign}{s}.{ns:09d}' + + +def pax_time_to_ns(value): + """Parse a tar PAX header timestamp (decimal seconds string) into nanoseconds, return None if invalid.""" + try: + return safe_ns(int(Decimal(value).scaleb(9))) + except (InvalidOperation, ValueError, OverflowError): + return None + + def safe_timestamp(item_timestamp_ns): t_ns = safe_ns(item_timestamp_ns) return datetime.fromtimestamp(t_ns / 1e9) diff --git a/src/borg/testsuite/archive.py b/src/borg/testsuite/archive.py index 7358a0454e..3378dc0fe5 100644 --- a/src/borg/testsuite/archive.py +++ b/src/borg/testsuite/archive.py @@ -1,4 +1,5 @@ import json +import os from collections import OrderedDict from datetime import datetime, timezone from io import StringIO @@ -9,11 +10,12 @@ from . import BaseTestCase from ..crypto.key import PlaintextKey from ..archive import Archive, CacheChunkBuffer, RobustUnpacker, valid_msgpacked_dict, ITEM_KEYS, Statistics -from ..archive import BackupOSError, backup_io, backup_io_iter, get_item_uid_gid +from ..archive import BackupOSError, backup_io, backup_io_iter, get_item_uid_gid, tar_acl_to_borg from ..helpers import Manifest from ..helpers import msgpack from ..item import Item, ArchiveItem from ..platform import uid2user, gid2group +from ..platformflags import is_win32 @pytest.fixture() @@ -368,3 +370,31 @@ def test_get_item_uid_gid(): # because item uid/gid seems valid, do not use the given uid/gid defaults assert uid == 9 assert gid == 10 + + +@pytest.mark.parametrize('acl, expected', [ + # GNU tar: newline separated, no numeric id for named entries + ('user::rw-\nuser:{user}:rw-\ngroup::r--\nmask::rw-\nother::r--\n', + 'user::rw-\nuser:{user}:rw-:{uid}\ngroup::r--\nmask::rw-\nother::r--'), + # star: comma separated, numeric id appended + ('user::rw-,user:root:rw-:0,group::r--,mask::rw-,other::r--', + 'user::rw-\nuser:root:rw-:0\ngroup::r--\nmask::rw-\nother::r--'), + # borg export-tar: newline separated, numeric id appended + ('user::rw-\nuser:root:rw-:0\ngroup::r--\nmask::rw-\nother::r--', + 'user::rw-\nuser:root:rw-:0\ngroup::r--\nmask::rw-\nother::r--'), + # unknown names fall back to the name (no name lookups on Windows) + pytest.param('group:nosuchgroup-borgtest:r--', 'group:nosuchgroup-borgtest:r--:nosuchgroup-borgtest', + marks=pytest.mark.skipif(is_win32, reason='no name lookups on Windows')), + # comments get removed + ('user:{user}:r--\t#effective:r--\n', 'user:{user}:r--:{uid}'), + ('', ''), +]) +def test_tar_acl_to_borg(acl, expected): + # the name lookups need an existing user, e.g. Haiku has no "root" user. + try: + uid = os.getuid() # UNIX only + except AttributeError: + uid = 0 + user = uid2user(uid) + acl, expected = acl.format(user=user, uid=uid), expected.format(user=user, uid=uid) + assert tar_acl_to_borg(acl) == expected.encode() diff --git a/src/borg/testsuite/archiver.py b/src/borg/testsuite/archiver.py index 6102b8419d..e265364484 100644 --- a/src/borg/testsuite/archiver.py +++ b/src/borg/testsuite/archiver.py @@ -13,6 +13,7 @@ import stat import subprocess import sys +import tarfile import tempfile import time import unittest @@ -59,7 +60,8 @@ from . import has_lchflags, has_mknod, llfuse from . import BaseTestCase, changedir, environment_variable, filter_xattrs, same_ts_ns, granularity_sleep from . import are_symlinks_supported, are_hardlinks_supported, are_fifos_supported, is_utime_fully_supported, is_birthtime_fully_supported -from .platform import fakeroot_detected, is_darwin, is_freebsd, is_netbsd, is_win32, is_haiku +from .platform import fakeroot_detected, is_darwin, is_freebsd, is_linux, is_netbsd, is_win32, is_haiku +from .platform import are_acls_working from .upgrader import make_attic_repo from . import key @@ -4202,6 +4204,192 @@ def test_import_tar(self): self.cmd('extract', self.repository_location + '::dst') self.assert_dirs_equal('input', 'output/input', ignore_ns=True, ignore_xattrs=True) + def test_import_tar_pax(self): + self.create_test_files() + os.unlink('input/flagfile') + self.cmd('init', '--encryption=none', self.repository_location) + self.cmd('create', self.repository_location + '::src', 'input') + self.cmd('export-tar', '--tar-format=PAX', self.repository_location + '::src', 'simple.tar') + self.cmd('import-tar', self.repository_location + '::dst', 'simple.tar') + with changedir(self.output_path): + self.cmd('extract', self.repository_location + '::dst') + # PAX transfers mtime with ns resolution and xattrs. + self.assert_dirs_equal('input', 'output/input') + + def test_roundtrip_pax_xattrs(self): + if not xattr.is_enabled(self.input_path): + pytest.skip('xattrs not supported') + self.create_regular_file('file') + original_path = os.path.join(self.input_path, 'file') + xa_key, xa_value = b'user.xattrtest', b'not valid utf-8: \xff' + xattr.setxattr(original_path.encode(), xa_key, xa_value) + self.cmd('init', '--encryption=none', self.repository_location) + self.cmd('create', self.repository_location + '::src', 'input') + self.cmd('export-tar', self.repository_location + '::src', 'xattrs.tar', '--tar-format=PAX') + self.cmd('import-tar', self.repository_location + '::dst', 'xattrs.tar') + with changedir(self.output_path): + self.cmd('extract', self.repository_location + '::dst') + extracted_path = os.path.abspath('input/file') + xa_value_extracted = xattr.getxattr(extracted_path.encode(), xa_key) + assert xa_value_extracted == xa_value + + def test_roundtrip_pax_timestamps(self): + """export-tar --tar-format=PAX and import-tar keep the timestamps with exact ns precision.""" + self.create_regular_file('file') + mtime_ns = 1700000000_987654321 # float seconds would round this to ~240 ns + os.utime(os.path.join(self.input_path, 'file'), ns=(mtime_ns, mtime_ns)) + self.cmd('init', '--encryption=none', self.repository_location) + self.cmd('create', self.repository_location + '::src', 'input') + self.cmd('export-tar', self.repository_location + '::src', 'pax.tar', '--tar-format=PAX') + self.cmd('import-tar', self.repository_location + '::dst', 'pax.tar') + + def get_times(archive_name): + archive, repository = self.open_archive(archive_name) + with repository: + item = next(item for item in archive.iter_items() if item.path == 'input/file') + return {name: item.get(name) for name in ('atime', 'ctime', 'mtime')} + + src_times, dst_times = get_times('src'), get_times('dst') + assert dst_times == src_times + with tarfile.open('pax.tar') as tar: + pax_mtime = tar.getmember('input/file').pax_headers['mtime'] + assert pax_mtime == f"{src_times['mtime'] // 10**9}.{src_times['mtime'] % 10**9:09d}" + + def test_import_tar_invalid_pax_timestamp(self): + """import-tar ignores invalid PAX timestamps (tarfile ignores them, too).""" + with tarfile.open('input.tar', 'w', format=tarfile.PAX_FORMAT) as tar: + tarinfo = tarfile.TarInfo('file') + tarinfo.pax_headers = {'atime': 'garbage', 'ctime': '1700000000.5'} + tar.addfile(tarinfo, io.BytesIO()) + self.cmd('init', '--encryption=none', self.repository_location) + self.cmd('import-tar', self.repository_location + '::dst', 'input.tar') + archive, repository = self.open_archive('dst') + with repository: + item = next(archive.iter_items()) + assert 'atime' not in item + assert item.ctime == 1700000000_500000000 + + @pytest.mark.skipif(not is_linux, reason='POSIX ACL test, Linux only') + @pytest.mark.skipif(not are_acls_working(), reason='ACLs do not work') + def test_acl_roundtrip(self): + """Test the complete workflow for POSIX ACLs with export-tar and import-tar. + + This test follows the workflow: + 1. set filesystem ACLs + 2. create a Borg archive + 3. export-tar this archive + 4. import-tar the resulting tar file + 5. extract the imported archive + 6. check the expected ACLs in the filesystem + """ + # Define helper functions for working with ACLs + def get_acl(path): + item = {} + platform.acl_get(path, item, os.stat(path)) + return item + + def set_acl(path, access=None, default=None): + item = {'acl_access': access, 'acl_default': default} + platform.acl_set(path, item) + + # Define example ACLs + ACCESS_ACL = b'user::rw-\nuser:root:rw-:0\ngroup::r--\ngroup:root:r--:0\nmask::rw-\nother::r--' + DEFAULT_ACL = b'user::rw-\nuser:root:r--:0\ngroup::r--\ngroup:root:r--:0\nmask::rw-\nother::r--' + + # 1. Set filesystem ACLs + # Create test files with ACLs + self.create_regular_file('file') + os.mkdir(os.path.join(self.input_path, 'dir')) + + file_path = os.path.join(self.input_path, 'file') + dir_path = os.path.join(self.input_path, 'dir') + + # Set ACLs on the test files + try: + set_acl(file_path, access=ACCESS_ACL) + set_acl(dir_path, access=ACCESS_ACL, default=DEFAULT_ACL) + except OSError as e: + pytest.skip(f'Failed to set ACLs: {e}') + + file_acl = get_acl(file_path) + dir_acl = get_acl(dir_path) + + if not file_acl.get('acl_access') or not dir_acl.get('acl_access') or not dir_acl.get('acl_default'): + pytest.skip('ACLs not supported or not working correctly') + + # 2. Create a Borg archive + self.cmd('init', '--encryption=none', self.repository_location) + self.cmd('create', self.repository_location + '::original', 'input') + + # 3. export-tar this archive to a tar file + self.cmd('export-tar', self.repository_location + '::original', 'acls.tar', '--tar-format=PAX') + + # 4. import-tar the resulting tar file + self.cmd('import-tar', self.repository_location + '::imported', 'acls.tar') + + # 5. Extract the imported archive + with changedir(self.output_path): + self.cmd('extract', self.repository_location + '::imported') + + # 6. Check the expected ACLs in the filesystem + extracted_file_path = os.path.abspath('input/file') + extracted_dir_path = os.path.abspath('input/dir') + + extracted_file_acl = get_acl(extracted_file_path) + extracted_dir_acl = get_acl(extracted_dir_path) + + # Check that access ACLs were preserved + assert 'acl_access' in extracted_file_acl + assert extracted_file_acl['acl_access'] == file_acl['acl_access'] + assert b'user:root:rw-' in file_acl['acl_access'] + + assert 'acl_access' in extracted_dir_acl + assert extracted_dir_acl['acl_access'] == dir_acl['acl_access'] + assert b'user:root:rw-' in dir_acl['acl_access'] + + # Check that default ACLs were preserved for directories + assert 'acl_default' in extracted_dir_acl + assert extracted_dir_acl['acl_default'] == dir_acl['acl_default'] + assert b'user:root:r--' in dir_acl['acl_default'] + + @pytest.mark.skipif(not is_linux, reason='POSIX ACL test, Linux only') + @pytest.mark.skipif(not are_acls_working(), reason='ACLs do not work') + def test_import_tar_gnu_tar_acls(self): + """Test import-tar with POSIX ACLs in PAX headers like GNU tar writes them (no numeric ids).""" + def get_acl(path): + item = {} + platform.acl_get(path, item, os.stat(path)) + return item + + # GNU tar --format=posix --acls writes these headers, see also tar_acl_to_borg. + with tarfile.open('gnu.tar', 'w', format=tarfile.PAX_FORMAT) as tar: + tarinfo = tarfile.TarInfo('dir') + tarinfo.type, tarinfo.mode = tarfile.DIRTYPE, 0o755 + tarinfo.pax_headers = { + 'SCHILY.acl.access': 'user::rwx\ngroup::r-x\nother::r-x\n', + 'SCHILY.acl.default': 'user::rwx\nuser:root:r-x\ngroup::r-x\nmask::r-x\nother::r-x\n', + } + tar.addfile(tarinfo) + tarinfo = tarfile.TarInfo('dir/file') + tarinfo.mode = 0o644 + tarinfo.pax_headers = { + 'SCHILY.acl.access': 'user::rw-\nuser:root:rw-\ngroup::r--\nmask::rw-\nother::r--\n', + # GNU tar --xattrs-include='*' also stores the ACLs as raw xattrs, these must be ignored. + 'SCHILY.xattr.system.posix_acl_access': 'not a valid binary ACL', + } + tar.addfile(tarinfo, io.BytesIO()) + + self.cmd('init', '--encryption=none', self.repository_location) + self.cmd('import-tar', self.repository_location + '::dst', 'gnu.tar') + with changedir(self.output_path): + self.cmd('extract', self.repository_location + '::dst') + file_acl = get_acl(os.path.abspath('dir/file')) + dir_acl = get_acl(os.path.abspath('dir')) + if not file_acl.get('acl_access') or not dir_acl.get('acl_default'): + pytest.skip('ACLs not supported or not working correctly') + assert b'user:root:rw-:0' in file_acl['acl_access'] + assert b'user:root:r-x:0' in dir_acl['acl_default'] + def test_import_tar_quick_stats(self): self.create_regular_file('file1', size=1024) self.cmd('init', '--encryption=none', self.repository_location) diff --git a/src/borg/testsuite/helpers.py b/src/borg/testsuite/helpers.py index 0ab91805de..03abda8651 100644 --- a/src/borg/testsuite/helpers.py +++ b/src/borg/testsuite/helpers.py @@ -29,6 +29,7 @@ from ..helpers import ellipsis_truncate from ..helpers import chunkit from ..helpers import safe_ns, safe_s, SUPPORT_32BIT_PLATFORMS +from ..helpers import ns_to_pax_time, pax_time_to_ns from ..helpers import popen_with_error_handling from ..helpers import dash_open from ..helpers import iter_separated @@ -1176,6 +1177,31 @@ def test_safe_timestamps(): assert utcfromtimestamp(safe_ns(beyond_y10k) / 1000000000) > datetime(2262, 1, 1) +@pytest.mark.parametrize('ns, expected', [ + (0, '0.000000000'), + (1, '0.000000001'), + (1700000000_987654321, '1700000000.987654321'), + (-1, '-0.000000001'), + (-1500000000, '-1.500000000'), +]) +def test_ns_to_pax_time(ns, expected): + assert ns_to_pax_time(ns) == expected + + +@pytest.mark.parametrize('value, expected', [ + ('1700000000.987654321', 1700000000_987654321), + ('1700000000', 1700000000_000000000), + ('1700000000.5', 1700000000_500000000), + ('-1.5', 0), # safe_ns clamps + ('', None), + ('garbage', None), + ('NaN', None), + ('Infinity', None), +]) +def test_pax_time_to_ns(value, expected): + assert pax_time_to_ns(value) == expected + + class TestPopenWithErrorHandling: @pytest.mark.skipif(not shutil.which('test'), reason='"test" binary is needed') def test_simple(self):