1- import typing as t
1+ import locale
22import os
3+ import subprocess
4+ import typing as t
35
46import psycopg
7+ import furl
58import pytest
69
710
11+ ConnectionFactory = t .Callable [[str ], psycopg .Connection ]
12+
13+
14+ @pytest .fixture (scope = "function" )
15+ def connection_uri () -> str :
16+ """Read and return connection URI from environment."""
17+
18+ connection_uri = os .getenv ("CONNECTION_URI" )
19+ if connection_uri is None :
20+ pytest .fail ("CONNECTION_URI: environment variable is not set" )
21+ return connection_uri
22+
23+
824@pytest .fixture (scope = "function" )
9- def connection_factory () -> t .Callable [[str ], psycopg .Connection ]:
25+ def connection_factory () -> ConnectionFactory :
26+ """Return 'psycopg.Connection' factory."""
27+
1028 def factory (connection_uri : str ) -> psycopg .Connection :
1129 return psycopg .connect (connection_uri )
1230 return factory
1331
1432
1533@pytest .fixture (scope = "function" )
16- def connection (connection_factory ) -> psycopg .Connection :
17- return connection_factory (os .getenv ("CONNECTION_URI" ))
34+ def connection (connection_uri : str , connection_factory : ConnectionFactory ) -> psycopg .Connection :
35+ """Return 'psycopg.Connection' for connection URI set in environment."""
36+
37+ return connection_factory (connection_uri )
1838
1939
2040def test_connection_uri ():
@@ -34,12 +54,15 @@ def test_server_encoding(connection: psycopg.Connection):
3454def test_locale (connection : psycopg .Connection ):
3555 """Test that PostgreSQL's locale is 'en_US.UTF-8'."""
3656
37- assert connection .execute ("SHOW LC_COLLATE" ).fetchone ()[0 ] == "en_US.UTF-8"
38- assert connection .execute ("SHOW LC_CTYPE" ).fetchone ()[0 ] == "en_US.UTF-8"
57+ lc_collate = connection .execute ("SHOW LC_COLLATE" ).fetchone ()[0 ]
58+ lc_ctype = connection .execute ("SHOW LC_CTYPE" ).fetchone ()[0 ]
59+
60+ assert locale .normalize (lc_collate ) == "en_US.UTF-8"
61+ assert locale .normalize (lc_ctype ) == "en_US.UTF-8"
3962
4063
4164def test_user_permissions (connection : psycopg .Connection ):
42- """Test that a user can create databases but is not a superuser ."""
65+ """Test that a user has super/createdb permissions ."""
4366
4467 with connection :
4568 record = connection \
@@ -49,7 +72,7 @@ def test_user_permissions(connection: psycopg.Connection):
4972
5073 usecreatedb , usesuper = record
5174 assert usecreatedb
52- assert not usesuper
75+ assert usesuper
5376
5477
5578def test_user_create_insert_select (connection : psycopg .Connection ):
@@ -82,12 +105,65 @@ def test_user_create_insert_non_ascii(connection: psycopg.Connection):
82105
83106
84107def test_user_create_drop_database (connection : psycopg .Connection ):
85- """Test that a user has no permissions to create databases."""
108+ """Test that a user has permissions to create databases."""
86109
87110 # CREATE/DROP DATABASE statements don't work within transactions, and with
88111 # autocommit disabled transactions are created by psycopg automatically.
89112 connection .autocommit = True
90113
91- database_name = "foobar42"
92- connection .execute (f"CREATE DATABASE { database_name } " )
93- connection .execute (f"DROP DATABASE { database_name } " )
114+ database = "databas3"
115+ connection .execute (f"CREATE DATABASE { database } " )
116+ connection .execute (f"DROP DATABASE { database } " )
117+
118+
119+ def test_user_create_drop_user (
120+ connection : psycopg .Connection ,
121+ connection_factory : ConnectionFactory ,
122+ connection_uri : str
123+ ):
124+ """Test that a user has permissions to create users."""
125+
126+ # CREATE/DROP USER statements don't work within transactions, and with
127+ # autocommit disabled transactions are created by psycopg automatically.
128+ connection .autocommit = True
129+
130+ username = "us3rname"
131+ password = "passw0rd"
132+ database = "databas3"
133+
134+ connection .execute (f"CREATE USER { username } WITH PASSWORD '{ password } '" )
135+ connection .execute (f"CREATE DATABASE { database } WITH OWNER '{ username } '" )
136+
137+ try :
138+ # Smoke test that created user can successfully log-in and execute
139+ # queries for its own database.
140+ connection_uri = furl .furl (
141+ connection_uri , username = username , password = password , path = database ).url
142+ test_user_create_insert_select (connection_factory (connection_uri ))
143+
144+ finally :
145+ connection .execute (f"DROP DATABASE { database } " )
146+ connection .execute (f"DROP USER { username } " )
147+
148+
149+ def test_client_applications (connection_uri , connection_factory ):
150+ """Test that PostgreSQL client applications can be used."""
151+
152+ username = "us3rname"
153+ password = "passw0rd"
154+ database = "databas3"
155+
156+ subprocess .check_call (["createuser" , username ])
157+ subprocess .check_call (["createdb" , "--owner" , username , database ])
158+ subprocess .check_call (["psql" , "-c" , f"ALTER USER { username } WITH PASSWORD '{ password } '" ])
159+
160+ try :
161+ # Smoke test that created user can successfully log-in and execute
162+ # queries for its own database.
163+ connection_uri = furl .furl (
164+ connection_uri , username = username , password = password , path = database ).url
165+ test_user_create_insert_select (connection_factory (connection_uri ))
166+
167+ finally :
168+ subprocess .check_call (["dropdb" , database ])
169+ subprocess .check_call (["dropuser" , username ])
0 commit comments