openstack/neutron
Code Issues Proposed changes

[sqlalchemy-20] Open a connection to execute a command

Browse Source 
``Engine`` class no longer inherits from ``Connectable``. In order to
execute a SQL command, it is needed to start a connection and then
being a transaction. At this point, it is possible to execute a
command from the connection object.

Closes-Bug: #2020114

Change-Id: Id1dcf338bf04e6eef73cc447698ec6d2100abd96
This commit is contained in:
Rodolfo Alonso Hernandez
2023-05-18 03:33:10 +02:00
committed by Rodolfo Alonso
parent 6fa3d8019f
commit 0b67da59c6
1 changed files with 16 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
neutron/tests/functional/db/migrations/test_c3e9d13c4367_add_binding_index_to_.py
View File

@@ -26,18 +26,20 @@ class NetworkDhcpAgentBindingMigrationMixin(object):
def _create_so(self, o_type, values):
"""create standard attr object."""
stan = db_utils.get_table(self.engine, 'standardattributes')
# find next available id taking into account existing records
rec_ids = [r.id for r in self.engine.execute(stan.select()).fetchall()]
next_id = max([0] + rec_ids) + 1
self.engine.execute(stan.insert().values({'id': next_id,
'resource_type': o_type}))
with self.engine.connect() as conn, conn.begin():
stan = db_utils.get_table(self.engine, 'standardattributes')
# find next available id taking into account existing records
rec_ids = [r.id for r in conn.execute(stan.select()).fetchall()]
next_id = max([0] + rec_ids) + 1
conn.execute(stan.insert().values({'id': next_id,
'resource_type': o_type}))
values['standard_attr_id'] = next_id
return self._create_rec(o_type, values)
def _create_rec(self, o_type, values):
otable = db_utils.get_table(self.engine, o_type)
self.engine.execute(otable.insert().values(values))
with self.engine.connect() as conn, conn.begin():
conn.execute(otable.insert().values(values))
def _make_network_agents_and_bindings(self, network_id):
self._create_so('networks', {'id': network_id})
@@ -68,14 +70,15 @@ class NetworkDhcpAgentBindingMigrationMixin(object):
def _check_c3e9d13c4367(self, engine, data):
bindings_table = db_utils.get_table(engine, 'networkdhcpagentbindings')
rows = engine.execute(bindings_table.select()).fetchall()
with self.engine.connect() as conn, conn.begin():
rows = conn.execute(bindings_table.select()).fetchall()
networks_to_bindings = collections.defaultdict(list)
for network_id, agent_id, binding_index in rows:
networks_to_bindings[network_id].append(binding_index)
networks_to_bindings = collections.defaultdict(list)
for network_id, agent_id, binding_index in rows:
networks_to_bindings[network_id].append(binding_index)
for binding_indices in networks_to_bindings.values():
self.assertEqual(list(range(1, 3)), sorted(binding_indices))
for binding_indices in networks_to_bindings.values():
self.assertEqual(list(range(1, 3)), sorted(binding_indices))
class TestNetworkDhcpAgentBindingMigrationMySQL(