453 lines
14 KiB
Python
453 lines
14 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
|
|
|
|
|
|
@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]]
|
|
elif field_to_remove[0] == "parameters":
|
|
del valid_osb_payload["parameters"][field_to_remove[1]]
|
|
else:
|
|
if field_to_remove in valid_osb_payload:
|
|
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
|