Allow ASN database update to create missing file

CHANGELOG.md

@@ -30,6 +30,7 @@ Please refer to the [NEWS](NEWS.md) for a list of changes which have an affect o
 - `intelmq.bots.parsers.generic_csv.parser_csv`: Handle empty string parameter `columns_required` as unset (PR#2680 by Sebastian Wagner, fixes #2679).
 
 #### Experts
+- `intelmq.bots.experts.asn_lookup.expert`: Allow `--update-database` to create a missing database file (fixes #2689 by Haitao Zheng).
 
 #### Outputs
 - `intelmq.bots.outputs.smtp_batch.output`:

NEWS.md

@@ -16,6 +16,7 @@ Please refer to the change log for a full list of changes.
 ### Requirements
 
 ### Tools
+- `intelmq.bots.experts.asn_lookup.expert --update-database` now creates the configured database file if it does not exist yet.
 
 ### Data Format
 

intelmq/bots/experts/asn_lookup/expert.py

@@ -123,9 +123,10 @@ def update_database(cls, verbose=False):
             raise MissingDependencyError("pyasn")
 
         for database_path in set(bots.values()):
-            if not Path(database_path).is_file():
-                raise ValueError('Database file does not exist or is not a file.')
-            elif not os.access(database_path, os.W_OK):
+            database_file = Path(database_path)
+            if database_file.exists() and not database_file.is_file():
+                raise ValueError('Database path exists but is not a file.')
+            elif database_file.exists() and not os.access(database_path, os.W_OK):
                 raise ValueError('Database file is not writeable.')
 
         try:

intelmq/tests/bots/experts/asn_lookup/test_expert.py

@@ -8,6 +8,10 @@
 """
 
 import unittest
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from types import SimpleNamespace
+from unittest import mock
 
 import pkg_resources
 
@@ -62,5 +66,65 @@ def test_ipv6_lookup(self):
         self.assertMessageEqual(0, EXAMPLE_OUTPUT6)
 
 
+class TestASNLookupUpdateDatabase(unittest.TestCase):
+    def test_update_database_creates_missing_database_file(self):
+        def dump_prefixes_to_file(prefixes, database_path):
+            Path(database_path).write_text("fake database", encoding="utf-8")
+
+        class Session:
+            def get(self, url):
+                if url.endswith("/RIBS/"):
+                    return SimpleNamespace(
+                        text='<a href="rib.20260713.0000.bz2">rib</a>',
+                        status_code=200,
+                        url=url,
+                    )
+                if url.endswith(".bz2"):
+                    return SimpleNamespace(text="", content=b"BZh91AY&SY", status_code=200, url=url)
+                return SimpleNamespace(
+                    text='<a href="2026.07/">2026.07</a><a href="2026.06/">2026.06</a>',
+                    status_code=200,
+                    url=url,
+                )
+
+        pyasn = SimpleNamespace(
+            mrtx=SimpleNamespace(
+                parse_mrt_file=mock.Mock(return_value=["prefixes"]),
+                dump_prefixes_to_file=mock.Mock(side_effect=dump_prefixes_to_file),
+            )
+        )
+
+        with TemporaryDirectory() as tempdir:
+            database_path = Path(tempdir) / "asn_lookup" / "ipasn.dat"
+            runtime_config = {
+                "asn-lookup-expert": {
+                    "module": "intelmq.bots.experts.asn_lookup.expert",
+                    "parameters": {"database": str(database_path)},
+                }
+            }
+            controller = mock.Mock()
+
+            with (
+                mock.patch(
+                    "intelmq.bots.experts.asn_lookup.expert.get_bots_settings",
+                    return_value=runtime_config,
+                ),
+                mock.patch(
+                    "intelmq.bots.experts.asn_lookup.expert.create_request_session",
+                    return_value=Session(),
+                ),
+                mock.patch("intelmq.bots.experts.asn_lookup.expert.pyasn", pyasn),
+                mock.patch(
+                    "intelmq.bots.experts.asn_lookup.expert.IntelMQController",
+                    return_value=controller,
+                ),
+            ):
+                ASNLookupExpertBot.update_database()
+
+            self.assertTrue(database_path.is_file())
+            pyasn.mrtx.dump_prefixes_to_file.assert_called_once_with(["prefixes"], str(database_path))
+            controller.bot_reload.assert_called_once_with("asn-lookup-expert")
+
+
 if __name__ == '__main__':  # pragma: no cover
     unittest.main()