mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-08 19:26:48 +00:00
88 lines
4.8 KiB
Python
88 lines
4.8 KiB
Python
from django.contrib.auth import get_user_model
|
|
from django.contrib.auth.models import Permission
|
|
from django.test import TransactionTestCase
|
|
|
|
from accounts.models import UserInstance
|
|
from computes.models import Compute
|
|
from instances.models import Instance
|
|
from operations import claims, jobs
|
|
from operations.models import Claim, Operation
|
|
from operations.tests import fakes
|
|
|
|
|
|
class OperationsApiTestCase(TransactionTestCase):
|
|
def setUp(self):
|
|
fakes.reset()
|
|
U = get_user_model()
|
|
self.owner = U.objects.create_user("api_owner", "o@example.com", "pw")
|
|
self.other = U.objects.create_user("api_other", "x@example.com", "pw")
|
|
compute = Compute.objects.create(name="api-c", hostname="192.0.2.20", login="root", password="", type=1)
|
|
self.vm = Instance.objects.create(compute=compute, name="api-vm", uuid="u1")
|
|
self.vm2 = Instance.objects.create(compute=compute, name="api-vm2", uuid="u2")
|
|
UserInstance.objects.create(user=self.other, instance=self.vm)
|
|
UserInstance.objects.create(user=self.owner, instance=self.vm2)
|
|
self.op = jobs.submit("test.noop", self.owner, {"uuid": "u1"}, "key-00000001")
|
|
|
|
def test_users_list_only_their_own_jobs(self):
|
|
self.client.force_login(self.other)
|
|
self.assertEqual(self.client.get("/api/v1/operations/").json()["count"], 0)
|
|
self.assertEqual(self.client.get(f"/api/v1/operations/{self.op.pk}/").status_code, 404)
|
|
self.client.force_login(self.owner)
|
|
self.assertEqual(self.client.get("/api/v1/operations/").json()["count"], 1)
|
|
|
|
def test_the_resource_view_shows_others_jobs_as_summaries(self):
|
|
self.client.force_login(self.other)
|
|
data = self.client.get("/api/v1/operations/resource/", {"key": "instance:u1"}).json()
|
|
self.assertEqual(data["jobs"], [])
|
|
[summary] = data["others"]
|
|
self.assertEqual(summary["kind"], "test.noop")
|
|
for private in ("id", "user", "username", "params"):
|
|
self.assertNotIn(private, summary)
|
|
|
|
def test_a_resource_the_user_cannot_see_is_refused(self):
|
|
stranger = get_user_model().objects.create_user("api_s", "s@example.com", "pw")
|
|
self.client.force_login(stranger)
|
|
self.assertEqual(self.client.get("/api/v1/operations/resource/", {"key": "instance:u1"}).status_code, 404)
|
|
|
|
def test_a_short_operation_is_flagged(self):
|
|
claims.admit({"instance:u2": Claim.EXCLUSIVE}, holder="another-request")
|
|
self.client.force_login(self.owner)
|
|
self.assertTrue(self.client.get("/api/v1/operations/resource/", {"key": "instance:u2"}).json()["short_operation"])
|
|
|
|
def test_cancel_through_the_api(self):
|
|
self.client.force_login(self.owner)
|
|
response = self.client.post(f"/api/v1/operations/{self.op.pk}/cancel/")
|
|
self.assertEqual((response.status_code, response.json()["state"]), (200, "cancelled"))
|
|
|
|
def test_a_job_that_cannot_stop_answers_400(self):
|
|
Operation.objects.filter(pk=self.op.pk).update(state=Operation.RUNNING)
|
|
self.client.force_login(self.owner)
|
|
self.assertEqual(self.client.post(f"/api/v1/operations/{self.op.pk}/cancel/").status_code, 400)
|
|
|
|
def _labels_seen_by(self, user, label="wvc-test-host/wvc-test-vm"):
|
|
op = jobs.submit("test.noop", self.owner, {"uuid": "u2"}, "key-00000002")
|
|
Operation.objects.filter(pk=op.pk).update(resource_label=label)
|
|
self.client.force_login(user)
|
|
listed = [row["resource_label"] for row in self.client.get("/api/v1/operations/").json()["results"]
|
|
if row["id"] == str(op.pk)]
|
|
detail = self.client.get(f"/api/v1/operations/{op.pk}/").json()["resource_label"]
|
|
[job] = self.client.get("/api/v1/operations/resource/", {"key": "instance:u2"}).json()["jobs"]
|
|
cancelled = self.client.post(f"/api/v1/operations/{op.pk}/cancel/").json()["resource_label"]
|
|
return listed + [detail, job["resource_label"], cancelled]
|
|
|
|
def test_a_plain_owner_sees_only_the_vm_name(self):
|
|
self.assertEqual(self._labels_seen_by(self.owner), ["wvc-test-vm"] * 4)
|
|
|
|
def test_a_compute_name_with_a_slash_stays_hidden(self):
|
|
# libvirt refuses "/" in VM names; compute names may contain it
|
|
self.assertEqual(self._labels_seen_by(self.owner, "wvc-test-region/wvc-test-host/wvc-test-vm"),
|
|
["wvc-test-vm"] * 4)
|
|
|
|
def test_users_who_may_see_hosts_see_the_compute_name(self):
|
|
self.owner.user_permissions.add(Permission.objects.get(codename="view_instances",
|
|
content_type__app_label="instances"))
|
|
self.assertEqual(self._labels_seen_by(self.owner), ["wvc-test-host/wvc-test-vm"] * 4)
|
|
|
|
def test_superusers_see_the_compute_name(self):
|
|
admin = get_user_model().objects.create_superuser("api_admin", "a@example.com", "pw")
|
|
self.assertEqual(self._labels_seen_by(admin), ["wvc-test-host/wvc-test-vm"] * 4)
|