test: add unit tests for on_triplestore edge cases [release]

Author: arcangelo7

src/piccione/upload/on_triplestore.py

@@ -72,7 +72,7 @@ def upload_sparql_updates(
                 save_failed_query_file(file, failed_file)
 
 
-def main():
+def main():  # pragma: no cover
     parser = argparse.ArgumentParser(
         description="Execute SPARQL update queries on a triple store."
     )
@@ -104,5 +104,5 @@ def main():
     )
 
 
-if __name__ == "__main__":
+if __name__ == "__main__":  # pragma: no cover
     main()

tests/on_internet_archive_test.py

@@ -37,7 +37,7 @@ def test_successful_upload(self, tmp_path):
             validate_identifier=True,
         )
 
-    def test_failed_upload(self, tmp_path):
+    def test_failed_upload_prints_message(self, tmp_path, capsys):
         config_file = tmp_path / "config.yaml"
         config_file.write_text("""
 identifier: test-item
@@ -54,3 +54,6 @@ def test_failed_upload(self, tmp_path):
 
         with patch("piccione.upload.on_internet_archive.upload", return_value=[mock_response]):
             upload_files(str(config_file))
+
+        captured = capsys.readouterr()
+        assert captured.out == "Upload failed.\n"

tests/on_triplestore_test.py

@@ -1,9 +1,10 @@
 import os
+from unittest.mock import MagicMock, patch
 
 import pytest
 from tests.conftest import REDIS_DB, REDIS_PORT
 from piccione.upload.cache_manager import CacheManager
-from piccione.upload.on_triplestore import save_failed_query_file, upload_sparql_updates
+from piccione.upload.on_triplestore import remove_stop_file, save_failed_query_file, upload_sparql_updates
 from sparqlite import SPARQLClient
 
 SPARQL_ENDPOINT = "http://localhost:28890/sparql"
@@ -24,8 +25,6 @@ def test_add_and_contains(self, clean_redis):
         test_file = "test.sparql"
         cache_manager.add(test_file)
 
-        assert test_file in cache_manager.processed_files
-        assert cache_manager._redis.sismember(CacheManager.REDIS_KEY, test_file)
         assert test_file in cache_manager
 
     def test_persistence(self, clean_redis):
@@ -153,3 +152,81 @@ def test_data_loaded_to_triplestore(self, temp_dir, clean_redis, clean_virtuoso)
         bindings = result["results"]["bindings"]
         assert len(bindings) == 1
         assert bindings[0]["o"]["value"] == "test value"
+
+    def test_nonexistent_folder_returns_early(self, temp_dir, clean_redis):
+        cache_manager = CacheManager(redis_port=REDIS_PORT, redis_db=REDIS_DB)
+        upload_sparql_updates(
+            SPARQL_ENDPOINT,
+            os.path.join(temp_dir, "nonexistent"),
+            cache_manager=cache_manager,
+        )
+        assert cache_manager.get_all() == set()
+
+    def test_empty_folder_returns_early(self, temp_dir, clean_redis, clean_virtuoso):
+        sparql_dir = os.path.join(temp_dir, "empty_sparql")
+        os.makedirs(sparql_dir)
+
+        cache_manager = CacheManager(redis_port=REDIS_PORT, redis_db=REDIS_DB)
+        upload_sparql_updates(
+            SPARQL_ENDPOINT,
+            sparql_dir,
+            cache_manager=cache_manager,
+        )
+        assert cache_manager.get_all() == set()
+
+    def test_empty_query_file_is_skipped(self, temp_dir, clean_redis, clean_virtuoso):
+        sparql_dir = os.path.join(temp_dir, "sparql_files")
+        os.makedirs(sparql_dir)
+
+        with open(os.path.join(sparql_dir, "empty.sparql"), "w") as f:
+            f.write("   \n  ")
+
+        cache_manager = CacheManager(redis_port=REDIS_PORT, redis_db=REDIS_DB)
+        upload_sparql_updates(
+            SPARQL_ENDPOINT,
+            sparql_dir,
+            cache_manager=cache_manager,
+            show_progress=False,
+        )
+        assert "empty.sparql" in cache_manager
+
+    def test_remove_stop_file_when_exists(self, temp_dir):
+        stop_file = os.path.join(temp_dir, ".stop_upload")
+        with open(stop_file, "w") as f:
+            f.write("")
+
+        assert os.path.exists(stop_file)
+        remove_stop_file(stop_file)
+        assert not os.path.exists(stop_file)
+
+    def test_remove_stop_file_when_not_exists(self, temp_dir):
+        stop_file = os.path.join(temp_dir, ".stop_upload")
+        assert not os.path.exists(stop_file)
+        remove_stop_file(stop_file)
+
+    def test_creates_cache_manager_when_none(self, temp_dir, clean_virtuoso):
+        sparql_dir = os.path.join(temp_dir, "sparql_files")
+        os.makedirs(sparql_dir)
+
+        query = """
+        INSERT DATA {
+            GRAPH <http://test.graph> {
+                <http://test.subject> <http://test.predicate> "value" .
+            }
+        }
+        """
+        with open(os.path.join(sparql_dir, "test.sparql"), "w") as f:
+            f.write(query)
+
+        mock_cache = MagicMock()
+        mock_cache.__contains__ = MagicMock(return_value=False)
+
+        with patch("piccione.upload.on_triplestore.CacheManager", return_value=mock_cache):
+            upload_sparql_updates(
+                SPARQL_ENDPOINT,
+                sparql_dir,
+                cache_manager=None,
+                show_progress=False,
+            )
+
+        mock_cache.add.assert_called_once_with("test.sparql")

tests/on_zenodo_test.py

@@ -61,7 +61,6 @@ def test_successful_upload(self, tmp_path):
                 )
 
         assert result == mock_response
-        assert mock_put.call_count == 1
         mock_put.assert_called_once()
         call_kwargs = mock_put.call_args[1]
         assert call_kwargs["headers"] == {"Authorization": "Bearer token123"}