Revize c7214bba

Re #8471 - Modified source
- deleted redundat files
- deleted '_impl' from repositories filename

src/dao/certificate_repository.py

@@ +1 @@
+from typing import List, Dict
+
+from ..db_objects.certificate import Certificate
+from ..dao.repository import IRepository
+from db_manager import DBManager
+from ..constants import *
+
+
+class CertificateRepositoryImpl(IRepository):
+
+    def create(self, common_name: str, valid_from: str, valid_to: str, pem_data: str,
+               private_key_id: int, type_id: int, parent_id: int, usages: Dict[int, bool]) -> bool:
+        """
+        Creates a certificate
+
+        :param common_name: private key of the certificate
+        :param valid_from: string of date (time) of validation after (from)
+        :param valid_to: string of date (time) of validation before (to)
+        :param pem_data: data of certificate in a PEM format
+        :param private_key_id: ID from PrivateKeys table for specific private key
+        :param type_id: ID from CertificateTypes table (ROOT_CA_ID=1, INTERMEDIATE_CA_ID=2, CERTIFICATE_ID=3)
+        :param parent_id: ID from Certificates table for specific parent certificate
+            for root CA use e.g. -1 (ID will be changed after created certificate)
+        :param usages: dictionary of usages IDs which certificate use
+            Dictionary[Integer, Boolean]
+                - Key = ID (CA_ID=1, SSL_ID=2, SIGNATURE_ID=3, AUTHENTICATION_ID=4)
+                - Value = used=True/unused=False
+
+        :return: the result of whether the creation was successful
+        """
+
+        sql = (f"INSERT INTO {TAB_CERTIFICATES} ({COL_COMMON_NAME},{COL_VALID_FROM},{COL_VALID_TO},{COL_PEM_DATA},"
+               f"{COL_PRIVATE_KEY_ID},{COL_TYPE_ID},{COL_PARENT_ID})"
+               f"VALUES(?,?,?,?,?,?,?)")
+        result = DBManager.create(sql, [common_name, valid_from, valid_to, pem_data,
+                                        private_key_id, type_id, parent_id], usages)
+        return result
+
+    def read(self, certificate_id: int = None):
+        """
+        Reads (selects) a certificate or certificates
+
+        :param certificate_id: ID of specific certificate
+
+        :return: instance of Certificate class if ID was used otherwise list of this instances
+        """
+
+        sql_extend = ""
+
+        if certificate_id is not None:
+            sql_extend = f" WHERE {COL_ID} = ?"
+
+        sql = f"SELECT * FROM {TAB_CERTIFICATES}{sql_extend}"
+        certificate_rows = DBManager.read(sql, certificate_id)
+
+        certificates: list = []
+
+        for certificate_row in certificate_rows:
+            sql = f"SELECT * FROM {TAB_CERTIFICATE_USAGES} WHERE {COL_CERTIFICATE_ID} = ?"
+            usage_rows = DBManager.read(sql, certificate_row[0])
+
+            usage_dict: Dict[int, bool] = {}
+            for usage_row in usage_rows:
+                usage_dict[usage_row[1]] = True
+
+            certificates.append(Certificate(certificate_row[0],
+                                            certificate_row[1],
+                                            certificate_row[2],
+                                            certificate_row[3],
+                                            certificate_row[4],
+                                            certificate_row[5],
+                                            certificate_row[6],
+                                            certificate_row[7],
+                                            usage_dict))
+
+        if certificate_id is not None:
+            return certificates[0]
+
+        return certificates
+
+    def update(self, certificate_id: int, common_name: str = None, valid_from: str = None, valid_to: str = None,
+               pem_data: str = None, private_key_id: int = None, type_id: int = None, parent_id: int = None,
+               usages: Dict[int, bool] = None) -> bool:
+        """
+        Updates a certificate
+
+        :param certificate_id: ID of specific certificate
+        :param common_name: private key of the certificate
+        :param valid_from: string of date (time) of validation after (from)
+        :param valid_to: string of date (time) of validation before (to)
+        :param pem_data: data of certificate in a PEM format
+        :param private_key_id: ID from PrivateKeys table for specific private key
+        :param type_id: ID from CertificateTypes table (ROOT_CA_ID=1, INTERMEDIATE_CA_ID=2, CERTIFICATE_ID=3)
+        :param parent_id: ID from Certificates table for specific parent certificate
+            for root CA use e.g. -1 (ID will be changed after created certificate)
+        :param usages: dictionary of usages IDs which certificate use
+            Dictionary[Integer, Boolean]
+                - Key = ID (CA_ID=1, SSL_ID=2, SIGNATURE_ID=3, AUTHENTICATION_ID=4)
+                - Value = used=True/unused=False
+
+        :return: the result of whether the updation was successful
+        """
+
+        updated_list = []
+        values = []
+        if common_name is not None:
+            updated_list.append(f"{COL_COMMON_NAME} = ?")
+            values.append(common_name)
+        if valid_from is not None:
+            updated_list.append(f"{COL_VALID_FROM} = ?")
+            values.append(valid_from)
+        if valid_to is not None:
+            updated_list.append(f"{COL_VALID_TO} = ?")
+            values.append(valid_to)
+        if pem_data is not None:
+            updated_list.append(f"{COL_PEM_DATA} = ?")
+            values.append(pem_data)
+        if private_key_id is not None:
+            updated_list.append(f"{COL_PRIVATE_KEY_ID} = ?")
+            values.append(private_key_id)
+        if type_id is not None:
+            updated_list.append(f"{COL_TYPE_ID} = ?")
+            values.append(type_id)
+        if parent_id is not None:
+            updated_list.append(f"{COL_PARENT_ID} = ?")
+            values.append(parent_id)
+
+        updated_str = ", ".join(updated_list)
+        sql = f"UPDATE {TAB_CERTIFICATES} SET {updated_str} WHERE {COL_ID} = ?"
+        result = DBManager.update(sql, certificate_id, values, usages)
+        return result
+
+    def delete(self, certificate_id: int) -> bool:
+        """
+        Deletes a certificate
+
+        :param certificate_id: ID of specific certificate
+
+        :return: the result of whether the deletion was successful
+        """
+
+        sql = f"DELETE FROM {TAB_CERTIFICATES} WHERE {COL_ID} = ?"
+        result = DBManager.delete(sql, certificate_id, True)
+        return result

src/dao/private_key_repository.py

@@ +1 @@
+from ..db_objects.private_key import PrivateKey
+from ..dao.repository import IRepository
+from db_manager import DBManager
+from ..constants import *
+
+
+class PrivateKeyRepositoryImpl(IRepository):
+
+    def create(self, private_key: str, password: str) -> bool:
+        """
+        Creates a private key
+
+        :param private_key: private key
+        :param password: passphrase for encode private key
+
+        :return: the result of whether the creation was successful
+        """
+
+        sql = f"INSERT INTO {TAB_PRIVATE_KEYS} ({COL_PRIVATE_KEY},{COL_PASSWORD}) VALUES(?,?)"
+        result = DBManager.create(sql, [private_key, password])
+        return result
+
+    def read(self, private_key_id: int = None):
+        """
+        Reads (selects) a private key
+
+        :param private_key_id: ID of specific private
+
+        :return: instance of PrivateKey class if ID was used otherwise list of this instances
+        """
+
+        sql_extend = ""
+
+        if private_key_id is not None:
+            sql_extend = f" WHERE {COL_ID} = ?"
+
+        sql = f"SELECT * FROM {TAB_PRIVATE_KEYS}{sql_extend}"
+        rows = DBManager.read(sql, private_key_id)
+
+        private_keys: list = []
+
+        for row in rows:
+            private_keys.append(PrivateKey(row))
+
+        if private_key_id is not None:
+            return private_keys[0]
+
+        return private_keys
+
+    def update(self, private_key_id: int, private_key: str = None, password: str = None) -> bool:
+        """
+        Updates a private key
+
+        :param private_key_id: ID of specific private key
+        :param private_key: private key
+        :param password: passphrase for encode private key
+
+        :return: the result of whether the updation was successful
+        """
+
+        updated_list = []
+        values = []
+        if private_key is not None:
+            updated_list.append(f"{COL_PRIVATE_KEY} = ?")
+            values.append(private_key)
+        if password is not None:
+            updated_list.append(f"{COL_PASSWORD} = ?")
+            values.append(password)
+
+        updated_str = ", ".join(updated_list)
+        sql = f"UPDATE {TAB_PRIVATE_KEYS} SET {updated_str} WHERE {COL_ID} = ?"
+        result = DBManager.update(sql, private_key_id, values)
+        return result
+
+    def delete(self, private_key_id: int) -> bool:
+        """
+        Deletes a private key
+
+        :param private_key_id: ID of specific private key
+
+        :return: the result of whether the deletion was successful
+        """
+
+        sql = f"DELETE FROM {TAB_PRIVATE_KEYS} WHERE {COL_ID} = ?"
+        result = DBManager.delete(sql, private_key_id, False)
+        return result

src/dao/repository.py

@@ -1 @@
-class IRepository:
-    def create(self, *args) -> bool:
-        pass
-
-    def read(self, *args):
-        pass
-
-    def update(self, *args) -> bool:
-        pass
-
-    def delete(self, *args) -> bool:
-        pass

src/impl/certificate_repository_impl.py

@@ -1 @@
-from typing import List, Dict
-
-from ..db_objects.certificate import Certificate
-from ..dao.repository import IRepository
-from db_manager import DBManager
-from ..constants import *
-
-
-class CertificateRepositoryImpl(IRepository):
-
-    def create(self, common_name: str, valid_from: str, valid_to: str, pem_data: str,
-               private_key_id: int, type_id: int, parent_id: int, usages: Dict[int, bool]) -> bool:
-        """
-        Creates a certificate
-
-        :param common_name: private key of the certificate
-        :param valid_from: string of date (time) of validation after (from)
-        :param valid_to: string of date (time) of validation before (to)
-        :param pem_data: data of certificate in a PEM format
-        :param private_key_id: ID from PrivateKeys table for specific private key
-        :param type_id: ID from CertificateTypes table (ROOT_CA_ID=1, INTERMEDIATE_CA_ID=2, CERTIFICATE_ID=3)
-        :param parent_id: ID from Certificates table for specific parent certificate
-            for root CA use e.g. -1 (ID will be changed after created certificate)
-        :param usages: dictionary of usages IDs which certificate use
-            Dictionary[Integer, Boolean]
-                - Key = ID (CA_ID=1, SSL_ID=2, SIGNATURE_ID=3, AUTHENTICATION_ID=4)
-                - Value = used=True/unused=False
-
-        :return: the result of whether the creation was successful
-        """
-
-        sql = (f"INSERT INTO {TAB_CERTIFICATES} ({COL_COMMON_NAME},{COL_VALID_FROM},{COL_VALID_TO},{COL_PEM_DATA},"
-               f"{COL_PRIVATE_KEY_ID},{COL_TYPE_ID},{COL_PARENT_ID})"
-               f"VALUES(?,?,?,?,?,?,?)")
-        result = DBManager.create(sql, [common_name, valid_from, valid_to, pem_data,
-                                        private_key_id, type_id, parent_id], usages)
-        return result
-
-    def read(self, certificate_id: int = None):
-        """
-        Reads (selects) a certificate or certificates
-
-        :param certificate_id: ID of specific certificate
-
-        :return: instance of Certificate class if ID was used otherwise list of this instances
-        """
-
-        sql_extend = ""
-
-        if certificate_id is not None:
-            sql_extend = f" WHERE {COL_ID} = ?"
-
-        sql = f"SELECT * FROM {TAB_CERTIFICATES}{sql_extend}"
-        certificate_rows = DBManager.read(sql, certificate_id)
-
-        certificates: list = []
-
-        for certificate_row in certificate_rows:
-            sql = f"SELECT * FROM {TAB_CERTIFICATE_USAGES} WHERE {COL_CERTIFICATE_ID} = ?"
-            usage_rows = DBManager.read(sql, certificate_row[0])
-
-            usage_dict: Dict[int, bool] = {}
-            for usage_row in usage_rows:
-                usage_dict[usage_row[1]] = True
-
-            certificates.append(Certificate(certificate_row[0],
-                                            certificate_row[1],
-                                            certificate_row[2],
-                                            certificate_row[3],
-                                            certificate_row[4],
-                                            certificate_row[5],
-                                            certificate_row[6],
-                                            certificate_row[7],
-                                            usage_dict))
-
-        if certificate_id is not None:
-            return certificates[0]
-
-        return certificates
-
-    def update(self, certificate_id: int, common_name: str = None, valid_from: str = None, valid_to: str = None,
-               pem_data: str = None, private_key_id: int = None, type_id: int = None, parent_id: int = None,
-               usages: Dict[int, bool] = None) -> bool:
-        """
-        Updates a certificate
-
-        :param certificate_id: ID of specific certificate
-        :param common_name: private key of the certificate
-        :param valid_from: string of date (time) of validation after (from)
-        :param valid_to: string of date (time) of validation before (to)
-        :param pem_data: data of certificate in a PEM format
-        :param private_key_id: ID from PrivateKeys table for specific private key
-        :param type_id: ID from CertificateTypes table (ROOT_CA_ID=1, INTERMEDIATE_CA_ID=2, CERTIFICATE_ID=3)
-        :param parent_id: ID from Certificates table for specific parent certificate
-            for root CA use e.g. -1 (ID will be changed after created certificate)
-        :param usages: dictionary of usages IDs which certificate use
-            Dictionary[Integer, Boolean]
-                - Key = ID (CA_ID=1, SSL_ID=2, SIGNATURE_ID=3, AUTHENTICATION_ID=4)
-                - Value = used=True/unused=False
-
-        :return: the result of whether the updation was successful
-        """
-
-        updated_list = []
-        values = []
-        if common_name is not None:
-            updated_list.append(f"{COL_COMMON_NAME} = ?")
-            values.append(common_name)
-        if valid_from is not None:
-            updated_list.append(f"{COL_VALID_FROM} = ?")
-            values.append(valid_from)
-        if valid_to is not None:
-            updated_list.append(f"{COL_VALID_TO} = ?")
-            values.append(valid_to)
-        if pem_data is not None:
-            updated_list.append(f"{COL_PEM_DATA} = ?")
-            values.append(pem_data)
-        if private_key_id is not None:
-            updated_list.append(f"{COL_PRIVATE_KEY_ID} = ?")
-            values.append(private_key_id)
-        if type_id is not None:
-            updated_list.append(f"{COL_TYPE_ID} = ?")
-            values.append(type_id)
-        if parent_id is not None:
-            updated_list.append(f"{COL_PARENT_ID} = ?")
-            values.append(parent_id)
-
-        updated_str = ", ".join(updated_list)
-        sql = f"UPDATE {TAB_CERTIFICATES} SET {updated_str} WHERE {COL_ID} = ?"
-        result = DBManager.update(sql, certificate_id, values, usages)
-        return result
-
-    def delete(self, certificate_id: int) -> bool:
-        """
-        Deletes a certificate
-
-        :param certificate_id: ID of specific certificate
-
-        :return: the result of whether the deletion was successful
-        """
-
-        sql = f"DELETE FROM {TAB_CERTIFICATES} WHERE {COL_ID} = ?"
-        result = DBManager.delete(sql, certificate_id, True)
-        return result

src/impl/db_manager.py

@@ -1 @@
-import sqlite3
-from sqlite3 import Error
-from typing import List, Dict
-
-from ..constants import *
-from ..dao.repository import IRepository
-
-
-class DBManager(IRepository):
-
-    @staticmethod
-    def create(sql: str, values: List, usages: Dict[int, bool] = None) -> bool:
-        """
-        Creates an item in database
-
-        :param sql: SQL query for creation (parameter values are replaced by question mark)
-        :param values: list of parameter values
-        :param usages: dictionary of usages IDs which certificate use
-            Dictionary[Integer, Boolean]
-                - Key = ID (CA_ID=1, SSL_ID=2, SIGNATURE_ID=3, AUTHENTICATION_ID=4)
-                - Value = used=True/unused=False
-
-        :return: the result of whether the creation was successful
-        """
-
-        conn = None
-        try:
-            conn = sqlite3.connect(DATABASE_FILE)
-            c = conn.cursor()
-            c.execute(sql, values)
-
-            if usages is not None:
-                for usage_id, usage_value in usages:
-                    if usage_value:
-                        c.execute(f"INSERT INTO {TAB_CERTIFICATE_USAGES} ({COL_CERTIFICATE_ID},{COL_USAGE_TYPE_ID}) "
-                                  f"VALUES (?,?)",
-                                  [c.lastrowid, usage_id])
-
-            conn.commit()
-            c.close()
-        except Error as e:
-            print(e)
-            return False
-        finally:
-            if conn:
-                conn.close()
-        return True
-
-    @staticmethod
-    def read(sql: str, item_id: int = None):
-        """
-        Reads (selecs) an item or items from database
-
-        :param sql: SQL query for selection (parameter values are replaced by question mark)
-        :param item_id: ID of specific item
-
-        :return: list of rows selected from database
-        """
-
-        conn = None
-        try:
-            conn = sqlite3.connect(DATABASE_FILE)
-            c = conn.cursor()
-
-            if item_id is not None:
-                c.execute(sql, [item_id])
-            else:
-                c.execute(sql)
-
-            records = c.fetchall()
-            c.close()
-        except Error as e:
-            print(e)
-            return []
-        finally:
-            if conn:
-                conn.close()
-        return records
-
-    @staticmethod
-    def update(sql: str, item_id: int, values: List, usages: Dict[int, bool] = None) -> bool:
-        """
-        Updates an item in database
-
-        :param sql: SQL query for creation (parameter values are replaced by question mark)
-        :param item_id: ID of specific item
-        :param values: list of parameter values
-        :param usages: dictionary of usages IDs which certificate use
-            Dictionary[Integer, Boolean]
-                - Key = ID (CA_ID=1, SSL_ID=2, SIGNATURE_ID=3, AUTHENTICATION_ID=4)
-                - Value = used=True/unused=False
-
-        :return: the result of whether the updation was successful
-        """
-
-        conn = None
-        try:
-            values.append(item_id)
-
-            conn = sqlite3.connect(DATABASE_FILE)
-            c = conn.cursor()
-            c.execute(sql, values)
-
-            if usages is not None:
-                for usage_id, usage_value in usages:
-                    c.execute(f"DELETE FROM {TAB_CERTIFICATE_USAGES} "
-                              f"WHERE {COL_CERTIFICATE_ID} = ? AND {COL_USAGE_TYPE_ID} = ?",
-                              [item_id, usage_id])
-
-                    if usage_value:
-                        c.execute(f"INSERT INTO {TAB_CERTIFICATE_USAGES} ({COL_CERTIFICATE_ID},{COL_USAGE_TYPE_ID}) "
-                                  f"VALUES (?,?)",
-                                  [item_id, usage_id])
-
-            conn.commit()
-            c.close()
-        except Error as e:
-            print(e)
-            return False
-        finally:
-            if conn:
-                conn.close()
-        return True
-
-    @staticmethod
-    def delete(sql: str, item_id: int, delete_usages: bool) -> bool:
-        """
-        Deletes an item from database
-
-        :param sql: SQL query for selection (parameter values are replaced by question mark)
-        :param item_id: ID of specific item
-        :param delete_usages: flag specifying whether to erase rows with specific id
-            from the table of usages (CertificateUsages)
-
-        :return: the result of whether the deletion was successful
-        """
-
-        conn = None
-        try:
-            conn = sqlite3.connect(DATABASE_FILE)
-            c = conn.cursor()
-
-            if delete_usages:
-                c.execute(f"DELETE FROM {TAB_CERTIFICATE_USAGES} WHERE {COL_CERTIFICATE_ID} = ?", [item_id])
-
-            c.execute(sql, [item_id])
-            conn.commit()
-            c.close()
-        except Error as e:
-            print(e)
-            return False
-        finally:
-            if conn:
-                conn.close()
-        return True

src/impl/private_key_repository_impl.py

@@ -1 @@
-from ..db_objects.private_key import PrivateKey
-from ..dao.repository import IRepository
-from db_manager import DBManager
-from ..constants import *
-
-
-class PrivateKeyRepositoryImpl(IRepository):
-
-    def create(self, private_key: str, password: str) -> bool:
-        """
-        Creates a private key
-
-        :param private_key: private key
-        :param password: passphrase for encode private key
-
-        :return: the result of whether the creation was successful
-        """
-
-        sql = f"INSERT INTO {TAB_PRIVATE_KEYS} ({COL_PRIVATE_KEY},{COL_PASSWORD}) VALUES(?,?)"
-        result = DBManager.create(sql, [private_key, password])
-        return result
-
-    def read(self, private_key_id: int = None):
-        """
-        Reads (selects) a private key
-
-        :param private_key_id: ID of specific private
-
-        :return: instance of PrivateKey class if ID was used otherwise list of this instances
-        """
-
-        sql_extend = ""
-
-        if private_key_id is not None:
-            sql_extend = f" WHERE {COL_ID} = ?"
-
-        sql = f"SELECT * FROM {TAB_PRIVATE_KEYS}{sql_extend}"
-        rows = DBManager.read(sql, private_key_id)
-
-        private_keys: list = []
-
-        for row in rows:
-            private_keys.append(PrivateKey(row))
-
-        if private_key_id is not None:
-            return private_keys[0]
-
-        return private_keys
-
-    def update(self, private_key_id: int, private_key: str = None, password: str = None) -> bool:
-        """
-        Updates a private key
-
-        :param private_key_id: ID of specific private key
-        :param private_key: private key
-        :param password: passphrase for encode private key
-
-        :return: the result of whether the updation was successful
-        """
-
-        updated_list = []
-        values = []
-        if private_key is not None:
-            updated_list.append(f"{COL_PRIVATE_KEY} = ?")
-            values.append(private_key)
-        if password is not None:
-            updated_list.append(f"{COL_PASSWORD} = ?")
-            values.append(password)
-
-        updated_str = ", ".join(updated_list)
-        sql = f"UPDATE {TAB_PRIVATE_KEYS} SET {updated_str} WHERE {COL_ID} = ?"
-        result = DBManager.update(sql, private_key_id, values)
-        return result
-
-    def delete(self, private_key_id: int) -> bool:
-        """
-        Deletes a private key
-
-        :param private_key_id: ID of specific private key
-
-        :return: the result of whether the deletion was successful
-        """
-
-        sql = f"DELETE FROM {TAB_PRIVATE_KEYS} WHERE {COL_ID} = ?"
-        result = DBManager.delete(sql, private_key_id, False)
-        return result

src/tests/impl/db_manager/conftest.py

@@ -1 @@
-from src.impl.db_manager import DBManager
-import pytest
-
-
[e-mail]
-def repository():
-    return DBManager()

src/tests/impl/db_manager/transaction_certificates.py

@@ -1 @@
-import pytest
-from src.constants import *
-
-
-def test_create_first(repository):
-    sql = f"INSERT INTO {TAB_PRIVATE_KEYS} ({COL_PRIVATE_KEY},{COL_PASSWORD}) VALUES (?,?)"
-    repository.create(sql, ["test_pk_1", "test_pwd_1"])
-
-    sql = f"SELECT * FROM {TAB_PRIVATE_KEYS} WHERE {COL_PRIVATE_KEY} LIKE 'test%'"
-    rows_private_keys = repository.read(sql)
-
-    sql = f"INSERT INTO {TAB_CERTIFICATES} ({COL_COMMON_NAME},{COL_VALID_FROM},{COL_VALID_TO},{COL_PEM_DATA}," \
-          f"{COL_PRIVATE_KEY_ID},{COL_TYPE_ID},{COL_PARENT_ID}) VALUES (?,?,?,?,?,?,?)"
-    check_create = repository.create(sql, ["test_common_name", "test_valid_from", "test_valid_to", "test_pem_data",
-                                           rows_private_keys[0][1], CA_ID, -1])
-
-    sql = f"SELECT * FROM {TAB_CERTIFICATES} WHERE {COL_COMMON_NAME} LIKE 'test%'"
-    rows_select = repository.read(sql)
-
-    assert check_create
-    assert rows_select[0][1] == "test_common_name" \
-           and rows_select[0][2] == "test_valid_from" \
-           and rows_select[0][3] == "test_valid_to" \
-           and rows_select[0][4] == "test_pem_data" \
-           and rows_select[0][5] == rows_private_keys[0][1] \
-           and rows_select[0][6] == ROOT_CA_ID \
-           and rows_select[0][7] == -1
-
-
-def test_update(repository):
-    sql = f"SELECT * FROM {TAB_CERTIFICATES} WHERE {COL_COMMON_NAME} LIKE 'test%'"
-    rows_select = repository.read(sql)
-
-    sql = f"UPDATE {TAB_CERTIFICATES} SET {COL_PARENT_ID} = ? WHERE {COL_ID} = ?"
-    check_update: bool = repository.update(sql, rows_select[0][0], [rows_select[0][0]])
-
-    sql = f"SELECT * FROM {TAB_CERTIFICATES} WHERE {COL_COMMON_NAME} LIKE 'test%'"
-    rows_select = repository.read(sql)
-
-    assert check_update
-    assert rows_select[0][7] == rows_select[0][0]
-
-
-def test_delete(repository):
-    sql = f"SELECT * FROM {TAB_CERTIFICATES} WHERE {COL_COMMON_NAME} LIKE 'test%'"
-    rows_select = repository.read(sql)
-
-    i = 1
-    print("\n")
-    for row in rows_select:
-        sql = f"DELETE FROM {TAB_CERTIFICATES} WHERE {COL_ID} = ?"
-        check_delete = repository.delete(sql, row[0], False)
-        assert check_delete
-
-    sql = f"SELECT * FROM {TAB_CERTIFICATES} WHERE {COL_COMMON_NAME} LIKE 'test%'"
-    rows_select = repository.read(sql)
-
-    assert len(rows_select) == 0

src/tests/impl/db_manager/transaction_private_keys.py

@@ -1 @@
-import pytest
-from src.constants import *
-
-
-def test_create_first(repository):
-    sql = f"INSERT INTO {TAB_PRIVATE_KEYS} ({COL_PRIVATE_KEY},{COL_PASSWORD}) VALUES (?,?)"
-    check_create: bool = repository.create(sql, ["test_pk_1", "test_pwd_1"])
-
-    sql = f"SELECT * FROM {TAB_PRIVATE_KEYS} WHERE {COL_PRIVATE_KEY} LIKE 'test%'"
-    rows_select = repository.read(sql)
-
-    assert check_create
-    assert rows_select[0][1] == "test_pk_1" and rows_select[0][2] == "test_pwd_1"
-
-
-def test_create_count(repository):
-    sql = f"INSERT INTO {TAB_PRIVATE_KEYS} ({COL_PRIVATE_KEY},{COL_PASSWORD}) VALUES (?,?)"
-    check_create: bool = repository.create(sql, ["test_pk_1", "test_pwd_1"])
-
-    sql = f"SELECT * FROM {TAB_PRIVATE_KEYS} WHERE {COL_PRIVATE_KEY} LIKE 'test%'"
-    rows_select = repository.read(sql)
-
-    assert check_create
-    assert len(rows_select) == 2
-
-
-def test_update(repository):
-    sql = f"SELECT * FROM {TAB_PRIVATE_KEYS} WHERE {COL_PRIVATE_KEY} LIKE 'test%'"
-    rows_select = repository.read(sql)
-
-    sql = f"UPDATE {TAB_PRIVATE_KEYS} SET {COL_PASSWORD} = ? WHERE {COL_ID} = ?"
-    check_update: bool = repository.update(sql, rows_select[0][0], ["test_pwd_updated"])
-
-    sql = f"SELECT * FROM {TAB_PRIVATE_KEYS} WHERE {COL_PRIVATE_KEY} LIKE 'test%'"
-    rows_select = repository.read(sql)
-
-    assert check_update
-    assert rows_select[0][2] == "test_pwd_updated"
-
-
-def test_delete(repository):
-    sql = f"SELECT * FROM {TAB_PRIVATE_KEYS} WHERE {COL_PRIVATE_KEY} LIKE 'test%'"
-    rows_select = repository.read(sql)
-
-    i = 1
-    print("\n")
-    for row in rows_select:
-        sql = f"DELETE FROM {TAB_PRIVATE_KEYS} WHERE {COL_ID} = ?"
-        check_delete = repository.delete(sql, row[0], False)
-        assert check_delete
-
-    sql = f"SELECT * FROM {TAB_PRIVATE_KEYS} WHERE {COL_PRIVATE_KEY} LIKE 'test%'"
-    rows_select = repository.read(sql)
-
-    assert len(rows_select) == 0