|
22 | 22 | from cryptography.hazmat.primitives.asymmetric import rsa
|
23 | 23 | from gabbi.driver import test_pytest
|
24 | 24 | from gabbi.handlers import base
|
| 25 | +from sqlalchemy import text |
25 | 26 |
|
26 | 27 |
|
27 | 28 | XSNIPPET_API_HOST = "127.0.0.1"
|
@@ -83,26 +84,27 @@ def setup_db(self):
|
83 | 84 |
|
84 | 85 | with self.management_db.connect() as conn:
|
85 | 86 | conn.execute(
|
86 |
| - "CREATE DATABASE {database} OWNER {username};".format( |
87 |
| - **self.test_db_url.translate_connect_args())) |
| 87 | + text("CREATE DATABASE {database} OWNER {username};".format( |
| 88 | + **self.test_db_url.translate_connect_args()))) |
88 | 89 |
|
89 | 90 | # apply schema migrations to the temporary database. Both
|
90 | 91 | # Alembic and XSnippet API expect the connection string to be
|
91 | 92 | # passed via the ROCKET_DATABASE_URL environment variable, so
|
92 | 93 | # we update the variable to point to the temporary database for
|
93 | 94 | # the duration of the test
|
94 |
| - os.environ["ROCKET_DATABASE_URL"] = str(self.test_db_url) |
| 95 | + os.environ["ROCKET_DATABASE_URL"] = self.test_db_url.render_as_string(hide_password=False) |
95 | 96 | alembic.config.main(["upgrade", "head"])
|
96 | 97 |
|
97 | 98 | def teardown_db(self):
|
98 | 99 | """Clean up the temporary database."""
|
99 | 100 |
|
100 | 101 | with self.management_db.connect() as conn:
|
101 |
| - conn.execute("DROP DATABASE IF EXISTS {};".format(self.test_db_url.database)) |
| 102 | + conn.execute( |
| 103 | + text("DROP DATABASE IF EXISTS {};".format(self.test_db_url.database))) |
102 | 104 |
|
103 | 105 | # restore the original value of ROCKET_DATABASE_URL, so that
|
104 | 106 | # it can be used by the fixture to create new databases again
|
105 |
| - os.environ["ROCKET_DATABASE_URL"] = str(self.management_db.url) |
| 107 | + os.environ["ROCKET_DATABASE_URL"] = self.management_db.url.render_as_string(hide_password=False) |
106 | 108 |
|
107 | 109 |
|
108 | 110 | def start_server(self):
|
|
0 commit comments