Skip to content

Instantly share code, notes, and snippets.

@WGH-
Created June 6, 2018 17:11
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save WGH-/cb61522716f2555901855121fd35c458 to your computer and use it in GitHub Desktop.
Save WGH-/cb61522716f2555901855121fd35c458 to your computer and use it in GitHub Desktop.
asynpgsa DDL problems bisection scripts
#!/usr/bin/env python3
import sys
import asyncio
import traceback
import sqlalchemy as sa
from sqlalchemy.schema import CreateTable
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.dialects.postgresql import TEXT, ARRAY
import asyncpgsa
Base = declarative_base()
class Foo(Base):
__tablename__ = "foo"
id = sa.Column(TEXT, primary_key=True, nullable=False)
nodes = sa.Column(ARRAY(TEXT), nullable=True)
async def test_bug():
pool = await asyncpgsa.create_pool(database="test")
async with pool.transaction() as conn:
await conn.execute("DROP TABLE IF EXISTS foo")
for table in Base.metadata.sorted_tables:
await conn.execute(CreateTable(table))
async def comain():
try:
await test_bug()
except AttributeError as e:
if "'NoneType' object has no attribute 'items'" in str(e):
traceback.print_exc()
print("EXPECTED BAD REVISION!", file=sys.stderr)
sys.exit(1)
raise
def main():
loop = asyncio.get_event_loop()
try:
loop.run_until_complete(comain())
except Exception as e:
traceback.print_exc()
sys.exit(128)
loop.close()
if __name__ == "__main__":
main()
#!/bin/sh
ENVDIR=/tmp/bisect_venv
REPODIR=/tmp/asyncpgsa
if [[ ! -x "$ENVDIR/bin/python3" ]]; then
virtualenv -p python3 "$ENVDIR"
fi
"$ENVDIR/bin/pip3" install -e "$REPODIR"
"$ENVDIR/bin/python3" check_revision.py
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment