Revize b2a9b83d

Re #8471 - Tested direct DB communication
- implemented tests for Certificates and PrivateKeys tables via DBManager

src/constants.py

@@ -1 @@
-DATABASE_FILE = r"C:\sqlite\db\pythonsqlite.db"
+DATABASE_FILE = "../../../../db/database_sqlite.db"
 
 ROOT_CA_ID = 1
 INTERMEDIATE_CA_ID = 2

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