Implement safeguard against problematic meta keys

ixmp4/data/db/meta/repository.py

@@ -1,12 +1,14 @@
 from typing import Union
 
+import numpy as np
 import pandas as pd
 import pandera as pa
 from pandera.typing import DataFrame, Series
 from sqlalchemy.exc import NoResultFound
 
 from ixmp4 import db
 from ixmp4.core.decorators import check_types
+from ixmp4.core.exceptions import InvalidRunMeta
 from ixmp4.data import abstract
 from ixmp4.data.auth.decorators import guard
 from ixmp4.data.db.model import Model
@@ -16,6 +18,8 @@
 from .. import base
 from .model import RunMetaEntry
 
+ILLEGAL_META_KEYS = {"model", "scenario", "id", "version", "is_default"}
+
 
 class RemoveRunMetaEntryFrameSchema(pa.DataFrameModel):
     key: Series[pa.String] = pa.Field(coerce=True)
@@ -60,6 +64,8 @@ def add(
                 default_only=False,
             )
 
+        if key in ILLEGAL_META_KEYS:
+            raise InvalidRunMeta("Illegal meta key: " + key)
         entry = RunMetaEntry(run__id=run__id, key=key, value=value)
         self.session.add(entry)
         return entry
@@ -205,6 +211,9 @@ def map_value_column(df: pd.DataFrame):
     @check_types
     @guard("edit")
     def bulk_upsert(self, df: DataFrame[AddRunMetaEntryFrameSchema]) -> None:
+        if illegal_keys := (set(np.unique(df.key.values)) & ILLEGAL_META_KEYS):
+            raise InvalidRunMeta("Illegal meta key(s): " + ", ".join(illegal_keys))
+
         self.check_df_access(df)
         df["type"] = df["value"].map(type).map(RunMetaEntry.Type.from_pytype)
 

tests/data/test_meta.py

@@ -2,7 +2,7 @@
 import pytest
 
 import ixmp4
-from ixmp4.core.exceptions import SchemaError
+from ixmp4.core.exceptions import InvalidRunMeta, SchemaError
 from ixmp4.data.abstract.meta import RunMetaEntry
 
 from ..utils import assert_unordered_equality
@@ -19,6 +19,8 @@
     columns=["id", "key", "type", "value"],
 )
 
+TEST_ILLEGAL_META_KEYS = {"model", "scenario", "id", "version", "is_default"}
+
 
 class TestDataMeta:
     def test_create_get_entry(self, platform: ixmp4.Platform):
@@ -37,6 +39,18 @@ def test_create_get_entry(self, platform: ixmp4.Platform):
             assert entry.value == value
             assert entry.type == type
 
+    def test_illegal_key(self, platform: ixmp4.Platform):
+        run = platform.runs.create("Model", "Scenario")
+        for key in TEST_ILLEGAL_META_KEYS:
+            with pytest.raises(InvalidRunMeta, match="Illegal meta key: " + key):
+                platform.backend.meta.create(run.id, key, "foo")
+
+            df = pd.DataFrame(
+                {"run__id": [run.id] * 2, "key": [key, "foo"], "value": ["bar", "baz"]},
+            )
+            with pytest.raises(InvalidRunMeta, match=r"Illegal meta key\(s\): " + key):
+                platform.backend.meta.bulk_upsert(df)
+
     def test_entry_unique(self, platform: ixmp4.Platform):
         run = platform.runs.create("Model", "Scenario")
         platform.backend.meta.create(run.id, "Key", "Value")