11import os
22import sqlite3
33import subprocess
4+ from typing import Mapping , Optional
45
56import pytest
67from django .conf import settings
1011# Construct names for the "inner" database used in runpytest tests
1112_settings = settings .DATABASES ["default" ]
1213
13- DB_NAME = _settings ["NAME" ]
14- TEST_DB_NAME = _settings ["TEST" ]["NAME" ]
14+ DB_NAME : str = _settings ["NAME" ]
15+ TEST_DB_NAME : str = _settings ["TEST" ]["NAME" ]
1516
1617if _settings ["ENGINE" ] == "django.db.backends.sqlite3" and TEST_DB_NAME is None :
1718 TEST_DB_NAME = ":memory:"
3132 SECOND_TEST_DB_NAME = TEST_DB_NAME + '_second' if DB_NAME is not None else None
3233
3334
34- def get_db_engine ():
35- return _settings ["ENGINE" ].split ("." )[- 1 ]
35+ def get_db_engine () -> str :
36+ db_engine : str = _settings ["ENGINE" ].split ("." )[- 1 ]
37+ return db_engine
3638
3739
3840class CmdResult :
39- def __init__ (self , status_code , std_out , std_err ) :
41+ def __init__ (self , status_code : int , std_out : bytes , std_err : bytes ) -> None :
4042 self .status_code = status_code
4143 self .std_out = std_out
4244 self .std_err = std_err
4345
4446
45- def run_cmd (* args , env = None ):
47+ def run_cmd (* args : str , env : Optional [ Mapping [ str , str ]] = None ) -> CmdResult :
4648 r = subprocess .Popen (
4749 args ,
4850 stdout = subprocess .PIPE ,
@@ -54,7 +56,7 @@ def run_cmd(*args, env=None):
5456 return CmdResult (ret , stdoutdata , stderrdata )
5557
5658
57- def run_psql (* args ) :
59+ def run_psql (* args : str ) -> CmdResult :
5860 env = {}
5961 user = _settings .get ("USER" )
6062 if user : # pragma: no branch
@@ -68,7 +70,7 @@ def run_psql(*args):
6870 return run_cmd ("psql" , * args , env = env )
6971
7072
71- def run_mysql (* args ) :
73+ def run_mysql (* args : str ) -> CmdResult :
7274 user = _settings .get ("USER" )
7375 if user : # pragma: no branch
7476 args = ("-u" , user , * args )
@@ -82,22 +84,22 @@ def run_mysql(*args):
8284 return run_cmd ("mysql" , * args )
8385
8486
85- def skip_if_sqlite_in_memory ():
87+ def skip_if_sqlite_in_memory () -> None :
8688 if (
8789 _settings ["ENGINE" ] == "django.db.backends.sqlite3"
8890 and _settings ["TEST" ]["NAME" ] is None
8991 ):
9092 pytest .skip ("Do not test db reuse since database does not support it" )
9193
9294
93- def _get_db_name (db_suffix = None ):
95+ def _get_db_name (db_suffix : Optional [ str ] = None ) -> str :
9496 name = TEST_DB_NAME
9597 if db_suffix :
9698 name = f"{ name } _{ db_suffix } "
9799 return name
98100
99101
100- def drop_database (db_suffix = None ):
102+ def drop_database (db_suffix : Optional [ str ] = None ) -> None :
101103 name = _get_db_name (db_suffix )
102104 db_engine = get_db_engine ()
103105
@@ -119,7 +121,7 @@ def drop_database(db_suffix=None):
119121 os .unlink (name )
120122
121123
122- def db_exists (db_suffix = None ):
124+ def db_exists (db_suffix : Optional [ str ] = None ) -> bool :
123125 name = _get_db_name (db_suffix )
124126 db_engine = get_db_engine ()
125127
@@ -137,7 +139,7 @@ def db_exists(db_suffix=None):
137139 return os .path .exists (name )
138140
139141
140- def mark_database ():
142+ def mark_database () -> None :
141143 db_engine = get_db_engine ()
142144
143145 if db_engine == "postgresql" :
@@ -162,7 +164,7 @@ def mark_database():
162164 conn .close ()
163165
164166
165- def mark_exists ():
167+ def mark_exists () -> bool :
166168 db_engine = get_db_engine ()
167169
168170 if db_engine == "postgresql" :
0 commit comments