|
| 1 | +import pytest |
| 2 | +import sqlalchemy as sa |
| 3 | +from types import MethodType |
| 4 | +from sqlalchemy import Column, Integer, Unicode |
| 5 | +from sqlalchemy.orm import declarative_base, sessionmaker |
| 6 | +from sqlalchemy.testing.fixtures import TablesTest, config |
| 7 | + |
| 8 | +class TestDirectories(TablesTest): |
| 9 | + __backend__ = True |
| 10 | + |
| 11 | + def prepare_table(self, engine): |
| 12 | + base = declarative_base() |
| 13 | + |
| 14 | + class Table(base): |
| 15 | + __tablename__ = "dir/test" |
| 16 | + id = Column(Integer, primary_key=True) |
| 17 | + text = Column(Unicode) |
| 18 | + |
| 19 | + base.metadata.create_all(engine) |
| 20 | + session = sessionmaker(bind=engine)() |
| 21 | + session.add(Table(id=2, text="foo")) |
| 22 | + session.commit() |
| 23 | + return base, Table, session |
| 24 | + |
| 25 | + def try_update(self, session, Table): |
| 26 | + row = session.query(Table).first() |
| 27 | + row.text = "bar" |
| 28 | + session.commit() |
| 29 | + return row |
| 30 | + |
| 31 | + def drop_table(self, base, engine): |
| 32 | + base.metadata.drop_all(engine) |
| 33 | + |
| 34 | + def bind_old_method_to_dialect(self, dialect): |
| 35 | + def _fix_variable_name(self, variable): |
| 36 | + return variable |
| 37 | + dialect._fix_variable_name = MethodType(_fix_variable_name, dialect) |
| 38 | + |
| 39 | + def test_directories(self): |
| 40 | + engine_good = sa.create_engine(config.db_url) |
| 41 | + base, Table, session = self.prepare_table(engine_good) |
| 42 | + row = self.try_update(session, Table) |
| 43 | + assert row.id == 2 |
| 44 | + assert row.text == "bar" |
| 45 | + self.drop_table(base, engine_good) |
| 46 | + |
| 47 | + engine_bad = sa.create_engine(config.db_url) |
| 48 | + self.bind_old_method_to_dialect(engine_bad.dialect) |
| 49 | + base, Table, session = self.prepare_table(engine_bad) |
| 50 | + with pytest.raises(Exception) as excinfo: |
| 51 | + self.try_update(session, Table) |
| 52 | + assert "Unknown name: $dir" in str(excinfo.value) |
| 53 | + self.drop_table(base, engine_bad) |
0 commit comments