#183 Add check constraint to meta_vars.net_var_name to prevent whites…

…pace

.gitignore

@@ -1,3 +1,8 @@
 *.pyc
 venv
 *.egg*
+# don't commit poetry cache
+.cache 
+# ignore ide specific folders
+.idea
+.vscode

pycds/alembic/versions/78260d36e42b_add_net_var_name_valid_identifier_check.py

@@ -0,0 +1,45 @@
+"""Add net_var_name valid identifier check constraint
+
+Revision ID: 78260d36e42b
+Revises: bb2a222a1d4a
+Create Date: 2024-01-04 23:22:47.992791
+
+"""
+from alembic import op
+from pycds import get_schema_name
+
+# revision identifiers, used by Alembic.
+revision = "78260d36e42b"
+down_revision = "bb2a222a1d4a"
+branch_labels = None
+depends_on = None
+
+schema_name = get_schema_name()
+table_name = "meta_vars"
+constraint_name = "ck_net_var_name_valid_identifier"
+
+
+def update_table():
+    op.execute(
+        f"""UPDATE {schema_name}.{table_name} SET net_var_name=regexp_replace(net_var_name, '\\W', '_', 'g') WHERE net_var_name ~ '\\W';"""
+    )
+
+
+def regress_table():
+    # kept for convention.
+    # basically impossible to reverse as we can't know what has been removed and what was an underscore already
+    op.noop()
+
+
+def upgrade():
+    update_table()
+    op.create_check_constraint(
+        constraint_name, table_name, """net_var_name !~ '\\W'""", schema=schema_name
+    )
+    pass
+
+
+def downgrade():
+    op.drop_constraint(constraint_name, table_name, schema=schema_name, type_="check")
+    # regress_table()
+    pass

pycds/orm/tables.py

@@ -33,6 +33,7 @@
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm import relationship, synonym
 from sqlalchemy.schema import UniqueConstraint
+from sqlalchemy.schema import CheckConstraint
 from geoalchemy2 import Geometry
 
 from citext import CIText
@@ -297,6 +298,7 @@ class Variable(Base):
         UniqueConstraint(
             "network_id", "net_var_name", name="network_variable_name_unique"
         ),
+        CheckConstraint("net_var_name ~ '\W'", name="ck_net_var_name_valid_identifier"),
     )
 
     def __repr__(self):

tests/alembic_migrations/versions/v_78260d36e42b_add_net_var_name_valid_identifier_check/conftest.py

@@ -0,0 +1,6 @@
+import pytest
+
+
+@pytest.fixture(scope="module")
+def target_revision():
+    return "78260d36e42b"

tests/alembic_migrations/versions/v_78260d36e42b_add_net_var_name_valid_identifier_check/test_smoke.py

@@ -0,0 +1,100 @@
+"""Smoke tests:
+- Upgrade adds check constraint to meta_vars.net_var_name such that they are valid sql identifiers 
+    (this boils down to values inserted into the column should not contain whitespace)
+- Check constraint can still enter valid data
+- Check constraint rejects bad data after migration
+"""
+
+# -*- coding: utf-8 -*-
+import logging
+import pytest
+from alembic import command
+from sqlalchemy import inspect
+from sqlalchemy.exc import IntegrityError
+from psycopg2.errors import CheckViolation
+
+logger = logging.getLogger("tests")
+
+# this is the revision *before* the migration is to be run.
+prior_revision = "bb2a222a1d4a"
+final_revision = "78260d36e42b"
+
+
+@pytest.mark.usefixtures("new_db_left")
+@pytest.mark.parametrize(
+    "prepared_schema_from_migrations_left", (prior_revision,), indirect=True
+)
+def test_upgrade(
+    prepared_schema_from_migrations_left,
+    alembic_config_left,
+    schema_name,
+):
+    """test migration from bb2a222a1d4a to 78260d36e42b."""
+
+    # Set up database to version bb2a222a1d4a (previous migration)
+    engine, script = prepared_schema_from_migrations_left
+
+    engine.execute(
+        f"INSERT INTO {schema_name}.meta_vars(vars_id, net_var_name, standard_name, cell_method, display_name)"
+        f"VALUES (1, 'bad var name with\nwhitespace', 'test', 'test', 'test')"
+    )
+
+    command.upgrade(alembic_config_left, "+1")
+
+    result = engine.execute(
+        f"SELECT vars_id, net_var_name FROM {schema_name}.meta_vars"
+    )
+
+    row = next(result)
+
+    # After migration is run, bad strings should have whitespace replaced with underscores
+    assert row == (1, "bad_var_name_with_whitespace")
+
+
+@pytest.mark.usefixtures("new_db_left")
+@pytest.mark.parametrize(
+    "prepared_schema_from_migrations_left", (final_revision,), indirect=True
+)
+@pytest.mark.parametrize(
+    "test_value", ["allow_underscores", "allowfullwords", "allowCapitals"]
+)
+def test_check_good_constraint_values(
+    prepared_schema_from_migrations_left, schema_name, test_value
+):
+    # Set up database to version 78260d36e42b (after migration)
+    engine, script = prepared_schema_from_migrations_left
+
+    engine.execute(
+        f"INSERT INTO {schema_name}.meta_vars(vars_id, net_var_name, standard_name, cell_method, display_name)"
+        f"VALUES (1, '{test_value}' , 'test', 'test', 'test')"
+    )
+
+    result = engine.execute(
+        f"SELECT vars_id, net_var_name FROM {schema_name}.meta_vars"
+    )
+
+    row = next(result)
+
+    assert row == (1, test_value)
+
+
+@pytest.mark.usefixtures("new_db_left")
+@pytest.mark.parametrize(
+    "prepared_schema_from_migrations_left", (final_revision,), indirect=True
+)
+@pytest.mark.parametrize("test_value", ["bad string", "bad\nnewline"])
+def test_check_bad_constraint_values(
+    prepared_schema_from_migrations_left, schema_name, test_value
+):
+    # Set up database to version 78260d36e42b (after migration)
+    engine, script = prepared_schema_from_migrations_left
+    # This test passes bad data and expects an integrity error back from SQLAlchemy when executed
+    with pytest.raises(IntegrityError) as excinfo:
+        engine.execute(
+            f"INSERT INTO {schema_name}.meta_vars(vars_id, net_var_name, standard_name, cell_method, display_name)"
+            f"VALUES (1, '{test_value}' , 'test', 'test', 'test')"
+        )
+
+    # The specific exception raised by psycopg2 is stored internally in SQLAlchemy's IntegrityError
+    # By checking this inner exception we can know that it is specifically a Check Constraint violation
+    assert isinstance(excinfo.value.orig, CheckViolation)