[Concept,02/32] patman: Reject database versions newer than supported

Message ID 20260226200106.1727176-3-sjg@u-boot.org
State New
Headers
Series patman: Add multi-upstream support |

Commit Message

Simon Glass Feb. 26, 2026, 8 p.m. UTC
  From: Simon Glass <simon.glass@canonical.com>

If a newer version of patman has upgraded the database schema beyond
what the current patman understands, it goes into an infinite loop,
trying to migrate.

Exit with an error message instead.

Signed-off-by: Simon Glass <simon.glass@canonical.com>
---

 tools/patman/database.py     |  6 ++++++
 tools/patman/test_cseries.py | 17 +++++++++++++++++
 2 files changed, 23 insertions(+)
  

Patch

diff --git a/tools/patman/database.py b/tools/patman/database.py
index e6684f990da..4557c4ff3fc 100644
--- a/tools/patman/database.py
+++ b/tools/patman/database.py
@@ -86,6 +86,12 @@  class Database:  # pylint:disable=R0904
     def start(self):
         """Open the database read for use, migrate to latest schema"""
         self.open_it()
+        old_version = self.get_schema_version()
+        if old_version > LATEST:
+            self.close()
+            tout.fatal(
+                f'Database version {old_version} is too new (max'
+                f' {LATEST}); please update patman')
         self.migrate_to(LATEST)
 
     def open_it(self):
diff --git a/tools/patman/test_cseries.py b/tools/patman/test_cseries.py
index 11724b1378f..3a9411033a6 100644
--- a/tools/patman/test_cseries.py
+++ b/tools/patman/test_cseries.py
@@ -3318,6 +3318,23 @@  Date:   .*
             self.assertEqual(version, db.get_schema_version())
         self.assertEqual(4, database.LATEST)
 
+    def test_migrate_future_version(self):
+        """Test that a database newer than patman is rejected"""
+        db = database.Database(f'{self.tmpdir}/.patman.db')
+        with terminal.capture():
+            db.start()
+
+        # Set the schema version beyond what patman supports
+        db.cur.execute(
+            f'UPDATE schema_version SET version = {database.LATEST + 1}')
+        db.commit()
+        db.close()
+
+        with self.assertRaises(SystemExit):
+            with terminal.capture() as (_, err):
+                db.start()
+        self.assertIn('is too new', err.getvalue())
+
     def test_series_scan(self):
         """Test scanning a series for updates"""
         cser, _ = self.setup_second()