Revize d65b022d

Re #8578 - Cover private_key_repository tests
- created tests for functions create, read, read_all, update, delete
- (+) edited certificate_repository to the same stylistics

src/dao/certificate_repository.py

@@ -1 +1 @@
 from typing import Dict, List
 from sqlite3 import Connection, Cursor, Error
 
+from src.exceptions.database_exception import DatabaseException
 from src.model.certificate import Certificate
 from src.constants import *
 
@@ -70 +71 @@
                         self.cursor.execute(sql, values)
                         self.connection.commit()
         except Error as e:
-            return e
+            raise DatabaseException(e)
 
         return last_id
 
@@ -114 +115 @@
                                                    certificate_row[5],
                                                    certificate_row[6])
         except Error as e:
-            return e
+            raise DatabaseException(e)
 
-        if len(certificate_row) > 0:
-            return certificate
-        else:
-            return None
+        return certificate
 
     def read_all(self, filter_type: int = None):
         """
@@ -166 +164 @@
                                                 certificate_row[5],
                                                 certificate_row[6]))
         except Error as e:
-            return e
+            raise DatabaseException(e)
 
-        if len(certificates) > 0:
-            return certificates
-        else:
-            return None
+        return certificates
 
     def update(self, certificate_id: int, certificate: Certificate):
         """
@@ -228 +223 @@
                     self.cursor.execute(sql, values)
                     self.connection.commit()
         except Error as e:
-            return e
+            raise DatabaseException(e)
 
-        if self.cursor.rowcount > 0:
-            return True
-
-        return False
+        return self.cursor.rowcount > 0
 
     def delete(self, certificate_id: int):
         """
@@ -251 +243 @@
             self.cursor.execute(sql, values)
             self.connection.commit()
         except Error as e:
-            return e
+            raise DatabaseException(e)
 
         return self.cursor.rowcount > 0
 
@@ -285 +277 @@
             self.cursor.execute(sql, values)
             self.connection.commit()
         except Error as e:
-            return e
-
-        if self.cursor.rowcount > 0:
-            return True
+            raise DatabaseException(e)
 
-        return False
+        return self.cursor.rowcount > 0
 
     def clear_certificate_revocation(self, certificate_id: int):
         """
@@ -312 +301 @@
             self.cursor.execute(sql, values)
             self.connection.commit()
         except Error as e:
-            return e
+            raise DatabaseException(e)
 
-        return True
+        return self.cursor.rowcount > 0
 
     def get_all_revoked_by(self, certificate_id: int):
         """
@@ -359 +348 @@
                                                 certificate_row[5],
                                                 certificate_row[6]))
         except Error as e:
-            return e
+            raise DatabaseException(e)
 
-        if len(certificates) > 0:
-            return certificates
-        else:
-            return None
+        return certificates
 
     def get_all_issued_by(self, certificate_id: int):
         """
@@ -409 +395 @@
                                                 certificate_row[5],
                                                 certificate_row[6]))
         except Error as e:
-            return e
+            raise DatabaseException(e)
 
-        if len(certificates) > 0:
-            return certificates
-        else:
-            return None
+        return certificates

src/dao/private_key_repository.py

@@ -1 +1 @@
 from sqlite3 import Connection, Cursor, Error
 from typing import List
 
+from src.exceptions.database_exception import DatabaseException
 from src.model.private_key import PrivateKey
 from src.constants import *
 
@@ -38 +39 @@
             last_id = self.cursor.lastrowid
             self.connection.commit()
         except Error as e:
-            print(e)
-            return None
+            raise DatabaseException(e)
 
         return last_id
 
@@ -59 +59 @@
             self.cursor.execute(sql, values)
             private_key_row = self.cursor.fetchone()
 
+            if private_key_row is None:
+                return None
+
             private_key: PrivateKey = PrivateKey(private_key_row[0],
                                                  private_key_row[1],
                                                  private_key_row[2])
         except Error as e:
-            print(e)
-            return None
+            raise DatabaseException(e)
 
-        if len(private_key_row) > 0:
-            return private_key
-        else:
-            return None
+        return private_key
 
     def read_all(self):
         """
@@ -89 +88 @@
                                                private_key_row[1],
                                                private_key_row[2]))
         except Error as e:
-            print(e)
-            return None
+            raise DatabaseException(e)
 
-        if len(private_keys) > 0:
-            return private_keys
-        else:
-            return None
+        return private_keys
 
     def update(self, private_key_id: int, private_key: PrivateKey) -> bool:
         """
@@ -118 +113 @@
             self.cursor.execute(sql, values)
             self.connection.commit()
         except Error as e:
-            print(e)
-            return False
+            raise DatabaseException(e)
 
-        return True
+        return self.cursor.rowcount > 0
 
     def delete(self, private_key_id: int) -> bool:
         """
@@ -139 +133 @@
             self.cursor.execute(sql, values)
             self.connection.commit()
         except Error as e:
-            print(e)
-            return False
+            raise DatabaseException
 
         return self.cursor.rowcount > 0

src/exceptions/database_exception.py

@@ +1 @@
+class DatabaseException(Exception):
+    pass

src/model/private_key.py

@@ -2 +2 @@
     def __init__(self,
                  private_key_id: int,
                  private_key: str,
-                 password: str):
+                 password: str = None):
         self.private_key_id = private_key_id
         self.private_key = private_key
         self.password = password

tests/integration_tests/dao/private_key_repository_test.py

@@ +1 @@
+import pytest
+from src.constants import *
+from src.model.certificate import Certificate
+from src.model.private_key import PrivateKey
+
+
+def test_create(private_key_repository):
+    private_key_1 = PrivateKey(private_key_id=-1, private_key="test_private_key_1", password="test_password_1")
+    private_key_2 = PrivateKey(private_key_id=-1, private_key="test_private_key_2")
+
+    assert private_key_1.private_key_id == -1
+    assert private_key_1.private_key == "test_private_key_1"
+    assert private_key_1.password == "test_password_1"
+
+    assert private_key_2.private_key_id == -1
+    assert private_key_2.private_key == "test_private_key_2"
+    assert private_key_2.password is None
+
+    private_key_1_id = private_key_repository.create(private_key=private_key_1)
+    private_key_2_id = private_key_repository.create(private_key=private_key_2)
+
+    assert type(private_key_1_id) is int
+    assert type(private_key_2_id) is int
+
+    private_key_1 = private_key_repository.read(private_key_id=private_key_1_id)
+    private_key_2 = private_key_repository.read(private_key_id=private_key_2_id)
+
+    assert private_key_1.private_key_id == private_key_1_id
+    assert private_key_1.private_key == "test_private_key_1"
+    assert private_key_1.password == "test_password_1"
+
+    assert private_key_2.private_key_id == private_key_2_id
+    assert private_key_2.private_key == "test_private_key_2"
+    assert private_key_2.password is None
+
+
+def test_read(private_key_repository):
+    private_key = PrivateKey(private_key_id=-1, private_key="test_private_key", password="test_password")
+
+    private_key_id = private_key_repository.create(private_key=private_key)
+
+    assert type(private_key_id) is int
+
+    private_key = private_key_repository.read(private_key_id=private_key_id)
+
+    assert private_key.private_key_id == private_key_id
+    assert private_key.private_key == "test_private_key"
+    assert private_key.password == "test_password"
+
+
+def test_read_invalid_id(private_key_repository):
+    none_id = 100
+    private_key = private_key_repository.read(private_key_id=none_id)
+
+    assert private_key is None
+
+
+def test_read_all(private_key_repository):
+    private_keys = private_key_repository.read_all()
+
+    assert len(private_keys) == 0
+
+    private_key_1 = PrivateKey(private_key_id=-1, private_key="test_private_key_1", password="test_password_1")
+    private_key_2 = PrivateKey(private_key_id=-1, private_key="test_private_key_2")
+
+    private_key_1_id = private_key_repository.create(private_key=private_key_1)
+    private_key_2_id = private_key_repository.create(private_key=private_key_2)
+
+    assert type(private_key_1_id) is int
+    assert type(private_key_2_id) is int
+
+    private_keys = private_key_repository.read_all()
+
+    assert len(private_keys) == 2
+
+    assert private_keys[0].private_key_id == private_key_1_id
+    assert private_keys[0].private_key == "test_private_key_1"
+    assert private_keys[0].password == "test_password_1"
+
+    assert private_keys[1].private_key_id == private_key_2_id
+    assert private_keys[1].private_key == "test_private_key_2"
+    assert private_keys[1].password is None
+
+
+def test_update(private_key_repository):
+    private_key = PrivateKey(private_key_id=-1, private_key="test_private_key")
+
+    private_key_id = private_key_repository.create(private_key=private_key)
+
+    assert type(private_key_id) is int
+
+    private_key = private_key_repository.read(private_key_id=private_key_id)
+
+    assert private_key.private_key_id == private_key_id
+    assert private_key.private_key == "test_private_key"
+    assert private_key.password is None
+
+    private_key.password = "test_password"
+
+    check_updated = private_key_repository.update(private_key_id=private_key_id, private_key=private_key)
+
+    assert check_updated is True
+
+    private_key = private_key_repository.read(private_key_id=private_key_id)
+
+    assert private_key.private_key_id == private_key_id
+    assert private_key.private_key == "test_private_key"
+    assert private_key.password == "test_password"
+
+
+def test_update_invalid_id(private_key_repository):
+    private_key = PrivateKey(private_key_id=-1, private_key="test_private_key")
+
+    private_key_id = private_key_repository.create(private_key=private_key)
+
+    assert type(private_key_id) is int
+
+    private_key = private_key_repository.read(private_key_id=private_key_id)
+
+    assert private_key.private_key_id == private_key_id
+    assert private_key.private_key == "test_private_key"
+    assert private_key.password is None
+
+    private_key.password = "test_password"
+
+    none_id = 100
+    check_updated = private_key_repository.update(private_key_id=none_id, private_key=private_key)
+
+    assert check_updated is False
+
+def test_delete(private_key_repository):
+    private_key_1 = PrivateKey(private_key_id=-1, private_key="test_private_key_1", password="test_password_1")
+    private_key_2 = PrivateKey(private_key_id=-1, private_key="test_private_key_2")
+
+    private_key_1_id = private_key_repository.create(private_key=private_key_1)
+    private_key_2_id = private_key_repository.create(private_key=private_key_2)
+
+    assert type(private_key_1_id) is int
+    assert type(private_key_2_id) is int
+
+    private_keys = private_key_repository.read_all()
+
+    assert len(private_keys) == 2
+
+    assert private_keys[0].private_key_id == private_key_1_id
+    assert private_keys[0].private_key == "test_private_key_1"
+    assert private_keys[0].password == "test_password_1"
+
+    assert private_keys[1].private_key_id == private_key_2_id
+    assert private_keys[1].private_key == "test_private_key_2"
+    assert private_keys[1].password is None
+
+    check_deleted = private_key_repository.delete(private_key_id=private_key_1_id)
+
+    private_keys = private_key_repository.read_all()
+
+    assert len(private_keys) == 1
+
+    assert private_keys[0].private_key_id == private_key_2_id
+    assert private_keys[0].private_key == "test_private_key_2"
+    assert private_keys[0].password is None