Why does my test (using pytest) fail and say the database exists?-CodePudding

I have some unit tests that I am testing using pytest. I want to have the tests create and then destroy an SQLite3 database for each test so that I can test each function with an actual database.

database.py:

import sqlite3

db = sqlite3.connect('geminon.db', check_same_thread=False)
cur = db.cursor()


def create_tables():
    """
    create tables
    """

    cur.execute("""
    CREATE TABLE ships (
    name                TEXT    PRIMARY KEY,
    classification      INTEGER DEFAULT   0,
    number              INTEGER DEFAULT   0,
    damage              INTEGER DEFAULT   0
    )
    """)

def add_ship(name):
    cur.execute("SELECT * FROM ships WHERE name=:name LIMIT 1", {'name': name})
    if cur.fetchone() is None:
        cur.execute("insert into ships (name) values (?)", (name,))
        db.commit()
        return True
    else:
        return False

test_database.py:

import unittest
import database as database
import os


class test_database(unittest.TestCase):

    def tearDown(self):
        os.remove('geminon.db')

    def test_create_tables(self):
        database.create_tables()
        database.cur.execute("insert into ships (name) values (?)", ('test_ship',))
        database.cur.execute("SELECT * FROM ships ", )
        self.assertIsNotNone(database.cur.fetchall())


class test_database_ships_functions(unittest.TestCase):

    def setUp(self):
        database.create_tables()

    def tearDown(self):
        os.remove('geminon.db')

    def test_add_ship(self):
        database.add_ship('test_ship1')
        database.db.commit()
        database.cur.execute("SELECT * FROM ships WHERE name=:name LIMIT 1", {'name': 'test_ship1'})
        self.assertIsNotNone(database.cur.fetchall())

When I run pytest, I get the following error:


self = <test_database.test_database_ships_functions testMethod=test_add_ship>

    def setUp(self):
>       database.create_tables()

test_database.py:21: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

    def create_tables():
        """
        create tables
        """
    
>       cur.execute("""
        CREATE TABLE ships (
        name                TEXT    PRIMARY KEY,
        classification      INTEGER DEFAULT   0,
        number              INTEGER DEFAULT   0,
        damage              INTEGER DEFAULT   0
        )
        """)
E       sqlite3.OperationalError: table ships already exists

database.py:12: OperationalError

Shouldn't the tearDown method in the test_database class destroy the database before test_add_ships runs? I must be misunderstanding how this is applied, but I'm not sure how.

CodePudding user response：

Could be that geminon.db already exists before you start the tests.

It can also be connected to the fact that both tests share a global database.

Here is how I would go about doing this:

database.py

import os
import sqlite3


class ShipDatabase:

    def __init__(self, db_name="geminon.db", fresh_db=False):
        self.db_name = db_name
        if fresh_db and os.path.exists(self.db_name):
            os.remove(self.db_name)
        self.con = sqlite3.connect(self.db_name, check_same_thread=False)

    def create_tables(self):
        cur = self.con.cursor()
        create_query = """
            CREATE TABLE ships (
                name                TEXT    PRIMARY KEY,
                classification      INTEGER DEFAULT   0,
                number              INTEGER DEFAULT   0,
                damage              INTEGER DEFAULT   0
            )
        """
        cur.execute(create_query)
        cur.close()

    def destroy_db(self):
        self.con.close()
        if os.path.exists(self.db_name):
            os.remove(self.db_name)

    def add_ship(self, name):
        cur = self.con.cursor()
        cur.execute("SELECT * FROM ships WHERE name=:name LIMIT 1", {'name': name})
        ship_does_not_exist = cur.fetchone() is None
        if ship_does_not_exist:
            cur.execute("insert into ships (name) values (?)", (name,))
            self.con.commit()
        cur.close()
        return ship_does_not_exist

database_test.py

import unittest
from database import ShipDatabase


class TestDatabase(unittest.TestCase):

    def setUp(self):
        self.sdb = ShipDatabase(fresh_db=True)
        self.sdb.create_tables()

    def tearDown(self):
        self.sdb.destroy_db()

    def test_create_tables(self):
        cur = self.sdb.con.cursor()
        cur.execute("insert into ships (name) values (?)", ('test_ship',))
        cur.execute("SELECT * FROM ships ", )
        self.assertIsNotNone(cur.fetchall())
        cur.close()


class TestDatabaseShipsFunctions(unittest.TestCase):

    def setUp(self):
        self.sdb = ShipDatabase(fresh_db=True)
        self.sdb.create_tables()

    def tearDown(self):
        self.sdb.destroy_db()

    def test_add_ship(self):
        self.sdb.add_ship('test_ship1')
        cur = self.sdb.con.cursor()
        cur.execute("SELECT * FROM ships WHERE name=:name LIMIT 1", {'name': 'test_ship1'})
        self.assertIsNotNone(cur.fetchall())
        cur.close()

This implementation is missing exception handling.