diff --git a/passbook/outposts/tests.py b/passbook/outposts/tests.py index 18675e4c0..de7454165 100644 --- a/passbook/outposts/tests.py +++ b/passbook/outposts/tests.py @@ -1,9 +1,16 @@ """outpost tests""" +from os import environ +from unittest.case import skipUnless +from unittest.mock import patch + from django.test import TestCase from guardian.models import UserObjectPermission from passbook.crypto.models import CertificateKeyPair from passbook.flows.models import Flow +from passbook.outposts.controllers.k8s.base import NeedsUpdate +from passbook.outposts.controllers.k8s.deployment import DeploymentReconciler +from passbook.outposts.controllers.kubernetes import KubernetesController from passbook.outposts.models import Outpost, OutpostDeploymentType, OutpostType from passbook.providers.proxy.models import ProxyProvider @@ -58,3 +65,50 @@ class OutpostTests(TestCase): permissions = UserObjectPermission.objects.filter(user=outpost.user) self.assertEqual(len(permissions), 1) self.assertEqual(permissions[0].object_pk, str(outpost.pk)) + + +@skipUnless("PB_TEST_K8S" in environ, "Kubernetes test cluster required") +class OutpostKubernetesTests(TestCase): + """Test Kubernetes Controllers""" + + def setUp(self): + super().setUp() + self.provider: ProxyProvider = ProxyProvider.objects.create( + name="test", + internal_host="http://localhost", + external_host="http://localhost", + authorization_flow=Flow.objects.first(), + ) + self.outpost: Outpost = Outpost.objects.create( + name="test", + type=OutpostType.PROXY, + deployment_type=OutpostDeploymentType.KUBERNETES, + ) + self.outpost.providers.add(self.provider) + self.outpost.save() + + def test_deployment_reconciler(self): + """test that deployment requires update""" + controller = KubernetesController(self.outpost) + deployment_reconciler = DeploymentReconciler(controller) + + self.assertIsNotNone(deployment_reconciler.retrieve()) + + config = self.outpost.config + config.kubernetes_replicas = 3 + self.outpost.config = config + + with self.assertRaises(NeedsUpdate): + deployment_reconciler.reconcile( + deployment_reconciler.retrieve(), + deployment_reconciler.get_reference_object(), + ) + + with patch.object(deployment_reconciler, "image_base", "test"): + with self.assertRaises(NeedsUpdate): + deployment_reconciler.reconcile( + deployment_reconciler.retrieve(), + deployment_reconciler.get_reference_object(), + ) + + deployment_reconciler.delete(deployment_reconciler.get_reference_object())