Test new Exoscale API
This commit is contained in:
parent
039b4a7031
commit
2e7143245a
2 changed files with 509 additions and 0 deletions
|
|
@ -1,3 +1,5 @@
|
|||
import base64
|
||||
|
||||
import pytest
|
||||
|
||||
from servala.core.models import (
|
||||
|
|
@ -6,6 +8,13 @@ from servala.core.models import (
|
|||
OrganizationOrigin,
|
||||
User,
|
||||
)
|
||||
from servala.core.models.service import (
|
||||
CloudProvider,
|
||||
Plan,
|
||||
Service,
|
||||
ServiceCategory,
|
||||
ServiceOffering,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
|
@ -30,3 +39,86 @@ def org_owner(organization):
|
|||
organization=organization, user=user, role="owner"
|
||||
)
|
||||
return user
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_service_category():
|
||||
return ServiceCategory.objects.create(
|
||||
name="Databases",
|
||||
description="Database services",
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_service(test_service_category):
|
||||
return Service.objects.create(
|
||||
name="Redis",
|
||||
slug="redis",
|
||||
category=test_service_category,
|
||||
description="Redis database service",
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_cloud_provider():
|
||||
return CloudProvider.objects.create(
|
||||
name="Exoscale",
|
||||
description="Exoscale cloud provider",
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_service_offering(test_service, test_cloud_provider):
|
||||
return ServiceOffering.objects.create(
|
||||
service=test_service,
|
||||
provider=test_cloud_provider,
|
||||
description="Redis on Exoscale",
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_plan(test_service_offering):
|
||||
return Plan.objects.create(
|
||||
name="Small",
|
||||
description="Small Redis plan",
|
||||
term=1,
|
||||
service_offering=test_service_offering,
|
||||
features={"memory": "1GB", "connections": 100},
|
||||
pricing={"monthly": 10.0},
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def osb_client(client):
|
||||
credentials = base64.b64encode(b"testuser:testpass").decode("ascii")
|
||||
client.defaults = {"HTTP_AUTHORIZATION": f"Basic {credentials}"}
|
||||
return client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_odoo_success(mocker):
|
||||
"""
|
||||
Mock Odoo client with successful responses for organization creation.
|
||||
Returns the mock object for further customization if needed.
|
||||
"""
|
||||
mock_client = mocker.patch("servala.core.models.organization.CLIENT")
|
||||
|
||||
# Default successful responses for organization creation
|
||||
mock_client.execute.side_effect = [
|
||||
123, # company_id
|
||||
456, # invoice_address_id
|
||||
789, # sale_order_id
|
||||
]
|
||||
mock_client.search_read.return_value = [{"name": "SO001"}]
|
||||
|
||||
return mock_client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_odoo_failure(mocker):
|
||||
"""
|
||||
Mock Odoo client that raises an exception to simulate failure.
|
||||
"""
|
||||
mock_client = mocker.patch("servala.core.models.organization.CLIENT")
|
||||
mock_client.execute.side_effect = Exception("Odoo connection failed")
|
||||
return mock_client
|
||||
|
|
|
|||
417
src/tests/test_api_exoscale.py
Normal file
417
src/tests/test_api_exoscale.py
Normal file
|
|
@ -0,0 +1,417 @@
|
|||
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_plan,
|
||||
valid_osb_payload,
|
||||
exoscale_origin,
|
||||
instance_id,
|
||||
):
|
||||
valid_osb_payload["service_id"] = test_service.id
|
||||
valid_osb_payload["plan_id"] = test_plan.id
|
||||
|
||||
response = osb_client.put(
|
||||
f"/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-")
|
||||
|
||||
user = User.objects.get(email="test@example.com")
|
||||
assert user.first_name == "Test"
|
||||
assert user.last_name == "User"
|
||||
with scopes_disabled():
|
||||
membership = org.memberships.get(user=user)
|
||||
assert membership.role == "owner"
|
||||
|
||||
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 len(mail.outbox) == 2
|
||||
invitation_email = mail.outbox[0]
|
||||
assert invitation_email.subject == "Welcome to Servala - Test Organization Display"
|
||||
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_duplicate_organization_returns_existing(
|
||||
osb_client,
|
||||
test_service,
|
||||
test_plan,
|
||||
valid_osb_payload,
|
||||
exoscale_origin,
|
||||
instance_id,
|
||||
):
|
||||
Organization.objects.create(
|
||||
name="Existing Org",
|
||||
osb_guid="test-org-guid-123",
|
||||
origin=exoscale_origin,
|
||||
)
|
||||
|
||||
valid_osb_payload["service_id"] = test_service.id
|
||||
valid_osb_payload["plan_id"] = test_plan.id
|
||||
|
||||
response = osb_client.put(
|
||||
f"/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) == 1 # Only one email was sent
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_unauthenticated_osb_api_request_fails(
|
||||
client,
|
||||
test_service,
|
||||
test_plan,
|
||||
valid_osb_payload,
|
||||
instance_id,
|
||||
):
|
||||
valid_osb_payload["service_id"] = test_service.id
|
||||
valid_osb_payload["plan_id"] = test_plan.id
|
||||
|
||||
response = client.put(
|
||||
f"/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_plan,
|
||||
valid_osb_payload,
|
||||
field_to_remove,
|
||||
expected_error,
|
||||
instance_id,
|
||||
):
|
||||
valid_osb_payload["service_id"] = test_service.id
|
||||
valid_osb_payload["plan_id"] = test_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"/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"/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.id
|
||||
valid_osb_payload["plan_id"] = 99999
|
||||
|
||||
response = osb_client.put(
|
||||
f"/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.id}"
|
||||
in response_data["error"]
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
def test_empty_users_array_error(
|
||||
osb_client, test_service, test_plan, valid_osb_payload, instance_id
|
||||
):
|
||||
valid_osb_payload["service_id"] = test_service.id
|
||||
valid_osb_payload["plan_id"] = test_plan.id
|
||||
valid_osb_payload["parameters"]["users"] = []
|
||||
|
||||
response = osb_client.put(
|
||||
f"/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_plan, valid_osb_payload, instance_id
|
||||
):
|
||||
valid_osb_payload["service_id"] = test_service.id
|
||||
valid_osb_payload["plan_id"] = test_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"/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_plan, valid_osb_payload, instance_id
|
||||
):
|
||||
valid_osb_payload["service_id"] = test_service.id
|
||||
valid_osb_payload["plan_id"] = test_plan.id
|
||||
valid_osb_payload["parameters"]["users"] = [
|
||||
{"email": "", "full_name": "User With No Email"},
|
||||
]
|
||||
|
||||
response = osb_client.put(
|
||||
f"/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"/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_plan,
|
||||
valid_osb_payload,
|
||||
exoscale_origin,
|
||||
instance_id,
|
||||
):
|
||||
valid_osb_payload["service_id"] = test_service.id
|
||||
valid_osb_payload["plan_id"] = test_plan.id
|
||||
valid_osb_payload["parameters"]["users"][0]["full_name"] = "John Doe Smith"
|
||||
|
||||
response = osb_client.put(
|
||||
f"/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_plan,
|
||||
valid_osb_payload,
|
||||
exoscale_origin,
|
||||
instance_id,
|
||||
):
|
||||
valid_osb_payload["service_id"] = test_service.id
|
||||
valid_osb_payload["plan_id"] = test_plan.id
|
||||
valid_osb_payload["parameters"]["users"][0]["email"] = " TEST@EXAMPLE.COM "
|
||||
|
||||
response = osb_client.put(
|
||||
f"/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_plan,
|
||||
valid_osb_payload,
|
||||
exoscale_origin,
|
||||
instance_id,
|
||||
):
|
||||
valid_osb_payload["service_id"] = test_service.id
|
||||
valid_osb_payload["plan_id"] = test_plan.id
|
||||
|
||||
response = osb_client.put(
|
||||
f"/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_plan, exoscale_origin, instance_id
|
||||
):
|
||||
payload = {
|
||||
"service_id": test_service.id,
|
||||
"plan_id": test_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"/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
|
||||
Loading…
Add table
Add a link
Reference in a new issue