mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-08 19:26:48 +00:00
4dd72766cf
The VM's keys page (instances:sshkeys) and the cloud-init user data collected the owners' keys with one query an owner. Both now read them in one query joined over the VM's owners, ordered owner by owner as before.
129 lines
6.1 KiB
Python
129 lines
6.1 KiB
Python
import json
|
|
|
|
import yaml
|
|
from unittest.mock import patch
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import RequestFactory, TestCase
|
|
from django.urls import reverse
|
|
|
|
from accounts.models import UserInstance, UserSSHKey
|
|
from computes.models import Compute
|
|
from instances.models import Instance
|
|
from datasource.views import get_client_ip, get_hostname_by_ip
|
|
|
|
|
|
class DataSourceTestCase(TestCase):
|
|
def setUp(self):
|
|
self.factory = RequestFactory()
|
|
self.client.login(request=RequestFactory().get("/"), username="admin", password="admin")
|
|
self.compute = Compute.objects.create(
|
|
name="test_compute",
|
|
hostname="localhost",
|
|
type=1,
|
|
)
|
|
|
|
def test_os_index(self):
|
|
response = self.client.get(reverse("ds_openstack_index"))
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn("latest", response.content.decode("utf-8"))
|
|
|
|
def test_os_metadata_json_latest(self):
|
|
response = self.client.get(reverse("ds_openstack_metadata", args=["latest"]))
|
|
self.assertEqual(response.status_code, 200)
|
|
data = json.loads(response.content.decode("utf-8"))
|
|
self.assertEqual(data["uuid"], "iid-dswebvirtcloud")
|
|
self.assertIn("hostname", data)
|
|
|
|
def test_os_metadata_json_invalid_version(self):
|
|
response = self.client.get(reverse("ds_openstack_metadata", args=["v1"]))
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
def test_os_userdata_invalid_version(self):
|
|
response = self.client.get(reverse("ds_openstack_userdata", args=["v1"]))
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
def test_get_client_ip_direct(self):
|
|
request = self.factory.get("/")
|
|
request.META["REMOTE_ADDR"] = "192.168.1.10"
|
|
self.assertEqual(get_client_ip(request), "192.168.1.10")
|
|
|
|
def test_get_client_ip_forwarded(self):
|
|
request = self.factory.get("/")
|
|
request.META["HTTP_X_FORWARDED_FOR"] = "10.0.0.1, 192.168.1.50"
|
|
self.assertEqual(get_client_ip(request), "192.168.1.50")
|
|
|
|
def test_get_hostname_by_ip_without_an_entry(self):
|
|
self.assertIsNone(get_hostname_by_ip("256.256.256.256"))
|
|
|
|
def test_vdi_url_not_found_compute(self):
|
|
response = self.client.get(reverse("vdi_url", args=[9999, "vm1"]))
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
|
|
class CloudInitTestCase(TestCase):
|
|
"""VMs fetch the cloud-init endpoints without logging in; keys go only to
|
|
a VM whose name is unique."""
|
|
|
|
KEY = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIOwner owner@example"
|
|
|
|
def setUp(self):
|
|
self.compute = Compute.objects.create(name="ci-a", hostname="a.example", login="", password="", type=4)
|
|
self.owner = get_user_model().objects.create_user("ci_owner", password="pw")
|
|
UserSSHKey.objects.create(user=self.owner, keyname="k", keypublic=self.KEY)
|
|
self.vm = Instance.objects.create(compute=self.compute, name="web1", uuid="11111111-2222-3333-4444-999999999991")
|
|
UserInstance.objects.create(instance=self.vm, user=self.owner)
|
|
patcher = patch("datasource.views.get_hostname_by_ip", return_value="web1.lan")
|
|
self.reverse_dns = patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
|
|
def userdata(self):
|
|
response = self.client.get(reverse("ds_openstack_userdata", args=["latest"]))
|
|
self.assertEqual(response.status_code, 200)
|
|
return response.content.decode()
|
|
|
|
def test_endpoints_need_no_login(self):
|
|
for name, args in (("ds_openstack_index", []), ("ds_openstack_metadata", ["latest"])):
|
|
self.assertEqual(self.client.get(reverse(name, args=args)).status_code, 200, name)
|
|
self.assertEqual(json.loads(self.client.get(reverse("ds_openstack_metadata", args=["latest"])).content)["hostname"], "web1.lan")
|
|
|
|
def test_owner_keys_for_a_unique_name(self):
|
|
self.assertIn(self.KEY, self.userdata())
|
|
|
|
def test_every_owner_s_keys_and_no_one_else_s(self):
|
|
second_key, stranger_key = (f"ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAI{who} {who}@example" for who in ("Second", "Stranger"))
|
|
second = get_user_model().objects.create_user("ci_second", password="pw")
|
|
UserSSHKey.objects.create(user=second, keyname="k", keypublic=second_key)
|
|
UserInstance.objects.create(instance=self.vm, user=second)
|
|
stranger = get_user_model().objects.create_user("ci_stranger", password="pw")
|
|
UserSSHKey.objects.create(user=stranger, keyname="k", keypublic=stranger_key)
|
|
other_vm = Instance.objects.create(compute=self.compute, name="web2", uuid="11111111-2222-3333-4444-999999999993")
|
|
UserInstance.objects.create(instance=other_vm, user=stranger)
|
|
UserInstance.objects.create(instance=other_vm, user=self.owner)
|
|
|
|
page = self.userdata()
|
|
self.assertLess(page.index(self.KEY), page.index(second_key))
|
|
self.assertEqual(page.count(self.KEY), 1)
|
|
self.assertNotIn(stranger_key, page)
|
|
|
|
def test_no_keys_when_the_name_is_taken_twice(self):
|
|
other = Compute.objects.create(name="ci-b", hostname="b.example", login="", password="", type=4)
|
|
Instance.objects.create(compute=other, name="web1", uuid="11111111-2222-3333-4444-999999999992")
|
|
self.assertNotIn("ssh", self.userdata())
|
|
|
|
def test_no_keys_for_an_unknown_name(self):
|
|
self.vm.delete()
|
|
self.assertNotIn("ssh", self.userdata())
|
|
|
|
def test_no_keys_for_an_address_without_a_reverse_dns_entry(self):
|
|
# the address's first number is no VM's name: a VM named "10" gets nothing
|
|
self.reverse_dns.return_value = None
|
|
Instance.objects.create(compute=self.compute, name="127", uuid="11111111-2222-3333-4444-999999999993")
|
|
self.assertNotIn("ssh", self.userdata())
|
|
metadata = json.loads(self.client.get(reverse("ds_openstack_metadata", args=["latest"])).content)
|
|
self.assertEqual(metadata["hostname"], "127.0.0.1")
|
|
|
|
def test_a_key_comment_stays_one_string(self):
|
|
key = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIOwner it's: mine"
|
|
UserSSHKey.objects.filter(user=self.owner).update(keypublic=key)
|
|
self.assertEqual(yaml.safe_load(self.userdata())["ssh_authorized_keys"], [key])
|