Revize 1e07432d

Re #8476 - Implemented `list_certificates` with unit tests.

swagger_server/controllers/certificates_controller.py

@@ +1 @@
+from datetime import datetime
+
 import connexion
 import six
 
@@ -8 +10 @@
 from src.services.cryptography import CryptographyService           # TODO not the Controller's responsibility. 2
 from sqlite3 import Connection                                      # TODO not the Controller's responsibility. 3
 from src.constants import DICT_USAGES, CA_ID, \
-    DATABASE_FILE_LOCATION                                          # TODO DATABASE_FILE - not the Controller's
+    DATABASE_FILE_LOCATION, SSL_ID, SIGNATURE_ID, AUTHENTICATION_ID, \
+    DATETIME_FORMAT  # TODO DATABASE_FILE - not the Controller's
                                                                     #  responsibility. 4
 from src.services.key_service import KeyService
-from swagger_server.models import CertificateRequest
+from swagger_server.models import CertificateRequest, CertificateListItem, CAUsage, IssuerListItem
 
 from swagger_server.models.certificate import Certificate  # noqa: E501
 from swagger_server.models.certificate_list_response import CertificateListResponse  # noqa: E501
@@ -31 +34 @@
                                                                         #  the one corresponding to the connection.
                                                                         #  The cursor can always be generated from the
                                                                         #  connection instance.
+GENERAL_ERROR = "An error occured during processing of the request."
+CORRUPTED_DATABASE = "Internal server error (corrupted database)."
 
 __ = CryptographyService()                                              # TODO not the Controller's responsibility. 6
 CERTIFICATE_SERVICE = CertificateService(__, CertificateRepository(None, None))
@@ -104 +109 @@
             if issuer_key is None:
                 return ErrorResponse(
                     success=False,
-                    data="Internal server error (corrupted database)."
+                    data=CORRUPTED_DATABASE
                 ), 400
 
             f = CERTIFICATE_SERVICE.create_ca if CA_ID in usages_dict and usages_dict[CA_ID] else \
@@ -130 +135 @@
                 data="The certificate could not have been created."
             ), 400
     else:
-        return 400
+        return ErrorResponse(success=False, data="The request must be JSON-formatted."), 400
 
 
 def get_certificate_by_id(id):  # noqa: E501
@@ -175 +180 @@
     """
     setup()                                                             # TODO remove after issue fixed
 
-    if connexion.request.is_json:
-        filtering = Filtering.from_dict(connexion.request.get_json())  # noqa: E501
-        print(filtering)
-        return 200
+    key_map = {CA_ID: 'ca', SSL_ID: 'ssl', SIGNATURE_ID: 'digital_signature', AUTHENTICATION_ID: 'authentication'}
+
+    if len(connexion.request.data) == 0:
+        certs = CERTIFICATE_SERVICE.get_certificates()
+        if certs is None:
+            return ErrorResponse(success=False, data=GENERAL_ERROR), 500
+        elif len(certs) == 0:
+            return ErrorResponse(success=False, data="No certificates found."), 204
+        else:
+            ret = []
+            for c in certs:
+                c_issuer = CERTIFICATE_SERVICE.get_certificate(c.parent_id)
+                if c_issuer is None:
+                    return ErrorResponse(success=False, data=CORRUPTED_DATABASE)
+
+                ret.append(
+                    CertificateListItem(
+                        id=c.certificate_id,
+                        cn=c.common_name,
+                        not_before=datetime.strptime(c.valid_from, DATETIME_FORMAT).date(),
+                        not_after=datetime.strptime(c.valid_to, DATETIME_FORMAT).date(),
+                        usage=CAUsage(**{key_map[k]: v for k, v in c.usages.items()}),
+                        issuer=IssuerListItem(
+                            id=c_issuer.certificate_id,
+                            cn=c_issuer.common_name
+                        )
+                    )
+                )
+            return CertificateListResponse(success=True, data=ret)
     else:
-        return 400
+        # TODO fix filtering issue (somehow)
+        return ErrorResponse(success=False, data="The request must be JSON-formatted."), 400
 
 
 def get_certificate_root_by_id(id):  # noqa: E501