feat(ui,api): clean DB tab header, gray out enterprise features when unlicensed, enforce strict license gating on API
CI Test Suite / Run Component Tests & Pipeline Verification (push) Successful in 1m22s
CI Test Suite / Run Component Tests & Pipeline Verification (push) Successful in 1m22s
This commit is contained in:
+48
-13
@@ -710,7 +710,53 @@ class TestServerComponent(unittest.TestCase):
|
||||
decrypted_msg = Server.decrypt_field_from_storage(encrypted_msg)
|
||||
self.assertEqual(decrypted_msg, raw_msg)
|
||||
|
||||
# 2. Test updating config via api_db_update_config (direct -> buffered, encrypt_at_rest = True)
|
||||
# 2. Test that without enterprise_db license, buffering, encryption at rest, and postgres are blocked
|
||||
cfg_unlicensed_buf = Server.DbFullConfigRequest(
|
||||
storage_engine="sqlite",
|
||||
cache_mode="buffered",
|
||||
encrypt_at_rest=False
|
||||
)
|
||||
with self.assertRaises(Server.HTTPException) as cm:
|
||||
Server.api_db_update_config(cfg_unlicensed_buf)
|
||||
self.assertEqual(cm.exception.status_code, 403)
|
||||
self.assertIn("enterprise_db", cm.exception.detail)
|
||||
|
||||
cfg_unlicensed_enc = Server.DbFullConfigRequest(
|
||||
storage_engine="sqlite",
|
||||
cache_mode="direct",
|
||||
encrypt_at_rest=True
|
||||
)
|
||||
with self.assertRaises(Server.HTTPException) as cm:
|
||||
Server.api_db_update_config(cfg_unlicensed_enc)
|
||||
self.assertEqual(cm.exception.status_code, 403)
|
||||
self.assertIn("enterprise_db", cm.exception.detail)
|
||||
|
||||
pg_req = Server.DbFullConfigRequest(
|
||||
storage_engine="postgres",
|
||||
cache_mode="direct",
|
||||
pg_config=Server.DbConnectionTestRequest(host="db.corp.internal", port=5432, database="logar", user="postgres", password="")
|
||||
)
|
||||
with self.assertRaises(Server.HTTPException) as cm:
|
||||
Server.api_db_update_config(pg_req)
|
||||
self.assertEqual(cm.exception.status_code, 403)
|
||||
self.assertIn("enterprise_db", cm.exception.detail)
|
||||
|
||||
# 3. Standard local SQLite configuration succeeds when unlicensed
|
||||
cfg_standard = Server.DbFullConfigRequest(
|
||||
storage_engine="sqlite",
|
||||
cache_mode="direct",
|
||||
encrypt_at_rest=False
|
||||
)
|
||||
res_std = Server.api_db_update_config(cfg_standard)
|
||||
self.assertEqual(res_std["status"], "success")
|
||||
|
||||
# 4. Unlock enterprise_db feature in license_config
|
||||
conn = sqlite3.connect(self.test_db)
|
||||
conn.execute("UPDATE license_config SET features = ? WHERE id = 1", (json.dumps(["enterprise_db"]),))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
# 5. Test updating config with enterprise_db unlocked (direct -> buffered, encrypt_at_rest = True)
|
||||
cfg_req = Server.DbFullConfigRequest(
|
||||
storage_engine="sqlite",
|
||||
cache_mode="buffered",
|
||||
@@ -726,18 +772,7 @@ class TestServerComponent(unittest.TestCase):
|
||||
self.assertEqual(Server.INGEST_QUEUE.flush_interval, 4.0)
|
||||
self.assertEqual(Server.INGEST_QUEUE.batch_size, 200)
|
||||
|
||||
# 3. Test that switching to external PostgreSQL requires enterprise_db license
|
||||
pg_req = Server.DbFullConfigRequest(
|
||||
storage_engine="postgres",
|
||||
cache_mode="direct",
|
||||
pg_config=Server.DbConnectionTestRequest(host="db.corp.internal", port=5432, database="logar", user="postgres", password="")
|
||||
)
|
||||
with self.assertRaises(Server.HTTPException) as cm:
|
||||
Server.api_db_update_config(pg_req)
|
||||
self.assertEqual(cm.exception.status_code, 403)
|
||||
self.assertIn("enterprise_db", cm.exception.detail)
|
||||
|
||||
# 4. Verify storage engine info output
|
||||
# 6. Verify storage engine info output
|
||||
info = Server.get_storage_engine_info(db_path=self.test_db)
|
||||
self.assertEqual(info["cache_mode"], "buffered")
|
||||
self.assertTrue(info["encrypt_at_rest"])
|
||||
|
||||
Reference in New Issue
Block a user