Liam Young
135 lines
4.7 KiB
Python
135 lines
4.7 KiB
Python
# Copyright 2021 Canonical Ltd.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
"""Encapsulating `ceph-dashboard` testing."""
|
|
|
|
import collections
|
|
import json
|
|
import requests
|
|
import uuid
|
|
|
|
import zaza
|
|
import zaza.openstack.charm_tests.test_utils as test_utils
|
|
import zaza.openstack.utilities.openstack as openstack_utils
|
|
|
|
|
|
class CephDashboardTest(test_utils.BaseCharmTest):
|
|
"""Class for `ceph-dashboard` tests."""
|
|
|
|
REMOTE_CERT_FILE = ('/usr/local/share/ca-certificates/'
|
|
'vault_ca_cert_dashboard.crt')
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
"""Run class setup for running ceph dashboard tests."""
|
|
super().setUpClass()
|
|
cls.application_name = 'ceph-dashboard'
|
|
cls.local_ca_cert = openstack_utils.get_remote_ca_cert_file(
|
|
cls.application_name)
|
|
|
|
def get_master_dashboard_url(self):
|
|
"""Get the url of the dashboard servicing requests.
|
|
|
|
Only one unit serves requests at any one time, the other units
|
|
redirect to that unit.
|
|
|
|
:returns: URL of dashboard on unit
|
|
:rtype: Union[str, None]
|
|
"""
|
|
units = zaza.model.get_units(self.application_name)
|
|
for unit in units:
|
|
r = requests.get(
|
|
'https://{}:8443'.format(unit.public_address),
|
|
verify=self.local_ca_cert,
|
|
allow_redirects=False)
|
|
if r.status_code == requests.codes.ok:
|
|
return 'https://{}:8443'.format(unit.public_address)
|
|
|
|
def test_dashboard_units(self):
|
|
"""Check dashboard units are configured correctly."""
|
|
verify = self.local_ca_cert
|
|
units = zaza.model.get_units(self.application_name)
|
|
rcs = collections.defaultdict(list)
|
|
for unit in units:
|
|
r = requests.get(
|
|
'https://{}:8443'.format(unit.public_address),
|
|
verify=verify,
|
|
allow_redirects=False)
|
|
rcs[r.status_code].append(unit.public_address)
|
|
self.assertEqual(len(rcs[requests.codes.ok]), 1)
|
|
self.assertEqual(len(rcs[requests.codes.see_other]), len(units) - 1)
|
|
|
|
def create_user(self, username, role='administrator'):
|
|
"""Create a dashboard user.
|
|
|
|
:param username: Username to create.
|
|
:type username: str
|
|
:param role: Role to grant to user.
|
|
:type role: str
|
|
:returns: Results from action.
|
|
:rtype: juju.action.Action
|
|
"""
|
|
action = zaza.model.run_action_on_leader(
|
|
'ceph-dashboard',
|
|
'add-user',
|
|
action_params={
|
|
'username': username,
|
|
'role': role})
|
|
return action
|
|
|
|
def get_random_username(self):
|
|
"""Generate a username to use in tests.
|
|
|
|
:returns: Username
|
|
:rtype: str
|
|
"""
|
|
return "zazauser-{}".format(uuid.uuid1())
|
|
|
|
def test_create_user(self):
|
|
"""Test create user action."""
|
|
test_user = self.get_random_username()
|
|
action = self.create_user(test_user)
|
|
self.assertEqual(action.status, "completed")
|
|
self.assertTrue(action.data['results']['password'])
|
|
action = self.create_user(test_user)
|
|
# Action should fail as the user already exists
|
|
self.assertEqual(action.status, "failed")
|
|
|
|
def access_dashboard(self, dashboard_url):
|
|
"""Test logging via a dashboard url.
|
|
|
|
:param dashboard_url: Base url to use to login to
|
|
:type dashboard_url: str
|
|
"""
|
|
user = self.get_random_username()
|
|
action = self.create_user(username=user)
|
|
self.assertEqual(action.status, "completed")
|
|
password = action.data['results']['password']
|
|
path = "api/auth"
|
|
headers = {
|
|
'Content-type': 'application/json',
|
|
'Accept': 'application/vnd.ceph.api.v1.0'}
|
|
payload = {"username": user, "password": password}
|
|
verify = self.local_ca_cert
|
|
r = requests.post(
|
|
"{}/{}".format(dashboard_url, path),
|
|
data=json.dumps(payload),
|
|
headers=headers,
|
|
verify=verify)
|
|
self.assertEqual(r.status_code, requests.codes.created)
|
|
|
|
def test_access_dashboard(self):
|
|
"""Test logging in to the dashboard."""
|
|
self.access_dashboard(self.get_master_dashboard_url())
|