主页 / 问题 / 1600029
Accepted
Евграф Котовский
Asked:2024-11-17 07:05:06 +0000 UTC2024-11-17 07:05:06 +0000 UTC

为什么数据库在执行过程中会被阻塞？

  • 772

我正在为我的项目编写测试

这是导致错误的测试：

# adding classes to path
import sys
sys.path.append('..')
from time import sleep
from classes.database import *
import psycopg2
import sqlite3
from classes.tools import *

class UserClassAllMethodstest(object):
    #defining data of example user for tests User() class
    SimpleUserData = {
        "password":"123",
        'email': '[email protected]',
        "user":"user",
        "is_admin":"0",
        "is_banned":'0',
        "is_banned":"0",
        "logo_path":'',
        "citate":"",
        "format":"",

    }

    def __init__(self, DBWorker=DB()):
        self.DBWorker = DBWorker
        DBWorker.db.DBInit()

    # checking methods of User() class
    def TestClass(self):

        """
        checks create method

        #in first case we are checking creation of new user
        #in second case we are checking error from same emails
        #in third case we are checking errors from same user ID
        """

        assert self.DBWorker.User().create(**UserClassAllMethodstest.SimpleUserData)

        

        try:
            self.DBWorker.User().create(**UserClassAllMethodstest.SimpleUserData)
            self.DBWorker.User().create(password="1234567890", email="[email protected]", user="user1", is_admin="",
                                  is_banned="", logo_path="", citate="", format="obj")
        except Exception as e:

            assert type(e) == sqlite3.IntegrityError or type(e) == psycopg2.errors.UniqueViolation

        

        try:
            self.DBWorker.User().create(password="1234567890", email="[email protected]", user="user", is_admin="",
                                    is_banned="", logo_path="", citate="", format="obj")

            self.DBWorker.User().create(password="1234567890", email="[email protected]", user="user", is_admin="",\
                                  is_banned="", logo_path="",\
                                  citate="", format="obj")


        except Exception as e:
            
            print(e)
            assert type(e) == sqlite3.IntegrityError or type(e) == psycopg2.errors.UniqueViolation

        


        """
        checks get method
        
        in first case we checking receiving user by his user id
        in second case we checking receiving user by his user id and password
        in second case we checking receiving user by his token, created as a hash from password and user id
        in four case we checking getting user from num
        in case 5 we checking getting data in json (dict in python) format
        in case 6 we checking getting data from not existing user id
        
        """

        assert self.DBWorker.User().get(user = UserClassAllMethodstest.SimpleUserData["user"], format = "obj").UserId == \
               UserClassAllMethodstest.SimpleUserData["user"]

        

        assert self.DBWorker.User().get(user=UserClassAllMethodstest.SimpleUserData["user"], password = \
                                   UserClassAllMethodstest.SimpleUserData["password"], format = "obj").UserId == \
               UserClassAllMethodstest.SimpleUserData['user']

        


        token = generate_token(UserClassAllMethodstest.SimpleUserData["user"],UserClassAllMethodstest.SimpleUserData["password"])

        assert self.DBWorker().get(token=token, format = "obj").UserId ==\
               UserClassAllMethodstest["user"]
        


        TestUser = self.DBWorker.User().create(password="1234567890", email="[email protected]", user="user5", is_admin="",
                                  is_banned="", logo_path="",
                                  citate="", format="obj")

        

        assert self.DBWorker.User().get(num = TestUser.ActiveNum).__dict__ == TestUser.__dict__

        


        assert self.DBWorker.User().get(user = UserClassAllMethodstest.SimpleUserData["user"], format = "json")["UserId"] == \
               UserClassAllMethodstest.SimpleUserData["user"]

        

        """
        checks data changing via UserStorage() class
        """

        UserData = self.DBWorker.User().get(user = UserClassAllMethodstest.SimpleUserData["user"], format = "obj")

        try:

            UserData.email = "[email protected]"
            UserData.UserId = "user1337"
            UserData.IsAdmin = "1"
            UserData.IsBanned = "1"
            UserData.LogoPath = "sample_image"
            UserData.citate = "sample quote"
            UserData.time = "13:37:14"
            UserData.ActiveNum = "1337"
            self.IsActivated = "1"
            self.NumOfPosts = "666"

            UserData.save()

        except Exception as e:

            raise AssertionError

        """
        checks delete method
        
        in case 1 we checking deleting data via user id
        in case 2 we checking deleting data via user id and password
        in case 3 we checking deleting via token created as a hash from user id and password
        """

        assert self.DBWorker.User().delete(user == UserClassAllMethodstest.SimpleUserData["user"]) == 1
        assert self.DBWorker.User().create(**UserClassAllMethodstest.SimpleUserData)

        assert self.DBWorker.User().delete(user == UserClassAllMethodstest.SimpleUserData["user"], password = UserClassAllMethodstest.SimpleUserData["password"]) == 1
        assert self.DBWorker.User().create(**UserClassAllMethodstest.SimpleUserData)

        assert self.DBWorker.User().delete(token = token) == 1

if __name__ == '__main__':
    UserClassAllMethodstest().TestClass()

执行时，由于数据库被锁定，程序崩溃。我用的是sqlite3。数据库被阻塞可能是什么问题以及如何解决？

python

2 个回答

  1. 问题是您没有连接到数据库或者有多个连接。最有可能的是——第一个。我建议你这样做：

    def TestClass(self):
    with self.DBWorker.db.connect() as conn:
    • 0
  2. Best Answer

    问题是sqlite3只有一个连接。要关闭连接，需要修改SQLite3类的work方法的代码：

        def work(self, query, param=("",)):
            con = sqlite3.connect(self.path, timeout=7)
            cursor = con.cursor()
            try:
                cursor.execute(query, param)
                data = cursor.fetchall()
                con.commit()
                return data
            except Exception as e:
                print(f"Error executing query: {e}")
                con.rollback()  # Rollback if there's an error
                return None
            finally:
                cursor.close()  # Close cursor
                con.close()     # Close connection

    在此代码中，进行更改后，我们应用更改并关闭连接

    • 0

相关问题