Add multivalued insert context for defaultsreview/mike_bayer/ticket_4075

Added a new method :class:`.DefaultExecutionContext.current_parameters`
which is used within a function-based default value generator in
order to retrieve the current parameters being passed to the statement.
The new function differs from the ``.current_parameters`` attribute in
that it also provides for optional grouping of parameters that
correspond to a multi-valued "insert" construct.  Previously it was not
possible to identify the subset of parameters that were relevant to
the function call.

Change-Id: I6894c7b4a2bce3e83c3ade8af0e5b2f8df37b785
Fixes: #4075

1 files changed, 64 insertions, 0 deletions

diff --git a/test/sql/test_defaults.py b/test/sql/test_defaults.py
index 1ef49bf04..fc42d420f 100644
--- a/test/sql/test_defaults.py
+++ b/test/sql/test_defaults.py
@@ -15,6 +15,7 @@ from sqlalchemy.dialects import sqlite
 from sqlalchemy.testing import fixtures
 from sqlalchemy.util import u, b
 from sqlalchemy import util
+from sqlalchemy.testing import mock
 import itertools
 
 t = f = f2 = ts = currenttime = metadata = default_generator = None
@@ -1630,3 +1631,66 @@ class InsertFromSelectTest(fixtures.TestBase):
             testing.db.execute(table.select().order_by(table.c.x)).fetchall(),
             [(2, 1, 5), (7, 1, 12)]
         )
+
+class CurrentParametersTest(fixtures.TablesTest):
+    __backend__ = True
+
+    @classmethod
+    def define_tables(cls, metadata):
+        def gen_default(context):
+            pass
+
+        Table(
+            "some_table", metadata,
+            Column('x', String(50), default=gen_default),
+            Column('y', String(50)),
+        )
+
+    def _fixture(self, fn):
+
+        def gen_default(context):
+            fn(context)
+        some_table = self.tables.some_table
+        some_table.c.x.default.arg = gen_default
+        return fn
+
+    def _test(self, exec_type, usemethod):
+        collect = mock.Mock()
+
+        @self._fixture
+        def fn(context):
+            collect(context.get_current_parameters())
+
+        table = self.tables.some_table
+        if exec_type in ('multivalues', 'executemany'):
+            parameters = [{"y": "h1"}, {"y": "h2"}]
+        else:
+            parameters = [{"y": "hello"}]
+
+        if exec_type == 'multivalues':
+            stmt, params = table.insert().values(parameters), {}
+        else:
+            stmt, params = table.insert(), parameters
+
+        with testing.db.connect() as conn:
+            conn.execute(stmt, params)
+        eq_(
+            collect.mock_calls,
+            [mock.call({"y": param['y'], "x": None}) for param in parameters]
+        )
+
+    def test_single_w_attribute(self):
+        self._test("single", "attribute")
+
+    def test_single_w_method(self):
+        self._test("single", "method")
+
+    def test_executemany_w_attribute(self):
+        self._test("executemany", "attribute")
+
+    def test_executemany_w_method(self):
+        self._test("executemany", "method")
+
+    @testing.requires.multivalues_inserts
+    def test_multivalued_w_method(self):
+        self._test("multivalues", "method")