5899312502

Committed 18 Aug 2023 05:56AM UTC coverage: 95.2% (+0.4%) from 94.836%

Build # 5899312502

Build Type

push

github

Committed by

web-flow

Commit Message

Drop legacy python (#39)

* Bumped version for breaking release.
* Drop support for Python 2.7, 3.5, 3.6.
* Add support for Python 3.10, 3.11.

Run Details

94 of 109 branches covered (86.24%)

Branch coverage included in aggregate %.

17 of 17 new or added lines in 7 files covered. (100.0%)

501 of 516 relevant lines covered (97.09%)

0.97 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.26

/src/gocept/testdb/tests/test_postgres.py

import gocept.testdb
import gocept.testdb.testing
import gocept.testing.assertion
import os
import sqlalchemy.exc
import time


class PostgreSQLTests(gocept.testdb.testing.TestCase,
                      gocept.testing.assertion.String,
                      gocept.testing.assertion.Exceptions,
                      gocept.testing.assertion.Ellipsis):
    """Testing ..postgres.PostgreSQL."""

    db_template = 'gocept.testdb.tests-template-%s' % os.getpid()

    def tearDown(self):
        try:
            self.drop_all()
        finally:
            super().tearDown()

    def drop_all(self):
        self.makeOne(db_template=self.db_template, create_db=False).drop_all(
            drop_template=True)

    def makeOne(self, create_db=True, db_template=None, **kw):
        import gocept.testdb
        db = gocept.testdb.PostgreSQL(
            db_template=db_template, **kw)
        if create_db:
            db.create()
        return db

    def test_template_db_doesnt_need_schema(self):
        # regression test
        db = self.makeOne(create_db=False)
        with self.assertNothingRaised():
            db.create()

    def test_naming_scheme_matches_with_dash_in_prefix(self):
        # regression test
        db = self.makeOne(prefix='foo-bar', create_db=False)
        self.assertTrue(db._matches_db_naming_scheme(db.db_name))
        self.assertTrue(db._matches_db_naming_scheme('foo-bar-123abc456efg'))

    def test_can_add_database_with_special_lc_collate(self):
        try:
            db = self.makeOne(lc_collate='de_DE.UTF-8')
        except SystemExit:
            # When using Docker container as suggested in README this call
            # fails with 'invalid locale name: "de_DE.UTF-8"', so let's skip
            # the test in that environment:
            self.skipTest('"de_DE.UTF-8" not supported by database.')
        collate_by_db_name = {x[0]: x[3] for x in db.pg_list_db_items()}
        self.assertEqual('de_DE.UTF-8', collate_by_db_name[db.db_name])

    def test_takes_configuration_from_environment(self):
        db = self.makeOne(create_db=False)
        self.assertStartsWith('postgresql://', db.dsn)
        hostname = 'localhost'
        if db.db_port:  # pragma: no cover
            hostname = f'{hostname}:{db.db_port}'
        self.assertIn(f'{hostname}/testdb-PID', db.dsn)

    def test_created_database_contains_marker_table(self):
        db = self.makeOne()
        # The database is marked as a testing database by creating a table
        # called 'tmp_functest' in it:
        with self.assertNothingRaised():
            self.execute(db.dsn, 'SELECT * from tmp_functest')

    def test_conveniences_properties_are_set(self):
        db = self.makeOne()
        self.assertTrue(db.exists)
        self.assertTrue(db.is_testing)

    def test_schema_gets_loaded(self):
        db = self.makeOne(schema_path=self.schema)
        with self.assertNothingRaised():
            self.execute(db.dsn, 'SELECT * from foo')

    def test_name_of_database_can_be_specified(self):
        db = self.makeOne(db_name='mytestdb', create_db=False)
        self.assertEndsWith('/mytestdb', db.dsn)

    def test_drop_drops_database(self):
        db = self.makeOne()
        db.drop()
        with self.assertRaises(sqlalchemy.exc.OperationalError) as err:
            self.connect(db)
        self.assertEllipsis(
            '... database ... does not exist...', str(err.exception))

    def test_encoding_is_used(self):
        # An optional encoding parameter can be specified in the constructor.
        # It is used when creating the database.
        db = self.makeOne(schema_path=self.schema, encoding='UTF8')
        encoding = self.execute(
            db.dsn,
            """SELECT pg_catalog.pg_encoding_to_char(encoding) as encoding
               FROM pg_catalog.pg_database
               WHERE datname = '%s'""" % db.dsn.split('/')[-1], fetch=True)
        self.assertEqual([('UTF8',)], encoding)

    def test_db_template_creates_template_database(self):
        # An optional template parameter can be passed to the constructor.
        # It specifies the name of a template db which is used for the creation
        # of the database. If the template db does not exist, it is created
        # with the specified schema.

        # The first time you create the database with the db_template argument,
        # the template db is created (if it does not exist already) along with
        # the requested db:
        db = self.makeOne(
            schema_path=self.schema, db_template=self.db_template)
        self.assertEqual(['foo', 'tmp_functest'],
                         self.table_names(db.get_dsn(self.db_template)))
        self.assertEqual(['foo', 'tmp_functest'],
                         self.table_names(db.dsn))

        # Now with the template available, the schema is not used anymore to
        # create the database (it's re-created from the template). Let's modify
        # the template db before the next db creation run to demonstrate this:
        self.execute(db.get_dsn(self.db_template), 'DROP TABLE foo;')
        db2 = self.makeOne(
            schema_path=self.schema, db_template=self.db_template)
        self.assertEqual(['tmp_functest'],
                         self.table_names(db2.get_dsn(self.db_template)))
        self.assertEqual(['tmp_functest'], self.table_names(db2.dsn))

        # When creating the database, we can, however, force the template db to
        # be created afresh from the schema. Doing so now will leave us with
        # both a test db and a template db according to the modified schema:
        db3 = self.makeOne(
            schema_path=self.schema, db_template=self.db_template,
            force_template=True)
        self.assertEqual(['foo', 'tmp_functest'],
                         self.table_names(db3.get_dsn(self.db_template)))
        self.assertEqual(['foo', 'tmp_functest'],
                         self.table_names(db3.dsn))

        # The template db (and with it, the test db) ist also created anew if
        # the schema file is newer than the existing template db:
        time.sleep(1)  # XXX hack; either fake db mtime or see #12431
        self.write(self.schema, 'CREATE TABLE bar (dummy int);')
        db4 = self.makeOne(
            schema_path=self.schema, db_template=self.db_template,
            force_template=True)
        self.assertEqual(['bar', 'tmp_functest'],
                         self.table_names(db4.get_dsn(self.db_template)))
        self.assertEqual(['bar', 'tmp_functest'],
                         self.table_names(db4.dsn))

        # If, however, the template db cannot be set up properly, it is removed
        # altogether to avoid a broken template db interfering with subsequent
        # tests:
        broken_schema = self.schema + '-broken'
        self.write(broken_schema, 'foobar')
        db_broken = self.makeOne(
            schema_path=broken_schema, db_template=self.db_template,
            force_template=True, create_db=False)
        with self.assertRaises(SystemExit) as err:
            db_broken.create()
        self.assertEllipsis(
            "Could not initialize schema in database "
            "'gocept.testdb.tests-PID...-templatetest'.", str(err.exception))
        self.assertNotIn(self.db_template, db_broken.list_db_names())

    def test_drop_all_drops_all_databases(self):
        # There's a method to drop all test databases that may have been left
        # on the server by previous test runs by removing all (but only those)
        # databases whose name matches the test database naming scheme of
        # ``gocept.testdb`` and using the same name prefix as the Database
        # instance used for dropping them all. This clean-up method doesn't by
        # default drop the template database if one was created:
        db = self.makeOne(db_template=self.db_template)
        self.makeOne()
        db.create_db(self.pid_prefix + 'foo')
        db.create_db(self.pid_prefix + 'bar')
        self.assertEqual(5, len(self.list_testdb_names(db)))
        db.drop_all()
        self.assertEqual(3, len(self.list_testdb_names(db)))
        db.drop_db(self.pid_prefix + 'foo')
        db.drop_db(self.pid_prefix + 'bar')
        self.assertEqual(1, len(self.list_testdb_names(db)))

        # However, the clean-up method can be instructed to drop the template
        # database as well:
        self.assertIn(db.db_template, db.list_db_names())
        db.drop_all(drop_template=True)
        self.assertNotIn(db.db_template, db.list_db_names())
        self.assertEqual([], self.list_testdb_names(db))

1	import gocept.testdb	1✔
2	import gocept.testdb.testing	1✔
3	import gocept.testing.assertion	1✔
4	import os	1✔
5	import sqlalchemy.exc	1✔
6	import time	1✔
7
8
9	class PostgreSQLTests(gocept.testdb.testing.TestCase,	1✔
10	gocept.testing.assertion.String,
11	gocept.testing.assertion.Exceptions,
12	gocept.testing.assertion.Ellipsis):
13	"""Testing ..postgres.PostgreSQL."""
14
15	db_template = 'gocept.testdb.tests-template-%s' % os.getpid()	1✔
16
17	def tearDown(self):	1✔
18	try:	1✔
19	self.drop_all()	1✔
20	finally:
21	super().tearDown()	1✔
22
23	def drop_all(self):	1✔
24	self.makeOne(db_template=self.db_template, create_db=False).drop_all(	1✔
25	drop_template=True)
26
27	def makeOne(self, create_db=True, db_template=None, **kw):	1✔
28	import gocept.testdb	1✔
29	db = gocept.testdb.PostgreSQL(	1✔
30	db_template=db_template, **kw)
31	if create_db:	1✔
32	db.create()	1✔
33	return db	1✔
34
35	def test_template_db_doesnt_need_schema(self):	1✔
36	# regression test
37	db = self.makeOne(create_db=False)	1✔
38	with self.assertNothingRaised():	1✔
39	db.create()	1✔
40
41	def test_naming_scheme_matches_with_dash_in_prefix(self):	1✔
42	# regression test
43	db = self.makeOne(prefix='foo-bar', create_db=False)	1✔
44	self.assertTrue(db._matches_db_naming_scheme(db.db_name))	1✔
45	self.assertTrue(db._matches_db_naming_scheme('foo-bar-123abc456efg'))	1✔
46
47	def test_can_add_database_with_special_lc_collate(self):	1✔
48	try:	1✔
49	db = self.makeOne(lc_collate='de_DE.UTF-8')	1✔
50	except SystemExit:	1✔
51	# When using Docker container as suggested in README this call
52	# fails with 'invalid locale name: "de_DE.UTF-8"', so let's skip
53	# the test in that environment:
54	self.skipTest('"de_DE.UTF-8" not supported by database.')	1✔
55	collate_by_db_name = {x[0]: x[3] for x in db.pg_list_db_items()}	×
56	self.assertEqual('de_DE.UTF-8', collate_by_db_name[db.db_name])	×
57
58	def test_takes_configuration_from_environment(self):	1✔
59	db = self.makeOne(create_db=False)	1✔
60	self.assertStartsWith('postgresql://', db.dsn)	1✔
61	hostname = 'localhost'	1✔
62	if db.db_port: # pragma: no cover
63	hostname = f'{hostname}:{db.db_port}'
64	self.assertIn(f'{hostname}/testdb-PID', db.dsn)	1✔
65
66	def test_created_database_contains_marker_table(self):	1✔
67	db = self.makeOne()	1✔
68	# The database is marked as a testing database by creating a table
69	# called 'tmp_functest' in it:
70	with self.assertNothingRaised():	1✔
71	self.execute(db.dsn, 'SELECT * from tmp_functest')	1✔
72
73	def test_conveniences_properties_are_set(self):	1✔
74	db = self.makeOne()	1✔
75	self.assertTrue(db.exists)	1✔
76	self.assertTrue(db.is_testing)	1✔
77
78	def test_schema_gets_loaded(self):	1✔
79	db = self.makeOne(schema_path=self.schema)	1✔
80	with self.assertNothingRaised():	1✔
81	self.execute(db.dsn, 'SELECT * from foo')	1✔
82
83	def test_name_of_database_can_be_specified(self):	1✔
84	db = self.makeOne(db_name='mytestdb', create_db=False)	1✔
85	self.assertEndsWith('/mytestdb', db.dsn)	1✔
86
87	def test_drop_drops_database(self):	1✔
88	db = self.makeOne()	1✔
89	db.drop()	1✔
90	with self.assertRaises(sqlalchemy.exc.OperationalError) as err:	1✔
91	self.connect(db)	1✔
92	self.assertEllipsis(	1✔
93	'... database ... does not exist...', str(err.exception))
94
95	def test_encoding_is_used(self):	1✔
96	# An optional encoding parameter can be specified in the constructor.
97	# It is used when creating the database.
98	db = self.makeOne(schema_path=self.schema, encoding='UTF8')	1✔
99	encoding = self.execute(	1✔
100	db.dsn,
101	"""SELECT pg_catalog.pg_encoding_to_char(encoding) as encoding
102	FROM pg_catalog.pg_database
103	WHERE datname = '%s'""" % db.dsn.split('/')[-1], fetch=True)
104	self.assertEqual([('UTF8',)], encoding)	1✔
105
106	def test_db_template_creates_template_database(self):	1✔
107	# An optional template parameter can be passed to the constructor.
108	# It specifies the name of a template db which is used for the creation
109	# of the database. If the template db does not exist, it is created
110	# with the specified schema.
111
112	# The first time you create the database with the db_template argument,
113	# the template db is created (if it does not exist already) along with
114	# the requested db:
115	db = self.makeOne(	1✔
116	schema_path=self.schema, db_template=self.db_template)
117	self.assertEqual(['foo', 'tmp_functest'],	1✔
118	self.table_names(db.get_dsn(self.db_template)))
119	self.assertEqual(['foo', 'tmp_functest'],	1✔
120	self.table_names(db.dsn))
121
122	# Now with the template available, the schema is not used anymore to
123	# create the database (it's re-created from the template). Let's modify
124	# the template db before the next db creation run to demonstrate this:
125	self.execute(db.get_dsn(self.db_template), 'DROP TABLE foo;')	1✔
126	db2 = self.makeOne(	1✔
127	schema_path=self.schema, db_template=self.db_template)
128	self.assertEqual(['tmp_functest'],	1✔
129	self.table_names(db2.get_dsn(self.db_template)))
130	self.assertEqual(['tmp_functest'], self.table_names(db2.dsn))	1✔
131
132	# When creating the database, we can, however, force the template db to
133	# be created afresh from the schema. Doing so now will leave us with
134	# both a test db and a template db according to the modified schema:
135	db3 = self.makeOne(	1✔
136	schema_path=self.schema, db_template=self.db_template,
137	force_template=True)
138	self.assertEqual(['foo', 'tmp_functest'],	1✔
139	self.table_names(db3.get_dsn(self.db_template)))
140	self.assertEqual(['foo', 'tmp_functest'],	1✔
141	self.table_names(db3.dsn))
142
143	# The template db (and with it, the test db) ist also created anew if
144	# the schema file is newer than the existing template db:
145	time.sleep(1) # XXX hack; either fake db mtime or see #12431	1✔
146	self.write(self.schema, 'CREATE TABLE bar (dummy int);')	1✔
147	db4 = self.makeOne(	1✔
148	schema_path=self.schema, db_template=self.db_template,
149	force_template=True)
150	self.assertEqual(['bar', 'tmp_functest'],	1✔
151	self.table_names(db4.get_dsn(self.db_template)))
152	self.assertEqual(['bar', 'tmp_functest'],	1✔
153	self.table_names(db4.dsn))
154
155	# If, however, the template db cannot be set up properly, it is removed
156	# altogether to avoid a broken template db interfering with subsequent
157	# tests:
158	broken_schema = self.schema + '-broken'	1✔
159	self.write(broken_schema, 'foobar')	1✔
160	db_broken = self.makeOne(	1✔
161	schema_path=broken_schema, db_template=self.db_template,
162	force_template=True, create_db=False)
163	with self.assertRaises(SystemExit) as err:	1✔
164	db_broken.create()	1✔
165	self.assertEllipsis(	1✔
166	"Could not initialize schema in database "
167	"'gocept.testdb.tests-PID...-templatetest'.", str(err.exception))
168	self.assertNotIn(self.db_template, db_broken.list_db_names())	1✔
169
170	def test_drop_all_drops_all_databases(self):	1✔
171	# There's a method to drop all test databases that may have been left
172	# on the server by previous test runs by removing all (but only those)
173	# databases whose name matches the test database naming scheme of
174	# ``gocept.testdb`` and using the same name prefix as the Database
175	# instance used for dropping them all. This clean-up method doesn't by
176	# default drop the template database if one was created:
177	db = self.makeOne(db_template=self.db_template)	1✔
178	self.makeOne()	1✔
179	db.create_db(self.pid_prefix + 'foo')	1✔
180	db.create_db(self.pid_prefix + 'bar')	1✔
181	self.assertEqual(5, len(self.list_testdb_names(db)))	1✔
182	db.drop_all()	1✔
183	self.assertEqual(3, len(self.list_testdb_names(db)))	1✔
184	db.drop_db(self.pid_prefix + 'foo')	1✔
185	db.drop_db(self.pid_prefix + 'bar')	1✔
186	self.assertEqual(1, len(self.list_testdb_names(db)))	1✔
187
188	# However, the clean-up method can be instructed to drop the template
189	# database as well:
190	self.assertIn(db.db_template, db.list_db_names())	1✔
191	db.drop_all(drop_template=True)	1✔
192	self.assertNotIn(db.db_template, db.list_db_names())	1✔
193	self.assertEqual([], self.list_testdb_names(db))	1✔

gocept / gocept.testdb / 5899312502

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous