Feature/gefyra integration

requirements.txt

@@ -21,3 +21,4 @@ semantic-version~=2.9.0
 kubernetes>=11.0.0,<22.0.0
 retrying~=1.3.3
 oic==1.3.0
+gefyra~=0.8.1

setup.py

@@ -39,6 +39,7 @@
         "python-slugify>=5.0.2,<6.2.0",
         "click-didyoumean~=0.3.0",
         "requests-toolbelt~=0.9.1",
+        "gefyra~=0.8.1",
     ],
     python_requires="~=3.7",
     packages=find_packages(),

tests/cluster/storage/test_cluster_storage.py

@@ -0,0 +1,41 @@
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from uuid import UUID
+
+from unikube.cluster.storage.cluster_storage import ClusterStorage
+
+
+class ClusterStorageTest(unittest.TestCase):
+    def setUp(self):
+        self.temporary_path_object = tempfile.TemporaryDirectory()
+        self.temporary_path = self.temporary_path_object.name
+
+    def tearDown(self):
+        self.temporary_path_object.cleanup()
+
+    def test_missing_id(self):
+        with self.assertRaises(Exception):
+            _ = ClusterStorage(file_path=self.temporary_path)
+
+    def test_save(self):
+        cluster_storage = ClusterStorage(file_path=self.temporary_path, id=UUID("00000000-0000-0000-0000-000000000000"))
+        cluster_storage.save()
+
+        file = Path(os.path.join(cluster_storage.file_path, cluster_storage.file_name))
+
+        self.assertTrue(file.exists())
+        self.assertEqual(UUID("00000000-0000-0000-0000-000000000000"), cluster_storage.id)
+
+    def test_load(self):
+        cluster_storage_01 = ClusterStorage(
+            file_path=self.temporary_path, id=UUID("00000000-0000-0000-0000-000000000000")
+        )
+        cluster_storage_01.name = "test"
+        cluster_storage_01.save()
+
+        cluster_storage_02 = ClusterStorage(
+            file_path=self.temporary_path, id=UUID("00000000-0000-0000-0000-000000000000")
+        )
+        self.assertEqual("test", cluster_storage_02.name)

tests/console/test_input.py

@@ -70,7 +70,7 @@ def test_filter_none(self):
         filter_ = None
 
         choices_resolved = resolve_duplicates(choices=choices, identifiers=identifiers)
-        choices_filtered = filter_by_identifiers(choices=choices_resolved, identifiers=identifiers, filter=filter_)
+        choices_filtered = filter_by_identifiers(choices=choices_resolved, identifiers=identifiers, _filter=filter_)
         assert choices_filtered == [CHOICE_01, CHOICE_02]
 
     def test_filter_empty_list(self):
@@ -79,7 +79,7 @@ def test_filter_empty_list(self):
         filter_ = []
 
         choices_resolved = resolve_duplicates(choices=choices, identifiers=identifiers)
-        choices_filtered = filter_by_identifiers(choices=choices_resolved, identifiers=identifiers, filter=filter_)
+        choices_filtered = filter_by_identifiers(choices=choices_resolved, identifiers=identifiers, _filter=filter_)
         assert choices_filtered == []
 
     def test_filter_existing_01(self):
@@ -88,7 +88,7 @@ def test_filter_existing_01(self):
         filter_ = ["1"]
 
         choices_resolved = resolve_duplicates(choices=choices, identifiers=identifiers)
-        choices_filtered = filter_by_identifiers(choices=choices_resolved, identifiers=identifiers, filter=filter_)
+        choices_filtered = filter_by_identifiers(choices=choices_resolved, identifiers=identifiers, _filter=filter_)
         assert choices_filtered == [CHOICE_01]
 
     def test_filter_existing_02(self):
@@ -97,7 +97,7 @@ def test_filter_existing_02(self):
         filter_ = ["2"]
 
         choices_resolved = resolve_duplicates(choices=choices, identifiers=identifiers)
-        choices_filtered = filter_by_identifiers(choices=choices_resolved, identifiers=identifiers, filter=filter_)
+        choices_filtered = filter_by_identifiers(choices=choices_resolved, identifiers=identifiers, _filter=filter_)
         assert choices_filtered == ["choice"]
 
     def test_filter_non_existing(self):
@@ -106,7 +106,7 @@ def test_filter_non_existing(self):
         filter_ = ["3"]
 
         choices_resolved = resolve_duplicates(choices=choices, identifiers=identifiers)
-        choices_filtered = filter_by_identifiers(choices=choices_resolved, identifiers=identifiers, filter=filter_)
+        choices_filtered = filter_by_identifiers(choices=choices_resolved, identifiers=identifiers, _filter=filter_)
         assert choices_filtered == []
 
 

tests/gefyra/Dockerfile

@@ -0,0 +1,5 @@
+FROM ubuntu
+# run a server on port 8000
+RUN apt update && apt install -y iproute2 iputils-ping python3 traceroute wget curl
+COPY local.py local.py
+CMD python3 local.py

tests/gefyra/local.py

@@ -0,0 +1,48 @@
+#!/bin/env python
+
+import http.server
+import signal
+import socket
+import socketserver
+import sys
+from datetime import datetime
+
+if sys.argv[1:]:
+    port = int(sys.argv[1])
+else:
+    port = 8000
+
+
+class MyHttpRequestHandler(http.server.SimpleHTTPRequestHandler):
+    def do_GET(self):
+        self.send_response(200)
+        self.send_header("Content-type", "text/html")
+        self.end_headers()
+        hostname = socket.gethostname()
+        now = datetime.utcnow()
+        self.wfile.write(
+            bytes(f"<html><body><h1>Hello from Gefyra. It is {now} on {hostname}.</h1></body></html>".encode("utf-8"))
+        )
+
+
+my_handler = MyHttpRequestHandler
+server = socketserver.ThreadingTCPServer(("", port), my_handler)
+
+
+def signal_handler(signal, frame):
+    try:
+        if server:
+            server.server_close()
+    finally:
+        sys.exit(0)
+
+
+signal.signal(signal.SIGINT, signal_handler)
+try:
+    while True:
+        sys.stdout.flush()
+        server.serve_forever()
+except KeyboardInterrupt:
+    pass
+
+server.server_close()

tests/gefyra/unikube.yml

@@ -0,0 +1,15 @@
+# unikube switch configuration file
+apps:
+  local:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    deployment: buzzword-counter-web
+    container: buzzword-counter
+    command: python3 local.py
+    ports:
+      - 9000:8000
+    volumes:
+      - ./src:/app
+    env:
+      - DJANGO_DEBUG: "True"

tests/test_cache.py

@@ -0,0 +1,38 @@
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from uuid import UUID
+
+from unikube.cache import Cache
+
+
+class CacheTest(unittest.TestCase):
+    def setUp(self):
+        self.temporary_path_object = tempfile.TemporaryDirectory()
+        self.temporary_path = self.temporary_path_object.name
+
+    def tearDown(self):
+        self.temporary_path_object.cleanup()
+
+    def test_cache_empty(self):
+        cache = Cache(file_path=self.temporary_path)
+        self.assertEqual(UUID("00000000-0000-0000-0000-000000000000"), cache.userId)
+
+    def test_cache_save(self):
+        cache = Cache(file_path=self.temporary_path)
+        cache.file_path = self.temporary_path
+        cache.save()
+
+        file = Path(os.path.join(cache.file_path, cache.file_name))
+
+        self.assertTrue(file.exists())
+        self.assertEqual(UUID("00000000-0000-0000-0000-000000000000"), cache.userId)
+
+    def test_cache_load(self):
+        cache_01 = Cache(file_path=self.temporary_path)
+        cache_01.userId = UUID("00000000-0000-0000-0000-000000000001")
+        cache_01.save()
+
+        cache_02 = Cache(file_path=self.temporary_path)
+        self.assertEqual(UUID("00000000-0000-0000-0000-000000000001"), cache_02.userId)

tests/test_cli_app.py

@@ -1,26 +1,24 @@
+from unittest.mock import patch
+
 from tests.login_testcase import LoginTestCase
+from unikube.authentication.authentication import TokenAuthentication
 from unikube.cli import app
-from unikube.commands import ClickContext
-
-
-def check():
-    """Function used to mock check function"""
-    pass
+from unikube.context import ClickContext
 
 
 class AppTestCase(LoginTestCase):
-    def test_list(self):
+    @patch.object(TokenAuthentication, "check")
+    def test_list(self, *args, **kwargs):
         obj = ClickContext()
-        obj.auth.check = check
         result = self.runner.invoke(
             app.list,
             obj=obj,
         )
         assert result.exit_code == 1
 
-    def test_shell_invalid_arguments(self):
+    @patch.object(TokenAuthentication, "check")
+    def test_shell_invalid_arguments(self, *args, **kwargs):
         obj = ClickContext()
-        obj.auth.check = check
         result = self.runner.invoke(
             app.shell,
             [