Additional tests and docstrings.

borgmatic/actions/browse/archive.py

@@ -1,9 +1,11 @@
 import argparse
+import collections
 import json
 import logging
 import os
 
 import borgmatic.borg.extract
+import borgmatic.borg.list
 import borgmatic.borg.repo_list
 import borgmatic.borg.version
 
@@ -13,7 +15,17 @@ import binaryornot.helpers
 logger = logging.getLogger(__name__)
 
 
+Archive_path = collections.namedtuple(
+    'Archive_path',
+    ('path_type', 'file_path', 'link_target'),
+)
+
+
 def get_repository_archives(config, repository):
+    '''
+    Given a configuration dict and a repository dict, return a list of the repository's archives,
+    one dict per archive.
+    '''
     with borgmatic.logger.Log_prefix(repository.get('label', repository['path'])):
         logger.info('Listing repository')
         repo_list_arguments = argparse.Namespace(
@@ -45,7 +57,12 @@ def get_repository_archives(config, repository):
         )
 
 
-def get_archive_files(config, repository, archive_name):
+def get_archive_paths(config, repository, archive_name):
+    '''
+    Given a configuration dict, a repository dict, and an archive name in that repository, get a
+    list of files, directories, symlinks, etc. found in the archive, each as an Archive_path
+    instance.
+    '''
     with borgmatic.logger.Log_prefix(repository.get('label', repository['path'])):
         logger.info(f'Listing archive {archive_name}')
 
@@ -53,10 +70,9 @@ def get_archive_files(config, repository, archive_name):
         local_path = config.get('local_path', 'borg')
         remote_path = config.get('remote_path')
         local_borg_version = borgmatic.borg.version.local_borg_version(config, local_path)
-        seen = set()
 
         return (
-            (
+            Archive_path(
                 path_data['type'],
                 path_data['path'],
                 path_data.get('linktarget'),
@@ -74,12 +90,16 @@ def get_archive_files(config, repository, archive_name):
 
 
 READLINES_HINT_BYTES = 100000
+TRUNCATION_MESSAGE = '[... truncated for display ...]'
 
 
 def get_archive_file_content(config, repository, archive_name, file_path):
     '''
     Given a configuration dict, a repository dict, an archive name in that repository, and a file
-    path in that archive, return the file's contents or None if the file can't be loaded.
+    path in that archive, return the file's contents or None if the file can't be loaded, e.g.
+    because it's binary or can't be decoded.
+
+    If the file is too large, then truncate the returned content.
     '''
     with borgmatic.logger.Log_prefix(repository.get('label', repository['path'])):
         logger.info(f'Getting archive content of file {file_path}')
@@ -110,7 +130,7 @@ def get_archive_file_content(config, repository, archive_name, file_path):
             return (
                 content.decode()
                 if len(content) < READLINES_HINT_BYTES
-                else f'{content.decode()}\n[... truncated for display ...]'
+                else f'{content.decode()}\n{TRUNCATION_MESSAGE}'
             )
         except UnicodeDecodeError:
             return None

borgmatic/actions/browse/workers.py

@@ -1,4 +1,3 @@
-import collections
 import contextlib
 import logging
 import os
@@ -45,12 +44,6 @@ def add_repository_archives(browse_app, archives_list, config, repository, timer
     browse_app.call_from_thread(timer.stop)
 
 
-Archive_path = collections.namedtuple(
-    'Archive_path',
-    ('path_type', 'file_path', 'link_target'),
-)
-
-
 def record_path(archive_path, hierarchy, path_components):
     if len(path_components) == 1:
         hierarchy[path_components[0]] = {} if archive_path.path_type == 'd' else archive_path
@@ -66,8 +59,10 @@ def get_paths(hierarchy, path_components, full_path_components=None):
     if len(path_components) == 1:
         return (
             archive_path
-            if isinstance(archive_path, Archive_path)
-            else Archive_path('d', os.path.join(*full_path_components, component), '')
+            if isinstance(archive_path, borgmatic.actions.browse.archive.Archive_path)
+            else borgmatic.actions.browse.archive.Archive_path(
+                'd', os.path.join(*full_path_components, component), ''
+            )
             for component, archive_path in hierarchy[path_components[0]].items()
         )
 
@@ -95,10 +90,10 @@ class Archive_path_loaded(textual.signal.Signal):
 
 @textual.work(thread=True)
 def load_archive_files(browse_app, directory_list, config, repository, archive_name):
-    for path_type, file_path, link_target in borgmatic.actions.browse.archive.get_archive_files(
+    for archive_path in borgmatic.actions.browse.archive.get_archive_paths(
         config, repository, archive_name
     ):
-        directory_list.path_loaded.publish(Archive_path(path_type, file_path, link_target))
+        directory_list.path_loaded.publish(archive_path)
 
     directory_list.path_loaded.publish(LOADING_DONE)
 

tests/unit/actions/browse/test_archive.py

@@ -0,0 +1,90 @@
+from flexmock import flexmock
+
+from borgmatic.actions.browse import archive as module
+
+
+def test_get_repository_archives_does_not_raise():
+    config = {'repositories': [{'path': 'test.borg'}]}
+    flexmock(module.borgmatic.logger).should_receive('Log_prefix').and_return(flexmock())
+    flexmock(module.borgmatic.borg.version).should_receive('local_borg_version').and_return('3.0')
+    flexmock(module.borgmatic.borg.repo_list).should_receive('list_repository').and_return('{}')
+
+    assert module.get_repository_archives(config, config['repositories'][0]) == {}
+
+
+def test_get_archive_paths_returns_each_as_archive_path_with_metadata():
+    config = {'repositories': [{'path': 'test.borg'}]}
+    flexmock(module.borgmatic.logger).should_receive('Log_prefix').and_return(flexmock())
+    flexmock(module.borgmatic.borg.version).should_receive('local_borg_version').and_return('3.0')
+    flexmock(module.borgmatic.borg.list).should_receive('capture_archive_listing').and_yield(
+        {'path': 'foo.txt', 'type': '-', 'linktarget': ''},
+        {'path': 'bar.txt', 'type': 'l', 'linktarget': 'foo.txt'},
+        {'path': 'etc', 'type': 'd', 'linktarget': ''},
+    )
+
+    assert tuple(module.get_archive_paths(config, config['repositories'][0], 'archive')) == (
+        module.Archive_path('-', 'foo.txt', ''),
+        module.Archive_path('l', 'bar.txt', 'foo.txt'),
+        module.Archive_path('d', 'etc', ''),
+    )
+
+
+def test_get_archive_file_content_with_binary_file_bails():
+    config = {'repositories': [{'path': 'test.borg'}]}
+    flexmock(module.borgmatic.logger).should_receive('Log_prefix').and_return(flexmock())
+    flexmock(module.borgmatic.borg.version).should_receive('local_borg_version').and_return('3.0')
+    flexmock(module.borgmatic.borg.extract).should_receive('extract_archive').and_return(
+        flexmock(stdout=flexmock(readlines=lambda hint: [b'foo\n', b'bar\n'])),
+    )
+    flexmock(module.binaryornot.helpers).should_receive('is_binary_string').and_return(True)
+
+    assert (
+        module.get_archive_file_content(config, config['repositories'][0], 'archive', 'etc/foo.txt')
+        is None
+    )
+
+
+def test_get_archive_file_content_decodes_content():
+    config = {'repositories': [{'path': 'test.borg'}]}
+    flexmock(module.borgmatic.logger).should_receive('Log_prefix').and_return(flexmock())
+    flexmock(module.borgmatic.borg.version).should_receive('local_borg_version').and_return('3.0')
+    flexmock(module.borgmatic.borg.extract).should_receive('extract_archive').and_return(
+        flexmock(stdout=flexmock(readlines=lambda hint: [b'foo\n', b'bar\n'])),
+    )
+    flexmock(module.binaryornot.helpers).should_receive('is_binary_string').and_return(False)
+
+    assert (
+        module.get_archive_file_content(config, config['repositories'][0], 'archive', 'etc/foo.txt')
+        == 'foo\nbar\n'
+    )
+
+
+def test_get_archive_file_content_with_large_file_truncates_content():
+    config = {'repositories': [{'path': 'test.borg'}]}
+    flexmock(module.borgmatic.logger).should_receive('Log_prefix').and_return(flexmock())
+    flexmock(module.borgmatic.borg.version).should_receive('local_borg_version').and_return('3.0')
+    flexmock(module).READLINES_HINT_BYTES = 3
+    flexmock(module.borgmatic.borg.extract).should_receive('extract_archive').and_return(
+        flexmock(stdout=flexmock(readlines=lambda hint: [b'foo\n', b'bar\n'])),
+    )
+    flexmock(module.binaryornot.helpers).should_receive('is_binary_string').and_return(False)
+
+    assert (
+        module.get_archive_file_content(config, config['repositories'][0], 'archive', 'etc/foo.txt')
+        == f'foo\nbar\n\n{module.TRUNCATION_MESSAGE}'
+    )
+
+
+def test_get_archive_file_content_with_unicode_decode_error_does_not_raise():
+    config = {'repositories': [{'path': 'test.borg'}]}
+    flexmock(module.borgmatic.logger).should_receive('Log_prefix').and_return(flexmock())
+    flexmock(module.borgmatic.borg.version).should_receive('local_borg_version').and_return('3.0')
+    flexmock(module.borgmatic.borg.extract).should_receive('extract_archive').and_return(
+        flexmock(stdout=flexmock(readlines=lambda hint: [b'foo\n', b'\xc3\n'])),
+    )
+    flexmock(module.binaryornot.helpers).should_receive('is_binary_string').and_return(False)
+
+    assert (
+        module.get_archive_file_content(config, config['repositories'][0], 'archive', 'etc/foo.txt')
+        is None
+    )