| # Copyright 2016 The Chromium Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style |
| # license that can be found in the LICENSE file or at |
| # https://developers.google.com/open-source/licenses/bsd |
| |
| """Unit tests for the sql module.""" |
| from __future__ import print_function |
| from __future__ import division |
| from __future__ import absolute_import |
| |
| import logging |
| import mock |
| import time |
| import unittest |
| |
| import settings |
| from framework import exceptions |
| from framework import sql |
| |
| |
| class MockSQLCnxn(object): |
| """This class mocks the connection and cursor classes.""" |
| |
| def __init__(self, instance, database): |
| self.instance = instance |
| self.database = database |
| self.last_executed = None |
| self.last_executed_args = None |
| self.result_rows = None |
| self.rowcount = 0 |
| self.lastrowid = None |
| self.pool_key = instance + '/' + database |
| self.is_bad = False |
| self.has_uncommitted = False |
| |
| def execute(self, stmt_str, args=None): |
| self.last_executed = stmt_str % tuple(args or []) |
| if not stmt_str.startswith(('SET', 'SELECT')): |
| self.has_uncommitted = True |
| |
| def executemany(self, stmt_str, args): |
| # We cannot format the string because args has many values for each %s. |
| self.last_executed = stmt_str |
| self.last_executed_args = tuple(args) |
| |
| # sql.py only calls executemany() for INSERT. |
| assert stmt_str.startswith('INSERT') |
| self.lastrowid = 123 |
| |
| def fetchall(self): |
| return self.result_rows |
| |
| def cursor(self): |
| return self |
| |
| def commit(self): |
| self.has_uncommitted = False |
| |
| def close(self): |
| assert not self.has_uncommitted |
| |
| def rollback(self): |
| self.has_uncommitted = False |
| |
| def ping(self): |
| if self.is_bad: |
| raise BaseException('connection error!') |
| |
| |
| sql.cnxn_ctor = MockSQLCnxn |
| |
| |
| class ConnectionPoolingTest(unittest.TestCase): |
| |
| def testGet(self): |
| pool_size = 2 |
| num_dbs = 2 |
| p = sql.ConnectionPool(pool_size) |
| |
| for i in range(num_dbs): |
| for _ in range(pool_size): |
| c = p.get('test', 'db%d' % i) |
| self.assertIsNotNone(c) |
| p.release(c) |
| |
| cnxn1 = p.get('test', 'db0') |
| q = p.queues[cnxn1.pool_key] |
| self.assertIs(q.qsize(), 0) |
| |
| p.release(cnxn1) |
| self.assertIs(q.qsize(), pool_size - 1) |
| self.assertIs(q.full(), False) |
| self.assertIs(q.empty(), False) |
| |
| cnxn2 = p.get('test', 'db0') |
| q = p.queues[cnxn2.pool_key] |
| self.assertIs(q.qsize(), 0) |
| self.assertIs(q.full(), False) |
| self.assertIs(q.empty(), True) |
| |
| def testGetAndReturnPooledCnxn(self): |
| p = sql.ConnectionPool(2) |
| |
| cnxn1 = p.get('test', 'db1') |
| self.assertIs(len(p.queues), 1) |
| |
| cnxn2 = p.get('test', 'db2') |
| self.assertIs(len(p.queues), 2) |
| |
| # Should use the existing pool. |
| cnxn3 = p.get('test', 'db1') |
| self.assertIs(len(p.queues), 2) |
| |
| p.release(cnxn3) |
| p.release(cnxn2) |
| |
| cnxn1.is_bad = True |
| p.release(cnxn1) |
| # cnxn1 should not be returned from the pool if we |
| # ask for a connection to its database. |
| |
| cnxn4 = p.get('test', 'db1') |
| |
| self.assertIsNot(cnxn1, cnxn4) |
| self.assertIs(len(p.queues), 2) |
| self.assertIs(cnxn4.is_bad, False) |
| |
| def testGetAndReturnPooledCnxn_badCnxn(self): |
| p = sql.ConnectionPool(2) |
| |
| cnxn1 = p.get('test', 'db1') |
| cnxn2 = p.get('test', 'db2') |
| cnxn3 = p.get('test', 'db1') |
| |
| cnxn3.is_bad = True |
| |
| p.release(cnxn3) |
| q = p.queues[cnxn3.pool_key] |
| self.assertIs(q.qsize(), 1) |
| |
| with self.assertRaises(BaseException): |
| cnxn3 = p.get('test', 'db1') |
| |
| q = p.queues[cnxn2.pool_key] |
| self.assertIs(q.qsize(), 0) |
| p.release(cnxn2) |
| self.assertIs(q.qsize(), 1) |
| |
| p.release(cnxn1) |
| q = p.queues[cnxn1.pool_key] |
| self.assertIs(q.qsize(), 1) |
| |
| |
| class MonorailConnectionTest(unittest.TestCase): |
| |
| def setUp(self): |
| self.cnxn = sql.MonorailConnection() |
| self.orig_local_mode = settings.local_mode |
| self.orig_num_logical_shards = settings.num_logical_shards |
| settings.local_mode = False |
| |
| def tearDown(self): |
| settings.local_mode = self.orig_local_mode |
| settings.num_logical_shards = self.orig_num_logical_shards |
| |
| def testGetPrimaryConnection(self): |
| sql_cnxn = self.cnxn.GetPrimaryConnection() |
| self.assertEqual(settings.db_instance, sql_cnxn.instance) |
| self.assertEqual(settings.db_database_name, sql_cnxn.database) |
| |
| sql_cnxn2 = self.cnxn.GetPrimaryConnection() |
| self.assertIs(sql_cnxn2, sql_cnxn) |
| |
| def testGetConnectionForShard(self): |
| sql_cnxn = self.cnxn.GetConnectionForShard(1) |
| replica_name = settings.db_replica_names[ |
| 1 % len(settings.db_replica_names)] |
| self.assertEqual(settings.physical_db_name_format % replica_name, |
| sql_cnxn.instance) |
| self.assertEqual(settings.db_database_name, sql_cnxn.database) |
| |
| sql_cnxn2 = self.cnxn.GetConnectionForShard(1) |
| self.assertIs(sql_cnxn2, sql_cnxn) |
| |
| def testClose(self): |
| sql_cnxn = self.cnxn.GetPrimaryConnection() |
| self.cnxn.Close() |
| self.assertFalse(sql_cnxn.has_uncommitted) |
| |
| def testExecute_Primary(self): |
| """Execute() with no shard passes the statement to the primary sql cnxn.""" |
| sql_cnxn = self.cnxn.GetPrimaryConnection() |
| with mock.patch.object(self.cnxn, '_ExecuteWithSQLConnection') as ewsc: |
| ewsc.return_value = 'db result' |
| actual_result = self.cnxn.Execute('statement', []) |
| self.assertEqual('db result', actual_result) |
| ewsc.assert_called_once_with(sql_cnxn, 'statement', [], commit=True) |
| |
| def testExecute_Shard(self): |
| """Execute() with a shard passes the statement to the shard sql cnxn.""" |
| shard_id = 1 |
| sql_cnxn_1 = self.cnxn.GetConnectionForShard(shard_id) |
| with mock.patch.object(self.cnxn, '_ExecuteWithSQLConnection') as ewsc: |
| ewsc.return_value = 'db result' |
| actual_result = self.cnxn.Execute('statement', [], shard_id=shard_id) |
| self.assertEqual('db result', actual_result) |
| ewsc.assert_called_once_with(sql_cnxn_1, 'statement', [], commit=True) |
| |
| def testExecute_Shard_Unavailable(self): |
| """If a shard is unavailable, we try the next one.""" |
| shard_id = 1 |
| sql_cnxn_1 = self.cnxn.GetConnectionForShard(shard_id) |
| sql_cnxn_2 = self.cnxn.GetConnectionForShard(shard_id + 1) |
| |
| # Simulate a recent failure on shard 1. |
| self.cnxn.unavailable_shards[1] = int(time.time()) - 3 |
| |
| with mock.patch.object(self.cnxn, '_ExecuteWithSQLConnection') as ewsc: |
| ewsc.return_value = 'db result' |
| actual_result = self.cnxn.Execute('statement', [], shard_id=shard_id) |
| self.assertEqual('db result', actual_result) |
| ewsc.assert_called_once_with(sql_cnxn_2, 'statement', [], commit=True) |
| |
| # Even a new MonorailConnection instance shares the same state. |
| other_cnxn = sql.MonorailConnection() |
| other_sql_cnxn_2 = other_cnxn.GetConnectionForShard(shard_id + 1) |
| |
| with mock.patch.object(other_cnxn, '_ExecuteWithSQLConnection') as ewsc: |
| ewsc.return_value = 'db result' |
| actual_result = other_cnxn.Execute('statement', [], shard_id=shard_id) |
| self.assertEqual('db result', actual_result) |
| ewsc.assert_called_once_with( |
| other_sql_cnxn_2, 'statement', [], commit=True) |
| |
| # Simulate an old failure on shard 1, allowing us to try using it again. |
| self.cnxn.unavailable_shards[1] = ( |
| int(time.time()) - sql.BAD_SHARD_AVOIDANCE_SEC - 2) |
| |
| with mock.patch.object(self.cnxn, '_ExecuteWithSQLConnection') as ewsc: |
| ewsc.return_value = 'db result' |
| actual_result = self.cnxn.Execute('statement', [], shard_id=shard_id) |
| self.assertEqual('db result', actual_result) |
| ewsc.assert_called_once_with(sql_cnxn_1, 'statement', [], commit=True) |
| |
| |
| class TableManagerTest(unittest.TestCase): |
| |
| def setUp(self): |
| self.emp_tbl = sql.SQLTableManager('Employee') |
| self.cnxn = sql.MonorailConnection() |
| self.primary_cnxn = self.cnxn.GetPrimaryConnection() |
| |
| def testSelect_Trivial(self): |
| self.primary_cnxn.result_rows = [(111, True), (222, False)] |
| rows = self.emp_tbl.Select(self.cnxn) |
| self.assertEqual('SELECT * FROM Employee', self.primary_cnxn.last_executed) |
| self.assertEqual([(111, True), (222, False)], rows) |
| |
| def testSelect_Conditions(self): |
| self.primary_cnxn.result_rows = [(111,)] |
| rows = self.emp_tbl.Select( |
| self.cnxn, cols=['emp_id'], fulltime=True, dept_id=[10, 20]) |
| self.assertEqual( |
| 'SELECT emp_id FROM Employee' |
| '\nWHERE dept_id IN (10,20)' |
| '\n AND fulltime = 1', self.primary_cnxn.last_executed) |
| self.assertEqual([(111,)], rows) |
| |
| def testSelectRow(self): |
| self.primary_cnxn.result_rows = [(111,)] |
| row = self.emp_tbl.SelectRow( |
| self.cnxn, cols=['emp_id'], fulltime=True, dept_id=[10, 20]) |
| self.assertEqual( |
| 'SELECT DISTINCT emp_id FROM Employee' |
| '\nWHERE dept_id IN (10,20)' |
| '\n AND fulltime = 1', self.primary_cnxn.last_executed) |
| self.assertEqual((111,), row) |
| |
| def testSelectRow_NoMatches(self): |
| self.primary_cnxn.result_rows = [] |
| row = self.emp_tbl.SelectRow( |
| self.cnxn, cols=['emp_id'], fulltime=True, dept_id=[99]) |
| self.assertEqual( |
| 'SELECT DISTINCT emp_id FROM Employee' |
| '\nWHERE dept_id IN (99)' |
| '\n AND fulltime = 1', self.primary_cnxn.last_executed) |
| self.assertEqual(None, row) |
| |
| row = self.emp_tbl.SelectRow( |
| self.cnxn, cols=['emp_id'], fulltime=True, dept_id=[99], |
| default=(-1,)) |
| self.assertEqual((-1,), row) |
| |
| def testSelectValue(self): |
| self.primary_cnxn.result_rows = [(111,)] |
| val = self.emp_tbl.SelectValue( |
| self.cnxn, 'emp_id', fulltime=True, dept_id=[10, 20]) |
| self.assertEqual( |
| 'SELECT DISTINCT emp_id FROM Employee' |
| '\nWHERE dept_id IN (10,20)' |
| '\n AND fulltime = 1', self.primary_cnxn.last_executed) |
| self.assertEqual(111, val) |
| |
| def testSelectValue_NoMatches(self): |
| self.primary_cnxn.result_rows = [] |
| val = self.emp_tbl.SelectValue( |
| self.cnxn, 'emp_id', fulltime=True, dept_id=[99]) |
| self.assertEqual( |
| 'SELECT DISTINCT emp_id FROM Employee' |
| '\nWHERE dept_id IN (99)' |
| '\n AND fulltime = 1', self.primary_cnxn.last_executed) |
| self.assertEqual(None, val) |
| |
| val = self.emp_tbl.SelectValue( |
| self.cnxn, 'emp_id', fulltime=True, dept_id=[99], |
| default=-1) |
| self.assertEqual(-1, val) |
| |
| def testInsertRow(self): |
| self.primary_cnxn.rowcount = 1 |
| generated_id = self.emp_tbl.InsertRow(self.cnxn, emp_id=111, fulltime=True) |
| self.assertEqual( |
| 'INSERT INTO Employee (emp_id, fulltime)' |
| '\nVALUES (%s,%s)', self.primary_cnxn.last_executed) |
| self.assertEqual(([111, 1],), self.primary_cnxn.last_executed_args) |
| self.assertEqual(123, generated_id) |
| |
| def testInsertRows_Empty(self): |
| generated_id = self.emp_tbl.InsertRows( |
| self.cnxn, ['emp_id', 'fulltime'], []) |
| self.assertIsNone(self.primary_cnxn.last_executed) |
| self.assertIsNone(self.primary_cnxn.last_executed_args) |
| self.assertEqual(None, generated_id) |
| |
| def testInsertRows(self): |
| self.primary_cnxn.rowcount = 2 |
| generated_ids = self.emp_tbl.InsertRows( |
| self.cnxn, ['emp_id', 'fulltime'], [(111, True), (222, False)]) |
| self.assertEqual( |
| 'INSERT INTO Employee (emp_id, fulltime)' |
| '\nVALUES (%s,%s)', self.primary_cnxn.last_executed) |
| self.assertEqual(([111, 1], [222, 0]), self.primary_cnxn.last_executed_args) |
| self.assertEqual([], generated_ids) |
| |
| def testUpdate(self): |
| self.primary_cnxn.rowcount = 2 |
| rowcount = self.emp_tbl.Update( |
| self.cnxn, {'fulltime': True}, emp_id=[111, 222]) |
| self.assertEqual( |
| 'UPDATE Employee SET fulltime=1' |
| '\nWHERE emp_id IN (111,222)', self.primary_cnxn.last_executed) |
| self.assertEqual(2, rowcount) |
| |
| def testUpdate_Limit(self): |
| self.emp_tbl.Update( |
| self.cnxn, {'fulltime': True}, limit=8, emp_id=[111, 222]) |
| self.assertEqual( |
| 'UPDATE Employee SET fulltime=1' |
| '\nWHERE emp_id IN (111,222)' |
| '\nLIMIT 8', self.primary_cnxn.last_executed) |
| |
| def testIncrementCounterValue(self): |
| self.primary_cnxn.rowcount = 1 |
| self.primary_cnxn.lastrowid = 9 |
| new_counter_val = self.emp_tbl.IncrementCounterValue( |
| self.cnxn, 'years_worked', emp_id=111) |
| self.assertEqual( |
| 'UPDATE Employee SET years_worked = LAST_INSERT_ID(years_worked + 1)' |
| '\nWHERE emp_id = 111', self.primary_cnxn.last_executed) |
| self.assertEqual(9, new_counter_val) |
| |
| def testDelete(self): |
| self.primary_cnxn.rowcount = 1 |
| rowcount = self.emp_tbl.Delete(self.cnxn, fulltime=True) |
| self.assertEqual( |
| 'DELETE FROM Employee' |
| '\nWHERE fulltime = 1', self.primary_cnxn.last_executed) |
| self.assertEqual(1, rowcount) |
| |
| def testDelete_Limit(self): |
| self.emp_tbl.Delete(self.cnxn, fulltime=True, limit=3) |
| self.assertEqual( |
| 'DELETE FROM Employee' |
| '\nWHERE fulltime = 1' |
| '\nLIMIT 3', self.primary_cnxn.last_executed) |
| |
| |
| class StatementTest(unittest.TestCase): |
| |
| def testMakeSelect(self): |
| stmt = sql.Statement.MakeSelect('Employee', ['emp_id', 'fulltime']) |
| stmt_str, args = stmt.Generate() |
| self.assertEqual( |
| 'SELECT emp_id, fulltime FROM Employee', |
| stmt_str) |
| self.assertEqual([], args) |
| |
| stmt = sql.Statement.MakeSelect( |
| 'Employee', ['emp_id', 'fulltime'], distinct=True) |
| stmt_str, args = stmt.Generate() |
| self.assertEqual( |
| 'SELECT DISTINCT emp_id, fulltime FROM Employee', |
| stmt_str) |
| self.assertEqual([], args) |
| |
| def testMakeInsert(self): |
| stmt = sql.Statement.MakeInsert( |
| 'Employee', ['emp_id', 'fulltime'], [(111, True), (222, False)]) |
| stmt_str, args = stmt.Generate() |
| self.assertEqual( |
| 'INSERT INTO Employee (emp_id, fulltime)' |
| '\nVALUES (%s,%s)', |
| stmt_str) |
| self.assertEqual([[111, 1], [222, 0]], args) |
| |
| stmt = sql.Statement.MakeInsert( |
| 'Employee', ['emp_id', 'fulltime'], [(111, False)], replace=True) |
| stmt_str, args = stmt.Generate() |
| self.assertEqual( |
| 'INSERT INTO Employee (emp_id, fulltime)' |
| '\nVALUES (%s,%s)' |
| '\nON DUPLICATE KEY UPDATE ' |
| 'emp_id=VALUES(emp_id), fulltime=VALUES(fulltime)', |
| stmt_str) |
| self.assertEqual([[111, 0]], args) |
| |
| stmt = sql.Statement.MakeInsert( |
| 'Employee', ['emp_id', 'fulltime'], [(111, False)], ignore=True) |
| stmt_str, args = stmt.Generate() |
| self.assertEqual( |
| 'INSERT IGNORE INTO Employee (emp_id, fulltime)' |
| '\nVALUES (%s,%s)', |
| stmt_str) |
| self.assertEqual([[111, 0]], args) |
| |
| def testMakeInsert_InvalidString(self): |
| with self.assertRaises(exceptions.InputException): |
| sql.Statement.MakeInsert( |
| 'Employee', ['emp_id', 'name'], [(111, 'First \x00 Last')]) |
| |
| def testMakeUpdate(self): |
| stmt = sql.Statement.MakeUpdate('Employee', {'fulltime': True}) |
| stmt_str, args = stmt.Generate() |
| self.assertEqual( |
| 'UPDATE Employee SET fulltime=%s', |
| stmt_str) |
| self.assertEqual([1], args) |
| |
| def testMakeUpdate_InvalidString(self): |
| with self.assertRaises(exceptions.InputException): |
| sql.Statement.MakeUpdate('Employee', {'name': 'First \x00 Last'}) |
| |
| def testMakeIncrement(self): |
| stmt = sql.Statement.MakeIncrement('Employee', 'years_worked') |
| stmt_str, args = stmt.Generate() |
| self.assertEqual( |
| 'UPDATE Employee SET years_worked = LAST_INSERT_ID(years_worked + %s)', |
| stmt_str) |
| self.assertEqual([1], args) |
| |
| stmt = sql.Statement.MakeIncrement('Employee', 'years_worked', step=5) |
| stmt_str, args = stmt.Generate() |
| self.assertEqual( |
| 'UPDATE Employee SET years_worked = LAST_INSERT_ID(years_worked + %s)', |
| stmt_str) |
| self.assertEqual([5], args) |
| |
| def testMakeDelete(self): |
| stmt = sql.Statement.MakeDelete('Employee') |
| stmt_str, args = stmt.Generate() |
| self.assertEqual( |
| 'DELETE FROM Employee', |
| stmt_str) |
| self.assertEqual([], args) |
| |
| def testAddUseClause(self): |
| stmt = sql.Statement.MakeSelect('Employee', ['emp_id', 'fulltime']) |
| stmt.AddUseClause('USE INDEX (emp_id) USE INDEX FOR ORDER BY (emp_id)') |
| stmt.AddOrderByTerms([('emp_id', [])]) |
| stmt_str, args = stmt.Generate() |
| self.assertEqual( |
| 'SELECT emp_id, fulltime FROM Employee' |
| '\nUSE INDEX (emp_id) USE INDEX FOR ORDER BY (emp_id)' |
| '\nORDER BY emp_id', |
| stmt_str) |
| self.assertEqual([], args) |
| |
| def testAddJoinClause_Empty(self): |
| stmt = sql.Statement.MakeSelect('Employee', ['emp_id', 'fulltime']) |
| stmt.AddJoinClauses([]) |
| stmt_str, args = stmt.Generate() |
| self.assertEqual( |
| 'SELECT emp_id, fulltime FROM Employee', |
| stmt_str) |
| self.assertEqual([], args) |
| |
| def testAddJoinClause(self): |
| stmt = sql.Statement.MakeSelect('Employee', ['emp_id', 'fulltime']) |
| stmt.AddJoinClauses([('CorporateHoliday', [])]) |
| stmt.AddJoinClauses( |
| [('Product ON Project.inventor_id = emp_id', [])], left=True) |
| stmt_str, args = stmt.Generate() |
| self.assertEqual( |
| 'SELECT emp_id, fulltime FROM Employee' |
| '\n JOIN CorporateHoliday' |
| '\n LEFT JOIN Product ON Project.inventor_id = emp_id', |
| stmt_str) |
| self.assertEqual([], args) |
| |
| def testAddGroupByTerms_Empty(self): |
| stmt = sql.Statement.MakeSelect('Employee', ['emp_id', 'fulltime']) |
| stmt.AddGroupByTerms([]) |
| stmt_str, args = stmt.Generate() |
| self.assertEqual( |
| 'SELECT emp_id, fulltime FROM Employee', |
| stmt_str) |
| self.assertEqual([], args) |
| |
| def testAddGroupByTerms(self): |
| stmt = sql.Statement.MakeSelect('Employee', ['emp_id', 'fulltime']) |
| stmt.AddGroupByTerms(['dept_id', 'location_id']) |
| stmt_str, args = stmt.Generate() |
| self.assertEqual( |
| 'SELECT emp_id, fulltime FROM Employee' |
| '\nGROUP BY dept_id, location_id', |
| stmt_str) |
| self.assertEqual([], args) |
| |
| def testAddOrderByTerms_Empty(self): |
| stmt = sql.Statement.MakeSelect('Employee', ['emp_id', 'fulltime']) |
| stmt.AddOrderByTerms([]) |
| stmt_str, args = stmt.Generate() |
| self.assertEqual( |
| 'SELECT emp_id, fulltime FROM Employee', |
| stmt_str) |
| self.assertEqual([], args) |
| |
| def testAddOrderByTerms(self): |
| stmt = sql.Statement.MakeSelect('Employee', ['emp_id', 'fulltime']) |
| stmt.AddOrderByTerms([('dept_id', []), ('emp_id DESC', [])]) |
| stmt_str, args = stmt.Generate() |
| self.assertEqual( |
| 'SELECT emp_id, fulltime FROM Employee' |
| '\nORDER BY dept_id, emp_id DESC', |
| stmt_str) |
| self.assertEqual([], args) |
| |
| def testSetLimitAndOffset(self): |
| stmt = sql.Statement.MakeSelect('Employee', ['emp_id', 'fulltime']) |
| stmt.SetLimitAndOffset(100, 0) |
| stmt_str, args = stmt.Generate() |
| self.assertEqual( |
| 'SELECT emp_id, fulltime FROM Employee' |
| '\nLIMIT 100', |
| stmt_str) |
| self.assertEqual([], args) |
| |
| stmt.SetLimitAndOffset(100, 500) |
| stmt_str, args = stmt.Generate() |
| self.assertEqual( |
| 'SELECT emp_id, fulltime FROM Employee' |
| '\nLIMIT 100 OFFSET 500', |
| stmt_str) |
| self.assertEqual([], args) |
| |
| def testAddWhereTerms_Select(self): |
| stmt = sql.Statement.MakeSelect('Employee', ['emp_id', 'fulltime']) |
| stmt.AddWhereTerms([], emp_id=[111, 222]) |
| stmt_str, args = stmt.Generate() |
| self.assertEqual( |
| 'SELECT emp_id, fulltime FROM Employee' |
| '\nWHERE emp_id IN (%s,%s)', |
| stmt_str) |
| self.assertEqual([111, 222], args) |
| |
| def testAddWhereTerms_Update(self): |
| stmt = sql.Statement.MakeUpdate('Employee', {'fulltime': True}) |
| stmt.AddWhereTerms([], emp_id=[111, 222]) |
| stmt_str, args = stmt.Generate() |
| self.assertEqual( |
| 'UPDATE Employee SET fulltime=%s' |
| '\nWHERE emp_id IN (%s,%s)', |
| stmt_str) |
| self.assertEqual([1, 111, 222], args) |
| |
| def testAddWhereTerms_Delete(self): |
| stmt = sql.Statement.MakeDelete('Employee') |
| stmt.AddWhereTerms([], emp_id=[111, 222]) |
| stmt_str, args = stmt.Generate() |
| self.assertEqual( |
| 'DELETE FROM Employee' |
| '\nWHERE emp_id IN (%s,%s)', |
| stmt_str) |
| self.assertEqual([111, 222], args) |
| |
| def testAddWhereTerms_Empty(self): |
| """Add empty terms should have no effect.""" |
| stmt = sql.Statement.MakeSelect('Employee', ['emp_id', 'fulltime']) |
| stmt.AddWhereTerms([]) |
| stmt_str, args = stmt.Generate() |
| self.assertEqual( |
| 'SELECT emp_id, fulltime FROM Employee', |
| stmt_str) |
| self.assertEqual([], args) |
| |
| def testAddWhereTerms_UpdateEmptyArray(self): |
| """Add empty array should throw an exception.""" |
| stmt = sql.Statement.MakeUpdate('SpamVerdict', {'user_id': 1}) |
| # See https://crbug.com/monorail/6735. |
| with self.assertRaises(exceptions.InputException): |
| stmt.AddWhereTerms([], user_id=[]) |
| mock_log.assert_called_once_with('Invalid update DB value %r', 'user_id') |
| |
| def testAddWhereTerms_MulitpleTerms(self): |
| stmt = sql.Statement.MakeSelect('Employee', ['emp_id', 'fulltime']) |
| stmt.AddWhereTerms( |
| [('emp_id %% %s = %s', [2, 0])], fulltime=True, emp_id_not=222) |
| stmt_str, args = stmt.Generate() |
| self.assertEqual( |
| 'SELECT emp_id, fulltime FROM Employee' |
| '\nWHERE emp_id %% %s = %s' |
| '\n AND emp_id != %s' |
| '\n AND fulltime = %s', |
| stmt_str) |
| self.assertEqual([2, 0, 222, 1], args) |
| |
| def testAddHavingTerms_NoGroupBy(self): |
| stmt = sql.Statement.MakeSelect('Employee', ['emp_id', 'fulltime']) |
| stmt.AddHavingTerms([('COUNT(*) > %s', [10])]) |
| self.assertRaises(AssertionError, stmt.Generate) |
| |
| def testAddHavingTerms_WithGroupBy(self): |
| stmt = sql.Statement.MakeSelect('Employee', ['emp_id', 'fulltime']) |
| stmt.AddGroupByTerms(['dept_id', 'location_id']) |
| stmt.AddHavingTerms([('COUNT(*) > %s', [10])]) |
| stmt_str, args = stmt.Generate() |
| self.assertEqual( |
| 'SELECT emp_id, fulltime FROM Employee' |
| '\nGROUP BY dept_id, location_id' |
| '\nHAVING COUNT(*) > %s', |
| stmt_str) |
| self.assertEqual([10], args) |
| |
| |
| class FunctionsTest(unittest.TestCase): |
| |
| def testIsValidDBValue_NonString(self): |
| self.assertTrue(sql._IsValidDBValue(12)) |
| self.assertTrue(sql._IsValidDBValue(True)) |
| self.assertTrue(sql._IsValidDBValue(False)) |
| self.assertTrue(sql._IsValidDBValue(None)) |
| |
| def testIsValidDBValue_String(self): |
| self.assertTrue(sql._IsValidDBValue('')) |
| self.assertTrue(sql._IsValidDBValue('hello')) |
| self.assertTrue(sql._IsValidDBValue(u'hello')) |
| self.assertFalse(sql._IsValidDBValue('null \x00 byte')) |
| |
| def testBoolsToInts_NoChanges(self): |
| self.assertEqual(['hello'], sql._BoolsToInts(['hello'])) |
| self.assertEqual([['hello']], sql._BoolsToInts([['hello']])) |
| self.assertEqual([['hello']], sql._BoolsToInts([('hello',)])) |
| self.assertEqual([12], sql._BoolsToInts([12])) |
| self.assertEqual([[12]], sql._BoolsToInts([[12]])) |
| self.assertEqual([[12]], sql._BoolsToInts([(12,)])) |
| self.assertEqual( |
| [12, 13, 'hi', [99, 'yo']], |
| sql._BoolsToInts([12, 13, 'hi', [99, 'yo']])) |
| |
| def testBoolsToInts_WithChanges(self): |
| self.assertEqual([1, 0], sql._BoolsToInts([True, False])) |
| self.assertEqual([[1, 0]], sql._BoolsToInts([[True, False]])) |
| self.assertEqual([[1, 0]], sql._BoolsToInts([(True, False)])) |
| self.assertEqual( |
| [12, 1, 'hi', [0, 'yo']], |
| sql._BoolsToInts([12, True, 'hi', [False, 'yo']])) |
| |
| def testRandomShardID(self): |
| """A random shard ID must always be a valid shard ID.""" |
| shard_id = sql.RandomShardID() |
| self.assertTrue(0 <= shard_id < settings.num_logical_shards) |