databases/tests/test_databases.py at master · pythonran/databases · GitHub

History

1244 lines (1014 loc) · 43 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

370

371

372

373

374

375

376

377

378

379

380

381

382

383

384

385

386

387

388

389

390

391

392

393

394

395

396

397

398

399

400

401

402

403

404

405

406

407

408

409

410

411

412

413

414

415

416

417

418

419

420

421

422

423

424

425

426

427

428

429

430

431

432

433

434

435

436

437

438

439

440

441

442

443

444

445

446

447

448

449

450

451

452

453

454

455

456

457

458

459

460

461

462

463

464

465

466

467

468

469

470

471

472

473

474

475

476

477

478

479

480

481

482

483

484

485

486

487

488

489

490

491

492

493

494

495

496

497

498

499

500

501

502

503

504

505

506

507

508

509

510

511

512

513

514

515

516

517

518

519

520

521

522

523

524

525

526

527

528

529

530

531

532

533

534

535

536

537

538

539

540

541

542

543

544

545

546

547

548

549

550

551

552

553

554

555

556

557

558

559

560

561

562

563

564

565

566

567

568

569

570

571

572

573

574

575

576

577

578

579

580

581

582

583

584

585

586

587

588

589

590

591

592

593

594

595

596

597

598

599

600

601

602

603

604

605

606

607

608

609

610

611

612

613

614

615

616

617

618

619

620

621

622

623

624

625

626

627

628

629

630

631

632

633

634

635

636

637

638

639

640

641

642

643

644

645

646

647

648

649

650

651

652

653

654

655

656

657

658

659

660

661

662

663

664

665

666

667

668

669

670

671

672

673

674

675

676

677

678

679

680

681

682

683

684

685

686

687

688

689

690

691

692

693

694

695

696

697

698

699

700

701

702

703

704

705

706

707

708

709

710

711

712

713

714

715

716

717

718

719

720

721

722

723

724

725

726

727

728

729

730

731

732

733

734

735

736

737

738

739

740

741

742

743

744

745

746

747

748

749

750

751

752

753

754

755

756

757

758

759

760

761

762

763

764

765

766

767

768

769

770

771

772

773

774

775

776

777

778

779

780

781

782

783

784

785

786

787

788

789

790

791

792

793

794

795

796

797

798

799

800

801

802

803

804

805

806

807

808

809

810

811

812

813

814

815

816

817

818

819

820

821

822

823

824

825

826

827

828

829

830

831

832

833

834

835

836

837

838

839

840

841

842

843

844

845

846

847

848

849

850

851

852

853

854

855

856

857

858

859

860

861

862

863

864

865

866

867

868

869

870

871

872

873

874

875

876

877

878

879

880

881

882

883

884

885

886

887

888

889

890

891

892

893

894

895

896

897

898

899

900

901

902

903

904

905

906

907

908

909

910

911

912

913

914

915

916

917

918

919

920

921

922

923

924

925

926

927

928

929

930

931

932

933

934

935

936

937

938

939

940

941

942

943

944

945

946

947

948

949

950

951

952

953

954

955

956

957

958

959

960

961

962

963

964

965

966

967

968

969

970

971

972

973

974

975

976

977

978

979

980

981

982

983

984

985

986

987

988

989

990

991

992

993

994

995

996

997

998

999

1000

import asyncio

import datetime

import decimal

import functools

import os

import re

import sys

from unittest.mock import MagicMock, patch

import pytest

import sqlalchemy

from databases import Database, DatabaseURL

assert "TEST_DATABASE_URLS" in os.environ, "TEST_DATABASE_URLS is not set."

DATABASE_URLS = [url.strip() for url in os.environ["TEST_DATABASE_URLS"].split(",")]

def mysql_versions(wrapped_func):

"""

Decorator used to handle multiple versions of Python for mysql drivers

"""

@functools.wraps(wrapped_func)

def check(*args, **kwargs): # pragma: no cover

url = DatabaseURL(kwargs["database_url"])

if url.scheme in ["mysql", "mysql+aiomysql"] and sys.version_info >= (3, 10):

pytest.skip("aiomysql supports python 3.9 and lower")

if url.scheme == "mysql+asyncmy" and sys.version_info < (3, 7):

pytest.skip("asyncmy supports python 3.7 and higher")

return wrapped_func(*args, **kwargs)

return check

class AsyncMock(MagicMock):

async def __call__(self, *args, **kwargs):

return super(AsyncMock, self).__call__(*args, **kwargs)

class MyEpochType(sqlalchemy.types.TypeDecorator):

impl = sqlalchemy.Integer

epoch = datetime.date(1970, 1, 1)

def process_bind_param(self, value, dialect):

return (value - self.epoch).days

def process_result_value(self, value, dialect):

return self.epoch + datetime.timedelta(days=value)

metadata = sqlalchemy.MetaData()

notes = sqlalchemy.Table(

"notes",

metadata,

sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),

sqlalchemy.Column("text", sqlalchemy.String(length=100)),

sqlalchemy.Column("completed", sqlalchemy.Boolean),

)

# Used to test DateTime

articles = sqlalchemy.Table(

"articles",

metadata,

sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),

sqlalchemy.Column("title", sqlalchemy.String(length=100)),

sqlalchemy.Column("published", sqlalchemy.DateTime),

)

# Used to test JSON

session = sqlalchemy.Table(

"session",

metadata,

sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),

sqlalchemy.Column("data", sqlalchemy.JSON),

)

# Used to test custom column types

custom_date = sqlalchemy.Table(

"custom_date",

metadata,

sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),

sqlalchemy.Column("title", sqlalchemy.String(length=100)),

sqlalchemy.Column("published", MyEpochType),

)

# Used to test Numeric

prices = sqlalchemy.Table(

"prices",

metadata,

sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),

sqlalchemy.Column("price", sqlalchemy.Numeric(precision=30, scale=20)),

)

@pytest.fixture(autouse=True, scope="function")

def create_test_database():

# Create test databases with tables creation

for url in DATABASE_URLS:

database_url = DatabaseURL(url)

if database_url.scheme in ["mysql", "mysql+aiomysql", "mysql+asyncmy"]:

url = str(database_url.replace(driver="pymysql"))

elif database_url.scheme in [

"postgresql+aiopg",

"sqlite+aiosqlite",

"postgresql+asyncpg",

url = str(database_url.replace(driver=None))

engine = sqlalchemy.create_engine(url)

metadata.create_all(engine)

# Run the test suite

yield

# Drop test databases

for url in DATABASE_URLS:

database_url = DatabaseURL(url)

if database_url.scheme in ["mysql", "mysql+aiomysql", "mysql+asyncmy"]:

url = str(database_url.replace(driver="pymysql"))

elif database_url.scheme in [

"postgresql+aiopg",

"sqlite+aiosqlite",

"postgresql+asyncpg",

url = str(database_url.replace(driver=None))

engine = sqlalchemy.create_engine(url)

metadata.drop_all(engine)

def async_adapter(wrapped_func):

"""

Decorator used to run async test cases.

"""

@functools.wraps(wrapped_func)

def run_sync(*args, **kwargs):

loop = asyncio.new_event_loop()

task = wrapped_func(*args, **kwargs)

return loop.run_until_complete(task)

return run_sync

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_queries(database_url):

"""

Test that the basic `execute()`, `execute_many()`, `fetch_all()``, and

`fetch_one()` interfaces are all supported (using SQLAlchemy core).

"""

async with Database(database_url) as database:

async with database.transaction(force_rollback=True):

# execute()

query = notes.insert()

values = {"text": "example1", "completed": True}

await database.execute(query, values)

# execute_many()

query = notes.insert()

values = [

{"text": "example2", "completed": False},

{"text": "example3", "completed": True},

]

await database.execute_many(query, values)

# fetch_all()

query = notes.select()

results = await database.fetch_all(query=query)

assert len(results) == 3

assert results[0]["text"] == "example1"

assert results[0]["completed"] == True

assert results[1]["text"] == "example2"

assert results[1]["completed"] == False

assert results[2]["text"] == "example3"

assert results[2]["completed"] == True

# fetch_one()

query = notes.select()

result = await database.fetch_one(query=query)

assert result["text"] == "example1"

assert result["completed"] == True

# fetch_val()

query = sqlalchemy.sql.select([notes.c.text])

result = await database.fetch_val(query=query)

assert result == "example1"

# fetch_val() with no rows

query = sqlalchemy.sql.select([notes.c.text]).where(

notes.c.text == "impossible"

)

result = await database.fetch_val(query=query)

assert result is None

# fetch_val() with a different column

query = sqlalchemy.sql.select([notes.c.id, notes.c.text])

result = await database.fetch_val(query=query, column=1)

assert result == "example1"

# row access (needed to maintain test coverage for Record.__getitem__ in postgres backend)

query = sqlalchemy.sql.select([notes.c.text])

result = await database.fetch_one(query=query)

assert result["text"] == "example1"

assert result[0] == "example1"

# iterate()

query = notes.select()

iterate_results = []

async for result in database.iterate(query=query):

iterate_results.append(result)

assert len(iterate_results) == 3

assert iterate_results[0]["text"] == "example1"

assert iterate_results[0]["completed"] == True

assert iterate_results[1]["text"] == "example2"

assert iterate_results[1]["completed"] == False

assert iterate_results[2]["text"] == "example3"

assert iterate_results[2]["completed"] == True

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_queries_raw(database_url):

"""

Test that the basic `execute()`, `execute_many()`, `fetch_all()``, and

`fetch_one()` interfaces are all supported (raw queries).

"""

async with Database(database_url) as database:

async with database.transaction(force_rollback=True):

# execute()

query = "INSERT INTO notes(text, completed) VALUES (:text, :completed)"

values = {"text": "example1", "completed": True}

await database.execute(query, values)

# execute_many()

query = "INSERT INTO notes(text, completed) VALUES (:text, :completed)"

values = [

{"text": "example2", "completed": False},

{"text": "example3", "completed": True},

]

await database.execute_many(query, values)

# fetch_all()

query = "SELECT * FROM notes WHERE completed = :completed"

results = await database.fetch_all(query=query, values={"completed": True})

assert len(results) == 2

assert results[0]["text"] == "example1"

assert results[0]["completed"] == True

assert results[1]["text"] == "example3"

assert results[1]["completed"] == True

# fetch_one()

query = "SELECT * FROM notes WHERE completed = :completed"

result = await database.fetch_one(query=query, values={"completed": False})

assert result["text"] == "example2"

assert result["completed"] == False

# fetch_val()

query = "SELECT completed FROM notes WHERE text = :text"

result = await database.fetch_val(query=query, values={"text": "example1"})

assert result == True

query = "SELECT * FROM notes WHERE text = :text"

result = await database.fetch_val(

query=query, values={"text": "example1"}, column="completed"

)

assert result == True

# iterate()

query = "SELECT * FROM notes"

iterate_results = []

async for result in database.iterate(query=query):

iterate_results.append(result)

assert len(iterate_results) == 3

assert iterate_results[0]["text"] == "example1"

assert iterate_results[0]["completed"] == True

assert iterate_results[1]["text"] == "example2"

assert iterate_results[1]["completed"] == False

assert iterate_results[2]["text"] == "example3"

assert iterate_results[2]["completed"] == True

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_ddl_queries(database_url):

"""

Test that the built-in DDL elements such as `DropTable()`,

`CreateTable()` are supported (using SQLAlchemy core).

"""

async with Database(database_url) as database:

async with database.transaction(force_rollback=True):

# DropTable()

query = sqlalchemy.schema.DropTable(notes)

await database.execute(query)

# CreateTable()

query = sqlalchemy.schema.CreateTable(notes)

await database.execute(query)

@pytest.mark.parametrize("exception", [Exception, asyncio.CancelledError])

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_queries_after_error(database_url, exception):

"""

Test that the basic `execute()` works after a previous error.

"""

async with Database(database_url) as database:

with patch.object(

database.connection()._connection,

"acquire",

new=AsyncMock(side_effect=exception),

with pytest.raises(exception):

query = notes.select()

await database.fetch_all(query)

query = notes.select()

await database.fetch_all(query)

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_results_support_mapping_interface(database_url):

"""

Casting results to a dict should work, since the interface defines them

as supporting the mapping interface.

"""

async with Database(database_url) as database:

async with database.transaction(force_rollback=True):

# execute()

query = notes.insert()

values = {"text": "example1", "completed": True}

await database.execute(query, values)

# fetch_all()

query = notes.select()

results = await database.fetch_all(query=query)

results_as_dicts = [dict(item) for item in results]

assert len(results[0]) == 3

assert len(results_as_dicts[0]) == 3

assert isinstance(results_as_dicts[0]["id"], int)

assert results_as_dicts[0]["text"] == "example1"

assert results_as_dicts[0]["completed"] == True

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_results_support_column_reference(database_url):

"""

Casting results to a dict should work, since the interface defines them

as supporting the mapping interface.

"""

async with Database(database_url) as database:

async with database.transaction(force_rollback=True):

now = datetime.datetime.now().replace(microsecond=0)

today = datetime.date.today()

# execute()

query = articles.insert()

values = {"title": "Hello, world Article", "published": now}

await database.execute(query, values)

query = custom_date.insert()

values = {"title": "Hello, world Custom", "published": today}

await database.execute(query, values)

# fetch_all()

query = sqlalchemy.select([articles, custom_date])

results = await database.fetch_all(query=query)

assert len(results) == 1

assert results[0][articles.c.title] == "Hello, world Article"

assert results[0][articles.c.published] == now

assert results[0][custom_date.c.title] == "Hello, world Custom"

assert results[0][custom_date.c.published] == today

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_result_values_allow_duplicate_names(database_url):

"""

The values of a result should respect when two columns are selected

with the same name.

"""

async with Database(database_url) as database:

async with database.transaction(force_rollback=True):

query = "SELECT 1 AS id, 2 AS id"

row = await database.fetch_one(query=query)

assert list(row._mapping.keys()) == ["id", "id"]

assert list(row._mapping.values()) == [1, 2]

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_fetch_one_returning_no_results(database_url):

"""

fetch_one should return `None` when no results match.

"""

async with Database(database_url) as database:

async with database.transaction(force_rollback=True):

# fetch_all()

query = notes.select()

result = await database.fetch_one(query=query)

assert result is None

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_execute_return_val(database_url):

"""

Test using return value from `execute()` to get an inserted primary key.

"""

async with Database(database_url) as database:

async with database.transaction(force_rollback=True):

query = notes.insert()

values = {"text": "example1", "completed": True}

pk = await database.execute(query, values)

assert isinstance(pk, int)

# Apparently for `aiopg` it's OID that will always 0 in this case

# As it's only one action within this cursor life cycle

# It's recommended to use the `RETURNING` clause

# For obtaining the record id

if database.url.scheme == "postgresql+aiopg":

assert pk == 0

else:

query = notes.select().where(notes.c.id == pk)

result = await database.fetch_one(query)

assert result["text"] == "example1"

assert result["completed"] == True

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_rollback_isolation(database_url):

"""

Ensure that `database.transaction(force_rollback=True)` provides strict isolation.

"""

async with Database(database_url) as database:

# Perform some INSERT operations on the database.

async with database.transaction(force_rollback=True):

query = notes.insert().values(text="example1", completed=True)

await database.execute(query)

# Ensure INSERT operations have been rolled back.

query = notes.select()

results = await database.fetch_all(query=query)

assert len(results) == 0

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_rollback_isolation_with_contextmanager(database_url):

"""

Ensure that `database.force_rollback()` provides strict isolation.

"""

database = Database(database_url)

with database.force_rollback():

async with database:

# Perform some INSERT operations on the database.

query = notes.insert().values(text="example1", completed=True)

await database.execute(query)

async with database:

# Ensure INSERT operations have been rolled back.

query = notes.select()

results = await database.fetch_all(query=query)

assert len(results) == 0

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_transaction_commit(database_url):

"""

Ensure that transaction commit is supported.

"""

async with Database(database_url) as database:

async with database.transaction(force_rollback=True):

async with database.transaction():

query = notes.insert().values(text="example1", completed=True)

await database.execute(query)

query = notes.select()

results = await database.fetch_all(query=query)

assert len(results) == 1

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_transaction_commit_serializable(database_url):

"""

Ensure that serializable transaction commit via extra parameters is supported.

"""

database_url = DatabaseURL(database_url)

if database_url.scheme not in ["postgresql", "postgresql+asyncpg"]:

pytest.skip("Test (currently) only supports asyncpg")

if database_url.scheme == "postgresql+asyncpg":

database_url = database_url.replace(driver=None)

def insert_independently():

engine = sqlalchemy.create_engine(str(database_url))

conn = engine.connect()

query = notes.insert().values(text="example1", completed=True)

conn.execute(query)

def delete_independently():

engine = sqlalchemy.create_engine(str(database_url))

conn = engine.connect()

query = notes.delete()

conn.execute(query)

async with Database(database_url) as database:

async with database.transaction(force_rollback=True, isolation="serializable"):

query = notes.select()

results = await database.fetch_all(query=query)

assert len(results) == 0

insert_independently()

query = notes.select()

results = await database.fetch_all(query=query)

assert len(results) == 0

delete_independently()

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_transaction_rollback(database_url):

"""

Ensure that transaction rollback is supported.

"""

async with Database(database_url) as database:

async with database.transaction(force_rollback=True):

try:

async with database.transaction():

query = notes.insert().values(text="example1", completed=True)

await database.execute(query)

raise RuntimeError()

except RuntimeError:

pass

query = notes.select()

results = await database.fetch_all(query=query)

assert len(results) == 0

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_transaction_commit_low_level(database_url):

"""

Ensure that an explicit `await transaction.commit()` is supported.

"""

async with Database(database_url) as database:

async with database.transaction(force_rollback=True):

transaction = await database.transaction()

try:

query = notes.insert().values(text="example1", completed=True)

await database.execute(query)

except: # pragma: no cover

await transaction.rollback()

else:

await transaction.commit()

query = notes.select()

results = await database.fetch_all(query=query)

assert len(results) == 1

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_transaction_rollback_low_level(database_url):

"""

Ensure that an explicit `await transaction.rollback()` is supported.

"""

async with Database(database_url) as database:

async with database.transaction(force_rollback=True):

transaction = await database.transaction()

try:

query = notes.insert().values(text="example1", completed=True)

await database.execute(query)

raise RuntimeError()

except:

await transaction.rollback()

else: # pragma: no cover

await transaction.commit()

query = notes.select()

results = await database.fetch_all(query=query)

assert len(results) == 0

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_transaction_decorator(database_url):

"""

Ensure that @database.transaction() is supported.

"""

database = Database(database_url, force_rollback=True)

@database.transaction()

async def insert_data(raise_exception):

query = notes.insert().values(text="example", completed=True)

await database.execute(query)

if raise_exception:

raise RuntimeError()

async with database:

with pytest.raises(RuntimeError):

await insert_data(raise_exception=True)

query = notes.select()

results = await database.fetch_all(query=query)

assert len(results) == 0

await insert_data(raise_exception=False)

query = notes.select()

results = await database.fetch_all(query=query)

assert len(results) == 1

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_datetime_field(database_url):

"""

Test DataTime columns, to ensure records are coerced to/from proper Python types.

"""

async with Database(database_url) as database:

async with database.transaction(force_rollback=True):

now = datetime.datetime.now().replace(microsecond=0)

# execute()

query = articles.insert()

values = {"title": "Hello, world", "published": now}

await database.execute(query, values)

# fetch_all()

query = articles.select()

results = await database.fetch_all(query=query)

assert len(results) == 1

assert results[0]["title"] == "Hello, world"

assert results[0]["published"] == now

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_decimal_field(database_url):

"""

Test Decimal (NUMERIC) columns, to ensure records are coerced to/from proper Python types.

"""

async with Database(database_url) as database:

async with database.transaction(force_rollback=True):

price = decimal.Decimal("0.700000000000001")

# execute()

query = prices.insert()

values = {"price": price}

await database.execute(query, values)

# fetch_all()

query = prices.select()

results = await database.fetch_all(query=query)

assert len(results) == 1

if database_url.startswith("sqlite"):

# aiosqlite does not support native decimals --> a roud-off error is expected

assert results[0]["price"] == pytest.approx(price)

else:

assert results[0]["price"] == price

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_json_field(database_url):

"""

Test JSON columns, to ensure correct cross-database support.

"""

async with Database(database_url) as database:

async with database.transaction(force_rollback=True):

# execute()

data = {"text": "hello", "boolean": True, "int": 1}

values = {"data": data}

query = session.insert()

await database.execute(query, values)

# fetch_all()

query = session.select()

results = await database.fetch_all(query=query)

assert len(results) == 1

assert results[0]["data"] == {"text": "hello", "boolean": True, "int": 1}

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_custom_field(database_url):

"""

Test custom column types.

"""

async with Database(database_url) as database:

async with database.transaction(force_rollback=True):

today = datetime.date.today()

# execute()

query = custom_date.insert()

values = {"title": "Hello, world", "published": today}

await database.execute(query, values)

# fetch_all()

query = custom_date.select()

results = await database.fetch_all(query=query)

assert len(results) == 1

assert results[0]["title"] == "Hello, world"

assert results[0]["published"] == today

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_connections_isolation(database_url):

"""

Ensure that changes are visible between different connections.

To check this we have to not create a transaction, so that

each query ends up on a different connection from the pool.

"""

async with Database(database_url) as database:

try:

query = notes.insert().values(text="example1", completed=True)

await database.execute(query)

query = notes.select()

results = await database.fetch_all(query=query)

assert len(results) == 1

finally:

query = notes.delete()

await database.execute(query)

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_commit_on_root_transaction(database_url):

"""

Because our tests are generally wrapped in rollback-islation, they

don't have coverage for commiting the root transaction.

Deal with this here, and delete the records rather than rolling back.

"""

async with Database(database_url) as database:

try:

async with database.transaction():

query = notes.insert().values(text="example1", completed=True)

await database.execute(query)

query = notes.select()

results = await database.fetch_all(query=query)

assert len(results) == 1

finally:

query = notes.delete()

await database.execute(query)

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_connect_and_disconnect(database_url):

"""

Test explicit connect() and disconnect().

"""

database = Database(database_url)

assert not database.is_connected

await database.connect()

assert database.is_connected

await database.disconnect()

assert not database.is_connected

# connect and disconnect idempotence

await database.connect()

assert database.is_connected

await database.disconnect()

assert not database.is_connected

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_connection_context(database_url):

"""

Test connection contexts are task-local.

"""

async with Database(database_url) as database:

async with database.connection() as connection_1:

async with database.connection() as connection_2:

assert connection_1 is connection_2

async with Database(database_url) as database:

connection_1 = None

connection_2 = None

test_complete = asyncio.Event()

async def get_connection_1():

nonlocal connection_1

async with database.connection() as connection:

connection_1 = connection

await test_complete.wait()

async def get_connection_2():

nonlocal connection_2

async with database.connection() as connection:

connection_2 = connection

await test_complete.wait()

loop = asyncio.get_event_loop()

task_1 = loop.create_task(get_connection_1())

task_2 = loop.create_task(get_connection_2())

while connection_1 is None or connection_2 is None:

await asyncio.sleep(0.000001)

assert connection_1 is not connection_2

test_complete.set()

await task_1

await task_2

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_connection_context_with_raw_connection(database_url):

"""

Test connection contexts with respect to the raw connection.

"""

async with Database(database_url) as database:

async with database.connection() as connection_1:

async with database.connection() as connection_2:

assert connection_1 is connection_2

assert connection_1.raw_connection is connection_2.raw_connection

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_queries_with_expose_backend_connection(database_url):

"""

Replication of `execute()`, `execute_many()`, `fetch_all()``, and

`fetch_one()` using the raw driver interface.

"""

async with Database(database_url) as database:

async with database.connection() as connection:

async with connection.transaction(force_rollback=True):

# Get the raw connection

raw_connection = connection.raw_connection

# Insert query

if database.url.scheme in [

"mysql",

"mysql+asyncmy",

"mysql+aiomysql",

"postgresql+aiopg",

insert_query = "INSERT INTO notes (text, completed) VALUES (%s, %s)"

else:

insert_query = "INSERT INTO notes (text, completed) VALUES ($1, $2)"

# execute()

values = ("example1", True)

if database.url.scheme in [

"mysql",

"mysql+aiomysql",

"postgresql+aiopg",

cursor = await raw_connection.cursor()

await cursor.execute(insert_query, values)

elif database.url.scheme == "mysql+asyncmy":

async with raw_connection.cursor() as cursor:

await cursor.execute(insert_query, values)

elif database.url.scheme in ["postgresql", "postgresql+asyncpg"]:

await raw_connection.execute(insert_query, *values)

elif database.url.scheme in ["sqlite", "sqlite+aiosqlite"]:

await raw_connection.execute(insert_query, values)

# execute_many()

values = [("example2", False), ("example3", True)]

if database.url.scheme in ["mysql", "mysql+aiomysql"]:

cursor = await raw_connection.cursor()

await cursor.executemany(insert_query, values)

elif database.url.scheme == "mysql+asyncmy":

async with raw_connection.cursor() as cursor:

await cursor.executemany(insert_query, values)

elif database.url.scheme == "postgresql+aiopg":

cursor = await raw_connection.cursor()

# No async support for `executemany`

for value in values:

await cursor.execute(insert_query, value)

else:

await raw_connection.executemany(insert_query, values)

# Select query

select_query = "SELECT notes.id, notes.text, notes.completed FROM notes"

# fetch_all()

if database.url.scheme in [

"mysql",

"mysql+aiomysql",

"postgresql+aiopg",

cursor = await raw_connection.cursor()

await cursor.execute(select_query)

results = await cursor.fetchall()

elif database.url.scheme == "mysql+asyncmy":

async with raw_connection.cursor() as cursor:

await cursor.execute(select_query)

results = await cursor.fetchall()

elif database.url.scheme in ["postgresql", "postgresql+asyncpg"]:

results = await raw_connection.fetch(select_query)

elif database.url.scheme in ["sqlite", "sqlite+aiosqlite"]:

results = await raw_connection.execute_fetchall(select_query)

assert len(results) == 3

# Raw output for the raw request

assert results[0][1] == "example1"

assert results[0][2] == True

assert results[1][1] == "example2"

assert results[1][2] == False

assert results[2][1] == "example3"

assert results[2][2] == True

# fetch_one()

if database.url.scheme in ["postgresql", "postgresql+asyncpg"]:

result = await raw_connection.fetchrow(select_query)

elif database.url.scheme == "mysql+asyncmy":

async with raw_connection.cursor() as cursor:

await cursor.execute(select_query)

result = await cursor.fetchone()

else:

cursor = await raw_connection.cursor()

await cursor.execute(select_query)

result = await cursor.fetchone()

# Raw output for the raw request

assert result[1] == "example1"

assert result[2] == True

@pytest.mark.parametrize("database_url", DATABASE_URLS)

@mysql_versions

@async_adapter

async def test_database_url_interface(database_url):

"""

Test that Database instances expose a `.url` attribute.

View remainder of file in raw view

databases/tests/test_databases.py at master · pythonran/databases · GitHub

Expand file tree

Breadcrumbs

test_databases.py

Latest commit

History

Breadcrumbs

test_databases.py

File metadata and controls