mirror of
https://github.com/neondatabase/neon.git
synced 2025-12-28 00:23:00 +00:00
257ade0688
This patch aims to:
* Unify connection & querying logic of ZenithPagerserver and Postgres.
* Mitigate changes to transaction machinery introduced in `psycopg2 >= 2.9`.
Now it's possible to acquire db connection using the corresponding
method:
```python
pg = postgres.create_start('main')
conn = pg.connect()
...
conn.close()
```
This pattern can be further improved with the help of `closing`:
```python
from contextlib import closing
pg = postgres.create_start('main')
with closing(pg.connect()) as conn:
...
```
All connections produced by this method will have autocommit
enabled by default.
49 lines
1.7 KiB
Python
49 lines
1.7 KiB
Python
import json
|
|
|
|
pytest_plugins = ("fixtures.zenith_fixtures")
|
|
|
|
|
|
def helper_compare_branch_list(page_server_cur, zenith_cli):
|
|
"""
|
|
Compare branches list returned by CLI and directly via API.
|
|
Filters out branches created by other tests.
|
|
"""
|
|
|
|
page_server_cur.execute('branch_list')
|
|
branches_api = sorted(map(lambda b: b['name'], json.loads(page_server_cur.fetchone()[0])))
|
|
branches_api = [b for b in branches_api if b.startswith('test_cli_') or b in ('empty', 'main')]
|
|
|
|
res = zenith_cli.run(["branch"])
|
|
assert res.stderr == ''
|
|
branches_cli = sorted(map(lambda b: b.split(':')[-1].strip(), res.stdout.strip().split("\n")))
|
|
branches_cli = [b for b in branches_cli if b.startswith('test_cli_') or b in ('empty', 'main')]
|
|
|
|
assert branches_api == branches_cli
|
|
|
|
|
|
def test_cli_branch_list(pageserver, zenith_cli):
|
|
|
|
page_server_conn = pageserver.connect()
|
|
page_server_cur = page_server_conn.cursor()
|
|
|
|
# Initial sanity check
|
|
helper_compare_branch_list(page_server_cur, zenith_cli)
|
|
|
|
# Create a branch for us
|
|
res = zenith_cli.run(["branch", "test_cli_branch_list_main", "main"])
|
|
assert res.stderr == ''
|
|
helper_compare_branch_list(page_server_cur, zenith_cli)
|
|
|
|
# Create a nested branch
|
|
res = zenith_cli.run(["branch", "test_cli_branch_list_nested", "test_cli_branch_list_main"])
|
|
assert res.stderr == ''
|
|
helper_compare_branch_list(page_server_cur, zenith_cli)
|
|
|
|
# Check that all new branches are visible via CLI
|
|
res = zenith_cli.run(["branch"])
|
|
assert res.stderr == ''
|
|
branches_cli = sorted(map(lambda b: b.split(':')[-1].strip(), res.stdout.strip().split("\n")))
|
|
|
|
assert 'test_cli_branch_list_main' in branches_cli
|
|
assert 'test_cli_branch_list_nested' in branches_cli
|