Revize 2a90f4fd

Re #8472 - Implemented get_certificates method and added an integration test validating it

Added fixtures providing a unique DB connection for each test

src/services/certificate_service.py

@@ -52 +52 @@
         certificate = Certificate(-1, common_name, not_before_formatted, not_after_formatted, cert_pem,
                                   private_key_id, cert_type, parent_id, usages)
 
-        print(f"CW: {parent_id} {certificate.parent_id}")
         return certificate
 
     # TODO config parameter present in class diagram but not here (unused)
@@ -111 +110 @@
                                                         issuer_key_pass=issuer_key.password, extensions=extensions,
                                                         days=days)
 
-        print(f"test cert {issuer_cert.certificate_id}")
         # wrap the generated certificate using Certificate class
         certificate = self.__create_wrapper(cert_pem, subject_key.private_key_id, subject.common_name, usages,
                                             issuer_cert.certificate_id, CERTIFICATE_ID)
 
-        print(f"parent cert {certificate.parent_id}")
-
         created_id = self.certificate_repository.create(certificate)
 
         certificate.certificate_id = created_id
 
-        print(f"parent cert {certificate.parent_id}")
-
         return certificate
 
     def get_certificate(self, unique_id: int) -> Certificate:
         return self.certificate_repository.read(unique_id)
+
+    def get_certificates(self, cert_type=None) -> Certificate:
+        return self.certificate_repository.read_all(cert_type)

tests/integration_tests/services/certificate_key_service_test.py

@@ -50 +50 @@
     assert cert.usages == cert_loaded.usages
 
 
-def test_create_and_get_cert(private_key_service, certificate_service):
+def test_create_and_get_inter_cert(private_key_service, certificate_service):
     root_ca_private_key = private_key_service.create_new_key(passphrase="foobar")
     inter_ca_private_key = private_key_service.create_new_key()
 
@@ -121 +121 @@
     assert cert.parent_id == inter_ca_cert.certificate_id
     assert cert_loaded.parent_id == inter_ca_cert.certificate_id
     assert cert.usages == cert_loaded.usages
+
+
+def test_get_certificates(private_key_service_unique, certificate_service_unique):
+    root_ca_private_key = private_key_service_unique.create_new_key(passphrase="foobar")
+    inter_ca_private_key = private_key_service_unique.create_new_key(passphrase="barfoo")
+    end_cert_private_key = private_key_service_unique.create_new_key(passphrase="foofoo")
+
+    root_ca_cert = certificate_service_unique.create_root_ca(root_ca_private_key,
+                                                             Subject(common_name="RootFoo",
+                                                                     organization_unit="Department of Foo"))
+
+    inter_ca_cert = certificate_service_unique.create_ca(inter_ca_private_key, Subject(common_name="Intermediate CA"),
+                                                         root_ca_cert,
+                                                         root_ca_private_key, usages={SSL_ID: True})
+
+    cert = certificate_service_unique.create_end_cert(end_cert_private_key,
+                                                      Subject("Foo Child", email_address="[e-mail]"), inter_ca_cert,
+                                                      inter_ca_private_key, usages={AUTHENTICATION_ID: True})
+
+    all_certs = certificate_service_unique.get_certificates()
+    assert 3 == len(all_certs)
+    assert "RootFoo" == all_certs[0].common_name
+    assert "Intermediate CA" == all_certs[1].common_name
+    assert "Foo Child" == all_certs[2].common_name
+
+    assert 1 == len(certificate_service_unique.get_certificates(cert_type=ROOT_CA_ID))
+    assert 1 == len(certificate_service_unique.get_certificates(cert_type=INTERMEDIATE_CA_ID))
+    assert 1 == len(certificate_service_unique.get_certificates(cert_type=CERTIFICATE_ID))

tests/integration_tests/services/conftest.py

@@ -68 +68 @@
 @pytest.fixture
 def certificate_service(certificate_repository, cryptography_service):
     return CertificateService(cryptography_service, certificate_repository)
+
+
+# TODO improve this (lots of duplicated code) - some test cases need a DB connection that is not shared with other tests
[e-mail]
+def connection_unique():
+    print("Creating a new unique SQLITE connection to the test DB")
+    test_db_file = "test_unique.sqlite"
+    connection: Connection = sqlite3.connect(test_db_file)
+
+    # yield the created connection
+    yield connection
+
+    # after tests have finished delete the created db file
+    try:
+        print("Deleting the unique test DB")
+        os.unlink(test_db_file)
+    except FileNotFoundError:
+        print(f"Could not delete {test_db_file} file containing the unique test DB")
+        pass
+
+
[e-mail]
+def cursor_unique(connection_unique):
+    cursor = connection_unique.cursor()
+
+    # execute db initialisation script
+    cursor.executescript(SCHEMA_SQL)
+
+    # insert default values
+    cursor.executescript(DEFAULT_VALUES_SQL)
+
+    return cursor
+
+
+# scope defaults to "function" which means that the fixture is run once per test (function)
[e-mail]
+def certificate_repository_unique(connection_unique, cursor_unique):
+    return CertificateRepository(connection_unique, cursor_unique)
+
+
[e-mail]
+def private_key_repository_unique(connection_unique, cursor_unique):
+    return PrivateKeyRepository(connection_unique, cursor_unique)
+
+
[e-mail]
+def cryptography_service_unique():
+    return CryptographyService()
+
+
[e-mail]
+def private_key_service_unique(private_key_repository_unique, cryptography_service_unique):
+    return KeyService(cryptography_service_unique, private_key_repository_unique)
+
+
[e-mail]
+def certificate_service_unique(certificate_repository_unique, cryptography_service_unique):
+    return CertificateService(cryptography_service_unique, certificate_repository_unique)