servala-portal/src/tests/test_api_exoscale.py

745 lines
22 KiB
Python

import json
import pytest
from django.core import mail
from django_scopes import scopes_disabled
from servala.core.models import Organization, OrganizationOrigin, User
from servala.core.models.service import (
ControlPlane,
ControlPlaneCRD,
ServiceDefinition,
ServiceInstance,
)
@pytest.fixture
def exoscale_origin():
origin, _ = OrganizationOrigin.objects.get_or_create(
name="exoscale-marketplace",
defaults={
"description": "Organizations created via Exoscale marketplace onboarding"
},
)
return origin
@pytest.fixture
def instance_id():
return "test-instance-123"
@pytest.fixture
def valid_osb_payload():
return {
"service_id": None,
"plan_id": None,
"context": {
"organization_guid": "test-org-guid-123",
"organization_name": "Test Organization",
"organization_display_name": "Test Organization Display",
},
"parameters": {
"users": [
{
"email": "test@example.com",
"full_name": "Test User",
"role": "owner",
}
]
},
}
@pytest.mark.django_db
def test_successful_onboarding_new_organization(
mock_odoo_success,
osb_client,
test_service,
test_service_offering,
valid_osb_payload,
exoscale_origin,
instance_id,
):
valid_osb_payload["service_id"] = test_service.osb_service_id
valid_osb_payload["plan_id"] = test_service_offering.osb_plan_id
response = osb_client.put(
f"/api/osb/v2/service_instances/{instance_id}",
data=json.dumps(valid_osb_payload),
content_type="application/json",
)
assert response.status_code == 201
response_data = json.loads(response.content)
assert response_data["message"] == "Successfully enabled service"
org = Organization.objects.get(osb_guid="test-org-guid-123")
assert org.name == "Test Organization Display"
assert org.origin == exoscale_origin
assert org.namespace.startswith("org-")
with scopes_disabled():
assert org.invitations.all().filter(email="test@example.com").exists()
billing_entity = org.billing_entity
assert billing_entity.name == "Test Organization Display (Exoscale)"
assert billing_entity.odoo_company_id == 123
assert billing_entity.odoo_invoice_id == 456
assert org.odoo_sale_order_id == 789
assert org.odoo_sale_order_name == "SO001"
assert org.limit_osb_services.all().count() == 1
assert len(mail.outbox) == 2
invitation_email = mail.outbox[0]
assert (
invitation_email.subject
== "You're invited to join Test Organization Display on Servala"
)
assert "test@example.com" in invitation_email.to
welcome_email = mail.outbox[1]
assert welcome_email.subject == "Get started with Redis - Test Organization Display"
assert "redis/offering/" in welcome_email.body
@pytest.mark.django_db
def test_new_organization_inherits_origin(
osb_client,
test_service,
test_service_offering,
valid_osb_payload,
exoscale_origin,
instance_id,
billing_entity,
):
valid_osb_payload["service_id"] = test_service.osb_service_id
valid_osb_payload["plan_id"] = test_service_offering.osb_plan_id
exoscale_origin.billing_entity = billing_entity
exoscale_origin.save()
response = osb_client.put(
f"/api/osb/v2/service_instances/{instance_id}",
data=json.dumps(valid_osb_payload),
content_type="application/json",
)
assert response.status_code == 201
response_data = json.loads(response.content)
assert response_data["message"] == "Successfully enabled service"
org = Organization.objects.get(osb_guid="test-org-guid-123")
assert org.name == "Test Organization Display"
assert org.billing_entity == exoscale_origin.billing_entity
@pytest.mark.django_db
def test_duplicate_organization_returns_existing(
osb_client,
test_service,
test_service_offering,
valid_osb_payload,
exoscale_origin,
instance_id,
):
org = Organization.objects.create(
name="Existing Org",
osb_guid="test-org-guid-123",
origin=exoscale_origin,
)
org.limit_osb_services.add(test_service)
valid_osb_payload["service_id"] = test_service.osb_service_id
valid_osb_payload["plan_id"] = test_service_offering.osb_plan_id
response = osb_client.put(
f"/api/osb/v2/service_instances/{instance_id}",
data=json.dumps(valid_osb_payload),
content_type="application/json",
)
assert response.status_code == 200
response_data = json.loads(response.content)
assert response_data["message"] == "Service already enabled"
assert Organization.objects.filter(osb_guid="test-org-guid-123").count() == 1
assert len(mail.outbox) == 0 # No email necessary
@pytest.mark.django_db
def test_unauthenticated_osb_api_request_fails(
client,
test_service,
test_service_offering,
valid_osb_payload,
instance_id,
):
valid_osb_payload["service_id"] = test_service.osb_service_id
valid_osb_payload["plan_id"] = test_service_offering.osb_plan_id
response = client.put(
f"/api/osb/v2/service_instances/{instance_id}",
data=json.dumps(valid_osb_payload),
content_type="application/json",
)
assert response.status_code == 403
@pytest.mark.django_db
@pytest.mark.parametrize(
"field_to_remove,expected_error",
[
(
("context", "organization_guid"),
"organization_guid is required but missing",
),
(
("context", "organization_name"),
"organization_name is required but missing",
),
("service_id", "service_id is required but missing"),
("plan_id", "plan_id is required but missing"),
(("parameters", "users"), "users array is required but missing"),
],
)
def test_missing_required_fields_error(
osb_client,
test_service,
test_service_offering,
valid_osb_payload,
field_to_remove,
expected_error,
instance_id,
):
valid_osb_payload["service_id"] = test_service.osb_service_id
valid_osb_payload["plan_id"] = test_service_offering.osb_plan_id
if isinstance(field_to_remove, tuple):
if field_to_remove[0] == "context":
del valid_osb_payload["context"][field_to_remove[1]]
else:
del valid_osb_payload["parameters"][field_to_remove[1]]
else:
del valid_osb_payload[field_to_remove]
response = osb_client.put(
f"/api/osb/v2/service_instances/{instance_id}",
data=json.dumps(valid_osb_payload),
content_type="application/json",
)
assert response.status_code == 400
response_data = json.loads(response.content)
assert expected_error in response_data["error"]
@pytest.mark.django_db
def test_invalid_service_id_error(osb_client, valid_osb_payload, instance_id):
valid_osb_payload["service_id"] = 99999
valid_osb_payload["plan_id"] = 1
response = osb_client.put(
f"/api/osb/v2/service_instances/{instance_id}",
data=json.dumps(valid_osb_payload),
content_type="application/json",
)
assert response.status_code == 400
response_data = json.loads(response.content)
assert "Unknown service_id: 99999" in response_data["error"]
@pytest.mark.django_db
def test_invalid_plan_id_error(
osb_client, test_service, valid_osb_payload, instance_id
):
valid_osb_payload["service_id"] = test_service.osb_service_id
valid_osb_payload["plan_id"] = 99999
response = osb_client.put(
f"/api/osb/v2/service_instances/{instance_id}",
data=json.dumps(valid_osb_payload),
content_type="application/json",
)
assert response.status_code == 400
response_data = json.loads(response.content)
assert (
f"Unknown plan_id: 99999 for service_id: {test_service.osb_service_id}"
in response_data["error"]
)
@pytest.mark.django_db
def test_empty_users_array_error(
osb_client, test_service, test_service_offering, valid_osb_payload, instance_id
):
valid_osb_payload["service_id"] = test_service.osb_service_id
valid_osb_payload["plan_id"] = test_service_offering.osb_plan_id
valid_osb_payload["parameters"]["users"] = []
response = osb_client.put(
f"/api/osb/v2/service_instances/{instance_id}",
data=json.dumps(valid_osb_payload),
content_type="application/json",
)
assert response.status_code == 400
response_data = json.loads(response.content)
assert "users array is required but missing" in response_data["error"]
@pytest.mark.django_db
def test_multiple_users_error(
osb_client, test_service, test_service_offering, valid_osb_payload, instance_id
):
valid_osb_payload["service_id"] = test_service.osb_service_id
valid_osb_payload["plan_id"] = test_service_offering.osb_plan_id
valid_osb_payload["parameters"]["users"] = [
{"email": "user1@example.com", "full_name": "User One"},
{"email": "user2@example.com", "full_name": "User Two"},
]
response = osb_client.put(
f"/api/osb/v2/service_instances/{instance_id}",
data=json.dumps(valid_osb_payload),
content_type="application/json",
)
assert response.status_code == 400
response_data = json.loads(response.content)
assert "users array is expected to contain a single user" in response_data["error"]
@pytest.mark.django_db
def test_empty_email_address_error(
osb_client, test_service, test_service_offering, valid_osb_payload, instance_id
):
valid_osb_payload["service_id"] = test_service.osb_service_id
valid_osb_payload["plan_id"] = test_service_offering.osb_plan_id
valid_osb_payload["parameters"]["users"] = [
{"email": "", "full_name": "User With No Email"},
]
response = osb_client.put(
f"/api/osb/v2/service_instances/{instance_id}",
data=json.dumps(valid_osb_payload),
content_type="application/json",
)
assert response.status_code == 400
response_data = json.loads(response.content)
assert "Unable to create user:" in response_data["error"]
@pytest.mark.django_db
def test_invalid_json_error(osb_client, instance_id):
response = osb_client.put(
f"/api/osb/v2/service_instances/{instance_id}",
data="invalid json{",
content_type="application/json",
)
assert response.status_code == 400
response_data = json.loads(response.content)
assert "Invalid JSON in request body" in response_data["error"]
@pytest.mark.django_db
def test_user_creation_with_name_parsing(
mock_odoo_success,
osb_client,
test_service,
test_service_offering,
valid_osb_payload,
exoscale_origin,
instance_id,
):
valid_osb_payload["service_id"] = test_service.osb_service_id
valid_osb_payload["plan_id"] = test_service_offering.osb_plan_id
valid_osb_payload["parameters"]["users"][0]["full_name"] = "John Doe Smith"
response = osb_client.put(
f"/api/osb/v2/service_instances/{instance_id}",
data=json.dumps(valid_osb_payload),
content_type="application/json",
)
assert response.status_code == 201
user = User.objects.get(email="test@example.com")
assert user.first_name == "John"
assert user.last_name == "Doe Smith"
@pytest.mark.django_db
def test_email_normalization(
mock_odoo_success,
osb_client,
test_service,
test_service_offering,
valid_osb_payload,
exoscale_origin,
instance_id,
):
valid_osb_payload["service_id"] = test_service.osb_service_id
valid_osb_payload["plan_id"] = test_service_offering.osb_plan_id
valid_osb_payload["parameters"]["users"][0]["email"] = " TEST@EXAMPLE.COM "
response = osb_client.put(
f"/api/osb/v2/service_instances/{instance_id}",
data=json.dumps(valid_osb_payload),
content_type="application/json",
)
assert response.status_code == 201
user = User.objects.get(email="test@example.com")
assert user.email == "test@example.com"
@pytest.mark.django_db
def test_odoo_integration_failure_handling(
mock_odoo_failure,
osb_client,
test_service,
test_service_offering,
valid_osb_payload,
exoscale_origin,
instance_id,
):
valid_osb_payload["service_id"] = test_service.osb_service_id
valid_osb_payload["plan_id"] = test_service_offering.osb_plan_id
response = osb_client.put(
f"/api/osb/v2/service_instances/{instance_id}",
data=json.dumps(valid_osb_payload),
content_type="application/json",
)
assert response.status_code == 500
response_data = json.loads(response.content)
assert response_data["error"] == "Internal server error"
@pytest.mark.django_db
def test_organization_creation_with_context_only(
mock_odoo_success,
osb_client,
test_service,
test_service_offering,
exoscale_origin,
instance_id,
):
payload = {
"service_id": test_service.osb_service_id,
"plan_id": test_service_offering.osb_plan_id,
"context": {
"organization_guid": "fallback-org-guid",
"organization_name": "Fallback Organization",
},
"parameters": {
"users": [
{
"email": "fallback@example.com",
"full_name": "Fallback User",
}
]
},
}
response = osb_client.put(
f"/api/osb/v2/service_instances/{instance_id}",
data=json.dumps(payload),
content_type="application/json",
)
assert response.status_code == 201
org = Organization.objects.get(osb_guid="fallback-org-guid")
assert org is not None
@pytest.mark.django_db
def test_delete_offboarding_success(
mock_odoo_success,
osb_client,
test_service,
test_service_offering,
instance_id,
):
response = osb_client.delete(
f"/api/osb/v2/service_instances/{instance_id}"
f"?service_id={test_service.osb_service_id}&plan_id={test_service_offering.osb_plan_id}"
)
assert response.status_code == 200
assert response.content == b"{}"
@pytest.mark.django_db
def test_delete_missing_service_id(osb_client, test_service_offering, instance_id):
response = osb_client.delete(
f"/api/osb/v2/service_instances/{instance_id}?plan_id={test_service_offering.osb_plan_id}"
)
assert response.status_code == 400
response_data = json.loads(response.content)
assert "service_id is required but missing" in response_data["error"]
@pytest.mark.django_db
def test_delete_missing_plan_id(osb_client, test_service, instance_id):
response = osb_client.delete(
f"/api/osb/v2/service_instances/{instance_id}?service_id={test_service.osb_service_id}"
)
assert response.status_code == 400
response_data = json.loads(response.content)
assert "plan_id is required but missing" in response_data["error"]
@pytest.mark.django_db
def test_delete_invalid_service_id(osb_client, instance_id):
response = osb_client.delete(
f"/api/osb/v2/service_instances/{instance_id}?service_id=invalid&plan_id=invalid"
)
assert response.status_code == 400
response_data = json.loads(response.content)
assert "Unknown service_id: invalid" in response_data["error"]
@pytest.mark.django_db
def test_delete_invalid_plan_id(osb_client, test_service, instance_id):
response = osb_client.delete(
f"/api/osb/v2/service_instances/{instance_id}"
f"?service_id={test_service.osb_service_id}&plan_id=invalid"
)
assert response.status_code == 400
response_data = json.loads(response.content)
assert (
f"Unknown plan_id: invalid for service_id: {test_service.osb_service_id}"
in response_data["error"]
)
@pytest.mark.django_db
def test_patch_suspension_success(
mock_odoo_success,
osb_client,
test_service,
test_service_offering,
instance_id,
):
payload = {
"service_id": test_service.osb_service_id,
"plan_id": test_service_offering.osb_plan_id,
"parameters": {
"users": [
{
"email": "user@example.com",
"full_name": "Test User",
"role": "owner",
}
]
},
}
response = osb_client.patch(
f"/api/osb/v2/service_instances/{instance_id}",
data=json.dumps(payload),
content_type="application/json",
)
assert response.status_code == 200
assert response.content == b"{}"
@pytest.mark.django_db
def test_patch_missing_service_id(osb_client, test_service_offering, instance_id):
payload = {
"plan_id": test_service_offering.osb_plan_id,
"parameters": {"users": []},
}
response = osb_client.patch(
f"/api/osb/v2/service_instances/{instance_id}",
data=json.dumps(payload),
content_type="application/json",
)
assert response.status_code == 400
response_data = json.loads(response.content)
assert "service_id is required but missing" in response_data["error"]
@pytest.mark.django_db
def test_patch_missing_plan_id(osb_client, test_service, instance_id):
payload = {
"service_id": test_service.osb_service_id,
"parameters": {"users": []},
}
response = osb_client.patch(
f"/api/osb/v2/service_instances/{instance_id}",
data=json.dumps(payload),
content_type="application/json",
)
assert response.status_code == 400
response_data = json.loads(response.content)
assert "plan_id is required but missing" in response_data["error"]
@pytest.mark.django_db
def test_patch_invalid_json(osb_client, instance_id):
response = osb_client.patch(
f"/api/osb/v2/service_instances/{instance_id}",
data="invalid json{",
content_type="application/json",
)
assert response.status_code == 400
response_data = json.loads(response.content)
assert "Invalid JSON in request body" in response_data["error"]
@pytest.mark.django_db
def test_delete_creates_ticket_with_admin_links(
mocker,
mock_odoo_success,
osb_client,
test_service,
test_service_offering,
instance_id,
):
# Mock the create_helpdesk_ticket function
mock_create_ticket = mocker.patch("servala.api.views.create_helpdesk_ticket")
response = osb_client.delete(
f"/api/osb/v2/service_instances/{instance_id}"
f"?service_id={test_service.osb_service_id}&plan_id={test_service_offering.osb_plan_id}"
)
assert response.status_code == 200
# Verify the ticket was created with admin URL
mock_create_ticket.assert_called_once()
call_kwargs = mock_create_ticket.call_args[1]
# Check that the description contains an admin URL
assert "admin/core/serviceoffering" in call_kwargs["description"]
assert f"/{test_service_offering.pk}/" in call_kwargs["description"]
assert (
call_kwargs["title"]
== f"Exoscale OSB Offboard - {test_service.name} - {instance_id}"
)
@pytest.mark.django_db
def test_patch_creates_ticket_with_user_admin_links(
mocker,
mock_odoo_success,
osb_client,
test_service,
test_service_offering,
instance_id,
org_owner,
):
# Mock the create_helpdesk_ticket function
mock_create_ticket = mocker.patch("servala.api.views.create_helpdesk_ticket")
payload = {
"service_id": test_service.osb_service_id,
"plan_id": test_service_offering.osb_plan_id,
"parameters": {
"users": [
{
"email": org_owner.email,
"full_name": "Test User",
"role": "owner",
}
]
},
}
response = osb_client.patch(
f"/api/osb/v2/service_instances/{instance_id}",
data=json.dumps(payload),
content_type="application/json",
)
assert response.status_code == 200
# Verify the ticket was created with admin URLs
mock_create_ticket.assert_called_once()
call_kwargs = mock_create_ticket.call_args[1]
# Check that the description contains admin URLs
assert "admin/core/serviceoffering" in call_kwargs["description"]
assert "admin/core/user" in call_kwargs["description"]
assert f"/{org_owner.pk}/" in call_kwargs["description"]
assert (
call_kwargs["title"]
== f"Exoscale OSB Suspend - {test_service.name} - {instance_id}"
)
@pytest.mark.django_db
def test_ticket_includes_organization_and_instance_when_found(
mocker,
mock_odoo_success,
osb_client,
test_service,
test_service_offering,
organization,
):
# Mock the create_helpdesk_ticket function
mock_create_ticket = mocker.patch("servala.api.views.create_helpdesk_ticket")
service_definition = ServiceDefinition.objects.create(
name="Test Definition",
service=test_service,
api_definition={"group": "test.example.com", "version": "v1", "kind": "Test"},
)
control_plane = ControlPlane.objects.create(
name="Test Control Plane",
cloud_provider=test_service_offering.provider,
api_credentials={
"certificate-authority-data": "test",
"server": "https://test",
"token": "test",
},
)
crd = ControlPlaneCRD.objects.create(
service_offering=test_service_offering,
control_plane=control_plane,
service_definition=service_definition,
)
instance_name = "test-instance-123"
service_instance = ServiceInstance.objects.create(
name=instance_name,
organization=organization,
context=crd,
)
response = osb_client.delete(
f"/api/osb/v2/service_instances/{instance_name}"
f"?service_id={test_service.osb_service_id}&plan_id={test_service_offering.osb_plan_id}"
)
assert response.status_code == 200
# Verify the ticket was created with all admin URLs
mock_create_ticket.assert_called_once()
call_kwargs = mock_create_ticket.call_args[1]
# Check organization is included
assert f"Organization: {organization.name}" in call_kwargs["description"]
assert "admin/core/organization" in call_kwargs["description"]
assert f"/{organization.pk}/" in call_kwargs["description"]
# Check instance is included
assert f"Instance: {service_instance.name}" in call_kwargs["description"]
assert "admin/core/serviceinstance" in call_kwargs["description"]
assert f"/{service_instance.pk}/" in call_kwargs["description"]