test_runner: fix mypy errors and force it on CI (#774)

* Fix bugs found by mypy
* Add some missing types and runtime checks, remove unused code
* Make ZenithPageserver start right away for better type safety
* Add `types-*` packages to Pipfile
* Pin mypy version and run it on CircleCI
This commit is contained in:
Egor Suvorov
2021-10-21 13:51:54 +03:00
committed by GitHub
parent 7f9d2a7d05
commit ff563ff080
11 changed files with 112 additions and 79 deletions

View File

@@ -1,19 +1,24 @@
import json
import uuid
from psycopg2.extensions import cursor as PgCursor
from fixtures.zenith_fixtures import ZenithCli, ZenithPageserver
from typing import cast
pytest_plugins = ("fixtures.zenith_fixtures")
def helper_compare_branch_list(page_server_cur, zenith_cli, initial_tenant: str):
def helper_compare_branch_list(page_server_cur: PgCursor,
zenith_cli: ZenithCli,
initial_tenant: str):
"""
Compare branches list returned by CLI and directly via API.
Filters out branches created by other tests.
"""
page_server_cur.execute(f'branch_list {initial_tenant}')
branches_api = sorted(map(lambda b: b['name'], json.loads(page_server_cur.fetchone()[0])))
branches_api = sorted(
map(lambda b: cast(str, 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"])
@@ -32,7 +37,7 @@ def helper_compare_branch_list(page_server_cur, zenith_cli, initial_tenant: str)
assert branches_api == branches_cli == branches_cli_with_tenant_arg
def test_cli_branch_list(pageserver: ZenithPageserver, zenith_cli):
def test_cli_branch_list(pageserver: ZenithPageserver, zenith_cli: ZenithCli):
page_server_conn = pageserver.connect()
page_server_cur = page_server_conn.cursor()
@@ -58,9 +63,10 @@ def test_cli_branch_list(pageserver: ZenithPageserver, zenith_cli):
assert 'test_cli_branch_list_nested' in branches_cli
def helper_compare_tenant_list(page_server_cur, zenith_cli: ZenithCli):
def helper_compare_tenant_list(page_server_cur: PgCursor, zenith_cli: ZenithCli):
page_server_cur.execute(f'tenant_list')
tenants_api = sorted(map(lambda t: t['id'], json.loads(page_server_cur.fetchone()[0])))
tenants_api = sorted(
map(lambda t: cast(str, t['id']), json.loads(page_server_cur.fetchone()[0])))
res = zenith_cli.run(["tenant", "list"])
assert res.stderr == ''