Using unittest in python with docker and psycopg2

Question

BRIEF DESCRIPTION OF THE PROBLEM: Psycopg2 won't connect to a test DB within docker from unittest, but connects fine from console.

ERROR MESSAGE: psycopg2.OperationalError: server closed the connection unexpectedly This probably means the server terminated abnormally before or while processing the request.

DETAILS: I'm trying to set up a testing database in docker, that will be created and populated before testing and then removed after.

Here's the fuction to set up database:

def set_up_empty_test_db():
    client = docker.from_env()
    try:
        testdb = client.containers.get("TestDB")
        testdb.stop()
        testdb.remove()
        testdb = client.containers.run(
            "postgres",
            ports={5432: 5433},
            detach=True,
            name="TestDB",
            environment=["POSTGRES_PASSWORD=yourPassword"],
        )
    except NotFound:
        testdb = client.containers.run(
            "postgres",
            ports={5432: 5433},
            detach=True,
            name="TestDB",
            environment=["POSTGRES_PASSWORD=yourPassword"],
        )

    while testdb.status != "running":
        testdb = client.containers.get("TestDB")
    return

If I launch this function from console it works without an error and I can see TestDB container running. I can successfully initiate connection:

conn = psycopg2.connect("host='localhost' user='postgres' password='yourPassword' port='5433'")

But it doesn't work when unit testing. Here's the testing code:

class TestCreateCity(unittest.TestCase):
    def setUp(self):
        set_up_empty_test_db()
        con = psycopg2.connect("host='localhost' user='postgres' password='yourPassword' port='5433'")

        self.assertIsNone(con.isolation_level)

        cur = con.cursor()
        sql_file = open(os.path.join(ROOT_DIR + "/ddl/creates/schema_y.sql"), "r")
        cur.execute(sql_file.readline())
        con.commit()
        con.close()
        self.session = Session(creator=con)

    def test_create_city(self):
        cs.create_city("Test_CITY", "US")
        q = self.session.query(City).filter(City.city_name == "Test_CITY").one()
        self.assertIs(q)
        self.assertEqual(q.city_name, "Test_CITY")
        self.assertEqual(q.country_code, "US")

It breaks when trying to initiate connection. Please advise.

you get any error message?

RandomDude
– RandomDude

2018-09-06 13:47:10 +00:00
Commented Sep 6, 2018 at 13:47 — RandomDude
– RandomDude, Commented Sep 6, 2018 at 13:47

user2368291 · Accepted Answer · 2018-11-12 00:12:48Z

5

I know this is an old question, but I needed to do the same thing today. You try and connect to the postgres server too quickly after starting it - that's why it works in the console.

All you need to do is replace:

set_up_empty_test_db()
con = psycopg2.connect("host='localhost' user='postgres' password='yourPassword' port='5433'")

with:

set_up_empty_test_db()
con = None
while con == None:
    try:
        con = psycopg2.connect("host='localhost' user='postgres' password='yourPassword' port='5433'")
    except psycopg2.OperationalError:
        time.sleep(0.5);

Hope this helps someone else. Cheers!

answered Nov 12, 2018 at 0:12

user2368291

961 silver badge2 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

Myke Bilyanskyy Over a year ago

That's exactly what I did in my own solution. Thanks for sharing!

Collectives™ on Stack Overflow

Using unittest in python with docker and psycopg2

1 Answer 1

1 Comment

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

1 Comment

Your Answer

Sign up or log in

Post as a guest

Related