diff --git a/pyproject.toml b/pyproject.toml index 45be93d..c62bd58 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -56,10 +56,14 @@ extend_exclude = "src/servala/static/mazer" [tool.pytest.ini_options] DJANGO_SETTINGS_MODULE = "servala.settings_test" -addopts = "-p no:doctest -p no:pastebin -p no:nose --cov=./ --cov-report=term-missing:skip-covered" +addopts = "-p no:doctest -p no:pastebin -p no:nose --cov=./ --cov-report=term-missing:skip-covered --cov-config=pyproject.toml" testpaths = "src/tests" pythonpath = "src" +[tool.coverage.run] +branch = true +omit = ["*/admin.py","*/settings.py","*/wsgi.py", "*/migrations/*", "*/manage.py", "*/__init__.py"] + [tool.bumpver] current_version = "2025.11.17-0" version_pattern = "YYYY.0M.0D-INC0" diff --git a/src/servala/api/views.py b/src/servala/api/views.py index 015e091..9b0082d 100644 --- a/src/servala/api/views.py +++ b/src/servala/api/views.py @@ -239,9 +239,9 @@ The Servala Team""" service_offering = ServiceOffering.objects.get( osb_plan_id=plan_id, service=service ) - except Service.DoesNotExist: # pragma: no-cover + except Service.DoesNotExist: # pragma: no cover return self._error(f"Unknown service_id: {service_id}") - except ServiceOffering.DoesNotExist: # pragma: no-cover + except ServiceOffering.DoesNotExist: # pragma: no cover return self._error( f"Unknown plan_id: {plan_id} for service_id: {service_id}" ) diff --git a/src/servala/core/admin.py b/src/servala/core/admin.py index 29ddb2f..b686e2f 100644 --- a/src/servala/core/admin.py +++ b/src/servala/core/admin.py @@ -92,13 +92,43 @@ class OrganizationOriginAdmin(admin.ModelAdmin): list_display = ( "name", "billing_entity", + "invoice_grouping", "default_odoo_sale_order_id", "hide_billing_address", ) - list_filter = ("hide_billing_address",) + list_filter = ("hide_billing_address", "invoice_grouping") search_fields = ("name",) autocomplete_fields = ("billing_entity",) filter_horizontal = ("limit_cloudproviders",) + fieldsets = ( + ( + None, + { + "fields": ( + "name", + "description", + ) + }, + ), + ( + _("Billing"), + { + "fields": ( + "billing_entity", + "default_odoo_sale_order_id", + "invoice_grouping", + "hide_billing_address", + "billing_message", + ) + }, + ), + ( + _("Restrictions"), + { + "fields": ("limit_cloudproviders",), + }, + ), + ) @admin.register(OrganizationMembership) diff --git a/src/servala/core/crd/forms.py b/src/servala/core/crd/forms.py index 188ce3e..18df8bc 100644 --- a/src/servala/core/crd/forms.py +++ b/src/servala/core/crd/forms.py @@ -357,6 +357,16 @@ class CustomFormMixin(FormGeneratorMixin): max_val = field_config.get("max_value") unit = field_config.get("addon_text") + if not isinstance(field, forms.IntegerField): + widget = field.widget + self.fields[field_name] = forms.IntegerField( + label=field.label, + help_text=field.help_text, + required=field.required, + widget=widget, + ) + field = self.fields[field_name] + if unit: field.widget = NumberInputWithAddon(addon_text=unit) field.addon_text = unit diff --git a/src/servala/core/management/commands/make_staff_user.py b/src/servala/core/management/commands/make_staff_user.py index a8b76a2..1c09eff 100644 --- a/src/servala/core/management/commands/make_staff_user.py +++ b/src/servala/core/management/commands/make_staff_user.py @@ -30,11 +30,9 @@ class Command(BaseCommand): query = None if substring_match: + query = Q(email__icontains=emails[0]) for email in emails: - if query is None: - query = Q(email__icontains=email) - else: - query |= Q(email__icontains=email) + query |= Q(email__icontains=email) else: query = Q(email__in=emails) diff --git a/src/servala/core/management/commands/sync_billing_metadata.py b/src/servala/core/management/commands/sync_billing_metadata.py new file mode 100644 index 0000000..2093948 --- /dev/null +++ b/src/servala/core/management/commands/sync_billing_metadata.py @@ -0,0 +1,254 @@ +import kubernetes +from django.core.management.base import BaseCommand +from django_scopes import scopes_disabled + +from servala.core.models import ControlPlane, Organization, ServiceInstance + + +class Command(BaseCommand): + help = ( + "Sync billing labels and annotations to organization namespaces " + "and service instances on all or selected control planes." + ) + + def add_arguments(self, parser): + parser.add_argument( + "--control-plane", + type=int, + action="append", + dest="control_plane_ids", + help="Control plane ID to sync (can be specified multiple times). " + "If not specified, syncs all control planes.", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Show what would be changed without making actual changes.", + ) + parser.add_argument( + "--namespaces-only", + action="store_true", + help="Only sync organization namespace labels/annotations, skip service instances.", + ) + parser.add_argument( + "--instances-only", + action="store_true", + help="Only sync service instance annotations, skip namespaces.", + ) + + @scopes_disabled() + def handle(self, *args, **options): + control_plane_ids = options.get("control_plane_ids") + dry_run = options.get("dry_run", False) + namespaces_only = options.get("namespaces_only", False) + instances_only = options.get("instances_only", False) + + if namespaces_only and instances_only: + self.stdout.write( + self.style.ERROR( + "Cannot use both --namespaces-only and --instances-only" + ) + ) + return + + if dry_run: + self.stdout.write(self.style.WARNING("DRY RUN - no changes will be made")) + + if control_plane_ids: + control_planes = ControlPlane.objects.filter(id__in=control_plane_ids) + if not control_planes.exists(): # pragma: no cover + self.stdout.write( + self.style.ERROR("No control planes found with the specified IDs.") + ) + return + else: + control_planes = ControlPlane.objects.all() + + self.stdout.write( + f"Syncing billing metadata on {control_planes.count()} control plane(s)..." + ) + + for control_plane in control_planes: + self.stdout.write( + f"\nControl Plane: {control_plane.name} (ID: {control_plane.pk})" + ) + + if not instances_only: + self._sync_namespaces(control_plane, dry_run) + + if not namespaces_only: + self._sync_instances(control_plane, dry_run) + + self.stdout.write(self.style.SUCCESS("\nSync completed.")) + + def _build_namespace_metadata(self, organization): + labels = { + "servala.com/organization_id": str(organization.id), + } + annotations = { + "servala.com/organization": organization.name, + "servala.com/origin": organization.origin.name, + } + + if organization.billing_entity: + annotations["servala.com/billing"] = organization.billing_entity.name + + for field in ("company_id", "invoice_id"): + if value := getattr(organization.billing_entity, f"odoo_{field}"): + labels[f"servala.com/erp_{field}"] = str(value) + + if organization.odoo_sale_order_id: + labels["servala.com/erp_sale_order_id"] = str( + organization.odoo_sale_order_id + ) + + return labels, annotations + + def _sync_namespaces(self, control_plane, dry_run): + self.stdout.write(" Syncing organization namespaces...") + + try: + api_instance = kubernetes.client.CoreV1Api( + control_plane.get_kubernetes_client() + ) + except Exception as e: + self.stdout.write( + self.style.ERROR(f" Failed to connect to control plane: {e}") + ) + return + + organizations = Organization.objects.select_related( + "origin", "billing_entity" + ).all() + + synced = 0 + skipped = 0 + errors = 0 + + for org in organizations: + if not org.namespace: + continue + + try: + try: + api_instance.read_namespace(name=org.namespace) + except kubernetes.client.ApiException as e: + if e.status == 404: + skipped += 1 + continue + raise + + labels, annotations = self._build_namespace_metadata(org) + + if dry_run: + self.stdout.write( + f" [DRY RUN] Would update namespace {org.namespace}" + ) + self.stdout.write(f" Labels: {labels}") + self.stdout.write(f" Annotations: {annotations}") + else: + body = { + "metadata": { + "labels": labels, + "annotations": annotations, + } + } + api_instance.patch_namespace(name=org.namespace, body=body) + self.stdout.write(f" Updated namespace {org.namespace}") + + synced += 1 + + except Exception as e: + self.stdout.write( + self.style.ERROR( + f" Error syncing namespace {org.namespace}: {e}" + ) + ) + errors += 1 + + self.stdout.write( + f" Namespaces: {synced} synced, {skipped} skipped (not found), {errors} errors" + ) + + def _sync_instances(self, control_plane, dry_run): + self.stdout.write(" Syncing service instance annotations...") + instances = ServiceInstance.objects.filter( + context__control_plane=control_plane + ).select_related( + "organization", + "organization__origin", + "context", + "context__control_plane", + "context__control_plane__cloud_provider", + "context__service_offering", + "context__service_offering__service", + "compute_plan_assignment", + ) + + synced = 0 + skipped = 0 + errors = 0 + + for instance in instances: + try: + annotations = ServiceInstance._build_billing_annotations( + compute_plan_assignment=instance.compute_plan_assignment, + control_plane=instance.context.control_plane, + instance_name=instance.name, + organization=instance.organization, + service=instance.context.service_offering.service, + ) + + if not annotations: + skipped += 1 + continue + + if dry_run: + self.stdout.write( + f" [DRY RUN] Would update instance {instance.name} " + f"(org: {instance.organization.name})" + ) + self.stdout.write(f" Annotations: {annotations}") + else: + api_instance = instance.context.control_plane.custom_objects_api + patch_body = {"metadata": {"annotations": annotations}} + + try: + api_instance.patch_namespaced_custom_object( + group=instance.context.group, + version=instance.context.version, + namespace=instance.organization.namespace, + plural=instance.context.kind_plural, + name=instance.name, + body=patch_body, + ) + self.stdout.write( + f" Updated instance {instance.name} " + f"(org: {instance.organization.name})" + ) + except kubernetes.client.ApiException as e: + if e.status == 404: + self.stdout.write( + self.style.WARNING( + f" Instance {instance.name} not found in Kubernetes " + f"(org: {instance.organization.name})" + ) + ) + skipped += 1 + continue + raise + + synced += 1 + + except Exception as e: + self.stdout.write( + self.style.ERROR( + f" Error syncing instance {instance.name} " + f"(org: {instance.organization.name}): {e}" + ) + ) + errors += 1 + + self.stdout.write( + f" Instances: {synced} synced, {skipped} skipped, {errors} errors" + ) diff --git a/src/servala/core/migrations/0018_add_invoice_grouping_to_organization_origin.py b/src/servala/core/migrations/0018_add_invoice_grouping_to_organization_origin.py new file mode 100644 index 0000000..2e8b97a --- /dev/null +++ b/src/servala/core/migrations/0018_add_invoice_grouping_to_organization_origin.py @@ -0,0 +1,43 @@ +# Generated by Django 5.2.9 on 2025-12-04 12:46 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("core", "0017_add_unit_and_convert_odoo_ids_to_charfield"), + ] + + operations = [ + migrations.AddField( + model_name="organizationorigin", + name="invoice_grouping", + field=models.CharField( + choices=[ + ("by_service", "By Service"), + ("by_organization", "By Organization"), + ], + default="by_service", + help_text="Determines how service instances are grouped on invoices.", + max_length=20, + verbose_name="Invoice Line Item Grouping", + ), + ), + migrations.AlterField( + model_name="computeplanassignment", + name="unit", + field=models.CharField( + choices=[ + ("hour", "Hour"), + ("day", "Day"), + ("month", "Month (30 days / 720 hours)"), + ("year", "Year"), + ], + default="hour", + help_text="Unit for the price (e.g., price per hour)", + max_length=10, + verbose_name="Billing unit", + ), + ), + ] diff --git a/src/servala/core/models/__init__.py b/src/servala/core/models/__init__.py index 4cb8fd7..2a2e9d4 100644 --- a/src/servala/core/models/__init__.py +++ b/src/servala/core/models/__init__.py @@ -1,6 +1,7 @@ from .odoo_cache import OdooObjectCache from .organization import ( BillingEntity, + InvoiceGroupingChoice, Organization, OrganizationInvitation, OrganizationMembership, @@ -30,6 +31,7 @@ __all__ = [ "ComputePlanAssignment", "ControlPlane", "ControlPlaneCRD", + "InvoiceGroupingChoice", "OdooObjectCache", "Organization", "OrganizationInvitation", diff --git a/src/servala/core/models/organization.py b/src/servala/core/models/organization.py index be4f587..f3d6c5d 100644 --- a/src/servala/core/models/organization.py +++ b/src/servala/core/models/organization.py @@ -382,6 +382,11 @@ class BillingEntity(ServalaModelMixin, models.Model): return data +class InvoiceGroupingChoice(models.TextChoices): + BY_SERVICE = "by_service", _("By Service") + BY_ORGANIZATION = "by_organization", _("By Organization") + + class OrganizationOrigin(ServalaModelMixin, models.Model): """ Every organization has an origin, though origins may be @@ -433,6 +438,13 @@ class OrganizationOrigin(ServalaModelMixin, models.Model): "Optional message to display instead of billing address (e.g., 'You will be invoiced by Exoscale')." ), ) + invoice_grouping = models.CharField( + max_length=20, + choices=InvoiceGroupingChoice.choices, + default=InvoiceGroupingChoice.BY_SERVICE, + verbose_name=_("Invoice Line Item Grouping"), + help_text=_("Determines how service instances are grouped on invoices. "), + ) class Meta: verbose_name = _("Organization origin") diff --git a/src/servala/core/models/service.py b/src/servala/core/models/service.py index 3465d54..c6552c4 100644 --- a/src/servala/core/models/service.py +++ b/src/servala/core/models/service.py @@ -715,10 +715,18 @@ class ServiceInstance(ServalaModelMixin, models.Model): return spec_data @staticmethod - def _build_billing_annotations(compute_plan_assignment, control_plane): + def _build_billing_annotations( + compute_plan_assignment, + control_plane, + instance_name=None, + organization=None, + service=None, + ): """ Build Kubernetes annotations for billing integration. """ + from servala.core.models.organization import InvoiceGroupingChoice + annotations = {} if compute_plan_assignment: @@ -738,6 +746,32 @@ class ServiceInstance(ServalaModelMixin, models.Model): control_plane.storage_plan_odoo_unit_id ) + if organization and instance_name: + invoice_grouping = organization.origin.invoice_grouping + cloud_provider_name = control_plane.cloud_provider.name + control_plane_name = control_plane.name + + if invoice_grouping == InvoiceGroupingChoice.BY_SERVICE: + if service: + annotations["servala.com/erp_item_group_description"] = ( + f"Servala Service: {service.name}" + ) + annotations["servala.com/erp_item_description"] = ( + f"{instance_name} on {cloud_provider_name} {control_plane_name}" + ) + else: + group_description = f"Organization: {organization.name}" + item_description = f"{instance_name} on {control_plane_name}" + + if organization.osb_guid: + group_description += f" ({organization.osb_guid})" + item_description += f" [Org: {organization.osb_guid}]" + + annotations["servala.com/erp_item_group_description"] = ( + group_description + ) + annotations["servala.com/erp_item_description"] = item_description + return annotations @classmethod @@ -846,7 +880,11 @@ class ServiceInstance(ServalaModelMixin, models.Model): } annotations = cls._build_billing_annotations( - compute_plan_assignment, context.control_plane + compute_plan_assignment=compute_plan_assignment, + control_plane=context.control_plane, + instance_name=name, + organization=organization, + service=context.service_offering.service, ) if annotations: create_data["metadata"]["annotations"] = annotations @@ -900,7 +938,11 @@ class ServiceInstance(ServalaModelMixin, models.Model): patch_body = {"spec": spec_data} annotations = self._build_billing_annotations( - plan_to_use, self.context.control_plane + compute_plan_assignment=plan_to_use, + control_plane=self.context.control_plane, + instance_name=self.name, + organization=self.organization, + service=self.context.service_offering.service, ) if annotations: patch_body["metadata"] = {"annotations": annotations} diff --git a/src/servala/urls.py b/src/servala/urls.py index 9eb6705..d932643 100644 --- a/src/servala/urls.py +++ b/src/servala/urls.py @@ -24,7 +24,7 @@ urlpatterns = [ ] # Serve static and media files in development -if settings.DEBUG: +if settings.DEBUG: # pragma: no cover urlpatterns += static(settings.STATIC_URL, document_root=settings.STATIC_ROOT) urlpatterns += static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT) urlpatterns += [ diff --git a/src/setup.cfg b/src/setup.cfg new file mode 100644 index 0000000..48a09e7 --- /dev/null +++ b/src/setup.cfg @@ -0,0 +1,4 @@ +[flake8] +ignore = E203, E231, E266, E501, W503, W605, B028, R502, R503 +max-line-length = 160 +exclude = migrations,static,_static,build,*settings.py,.tox/*,local diff --git a/src/tests/conftest.py b/src/tests/conftest.py index 09db220..859b4ca 100644 --- a/src/tests/conftest.py +++ b/src/tests/conftest.py @@ -4,6 +4,8 @@ import pytest from servala.core.models import ( BillingEntity, + ComputePlan, + ComputePlanAssignment, Organization, OrganizationMembership, OrganizationOrigin, @@ -11,8 +13,11 @@ from servala.core.models import ( ) from servala.core.models.service import ( CloudProvider, + ControlPlane, + ControlPlaneCRD, Service, ServiceCategory, + ServiceDefinition, ServiceOffering, ) @@ -117,3 +122,68 @@ def mock_odoo_failure(mocker): mock_client = mocker.patch("servala.core.models.organization.CLIENT") mock_client.execute.side_effect = Exception("Odoo connection failed") return mock_client + + +@pytest.fixture +def test_control_plane(test_cloud_provider): + return ControlPlane.objects.create( + name="Geneva (CH-GVA-2)", + description="Geneva control plane", + cloud_provider=test_cloud_provider, + api_credentials={ + "server": "https://k8s.example.com", + "token": "test-token", + "certificate_authority_data": "test-ca-data", + }, + ) + + +@pytest.fixture +def test_service_definition(test_service): + return ServiceDefinition.objects.create( + name="Redis Standard", + service=test_service, + api_definition={ + "group": "vshn.appcat.vshn.io", + "version": "v1", + "kind": "VSHNRedis", + }, + ) + + +@pytest.fixture +def test_control_plane_crd( + test_service_offering, test_control_plane, test_service_definition +): + return ControlPlaneCRD.objects.create( + service_offering=test_service_offering, + control_plane=test_control_plane, + service_definition=test_service_definition, + ) + + +@pytest.fixture +def compute_plan(): + return ComputePlan.objects.create( + name="Medium", + description="Medium resource plan", + memory_requests="1Gi", + memory_limits="2Gi", + cpu_requests="500m", + cpu_limits="1000m", + is_active=True, + ) + + +@pytest.fixture +def compute_plan_assignment(compute_plan, test_control_plane_crd): + return ComputePlanAssignment.objects.create( + compute_plan=compute_plan, + control_plane_crd=test_control_plane_crd, + sla="besteffort", + odoo_product_id="test-product-id", + odoo_unit_id="test-unit-id", + price="10.00", + unit="hour", + is_active=True, + ) diff --git a/src/tests/test_api_exoscale.py b/src/tests/test_api_exoscale.py index 19f8b93..37542e5 100644 --- a/src/tests/test_api_exoscale.py +++ b/src/tests/test_api_exoscale.py @@ -218,11 +218,10 @@ def test_missing_required_fields_error( 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": + else: 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] + del valid_osb_payload[field_to_remove] response = osb_client.put( f"/api/osb/v2/service_instances/{instance_id}", diff --git a/src/tests/test_compute_plans.py b/src/tests/test_compute_plans.py index 0317229..283686c 100644 --- a/src/tests/test_compute_plans.py +++ b/src/tests/test_compute_plans.py @@ -5,6 +5,7 @@ import pytest from servala.core.models import ( ComputePlan, ComputePlanAssignment, + InvoiceGroupingChoice, ServiceInstance, ) @@ -125,7 +126,8 @@ def test_build_billing_annotations_complete(): control_plane.storage_plan_odoo_unit_id = "storage-unit-id" annotations = ServiceInstance._build_billing_annotations( - compute_plan_assignment, control_plane + compute_plan_assignment=compute_plan_assignment, + control_plane=control_plane, ) assert annotations["servala.com/erp_product_id_resource"] == "test-product-123" @@ -140,7 +142,10 @@ def test_build_billing_annotations_no_compute_plan(): control_plane.storage_plan_odoo_product_id = "storage-product-id" control_plane.storage_plan_odoo_unit_id = "storage-unit-id" - annotations = ServiceInstance._build_billing_annotations(None, control_plane) + annotations = ServiceInstance._build_billing_annotations( + compute_plan_assignment=None, + control_plane=control_plane, + ) assert "servala.com/erp_product_id_resource" not in annotations assert "servala.com/erp_unit_id_resource" not in annotations @@ -158,7 +163,8 @@ def test_build_billing_annotations_no_storage_plan(): control_plane.storage_plan_odoo_unit_id = None annotations = ServiceInstance._build_billing_annotations( - compute_plan_assignment, control_plane + compute_plan_assignment=compute_plan_assignment, + control_plane=control_plane, ) assert annotations["servala.com/erp_product_id_resource"] == "product-id" @@ -172,28 +178,183 @@ def test_build_billing_annotations_empty(): control_plane.storage_plan_odoo_product_id = None control_plane.storage_plan_odoo_unit_id = None - annotations = ServiceInstance._build_billing_annotations(None, control_plane) + annotations = ServiceInstance._build_billing_annotations( + compute_plan_assignment=None, + control_plane=control_plane, + ) assert annotations == {} @pytest.mark.django_db -def test_hour_unit(): +@pytest.mark.parametrize( + "unit_key,expected_display", + [ + ("hour", "Hour"), + ("day", "Day"), + ("year", "Year"), + ], +) +def test_billing_unit_choices(unit_key, expected_display): choices = dict(ComputePlanAssignment.BILLING_UNIT_CHOICES) - assert "hour" in choices - assert str(choices["hour"]) == "Hour" + assert unit_key in choices + assert str(choices[unit_key]) == expected_display @pytest.mark.django_db -def test_all_billing_units(): +def test_billing_unit_month(): + """Month has a special display format with additional info.""" choices = dict(ComputePlanAssignment.BILLING_UNIT_CHOICES) - - assert "hour" in choices - assert "day" in choices assert "month" in choices - assert "year" in choices - - assert str(choices["hour"]) == "Hour" - assert str(choices["day"]) == "Day" assert "Month" in str(choices["month"]) - assert str(choices["year"]) == "Year" + + +@pytest.mark.parametrize( + "invoice_grouping,org_name,osb_guid,expected_group_desc,expected_item_desc", + [ + pytest.param( + InvoiceGroupingChoice.BY_SERVICE, + "ACME Corp", + "some-guid", + "Servala Service: Redis", + "MyProdRedis on Exoscale Geneva (CH-GVA-2)", + id="by_service", + ), + pytest.param( + InvoiceGroupingChoice.BY_ORGANIZATION, + "ACME", + "01998651-dc86-7d43-9e49-cdb790fcc4f0", + "Organization: ACME (01998651-dc86-7d43-9e49-cdb790fcc4f0)", + "MyProdRedis on Geneva (CH-GVA-2) [Org: 01998651-dc86-7d43-9e49-cdb790fcc4f0]", + id="by_organization_with_guid", + ), + pytest.param( + InvoiceGroupingChoice.BY_ORGANIZATION, + "ACME Corp", + None, + "Organization: ACME Corp", + "MyProdRedis on Geneva (CH-GVA-2)", + id="by_organization_without_guid", + ), + pytest.param( + InvoiceGroupingChoice.BY_ORGANIZATION, + "ACME Corp", + "", + "Organization: ACME Corp", + "MyProdRedis on Geneva (CH-GVA-2)", + id="by_organization_with_empty_guid", + ), + ], +) +def test_build_billing_annotations_item_description( + invoice_grouping, org_name, osb_guid, expected_group_desc, expected_item_desc +): + control_plane = Mock() + control_plane.storage_plan_odoo_product_id = None + control_plane.storage_plan_odoo_unit_id = None + control_plane.cloud_provider = Mock() + control_plane.cloud_provider.name = "Exoscale" + control_plane.name = "Geneva (CH-GVA-2)" + + organization = Mock() + organization.name = org_name + organization.osb_guid = osb_guid + organization.origin = Mock() + organization.origin.invoice_grouping = invoice_grouping + + service = Mock() + service.name = "Redis" + + annotations = ServiceInstance._build_billing_annotations( + compute_plan_assignment=None, + control_plane=control_plane, + instance_name="MyProdRedis", + organization=organization, + service=service, + ) + + assert annotations["servala.com/erp_item_group_description"] == expected_group_desc + assert annotations["servala.com/erp_item_description"] == expected_item_desc + + +def test_build_billing_annotations_no_item_description_without_organization(): + control_plane = Mock() + control_plane.storage_plan_odoo_product_id = "storage-product" + control_plane.storage_plan_odoo_unit_id = "storage-unit" + + annotations = ServiceInstance._build_billing_annotations( + compute_plan_assignment=None, + control_plane=control_plane, + instance_name="MyProdRedis", + organization=None, + service=None, + ) + + assert "servala.com/erp_item_group_description" not in annotations + assert "servala.com/erp_item_description" not in annotations + assert annotations["servala.com/erp_product_id_storage"] == "storage-product" + + +@pytest.mark.django_db +def test_compute_plan_assignment_str( + compute_plan_assignment, +): + result = str(compute_plan_assignment) + # Format: "{plan_name} ({sla_display}) → {control_plane_crd}" + assert compute_plan_assignment.compute_plan.name in result + assert "→" in result + + +@pytest.mark.django_db +def test_compute_plan_assignment_get_odoo_reporting_product_id( + compute_plan_assignment, +): + compute_plan_assignment.odoo_product_id = "test-product-xyz" + compute_plan_assignment.save() + + result = compute_plan_assignment.get_odoo_reporting_product_id() + + assert result == "test-product-xyz" + + +def test_build_billing_annotations_combined(): + compute_plan_assignment = Mock() + compute_plan_assignment.odoo_product_id = "compute-product" + compute_plan_assignment.odoo_unit_id = "compute-unit" + + control_plane = Mock() + control_plane.storage_plan_odoo_product_id = "storage-product" + control_plane.storage_plan_odoo_unit_id = "storage-unit" + control_plane.cloud_provider = Mock() + control_plane.cloud_provider.name = "Exoscale" + control_plane.name = "Geneva (CH-GVA-2)" + + organization = Mock() + organization.name = "ACME" + organization.osb_guid = "test-guid" + organization.origin = Mock() + organization.origin.invoice_grouping = InvoiceGroupingChoice.BY_SERVICE + + service = Mock() + service.name = "PostgreSQL" + + annotations = ServiceInstance._build_billing_annotations( + compute_plan_assignment=compute_plan_assignment, + control_plane=control_plane, + instance_name="ProdDB", + organization=organization, + service=service, + ) + + assert annotations["servala.com/erp_product_id_resource"] == "compute-product" + assert annotations["servala.com/erp_unit_id_resource"] == "compute-unit" + assert annotations["servala.com/erp_product_id_storage"] == "storage-product" + assert annotations["servala.com/erp_unit_id_storage"] == "storage-unit" + assert ( + annotations["servala.com/erp_item_group_description"] + == "Servala Service: PostgreSQL" + ) + assert ( + annotations["servala.com/erp_item_description"] + == "ProdDB on Exoscale Geneva (CH-GVA-2)" + ) diff --git a/src/tests/test_form_config.py b/src/tests/test_form_config.py index 25f1bf2..014f1f2 100644 --- a/src/tests/test_form_config.py +++ b/src/tests/test_form_config.py @@ -5,7 +5,7 @@ from django.core.validators import MaxValueValidator, MinValueValidator from django.db import models from servala.core.crd import generate_custom_form_class -from servala.core.crd.forms import DEFAULT_FIELD_CONFIGS, MANDATORY_FIELDS +from servala.core.crd.forms import DEFAULT_FIELD_CONFIGS from servala.core.forms import ServiceDefinitionAdminForm from servala.core.models import ControlPlaneCRD @@ -806,10 +806,10 @@ def test_two_element_choices_work_correctly(): errors = [] for field in config_with_proper_choices["fieldsets"][0]["fields"]: - if field.get("type") == "choice": - form._validate_choice_field( - field, field["controlplane_field_mapping"], spec_schema, "spec", errors - ) + assert field["type"] == "choice" + form._validate_choice_field( + field, field["controlplane_field_mapping"], spec_schema, "spec", errors + ) assert len(errors) == 0, f"Expected no errors but got: {errors}" version_field = config_with_proper_choices["fieldsets"][0]["fields"][0] @@ -836,10 +836,10 @@ def test_empty_choices_fail_validation(): errors = [] for field in config_with_empty_choice["fieldsets"][0]["fields"]: - if field.get("type") == "choice": - form._validate_choice_field( - field, field["controlplane_field_mapping"], {}, "spec", errors - ) + assert field["type"] == "choice" + form._validate_choice_field( + field, field["controlplane_field_mapping"], {}, "spec", errors + ) assert len(errors) > 0 assert "must have 1 or 2 elements" in str(errors[0]) @@ -867,10 +867,10 @@ def test_three_plus_element_choices_fail_validation(): errors = [] for field in config_with_long_choice["fieldsets"][0]["fields"]: - if field.get("type") == "choice": - form._validate_choice_field( - field, field["controlplane_field_mapping"], {}, "spec", errors - ) + assert field["type"] == "choice" + form._validate_choice_field( + field, field["controlplane_field_mapping"], {}, "spec", errors + ) assert len(errors) > 0 assert "must have 1 or 2 elements" in str(errors[0]) @@ -936,88 +936,6 @@ def test_field_with_default_config_can_override_defaults(): assert name_field.help_text == DEFAULT_FIELD_CONFIGS["name"]["help_text"] -def test_admin_form_validates_mandatory_fields_present(): - - mock_crd = Mock() - mock_crd.resource_schema = { - "properties": { - "spec": { - "properties": { - "environment": { - "type": "string", - "enum": ["dev", "prod"], - } - } - } - } - } - - config_without_name = { - "fieldsets": [ - { - "fields": [ - { - "type": "choice", - "label": "Environment", - "controlplane_field_mapping": "spec.environment", - "choices": [["dev", "Development"]], - }, - ] - } - ] - } - - errors = [] - included_mappings = set() - for fieldset in config_without_name.get("fieldsets", []): - for field in fieldset.get("fields", []): - mapping = field.get("controlplane_field_mapping") - included_mappings.add(mapping) - - for mandatory_field in MANDATORY_FIELDS: - if mandatory_field not in included_mappings: - errors.append(f"Required field '{mandatory_field}' must be included") - - assert len(errors) > 0 - assert "name" in str(errors[0]).lower() - - -def test_admin_form_validates_fields_without_defaults_need_label_and_type(): - config_with_incomplete_field = { - "fieldsets": [ - { - "fields": [ - {"controlplane_field_mapping": "name"}, # Has defaults - OK - { - "controlplane_field_mapping": "spec.unknown", # No defaults - # Missing label and type - }, - ] - } - ] - } - - errors = [] - - for fieldset in config_with_incomplete_field.get("fieldsets", []): - for field in fieldset.get("fields", []): - mapping = field.get("controlplane_field_mapping") - - if mapping not in DEFAULT_FIELD_CONFIGS: - if not field.get("label"): - errors.append( - f"Field with mapping '{mapping}' must have a 'label' property" - ) - if not field.get("type"): - errors.append( - f"Field with mapping '{mapping}' must have a 'type' property" - ) - - assert len(errors) == 2 - assert any("label" in str(e) for e in errors) - assert any("type" in str(e) for e in errors) - - def test_empty_values_dont_override_default_configs(): class TestModel(models.Model): diff --git a/src/tests/test_management_commands.py b/src/tests/test_management_commands.py new file mode 100644 index 0000000..714d29a --- /dev/null +++ b/src/tests/test_management_commands.py @@ -0,0 +1,157 @@ +from io import StringIO + +import pytest +from django.core.management import call_command + +from servala.core.models import User + + +@pytest.mark.django_db +class TestMakeSuperuserCommand: + def test_make_superuser_success(self): + user = User.objects.create( + email="test@example.com", is_staff=False, is_superuser=False + ) + + out = StringIO() + call_command("make_superuser", "test@example.com", stdout=out) + + user.refresh_from_db() + assert user.is_superuser is True + assert user.is_staff is True + assert "is now a superuser" in out.getvalue() + + def test_make_superuser_case_insensitive(self): + user = User.objects.create( + email="Test@Example.com", is_staff=False, is_superuser=False + ) + + out = StringIO() + call_command("make_superuser", "TEST@EXAMPLE.COM", stdout=out) + + user.refresh_from_db() + assert user.is_superuser is True + + def test_make_superuser_user_not_found(self): + out = StringIO() + call_command("make_superuser", "nonexistent@example.com", stdout=out) + + assert "No matching user found" in out.getvalue() + + +@pytest.mark.django_db +class TestMakeStaffUserCommand: + def test_make_staff_user_success(self): + user = User.objects.create(email="staff@example.com", is_staff=False) + + out = StringIO() + call_command("make_staff_user", "staff@example.com", stdout=out) + + user.refresh_from_db() + assert user.is_staff is True + assert "Made 1 user(s) into staff users" in out.getvalue() + + def test_make_staff_user_multiple(self): + user1 = User.objects.create(email="staff1@example.com", is_staff=False) + user2 = User.objects.create(email="staff2@example.com", is_staff=False) + + out = StringIO() + call_command( + "make_staff_user", "staff1@example.com", "staff2@example.com", stdout=out + ) + + user1.refresh_from_db() + user2.refresh_from_db() + assert user1.is_staff is True + assert user2.is_staff is True + assert "Made 2 user(s) into staff users" in out.getvalue() + + def test_make_staff_user_no_emails(self): + out = StringIO() + call_command("make_staff_user", stdout=out) + + assert "No email addresses provided" in out.getvalue() + + def test_make_staff_user_already_staff(self): + User.objects.create(email="already@example.com", is_staff=True) + + out = StringIO() + call_command("make_staff_user", "already@example.com", stdout=out) + + output = out.getvalue() + assert "already staff" in output + assert "No matching non-staff users found" in output + + def test_make_staff_user_substring_match(self): + user1 = User.objects.create(email="user@company.com", is_staff=False) + user2 = User.objects.create(email="admin@company.com", is_staff=False) + user3 = User.objects.create(email="other@different.com", is_staff=False) + + out = StringIO() + call_command("make_staff_user", "@company.com", "--substring", stdout=out) + + user1.refresh_from_db() + user2.refresh_from_db() + user3.refresh_from_db() + assert user1.is_staff is True + assert user2.is_staff is True + assert user3.is_staff is False + assert "Made 2 user(s) into staff users" in out.getvalue() + + +@pytest.mark.django_db +class TestReencryptFieldsCommand: + def test_reencrypt_fields_no_control_planes(self): + out = StringIO() + call_command("reencrypt_fields", stdout=out) + + output = out.getvalue() + assert "Starting re-encryption" in output + assert "Re-encrypted 0 ControlPlane objects" in output + + def test_reencrypt_fields_with_control_plane(self, test_control_plane): + out = StringIO() + call_command("reencrypt_fields", stdout=out) + + output = out.getvalue() + assert "Re-encrypted 1 ControlPlane objects" in output + + +@pytest.mark.django_db +class TestSyncBillingMetadataCommand: + def test_sync_billing_metadata_no_control_planes(self): + out = StringIO() + call_command("sync_billing_metadata", "--dry-run", stdout=out) + + output = out.getvalue() + assert "DRY RUN" in output + assert "Syncing billing metadata on 0 control plane(s)" in output + assert "Sync completed" in output + + def test_sync_billing_metadata_conflicting_options(self): + out = StringIO() + call_command( + "sync_billing_metadata", + "--namespaces-only", + "--instances-only", + stdout=out, + ) + + assert ( + "Cannot use both --namespaces-only and --instances-only" in out.getvalue() + ) + + def test_sync_billing_metadata_invalid_control_plane_id(self): + out = StringIO() + call_command("sync_billing_metadata", "--control-plane", "99999", stdout=out) + + assert "No control planes found with the specified IDs" in out.getvalue() + + def test_sync_billing_metadata_dry_run_with_control_plane(self, test_control_plane): + out = StringIO() + call_command("sync_billing_metadata", "--dry-run", stdout=out) + + output = out.getvalue() + assert "DRY RUN" in output + assert "Syncing billing metadata on 1 control plane(s)" in output + assert test_control_plane.name in output