tests: test CRD versions

apiclient/kube_api/__init__.py

@@ -0,0 +1,7 @@
+# Copyright (c) 2024 SUSE LLC
+
+from .api import KubeAPI
+
+# This is just to make `tox -e pep8` stop complaining
+# about things that it really shouldn't complain about.
+_ = KubeAPI

apiclient/kube_api/api.py

@@ -0,0 +1,70 @@
+# Copyright (c) 2024 SUSE LLC
+#
+# pylint: disable=missing-function-docstring
+
+from urllib.parse import urljoin
+
+import kubernetes
+import requests
+import yaml
+
+
+class KubeAPI:
+    """
+    An abstraction of the Kubernetes API.
+
+    Example usage:
+
+    ```
+    with KubeAPI(endpoint, tls_verify=false) as api:
+      api.authenticate(username, password, verify=false)
+      kube_client = api.get_client()
+
+      corev1 = kubernetes.client.CoreV1Api(kube_client)
+
+      namespaces = corev1.list_namespace()
+    ```
+    """
+
+    HARVESTER_API_VERSION = "harvesterhci.io/v1beta1"
+
+    def __init__(self, endpoint, tls_verify, token=None, session=None):
+        self.session = session or requests.Session()
+        self.session.verify = tls_verify
+        self.session.headers.update(Authorization=token or "")
+
+        self.endpoint = endpoint
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_value, taceback):
+        pass
+
+    def _post(self, path, **kwargs):
+        url = self._get_url(path)
+        return self.session.post(url, **kwargs)
+
+    def _get_url(self, path):
+        return urljoin(self.endpoint, path).format(API_VERSION=self.HARVESTER_API_VERSION)
+
+    def get_client(self):
+        path = "/v1/management.cattle.io.clusters/local"
+        params = {"action": "generateKubeconfig"}
+
+        resp = self._post(path, params=params)
+        assert resp.status_code == 200, "Failed to generate kubeconfig"
+
+        kubeconfig = yaml.safe_load(resp.json()['config'])
+        return kubernetes.config.new_client_from_config_dict(kubeconfig)
+
+    def authenticate(self, user, passwd, **kwargs):
+        path = "v3-public/localProviders/local?action=login"
+        resp = self._post(path, json=dict(username=user, password=passwd), **kwargs)
+
+        assert resp.status_code == 201, "Failed to authenticate"
+
+        token = f"Bearer {resp.json()['token']}"
+        self.session.headers.update(Authorization=token)
+
+        return resp.json()

harvester_e2e_tests/apis/test_crd_versions.py

@@ -0,0 +1,81 @@
+# Copyright (c) 2024 SUSE LLC
+#
+# pylint: disable=missing-function-docstring, redefined-outer-name
+
+from urllib.parse import urljoin
+
+import kubernetes
+import pytest
+import requests
+import semver
+import yaml
+
+
+pytest_plugins = [
+    "harvester_e2e_tests.fixtures.kube_api_client",
+    "harvester_e2e_tests.fixtures.api_client"
+]
+
+
+@pytest.fixture(scope="session")
+def server_version(api_client):
+    code, data = api_client.settings.get(name="server-version")
+    assert code == 200
+    assert data.get("value") is not None
+
+    yield semver.VersionInfo.parse(data.get("value").lstrip("v"))
+
+
+@pytest.fixture(scope="module", params=[
+    ("csi-snapshotter", "volumesnapshotclasses"),
+    ("csi-snapshotter", "volumesnapshotcontents"),
+    ("csi-snapshotter", "volumesnapshots"),
+    ("kubevirt-operator", "crd-kubevirt"),
+    ("whereabouts", "whereabouts.cni.cncf.io_ippools"),
+    ("whereabouts", "whereabouts.cni.cncf.io_overlappingrangeipreservations")
+])
+def chart_and_file_name(request):
+    yield request.param
+
+
+@pytest.fixture(scope="module")
+def expected_crd(server_version, chart_and_file_name):
+    raw_url = "https://raw.githubusercontent.com/harvester/harvester/"
+    raw_url = urljoin(raw_url, f"v{server_version.major}.{server_version.minor}/")
+    raw_url = urljoin(raw_url, "deploy/charts/harvester/dependency_charts/")
+    raw_url = urljoin(raw_url, f"{chart_and_file_name[0]}/")
+    raw_url = urljoin(raw_url, "crds/")
+    raw_url = urljoin(raw_url, f"{chart_and_file_name[1]}.yaml")
+
+    resp = requests.get(raw_url, allow_redirects=True)
+    cont = resp.content.decode("utf-8")
+    data = yaml.safe_load(cont)
+    yield data
+
+
+@pytest.fixture(scope="module")
+def actual_crd(kube_api_client, expected_crd):
+    name = expected_crd['metadata']['name']
+    kube_client = kubernetes.client.ApiextensionsV1Api(kube_api_client)
+    yield kube_client.read_custom_resource_definition(name=name)
+
+
+@pytest.mark.api
+def test_api_version(expected_crd, actual_crd):
+    expected_versions = []
+    for ver in expected_crd['spec']['versions']:
+        expected_versions.append(ver['name'])
+
+    actual_versions = []
+    for ver in actual_crd.spec.versions:
+        actual_versions.append(ver.name)
+
+    assert expected_crd['metadata']['name'] == actual_crd.metadata.name
+
+    # Make sure all expected versions are there
+    for ver in expected_versions:
+        assert ver in actual_versions
+
+    # Make sure all installed versions are expected
+    for ver in actual_versions:
+        assert ver in expected_versions

harvester_e2e_tests/fixtures/kube_api_client.py

@@ -0,0 +1,20 @@
+# Copyright (c) 2024 SUSE LLC
+#
+# pylint: disable=missing-function-docstring
+
+import pytest
+
+from kube_api import KubeAPI
+
+
+@pytest.fixture(scope="session")
+def kube_api_client(request):
+    endpoint = request.config.getoption("--endpoint")
+    username = request.config.getoption("--username")
+    password = request.config.getoption("--password")
+    tls_verify = request.config.getoption("--ssl_verify", False)
+
+    with KubeAPI(endpoint, tls_verify) as api:
+        api.authenticate(username, password, verify=tls_verify)
+
+        yield api.get_client()

test-requirements.txt

@@ -7,6 +7,8 @@ pytest-json-report
 pytest-dependency
 jinja2
 bcrypt
+kubernetes
+semver
 requests
 paramiko
 pycryptodome