Merge remote-tracking branch 'origin/feat/crl'

conftest.py

@@ -1,2 +1,5 @@
+from tests.setup import run_all
+
+
 def pytest_sessionstart(session):
-    pass
+    run_all()

tests/setup/__init__.py

@@ -0,0 +1,7 @@
+from . import service
+import pook
+
+
+@pook.on
+def run_all():
+    service.run_setup()

tests/setup/service.py

@@ -0,0 +1,62 @@
+import pook
+from tests.service.utils import open_json
+from verificac19 import service
+from verificac19.service._settings import (
+    DSC_URL,
+    STATUS_URL,
+    SETTINGS_URL,
+    CHECK_CRL_URL,
+    DOWNLOAD_CRL_URL,
+)
+
+
+def _setup_mock_for_settings():
+    settings_data = open_json("settings.json")
+    pook.get(SETTINGS_URL, reply=200, response_json=settings_data)
+
+
+def _setup_mock_for_dsc_certificates():
+    FINISHED = "finished"
+    pook.get(STATUS_URL, reply=200, response_json=open_json("dsc_whitelist.json"))
+    dsc_validation: list = open_json("dsc_validation.json")
+    dsc_validation.append(FINISHED)
+    for index, dsc in enumerate(dsc_validation):
+        kwargs = {}
+        headers = {"content-type": "text/plain"}
+        kwargs["reply"] = 200
+        if 0 < index:
+            headers["X-RESUME-TOKEN"] = str(index)
+        kwargs["headers"] = headers
+        if dsc == FINISHED:
+            kwargs["reply"] = 204
+            pook.get(DSC_URL, **kwargs)
+            continue
+
+        kwargs["response_body"] = dsc["raw_data"]
+        kwargs["response_headers"] = {
+            "X-KID": dsc["kid"],
+            "X-RESUME-TOKEN": str(index + 1),
+        }
+        pook.get(DSC_URL, **kwargs)
+
+
+def _setup_mock_for_crl():
+    pook.get(CHECK_CRL_URL, reply=200, response_json=open_json("CRL-check-v1.json"))
+    pook.get(
+        f"{DOWNLOAD_CRL_URL}?chunk=1",
+        reply=200,
+        response_json=open_json("CRL-v1-c1.json"),
+    )
+    pook.get(
+        f"{DOWNLOAD_CRL_URL}?chunk=2",
+        reply=200,
+        response_json=open_json("CRL-v1-c2.json"),
+    )
+
+
+def run_setup():
+    _setup_mock_for_settings()
+    _setup_mock_for_dsc_certificates()
+    _setup_mock_for_crl()
+    service.clear_all_cache()
+    service.update_all()

tests/3 → tests/test_service.py

@@ -10,52 +10,53 @@
     DOWNLOAD_CRL_URL,
 )
 
-class TestService:
 
+class TestService:
     def open_json(self, name: str) -> dict:
         path = os.path.join("tests", "data", "mock_request", name)
         with open(path, "r") as file:
             return json.load(file)
 
     @pook.on
     def test_dsc_settings(self):
-        self.setup_mock_for_settings()
+        settings_data = self.open_json("settings.json")
+        self.setup_mock_for_settings(settings_data)
         self.setup_mock_for_dsc_certificates()
         service.update_settings()
         for setting in settings_data:
-            st_name = setting['name']
-            st_type = setting['type']
+            st_name = setting["name"]
+            st_type = setting["type"]
             assert service.get_setting(st_name, st_type) == setting
 
-    def setup_mock_for_settings(self):
-        settings_data = self.open_json("settings.json")
+    def setup_mock_for_settings(self, settings_data):
         pook.get(SETTINGS_URL, reply=200, response_json=settings_data)
 
     def setup_mock_for_dsc_certificates(self):
-        FINISHED = 'finished'
-        pook.get(STATUS_URL, reply=200, response_json=self.open_json("dsc_whitelist.json"))
+        FINISHED = "finished"
+        pook.get(
+            STATUS_URL, reply=200, response_json=self.open_json("dsc_whitelist.json")
+        )
         dsc_validation: list = self.open_json("dsc_validation.json")
         dsc_validation.append(FINISHED)
         for index, dsc in enumerate(dsc_validation):
             kwargs = {}
             headers = {"content-type": "text/plain"}
-            kwargs['reply'] = 200
+            kwargs["reply"] = 200
             if 0 < index:
-                headers["X-RESUME-TOKEN"] = str(index+1)
-            kwargs['headers'] = headers
+                headers["X-RESUME-TOKEN"] = str(index + 1)
+            kwargs["headers"] = headers
             if dsc == FINISHED:
-                kwargs['reply'] = 204
+                kwargs["reply"] = 204
                 pook.get(DSC_URL, **kwargs)
                 continue
 
-            kwargs['response_body'] = dsc["raw_data"]
-            kwargs['response_headers'] = {
+            kwargs["response_body"] = dsc["raw_data"]
+            kwargs["response_headers"] = {
                 "X-KID": dsc["kid"],
                 "X-RESUME-TOKEN": str(index + 1),
             }
             pook.get(DSC_URL, **kwargs)
 
-
     @pook.on
     def mtest_dsc_certificates(self):
         self.setup_mock_for_dsc_certificates()
@@ -64,20 +65,23 @@ def mtest_dsc_certificates(self):
         service.update_dsc()
         dsc_list = self.open_json("dsc_validation.json")
         for dsc in dsc_list:
-            kid = dsc['kid']
-            data = dsc['raw_data']
+            kid = dsc["kid"]
+            data = dsc["raw_data"]
             stored_dsc = service.get_dsc(kid)
             if kid in not_whitelisted_dsc:
                 assert stored_dsc is None
                 continue
 
             assert stored_dsc == data
 
-
     @pook.on
     def test_update_all(self):
         pook.get(SETTINGS_URL, reply=200, response_json=self.open_json("settings.json"))
-        pook.get(STATUS_URL, reply=200, response_json=self.open_json("certificate_status.json"))
+        pook.get(
+            STATUS_URL,
+            reply=200,
+            response_json=self.open_json("certificate_status.json"),
+        )
         dsc_validation = self.open_json("dsc_validation.json")
         for index, dsc in enumerate(dsc_validation):
             header = {"content-type": "text/plain"}
@@ -100,7 +104,9 @@ def test_update_all(self):
 
         service.clear_all_cache()
 
-        pook.get(CHECK_CRL_URL, reply=200, response_json=self.open_json("CRL-check-v1.json"))
+        pook.get(
+            CHECK_CRL_URL, reply=200, response_json=self.open_json("CRL-check-v1.json")
+        )
         pook.get(
             f"{DOWNLOAD_CRL_URL}?chunk=1",
             reply=200,
@@ -113,4 +119,3 @@ def test_update_all(self):
         )
 
         service.update_all()
-