tests/test_sql.py

@@ -2709,3 +2709,39 @@ def test_workflow_traces_initial_migration(pub):
 
     assert not any(isinstance(x, ActionsTracingEvolutionPart) for x in formdata.iter_evolution_parts())
     assert not any(isinstance(x, ActionsTracingEvolutionPart) for x in formdata2.iter_evolution_parts())
+
+
+def test_computed_field_bad_content(pub):
+    FormDef.wipe()
+    formdef = FormDef()
+    formdef.name = 'test'
+    formdef.fields = [
+        fields.ComputedField(
+            id='1',
+            label='computed',
+            varname='computed',
+        ),
+    ]
+    formdef.store()
+    formdef.data_class().wipe()
+
+    conn, cur = sql.get_connection_and_cursor()
+    sql.drop_views(formdef, conn, cur)
+    cur.execute(
+        '''ALTER TABLE %s ALTER COLUMN %s TYPE VARCHAR'''
+        % (formdef.table_name, sql.get_field_id(formdef.fields[0]))
+    )
+    conn.commit()
+    cur.close()
+
+    formdata = formdef.data_class()()
+    formdata.just_created()
+    formdata.data = {'1': 'bad value'}
+    formdata.store()
+
+    with pytest.raises(ValueError) as excinfo:
+        formdata = formdef.data_class().get(formdata.id)
+    assert (
+        str(excinfo.value)
+        == 'bad data {"data": "bad value", "@type": "computed-data"} (type <class \'str\'>) in computed field 1'
+    )

wcs/sql.py

@@ -2059,7 +2059,11 @@ class SqlMixin:
                     d[fmt] = force_str(val)
                 value = d
             elif field.key == 'computed':
-                if isinstance(value, dict) and value.get('@type') == 'computed-data':
+                if not isinstance(value, dict):
+                    raise ValueError(
+                        "bad data %s (type %s) in computed field %s" % (value, type(value), field.id)
+                    )
+                if value.get('@type') == 'computed-data':
                     value = value.get('data')
             if sql_type == 'date':
                 value = value.timetuple()