Add ability to ignore NoFilesToSync error in list command

solgate/cli.py

@@ -6,7 +6,8 @@
 import click
 
 from solgate import list_source, send, send_report, __version__ as version
-from .utils import serialize, logger, deserialize, read_general_config, NoFilesToSyncError, FilesFailedToSyncError
+from .utils import serialize, logger, deserialize, read_general_config, initialize_file
+from .utils import NoFilesToSyncError, FilesFailedToSyncError
 from .report import DEFAULT_RECIPIENT, DEFAULT_SENDER, DEFAULT_SMTP_SERVER
 
 
@@ -103,8 +104,10 @@ def _list(ctx, backfill: bool, silent: bool = False, output: str = None):
     if backfill:
         logger.info("Running in a backfill mode")
     try:
+        idx = 0
+        initialize_file(output)
         for idx, file in enumerate(list_source(ctx.obj["config"], backfill), start=1):
-            if output:
+            if output and file is not None:
                 serialize(file, output)
             if not silent:
                 click.echo(file)

solgate/lookup.py

@@ -56,6 +56,7 @@ def list_source(config: Dict[str, Any], backfill=False) -> Generator[Dict[str, A
 
     """
     general_config = read_general_config(**config)
+    ignore_alerts = general_config.pop('ignore_alerts', [])
 
     oldest_date = datetime.now(timezone.utc) - parse_timedelta(general_config.get("timedelta", DEFAULT_TIMEDELTA))
     s3 = S3FileSystem.from_config_file(config, S3ConfigSelector["source"])[0]
@@ -75,4 +76,8 @@ def list_source(config: Dict[str, Any], backfill=False) -> Generator[Dict[str, A
         yield {k: getattr(obj, k) for k in KEYS}
 
     if not is_files:
-        raise FileNotFoundError("No files found in given TIMEDELTA")
+        if "no_files" in ignore_alerts:
+            logger.warning("No files found in given TIMEDELTA, returning empty JSON")
+            return None
+        else:
+            raise FileNotFoundError("No files found in given TIMEDELTA")

solgate/utils/__init__.py

@@ -2,7 +2,7 @@
 
 # flake8: noqa
 
-from .io import deserialize, key_formatter, read_general_config, serialize
+from .io import deserialize, key_formatter, read_general_config, serialize, initialize_file
 from .logging import logger
 from .s3 import S3File, S3FileSystem, S3ConfigSelector
 from .exceptions import EXIT_CODES, NoFilesToSyncError, FilesFailedToSyncError

solgate/utils/io.py

@@ -73,6 +73,17 @@ def gen():
     return gen(), len(open(filename).readlines())
 
 
+def initialize_file(filename: str) -> None:
+    """Writes an empty file
+
+    Args:
+        filename (str): File name or path.
+
+    """
+    with open(filename, "w") as f:
+        pass
+
+
 def _read_yaml_file(filename: Union[str, Path]) -> Dict[str, Any]:
     """Read a file.
 

tests/cli_test.py

@@ -40,11 +40,13 @@ def test_list(run, mocker, cli_args, func_args, file_output):
     """Should call proper functions on list command."""
     mocked_list_source = mocker.patch("solgate.cli.list_source", return_value=["list", "of", "files"])
     mocked_serialize = mocker.patch("solgate.cli.serialize")
+    mocked_initialize_file = mocker.patch("solgate.cli.initialize_file")
 
     result = run(cli_args)
 
     assert result.exit_code == 0
     mocked_list_source.assert_called_once_with(*func_args)
+    mocked_initialize_file.assert_called_once()
 
     if file_output:
         calls = [mocker.call(f, "output.json") for f in ("list", "of", "files")]
@@ -60,6 +62,7 @@ def test_list(run, mocker, cli_args, func_args, file_output):
 def test_list_negative(run, mocker, side_effect, errno):
     """Should fail on list command."""
     mocker.patch("solgate.cli.list_source", side_effect=side_effect)
+    mocker.patch("solgate.cli.initialize_file")
 
     result = run("list")
 

tests/lookup_test.py

@@ -75,6 +75,22 @@ def test_list_source_no_objects(mocked_s3, mocker):
         list(lookup.list_source({}))
 
 
+@pytest.mark.parametrize("mocked_s3", ["same_client.yaml"], indirect=["mocked_s3"])
+def test_list_source_ignored_no_objects(mocked_s3, mocker):
+    """Should raise when no files found."""
+    fs = mocked_s3[0]
+    mocker.patch("solgate.lookup.read_general_config", return_value=dict(ignore_alerts=["no_files"]))
+    mocker.patch("solgate.lookup.S3FileSystem.from_config_file", return_value=[fs])
+
+    fs.s3fs.touch("BUCKET/old.csv")
+    s3_backend.buckets["BUCKET"].keys["old.csv"].last_modified = datetime(2020, 1, 1)
+
+    try:
+        list(lookup.list_source({}))
+    except:
+        pytest.fail("Unexpected FileNotFoundError raised despite ignore flag")
+
+
 @pytest.mark.parametrize("mocked_s3", ["same_client.yaml"], indirect=["mocked_s3"])
 def test_list_source_backfill(mocked_s3, mocker):
     """Should list all files when backfill is enabled."""

tests/utils/io_test.py

@@ -230,6 +230,16 @@ def test_serialize(mocker):
     args = "".join(c.args[0] for c in calls)
     assert args == '{"a": "b"}\n'
 
+def test_ititialize_file(mocker):
+    """Should serialize to JSON."""
+    mocked_open = mocker.patch("builtins.open")
+    io.initialize_file("file.json")
+
+    mocked_open.assert_called_once_with("file.json", "w")
+    calls = mocked_open.return_value.__enter__.return_value.write.call_args_list
+    args = "".join(c.args[0] for c in calls)
+    assert args == ''
+
 
 @pytest.mark.parametrize(
     "input,output",