diff --git a/src/tests/conftest.py b/src/tests/conftest.py index d88870e..3d176ad 100644 --- a/src/tests/conftest.py +++ b/src/tests/conftest.py @@ -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 diff --git a/src/tests/test_api_exoscale.py b/src/tests/test_api_exoscale.py new file mode 100644 index 0000000..fa6fc02 --- /dev/null +++ b/src/tests/test_api_exoscale.py @@ -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