# -*- coding: utf-8 -*- # # Copyright (C) 2010-2023 Edgewall Software # All rights reserved. # # This software is licensed as described in the file COPYING, which # you should have received as part of this distribution. The terms # are also available at https://trac.edgewall.org/wiki/TracLicense. # # This software consists of voluntary contributions made by many # individuals. For the exact contribution history, see the revision # history and logs, available at https://trac.edgewall.org/log/. import itertools import unittest from trac.db.api import DatabaseManager, get_column_names from trac.db.mysql_backend import MySQLConnector from trac.db.schema import Table, Column, Index from trac.test import EnvironmentStub, Mock, get_dburi, makeSuite class MySQLTableAlterationSQLTest(unittest.TestCase): def setUp(self): self.env = EnvironmentStub() def test_alter_column_types(self): connector = MySQLConnector(self.env) sql = connector.alter_column_types('milestone', {'due': ('int', 'int64'), 'completed': ('int', 'int64')}) sql = list(sql) self.assertEqual([ "ALTER TABLE milestone " "MODIFY completed bigint, " "MODIFY due bigint", ], sql) def test_alter_column_types_same(self): connector = MySQLConnector(self.env) sql = connector.alter_column_types('milestone', {'due': ('int', 'int'), 'completed': ('int', 'int64')}) sql = list(sql) self.assertEqual([ "ALTER TABLE milestone " "MODIFY completed bigint", ], sql) def test_alter_column_types_none(self): connector = MySQLConnector(self.env) sql = connector.alter_column_types('milestone', {'due': ('int', 'int')}) self.assertEqual([], list(sql)) def test_utf8_size(self): connector = MySQLConnector(self.env) self.assertEqual(3, connector._max_bytes(Mock(charset='utf8'))) self.assertEqual(4, connector._max_bytes(Mock(charset='utf8mb4'))) def test_to_sql(self): connector = MySQLConnector(self.env) tab = Table('blah', key=('col1', 'col2', 'col3', 'col4', 'col5')) \ [Column('col1'), Column('col2'), Column('col3'), Column('col4'), Column('col5'), Column('col6'), Index(['col2', 'col3', 'col4', 'col5'])] sql = list(connector.to_sql(tab, max_bytes=3)) self.assertEqual(2, len(sql)) self.assertIn(' PRIMARY KEY (`col1`(204),`col2`(204),`col3`(204),' '`col4`(204),`col5`(204))', sql[0]) self.assertIn(' `blah_col2_col3_col4_col5_idx` ON `blah` (`col2`(255),' '`col3`(255),`col4`(255),`col5`(255))', sql[1]) sql = list(connector.to_sql(tab, max_bytes=4)) self.assertEqual(2, len(sql)) self.assertIn(' PRIMARY KEY (`col1`(153),`col2`(153),`col3`(153),' '`col4`(153),`col5`(153))', sql[0]) self.assertIn(' `blah_col2_col3_col4_col5_idx` ON `blah` (`col2`(191),' '`col3`(191),`col4`(191),`col5`(191))', sql[1]) class MySQLConnectionTestCase(unittest.TestCase): def setUp(self): self.env = EnvironmentStub() self.schema = [ Table('test_simple', key='id')[ Column('id', auto_increment=True), Column('username'), Column('email'), Column('enabled', type='int'), Column('extra'), Index(['username'], unique=True), Index(['email'], unique=False), ], Table('test_composite', key=['id', 'name'])[ Column('id', type='int'), Column('name'), Column('value'), Column('enabled', type='int'), Index(['name', 'value'], unique=False), Index(['name', 'enabled'], unique=True), ], ] self.dbm = DatabaseManager(self.env) self.dbm.drop_tables(self.schema) self.dbm.create_tables(self.schema) self.dbm.insert_into_tables([ ('test_simple', ('username', 'email', 'enabled'), [('joe', 'joe@example.org', 1), ('joé', 'joe@example.org', 0)]), ('test_composite', ('id', 'name', 'value', 'enabled'), [(1, 'foo', '42', 1), (1, 'bar', '42', 1), (2, 'foo', '43', 0), (2, 'bar', '43', 0)]), ]) def tearDown(self): DatabaseManager(self.env).drop_tables(self.schema) self.env.reset_db() def _show_index(self, table): with self.env.db_query as db: cursor = db.cursor() cursor.execute("SHOW INDEX FROM " + db.quote(table)) columns = get_column_names(cursor) rows = [dict(zip(columns, row)) for row in cursor] results = {} for index, group in itertools.groupby(rows, lambda v: v['Key_name']): group = list(group) results[index] = { 'unique': not group[0]['Non_unique'], 'columns': [row['Column_name'] for row in group], } return results def _drop_column(self, table, column): with self.env.db_transaction as db: db.drop_column(table, column) def _query(self, stmt, *args): return self.env.db_query(stmt, args) def test_remove_simple_keys(self): indices_0 = self._show_index('test_simple') self.assertEqual(['PRIMARY', 'test_simple_email_idx', 'test_simple_username_idx'], sorted(indices_0)) self.assertEqual({'unique': True, 'columns': ['id']}, indices_0['PRIMARY']) self.assertEqual({'unique': True, 'columns': ['username']}, indices_0['test_simple_username_idx']) self.assertEqual({'unique': False, 'columns': ['email']}, indices_0['test_simple_email_idx']) self._drop_column('test_simple', 'enabled') self.assertEqual(indices_0, self._show_index('test_simple')) self._drop_column('test_simple', 'username') indices_1 = self._show_index('test_simple') self.assertEqual(['PRIMARY', 'test_simple_email_idx'], sorted(indices_1)) self._drop_column('test_simple', 'email') indices_2 = self._show_index('test_simple') self.assertEqual(['PRIMARY'], sorted(indices_2)) self._drop_column('test_simple', 'id') indices_3 = self._show_index('test_simple') self.assertEqual({}, indices_3) def test_remove_composite_keys(self): indices_0 = self._show_index('test_composite') self.assertEqual(['PRIMARY', 'test_composite_name_enabled_idx', 'test_composite_name_value_idx'], sorted(indices_0)) self.assertEqual({'unique': True, 'columns': ['id', 'name']}, indices_0['PRIMARY']) self.assertEqual({'unique': False, 'columns': ['name', 'value']}, indices_0['test_composite_name_value_idx']) self.assertEqual({'unique': True, 'columns': ['name', 'enabled']}, indices_0['test_composite_name_enabled_idx']) self._drop_column('test_composite', 'id') indices_1 = self._show_index('test_composite') self.assertEqual(['test_composite_name_enabled_idx', 'test_composite_name_value_idx'], sorted(indices_1)) self.assertEqual(indices_0['test_composite_name_value_idx'], indices_1['test_composite_name_value_idx']) self.assertEqual(indices_0['test_composite_name_enabled_idx'], indices_1['test_composite_name_enabled_idx']) rows = self._query("""SELECT * FROM test_composite ORDER BY name, value, enabled""") self.assertEqual([('bar', '42', 1), ('bar', '43', 0), ('foo', '42', 1), ('foo', '43', 0)], rows) self._drop_column('test_composite', 'name') self.assertEqual({}, self._show_index('test_composite')) rows = self._query("""SELECT * FROM test_composite ORDER BY value, enabled""") self.assertEqual([('42', 1), ('42', 1), ('43', 0), ('43', 0)], rows) def test_suite(): suite = unittest.TestSuite() suite.addTest(makeSuite(MySQLTableAlterationSQLTest)) if get_dburi().startswith('mysql:'): suite.addTest(makeSuite(MySQLConnectionTestCase)) return suite if __name__ == '__main__': unittest.main(defaultTest='test_suite')