# HG changeset patch
# User Denis Laxalde <denis.laxalde@logilab.fr>
# Date 1512039335 -3600
# Node ID 68ca7fe0ca29b92ce61c7bcfbf353cc9d87cea1d
# Parent  06deb43c23c3db14d3eb09d435714cfa5c970b7a
[ssplanner] Prevent execution of write queries involving computed relations

Previously, setting a computed relation upon entity creation or
modification (using the ORM or an RQL query) would usually fail with an
operational error in the backend ("no such table"). However, under some
mysterious circumstances (like passing a string as value in cw_set for a
computed relation), the RQL to SQL transformation would simply drop the
clause.

To prevent this to happen, we add a check for computed relation before
adding a relation to an execution plan. This check raises a QueryError.
It happens in several places:

* in querier.InsertPlan.add_relation_def() (called from several places
  in ssplanner steps) for INSERT queries,
* in ssplanner.UpdateStep.execute() for SET queries and,
* in ssplanner.SSplanner.build_delete_plan() for DELETE queries.

Tests added to unittest_querier.py because unittest_sslplanner.py looked
inappropriate (it has only unit tests) and the former already had a
NonRegressionTC class.

diff -r 06deb43c23c3 -r 68ca7fe0ca29 cubicweb/server/querier.py
--- a/cubicweb/server/querier.py	Thu Nov 30 11:00:01 2017 +0100
+++ b/cubicweb/server/querier.py	Thu Nov 30 11:55:35 2017 +0100
@@ -347,6 +347,8 @@
     def add_relation_def(self, rdef):
         """add an relation definition to build"""
         edef, rtype, value = rdef
+        if self.schema[rtype].rule:
+            raise QueryError("'%s' is a computed relation" % rtype)
         self.r_defs.add(rdef)
         if not isinstance(edef, int):
             self._r_subj_index.setdefault(edef, []).append(rdef)
diff -r 06deb43c23c3 -r 68ca7fe0ca29 cubicweb/server/ssplanner.py
--- a/cubicweb/server/ssplanner.py	Thu Nov 30 11:00:01 2017 +0100
+++ b/cubicweb/server/ssplanner.py	Thu Nov 30 11:55:35 2017 +0100
@@ -204,7 +204,10 @@
             step.children += self._sel_variable_step(plan, rqlst, etype, var)
             steps.append(step)
         for relation in rqlst.main_relations:
-            step = DeleteRelationsStep(plan, relation.r_type)
+            rtype = relation.r_type
+            if self.schema[rtype].rule:
+                raise QueryError("'%s' is a computed relation" % rtype)
+            step = DeleteRelationsStep(plan, rtype)
             step.children += self._sel_relation_steps(plan, rqlst, relation)
             steps.append(step)
         return steps
@@ -493,6 +496,9 @@
         for i, row in enumerate(result):
             newrow = []
             for (lhsinfo, rhsinfo, rschema) in self.updatedefs:
+                if rschema.rule:
+                    raise QueryError("'%s' is a computed relation"
+                                     % rschema.type)
                 lhsval = _handle_relterm(lhsinfo, row, newrow)
                 rhsval = _handle_relterm(rhsinfo, row, newrow)
                 if rschema.final or rschema.inlined:
diff -r 06deb43c23c3 -r 68ca7fe0ca29 cubicweb/server/test/unittest_querier.py
--- a/cubicweb/server/test/unittest_querier.py	Thu Nov 30 11:00:01 2017 +0100
+++ b/cubicweb/server/test/unittest_querier.py	Thu Nov 30 11:55:35 2017 +0100
@@ -19,6 +19,7 @@
 """unit tests for modules cubicweb.server.querier and cubicweb.server.ssplanner
 """
 
+from contextlib import contextmanager
 from datetime import date, datetime, timedelta, tzinfo
 import unittest
 
@@ -1667,5 +1668,33 @@
                 [[a1.eid]],
                 cnx.execute('Any A ORDERBY A WHERE U use_email A, U login "admin"').rows)
 
+    def test_computed_relation_in_write_queries(self):
+        """Computed relations are not allowed in main part of write queries."""
+        @contextmanager
+        def check(cnx):
+            with self.assertRaises(QueryError) as cm:
+                yield
+            self.assertIn("'user_login' is a computed relation",
+                          str(cm.exception))
+            cnx.rollback()
+
+        with self.admin_access.cnx() as cnx:
+            person = cnx.create_entity('Personne', nom=u'p')
+            cnx.commit()
+            # create
+            with check(cnx):
+                cnx.execute('INSERT CWUser X: X login "user", X user_login P'
+                            ' WHERE P is Personne, P nom "p"')
+            # update
+            bob = self.create_user(cnx, u'bob')
+            with check(cnx):
+                cnx.execute('SET U user_login P WHERE U login "bob", P nom "p"')
+            # delete
+            person.cw_set(login_user=bob)
+            cnx.commit()
+            with check(cnx):
+                cnx.execute('DELETE U user_login P WHERE U login "bob"')
+
+
 if __name__ == '__main__':
     unittest.main()