~cboylan/+junk/test_drizzle_dbapi : revision 3

1

#!/usr/bin/env python

2

'''

3

$Id: dbapi20.py,v 1.2 2003/02/12 09:48:46 zenzen Exp $

3

$Id: dbapi20.py,v 1.3 2003/02/12 22:54:38 zenzen Exp $

4

'''

5

6

__rcs_id__ = '$Id: dbapi20.py,v 1.2 2003/02/12 09:48:46 zenzen Exp $'

7

__version__ = '$Revision: 1.2 $'[11:-2]

6

__rcs_id__ = '$Id: dbapi20.py,v 1.3 2003/02/12 22:54:38 zenzen Exp $'

7

__version__ = '$Revision: 1.3 $'[11:-2]

8

__author__ = 'Stuart Bishop <zen@shangri-la.dropbear.id.au>'

9

10

import unittest

11

import time

11

12

13

class test_DBAPI20:

13

14

''' Test a database self.driver for DB API 2.0 compatibility.

47

48

def tearDown(self):

48

49

''' self.drivers should override this method to perform required cleanup

49

50

if any is necessary, such as deleting the dest database.

51

The default drops the tables that may be created.

50

52

'''

51

53

con = self._connect()

52

54

try:

75

77

76

78

def test_apilevel(self):

77

79

try:

80

# Must exist

78

81

apilevel = self.driver.apilevel

82

# Must equal 2.0

79

83

self.assertEqual(apilevel,'2.0')

80

84

except AttributeError:

81

85

self.fail("Driver doesn't define apilevel")

82

86

83

87

def test_threadsafety(self):

84

88

try:

89

# Must exist

85

90

threadsafety = self.driver.threadsafety

91

# Must be a valid value

86

92

self.failUnless(threadsafety in (0,1,2,3))

87

93

except AttributeError:

88

94

self.fail("Driver doesn't define threadsafety")

89

95

90

96

def test_paramstyle(self):

91

97

try:

98

# Must exist

92

99

paramstyle = self.driver.paramstyle

100

# Must be a valid value

93

101

self.failUnless(paramstyle in (

94

102

'qmark','numeric','named','format','pyformat'

95

103

))

97

105

self.fail("Driver doesn't define paramstyle")

98

106

99

107

def test_Exceptions(self):

108

# Make sure required exceptions exist, and are in the

109

# defined heirarchy.

100

110

self.failUnless(issubclass(self.driver.Warning,StandardError))

101

111

self.failUnless(issubclass(self.driver.Error,StandardError))

102

112

self.failUnless(issubclass(self.driver.InterfaceError,StandardError))

110

120

def test_commit(self):

111

121

con = self._connect()

112

122

try:

123

# Commit must work, even if it doesn't do anything

113

124

con.commit()

114

125

finally:

115

126

con.close()

116

127

117

128

def test_rollback(self):

118

129

con = self._connect()

130

# If rollback is defined, it should either work or throw

131

# the documented exception

119

132

if hasattr(con,'rollback'):

120

133

try:

121

134

con.rollback()

132

145

def test_cursor_isolation(self):

133

146

con = self._connect()

134

147

try:

148

# Make sure cursors created from the same connection have

149

# the documented transaction isolation level

135

150

cur1 = con.cursor()

136

151

cur2 = con.cursor()

137

152

cur1.execute(self.ddl1)

149

164

try:

150

165

cur = con.cursor()

151

166

cur.execute(self.ddl1)

152

self.assertEqual(cur.description,None)

167

self.assertEqual(cur.description,None,

168

'cursor.description should be none after executing a '

169

'statement that can return no rows (such as DDL)'

170

)

153

171

cur.execute('select name from booze')

154

self.assertEqual(len(cur.description),1)

155

self.assertEqual(len(cur.description[0]),7)

156

self.assertEqual(cur.description[0][0].lower(),'name')

157

self.failIfEqual(cur.description[0][1],self.driver.STRING)

172

self.assertEqual(len(cur.description),1,

173

'cursor.description describes too many columns'

174

)

175

self.assertEqual(len(cur.description[0]),7,

176

'cursor.description[x] tuples must have 7 elements'

177

)

178

self.assertEqual(cur.description[0][0].lower(),'name',

179

'cursor.description[x][0] must return column name'

180

)

181

self.failIfEqual(cur.description[0][1],self.driver.STRING,

182

'cursor.description[x][1] must return column type. Got %r'

183

% cur.description[0][1]

184

)

158

185

159

186

# Make sure self.description gets reset

160

187

cur.execute(self.ddl2)

161

self.assertEqual(cur.description,None)

188

self.assertEqual(cur.description,None,

189

'cursor.description not being set to None when executing '

190

'no-result statements (eg. DDL)'

191

)

162

192

finally:

163

193

con.close()

164

194

167

197

try:

168

198

cur = con.cursor()

169

199

cur.execute(self.ddl1)

170

self.assertEqual(cur.rowcount,-1)

200

self.assertEqual(cur.rowcount,-1,

201

'cursor.rowcount should be -1 after executing no-result '

202

'statements'

203

)

171

204

cur.execute("insert into booze values ('Victoria Bitter')")

172

self.failUnless(cur.rowcount in (-1,1))

205

self.failUnless(cur.rowcount in (-1,1),

206

'cursor.rowcount should == number or rows inserted, or '

207

'set to -1 after executing an insert statement'

208

)

173

209

cur.execute("select name from booze")

174

self.failUnless(cur.rowcount in (-1,1))

210

self.failUnless(cur.rowcount in (-1,1),

211

'cursor.rowcount should == number of rows returned, or '

212

'set to -1 after executing a select statement'

213

)

175

214

cur.execute(ddl2)

176

self.assertEqual(cur.rowcount,-1)

215

self.assertEqual(cur.rowcount,-1,

216

'cursor.rowcount not being reset to -1 after executing '

217

'no-result statements'

218

)

177

219

finally:

178

220

con.close()

179

221

189

231

cur = con.cursor()

190

232

finally:

191

233

con.close()

234

235

# cursor.execute should raise an Error if called after connection

236

# closed

192

237

self.assertRaises(self.driver.Error,cur.execute,self.ddl1)

238

239

# connection.commit should raise an Error if called after connection'

240

# closed.'

193

241

self.assertRaises(self.driver.Error,con.commit)

242

243

# connection.close should raise an Error if called more than once

194

244

self.assertRaises(self.driver.Error,con.close)

195

245

196

246

def test_execute(self):

219

269

'insert into booze values (%(beer)s)',{'beer':"Cooper's"}

220

270

)

221

271

else:

222

self.fail('Unknown paramstyle')

272

self.fail('Invalid paramstyle')

223

273

cur.execute('select name from booze')

224

274

res = cur.fetchall()

225

self.assertEqual(len(res),2)

275

self.assertEqual(len(res),2,'cursor.fetchall returned too few rows')

226

276

beers = [res[0][0],res[1][0]]

227

277

beers.sort()

228

self.assertEqual(beers[0],"Cooper's")

229

self.assertEqual(beers[1],"Victoria Bitter")

278

self.assertEqual(beers[0],"Cooper's",

279

'cursor.fetchall retrieved invalid data, or data inserted '

280

'incorrectly'

281

)

282

self.assertEqual(beers[1],"Victoria Bitter",

283

'cursor.fetchall retrieved invalid data, or data inserted '

284

'incorrectly'

285

)

230

286

231

287

def test_executemany(self):

232

288

con = self._connect()

249

305

self.fail('Unknown paramstyle')

250

306

cur.execute('select name from booze')

251

307

res = cur.fetchall()

252

self.assertEqual(len(res),2)

308

self.assertEqual(len(res),2,

309

'cursor.fetchall retrieved incorrect number of rows'

310

)

253

311

beers = [res[0][0],res[1][0]]

254

312

beers.sort()

255

self.assertEqual(beers[0],"Boag's")

256

self.assertEqual(beers[1],"Cooper's")

313

self.assertEqual(beers[0],"Boag's",'incorrect data retrieved')

314

self.assertEqual(beers[1],"Cooper's",'incorrect data retrieved')

257

315

finally:

258

316

con.close()

259

317

261

319

con = self._connect()

262

320

try:

263

321

cur = con.cursor()

322

323

# cursor.fetchone should raise an Error if called before

324

# executing a select-type query

264

325

self.assertRaises(self.driver.Error,cur.fetchone)

326

327

# cursor.fetchone should raise an Error if called after

328

# executing a query that cannnot return rows

265

329

cur.execute(self.ddl1)

266

330

self.assertRaises(self.driver.Error,cur.fetchone)

331

332

# cursor.fetchone should raise an Error if called after

333

# executing a query that cannnot return rows

267

334

cur.execute("insert into booze values ('Victoria Bitter')")

268

335

self.assertRaises(self.driver.Error,cur.fetchone)

336

269

337

cur.execute('select name from booze')

270

338

r = cur.fetchone()

271

self.assertEqual(len(r),1)

272

self.assertEqual(r[0],'Victoria Bitter')

273

self.assertEqual(cur.fetchone(),None)

339

self.assertEqual(len(r),1,

340

'cursor.fetchone should have retrieved a single row'

341

)

342

self.assertEqual(r[0],'Victoria Bitter',

343

'cursor.fetchone retrieved incorrect data'

344

)

345

self.assertEqual(cur.fetchone(),None,

346

'cursor.fetchone should return None if no more rows available'

347

)

274

348

finally:

275

349

con.close()

276

350

291

365

try:

292

366

cur = con.cursor()

293

367

294

self.assertRaises(self.driver.Error,cur.fetchmany,size=4)

368

# cursor.fetchmany should raise an Error if called without

369

#issuing a query

370

self.assertRaises(self.driver.Error,cur.fetchmany,4)

295

371

296

372

for sql in self.populate:

297

373

cur.execute(sql)

299

375

cur.arraysize=10

300

376

cur.execute('select name from booze')

301

377

r = cur.fetchmany(4) # Should get 4 rows

302

self.assertEqual(len(r),4)

378

self.assertEqual(len(r),4,

379

'cursor.fetchmany retrieved incorrect number of rows'

380

)

303

381

r = cur.fetchmany(4) # Should get 2 more

304

self.assertEqual(len(r),2)

382

self.assertEqual(len(r),2,

383

'cursor.fetchmany retrieved incorrect number of rows'

384

)

305

385

r = cur.fetchmany(4) # Should be an empty sequence

306

self.assertEqual(len(r),0)

307

308

cur.execute('select name from booze')

309

r = cur.fetchmany(size=4) # Should get 4 rows

310

self.assertEqual(len(r),4)

311

r = cur.fetchmany(size=1) # Should get 1 more

312

self.assertEqual(len(r),1)

313

r = cur.fetchmany(size=4) # Should get 1 more

314

self.assertEqual(len(r),1)

315

r = cur.fetchmany(size=4) # Should get an empty sequence

316

self.assertEqual(len(r),0)

317

386

self.assertEqual(len(r),0,

387

'cursor.fetchmany should return an empty sequence after '

388

'results are exhausted'

389

)

390

391

# Same as above, using cursor.arraysize

318

392

cur.arraysize=4

319

393

cur.execute('select name from booze')

320

394

r = cur.fetchmany() # Should get 4 rows

321

self.assertEqual(len(r),4)

395

self.assertEqual(len(r),4,

396

'cursor.arraysize not being honoured by fetchmany'

397

)

322

398

r = cur.fetchmany() # Should get 2 more

323

399

self.assertEqual(len(r),2)

324

400

r = cur.fetchmany() # Should be an empty sequence

327

403

cur.arraysize=6

328

404

cur.execute('select name from booze')

329

405

rows = cur.fetchmany() # Should get all rows

330

self.assertEqual(len(r),6)

331

self.assertEqual(len(r),6)

406

self.assertEqual(len(rows),6)

407

self.assertEqual(len(rows),6)

332

408

rows = [r[0] for r in rows]

333

409

rows.sort()

334

410

335

411

# Make sure we get the right data back out

336

412

for i in range(0,6):

337

self.assertEqual(rows[i],self.samples[i])

413

self.assertEqual(rows[i],self.samples[i],

414

'incorrect data retrieved by cursor.fetchmany'

415

)

338

416

339

417

cur.execute(self.ddl2)

340

418

cur.execute('select name from barflys')

341

419

r = cur.fetchmany() # Should get empty sequence

342

self.assertEqual(len(r),0)

420

self.assertEqual(len(r),0,

421

'cursor.fetchmany should return an empty sequence if '

422

'query retrieved no rows'

423

)

343

424

344

425

finally:

345

426

con.close()

348

429

con = self._connect()

349

430

try:

350

431

cur = con.cursor()

351

self.assertRaises(self.driver.Error,cur.fetchall)

432

# cursor.fetchall should raise an Error if called

433

# without executing a query that may return rows (such

434

# as a select)

435

self.assertRaises(self.driver.Error, cur.fetchall)

352

436

353

437

for sql in self.populate:

354

438

cur.execute(sql)

439

440

# cursor.fetchall should raise an Error if called

441

# after executing a a statement that cannot return rows

355

442

self.assertRaises(self.driver.Error,cur.fetchall)

356

443

357

444

cur.execute('select name from booze')

358

445

rows = cur.fetchall()

359

self.assertEqual(len(rows),len(self.samples))

446

self.assertEqual(len(rows),len(self.samples),

447

'cursor.fetchall did not retrieve all rows'

448

)

360

449

rows = [r[0] for r in rows]

361

450

rows.sort()

362

451

for i in range(0,len(self.samples)):

363

self.assertEqual(rows[i],self.samples[i])

452

self.assertEqual(rows[i],self.samples[i],

453

'cursor.fetchall retrieved incorrect rows'

454

)

364

455

365

456

cur.execute(self.ddl2)

366

457

cur.execute('select name from barflys')

367

458

rows = cur.fetchall()

368

self.assertEqual(len(rows),0)

459

self.assertEqual(len(rows),0,

460

'cursor.fetchall should return an empty list if '

461

'a select query returns no rows'

462

)

369

463

370

464

finally:

371

465

con.close()

382

476

rows23 = cur.fetchmany(2)

383

477

rows4 = cur.fetchone()

384

478

rows56 = cur.fetchall()

385

self.assertEqual(len(rows23),2)

386

self.assertEqual(len(rows56),2)

479

self.assertEqual(len(rows23),2,

480

'fetchmany returned incorrect number of rows'

481

)

482

self.assertEqual(len(rows56),2,

483

'fetchall returned incorrect number of rows'

484

)

387

485

388

486

rows = [rows1[0]]

389

487

rows.extend([rows23[0][0],rows23[1][0]])

391

489

rows.extend([rows56[0][0],rows56[1][0]])

392

490

rows.sort()

393

491

for i in range(0,len(self.samples)):

394

self.assertEqual(rows[i],self.samples[i])

492

self.assertEqual(rows[i],self.samples[i],

493

'incorrect data retrieved or inserted'

494

)

395

495

finally:

396

496

con.close()

397

497

403

503

con = self._connect()

404

504

try:

405

505

cur = con.cursor()

406

self.failUnless(hasattr(cur,'arraysize'))

506

self.failUnless(hasattr(cur,'arraysize'),

507

'cursor.arraysize must be defined'

508

)

407

509

finally:

408

510

con.close()

409

511

424

526

cur = con.cursor()

425

527

cur.execute(self.ddl1)

426

528

cur.setoutputsize(1000)

529

cur.setoutputsize(2000,0)

427

530

self._paraminsert(cur)

428

531

finally:

429

532

con.close()

434

537

435

538

def test_Date(self):

436

539

d1 = self.driver.Date(2002,12,25)

437

d2 = self.driver.DateFromTicks(1040823930)

540

d2 = self.driver.DateFromTicks(time.mktime((2002,12,25,0,0,0,0,0,0)))

438

541

# Can we assume this? API doesn't specify, but it seems implied

439

self.assertEqual(str(d1),str(d2))

542

# self.assertEqual(str(d1),str(d2))

440

543

441

544

def test_Time(self):

442

545

t1 = self.driver.Time(13,45,30)

443

t2 = self.driver.TimeFromTicks(1040823930)

546

t2 = self.driver.TimeFromTicks(time.mktime((0,0,0,13,45,30,0,0,0)))

444

547

# Can we assume this? API doesn't specify, but it seems implied

445

self.assertEqual(str(t1),str(t2))

548

# self.assertEqual(str(t1),str(t2))

446

549

447

550

def test_Timestamp(self):

448

551

t1 = self.driver.Timestamp(2002,12,25,13,45,30)

449

t2 = self.driver.TimestampFromTicks(1040823930)

552

t2 = self.driver.TimestampFromTicks(

553

time.mktime((2002,12,25,13,45,30,0,0,0))

554

)

450

555

# Can we assume this? API doesn't specify, but it seems implied

451

self.assertEqual(str(t1),str(t2))

556

# self.assertEqual(str(t1),str(t2))

452

557

453

558

def test_Binary(self):

454

559

b = self.driver.Binary('Something')

455

560

b = self.driver.Binary('')

456

561

457

562

def test_STRING(self):

458

self.failUnless(hasattr(self.driver,'STRING'))

563

self.failUnless(hasattr(self.driver,'STRING'),

564

'module.STRING must be defined'

565

)

459

566

460

567

def test_BINARY(self):

461

self.failUnless(hasattr(self.driver,'BINARY'))

568

self.failUnless(hasattr(self.driver,'BINARY'),

569

'module.BINARY must be defined.'

570

)

462

571

463

572

def test_NUMBER(self):

464

self.failUnless(hasattr(self.driver,'NUMBER'))

573

self.failUnless(hasattr(self.driver,'NUMBER'),

574

'module.NUMBER must be defined.'

575

)

465

576

466

577

def test_DATETIME(self):

467

self.failUnless(hasattr(self.driver,'DATETIME'))

578

self.failUnless(hasattr(self.driver,'DATETIME'),

579

'module.DATETIME must be defined.'

580

)

468

581

469

582

def test_ROWID(self):

470

self.failUnless(hasattr(self.driver,'ROWID'))

583

self.failUnless(hasattr(self.driver,'ROWID'),

584

'module.ROWID must be defined.'

585

)

471

586