chore(GH): add pytest-retry for operator tests (#428)

* fix: tests

Signed-off-by: Michael Schilonka <michael@blueshoe.io>

---------

Signed-off-by: Michael Schilonka <michael@blueshoe.io>
Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: Robert Stein <robert@blueshoe.io>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

.github/workflows/python-tester.yaml

@@ -408,6 +408,10 @@ jobs:
     name: Test Gefyra Operator
     runs-on: ubuntu-latest
     steps:
+      - name: Set up Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: '3.9'
       - name: Checkout
         uses: actions/checkout@v4
       - name: Install Poetry
@@ -423,7 +427,7 @@ jobs:
         run: curl --silent --fail https://raw.githubusercontent.com/k3d-io/k3d/main/install.sh | TAG=v5.5.1 bash
       - name: Run PyTest
         working-directory: ./operator
-        run: poetry run coverage run -m pytest -x
+        run: poetry run coverage run -m pytest -s -x --retries 5
       - name: Upload coverage data to coveralls.io
         if: github.repository == 'gefyrahq/gefyra'
         working-directory: ./operator

client/tests/conftest.py

@@ -66,17 +66,26 @@ def operator(k3d: AClusterManager, operator_image, stowaway_image):
             if event["reason"] == "Gefyra-Ready":
                 not_found = False
     if not_found:
-        print(
-            k3d.kubectl(
-                ["describe", "deployment", "-n", "gefyra", "gefyra-operator"],
-                as_dict=False,
+        try:
+            print(
+                k3d.kubectl(
+                    ["describe", "deployment", "-n", "gefyra", "gefyra-operator"],
+                    as_dict=False,
+                )
             )
-        )
-        print(k3d.kubectl(["logs", "-n", "gefyra", "deployment", "gefyra-operator"]))
-        print(
-            k3d.kubectl(
-                ["logs", "-n", "gefyra", "deployment", "gefyra-operator-webhook"]
+            print(
+                k3d.kubectl(
+                    ["logs", "-n", "gefyra", "deployment", "gefyra-operator"],
+                    as_dict=False,
+                )
             )
-        )
+            print(
+                k3d.kubectl(
+                    ["logs", "-n", "gefyra", "deployment", "gefyra-operator-webhook"],
+                    as_dict=False,
+                )
+            )
+        except Exception as e:
+            print(e)
         raise Exception("Gefyra-Ready event not found")
     yield k3d

client/tests/e2e/base.py

@@ -814,7 +814,7 @@ def test_o_reflect_occupied_port(self):
         with self.assertRaises(RuntimeError) as rte:
             reflect(**params)
 
-        self.assertIn("occupied", str(rte.exception))
+        self.assertIn("allocated", str(rte.exception))
         self._stop_container(container=container_name)
         self.gefyra_down()
         self.assert_namespace_not_found("gefyra")

operator/gefyra/connection/abstract.py

@@ -48,7 +48,7 @@ def remove_peer(self, peer_id: str):
         raise NotImplementedError
 
     @abstractmethod
-    def get_peer_config(self, peer_id: str) -> dict[str, str]:
+    def get_peer_config(self, peer_id: str) -> Dict[str, str]:
         """
         Returns a dict of configuration values for the peer to be stored in the Peer CRD
         """