fix(tools): publish NID database cache after download completes

CMakeLists.txt

@@ -53,6 +53,8 @@ if(BUILD_TESTING)
             ${CMAKE_CURRENT_SOURCE_DIR}/tools/hw-oracle/test_hw_oracle.py)
         add_test(NAME pr_overlap_exports COMMAND ${Python3_EXECUTABLE}
             ${CMAKE_CURRENT_SOURCE_DIR}/tools/tests/test_pr_overlap_exports.py)
+        add_test(NAME nid_database_cache COMMAND ${Python3_EXECUTABLE}
+            ${CMAKE_CURRENT_SOURCE_DIR}/tools/tests/test_nid_names.py)
     endif()
 endif()
 

tools/nid_names.py

@@ -41,7 +41,13 @@ def load_db(path):
     if not path.exists():
         path.parent.mkdir(parents=True, exist_ok=True)
         print("downloading NID database to %s ..." % path, file=sys.stderr)
-        urllib.request.urlretrieve(DB_URL, str(path))
+        with tempfile.NamedTemporaryFile(dir=path.parent, prefix=path.name + ".", suffix=".tmp", delete=False) as download:
+            temporary = Path(download.name)
+        try:
+            urllib.request.urlretrieve(DB_URL, str(temporary))
+            temporary.replace(path)
+        finally:
+            temporary.unlink(missing_ok=True)
     db = {}
     with open(path, errors="replace") as handle:
         for line in handle:

tools/tests/test_nid_names.py

@@ -0,0 +1,72 @@
+import tempfile
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+from urllib.error import URLError
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+import nid_names
+
+
+class DatabaseCacheTests(unittest.TestCase):
+    def test_interrupted_download_does_not_publish_a_partial_cache(self):
+        with tempfile.TemporaryDirectory() as directory:
+            cache = Path(directory) / "cache" / "aerolib.csv"
+
+            def interrupted(url, filename):
+                Path(filename).write_text("nid1 scePartial\n", encoding="utf-8")
+                raise URLError("connection interrupted")
+
+            with patch.object(nid_names.urllib.request, "urlretrieve", side_effect=interrupted):
+                with self.assertRaises(URLError):
+                    nid_names.load_db(cache)
+            self.assertFalse(cache.exists())
+            self.assertEqual(list(cache.parent.iterdir()), [])
+
+    def test_retry_after_interruption_downloads_the_complete_database(self):
+        with tempfile.TemporaryDirectory() as directory:
+            cache = Path(directory) / "aerolib.csv"
+
+            def interrupted(url, filename):
+                Path(filename).write_text("nid1 scePartial\n", encoding="utf-8")
+                raise URLError("connection interrupted")
+
+            def complete(url, filename):
+                Path(filename).write_text("nid1 sceComplete\nnid2 sceOther\n", encoding="utf-8")
+
+            with patch.object(nid_names.urllib.request, "urlretrieve", side_effect=interrupted):
+                with self.assertRaises(URLError):
+                    nid_names.load_db(cache)
+            with patch.object(nid_names.urllib.request, "urlretrieve", side_effect=complete) as download:
+                self.assertEqual(nid_names.load_db(cache), {"nid1": "sceComplete", "nid2": "sceOther"})
+                download.assert_called_once()
+            self.assertEqual(list(cache.parent.iterdir()), [cache])
+
+    def test_cache_is_published_only_after_download_completes(self):
+        with tempfile.TemporaryDirectory() as directory:
+            cache = Path(directory) / "aerolib.csv"
+
+            def complete(url, filename):
+                self.assertEqual(url, nid_names.DB_URL)
+                self.assertNotEqual(Path(filename), cache)
+                self.assertEqual(Path(filename).parent, cache.parent)
+                self.assertFalse(cache.exists())
+                Path(filename).write_text("nid1 sceComplete\n", encoding="utf-8")
+
+            with patch.object(nid_names.urllib.request, "urlretrieve", side_effect=complete):
+                self.assertEqual(nid_names.load_db(cache), {"nid1": "sceComplete"})
+            self.assertEqual(cache.read_text(encoding="utf-8"), "nid1 sceComplete\n")
+            self.assertEqual(list(cache.parent.iterdir()), [cache])
+
+    def test_existing_cache_is_used_without_downloading(self):
+        with tempfile.TemporaryDirectory() as directory:
+            cache = Path(directory) / "aerolib.csv"
+            cache.write_text("nid1 sceExisting\n", encoding="utf-8")
+            with patch.object(nid_names.urllib.request, "urlretrieve") as download:
+                self.assertEqual(nid_names.load_db(cache), {"nid1": "sceExisting"})
+                download.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()