migrate legacy source kind constraint

2026-09-04 15:51:59 +02:00
parent 0653e7366c
commit 1e4c3a1e15
2 changed files with 34 additions and 1 deletions
@@ -1,7 +1,7 @@
 import json, os, sqlite3, threading, unittest
 from http.client import HTTPConnection
 from tempfile import TemporaryDirectory
-from app.main import create_server
+from app.main import ORGANIZATION_ID, connect, create_server
 from app.sources import CsvSource, ManualSource, available_adapters

 class SourceAdapterTests(unittest.TestCase):
@@ -31,6 +31,17 @@ class SourceAdapterTests(unittest.TestCase):
        self.assertFalse(result.valid)
        self.assertIn('secret', result.errors[0].lower())

+    def test_legacy_source_kind_constraint_is_migrated(self):
+        with TemporaryDirectory() as tmp:
+            path=os.path.join(tmp,'legacy.db')
+            legacy=sqlite3.connect(path)
+            legacy.execute("CREATE TABLE sources (id INTEGER PRIMARY KEY AUTOINCREMENT, organization_id TEXT NOT NULL, name TEXT NOT NULL, kind TEXT NOT NULL CHECK(kind IN ('csv','manual')), enabled INTEGER NOT NULL DEFAULT 0, config_json TEXT NOT NULL DEFAULT '{}', health_status TEXT NOT NULL DEFAULT 'unknown', consecutive_failures INTEGER NOT NULL DEFAULT 0, circuit_open INTEGER NOT NULL DEFAULT 0, last_success_at TEXT, last_failure_at TEXT, last_error TEXT, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, source_code TEXT NOT NULL DEFAULT '', display_name TEXT NOT NULL DEFAULT '', approved INTEGER NOT NULL DEFAULT 0, policy_json TEXT NOT NULL DEFAULT '{}', quota_json TEXT NOT NULL DEFAULT '{}', UNIQUE(organization_id,name))")
+            legacy.execute("INSERT INTO sources(organization_id,name,kind) VALUES(?, 'Existing manual','manual')",(ORGANIZATION_ID,)); legacy.commit(); legacy.close()
+            db=connect(path)
+            self.assertEqual(db.execute("SELECT kind FROM sources WHERE name='Existing manual'").fetchone()[0],'manual')
+            db.execute("INSERT INTO sources(organization_id,name,kind,source_code) VALUES(?,?,?,?)",(ORGANIZATION_ID,'OpenStreetMap / Overpass · plumbers','approved_directory','openstreetmap'))
+            db.commit(); self.assertEqual(db.execute("SELECT kind FROM sources WHERE source_code='openstreetmap'").fetchone()[0],'approved_directory'); db.close()
+
 class SourceApiTests(unittest.TestCase):
    def setUp(self):
        self.tmp=TemporaryDirectory(); os.environ['BOOTSTRAP_ADMIN_EMAIL']='owner@example.test'; os.environ['BOOTSTRAP_ADMIN_PASSWORD']='development-password'