Nudge test coverage up to 60%

src/tests/test_service_models.py

@@ -0,0 +1,349 @@
+"""Tests for servala.core.models.service module."""
+
+from unittest.mock import MagicMock
+
+import pytest
+from django.core.exceptions import ValidationError
+
+from servala.core.models.service import (
+    ControlPlane,
+    Service,
+    ServiceInstance,
+    prune_empty_data,
+    validate_api_credentials,
+    validate_dict,
+)
+
+pytestmark = pytest.mark.django_db
+
+
+@pytest.fixture
+def service_with_external_links(service_category):
+    return Service.objects.create(
+        name="PostgreSQL",
+        slug="postgresql",
+        category=service_category,
+        description="PostgreSQL database",
+        external_links=[
+            {"url": "https://docs.example.com", "title": "Docs", "featured": True},
+            {"url": "https://github.com/example", "title": "GitHub", "featured": False},
+            {"url": "https://api.example.com", "title": "API", "featured": True},
+        ],
+    )
+
+
+@pytest.mark.parametrize("data", [None, {}])
+def test_validate_dict_allows_empty_by_default(data):
+    validate_dict(data)
+
+
+@pytest.mark.parametrize("data", [None, {}])
+def test_validate_dict_raises_when_empty_not_allowed(data):
+    with pytest.raises(ValidationError, match="Data may not be empty"):
+        validate_dict(data, allow_empty=False)
+
+
+def test_validate_dict_missing_required_fields_raises():
+    with pytest.raises(ValidationError, match="Missing required fields"):
+        validate_dict({"field1": "v"}, required_fields={"field1", "field2", "field3"})
+
+
+def test_validate_dict_all_required_fields_present_passes():
+    validate_dict({"a": 1, "b": 2, "extra": 3}, required_fields={"a", "b"})
+
+
+@pytest.mark.parametrize("data", [None, {}])
+def test_validate_api_credentials_allows_empty(data):
+    validate_api_credentials(data)
+
+
+@pytest.mark.parametrize(
+    "input_data,expected",
+    [
+        ({"a": 1, "b": None, "c": 3}, {"a": 1, "c": 3}),
+        ({"a": "hello", "b": "", "c": "world"}, {"a": "hello", "c": "world"}),
+        ({"a": [1, 2], "b": [], "c": [3]}, {"a": [1, 2], "c": [3]}),
+        (
+            {"a": {"nested": 1}, "b": {}, "c": {"x": 2}},
+            {"a": {"nested": 1}, "c": {"x": 2}},
+        ),
+        ({"outer": {"inner": {"empty": None}}}, {}),
+        (
+            {"false_val": False, "zero": 0, "none": None},
+            {"false_val": False, "zero": 0},
+        ),
+        ("string", "string"),
+        (42, 42),
+    ],
+)
+def test_prune_empty_data(input_data, expected):
+    assert prune_empty_data(input_data) == expected
+
+
+def test_prune_empty_data_nested_dicts():
+    data = {"level1": {"level2": {"keep": "value", "remove": None, "empty": ""}}}
+    assert prune_empty_data(data) == {"level1": {"level2": {"keep": "value"}}}
+
+
+def test_prune_empty_data_in_lists():
+    data = {"items": [{"keep": 1}, {"remove": None}, {"also_keep": 2}]}
+    assert prune_empty_data(data) == {"items": [{"keep": 1}, {"also_keep": 2}]}
+
+
+def test_service_featured_links_filters_correctly(service_with_external_links):
+    featured = service_with_external_links.featured_links
+    assert len(featured) == 2
+    assert all(link["featured"] for link in featured)
+    assert {link["title"] for link in featured} == {"Docs", "API"}
+
+
+@pytest.mark.parametrize(
+    "external_links,expected_count",
+    [
+        (None, 0),
+        ([], 0),
+        ([{"url": "https://x.com", "title": "X", "featured": False}], 0),
+    ],
+)
+def test_service_featured_links_empty_cases(
+    service_category, external_links, expected_count
+):
+    svc = Service.objects.create(
+        name="Test",
+        slug="test-svc",
+        category=service_category,
+        external_links=external_links,
+    )
+    assert len(svc.featured_links) == expected_count
+
+
+def test_service_str(service):
+    assert str(service) == "Redis"
+
+
+def test_service_category_str(service_category):
+    assert str(service_category) == "Databases"
+
+
+def test_cloud_provider_str(cloud_provider):
+    assert str(cloud_provider) == "Exoscale"
+
+
+def test_control_plane_str(control_plane):
+    assert str(control_plane) == "Geneva (CH-GVA-2)"
+
+
+def test_control_plane_test_connection_no_credentials(cloud_provider):
+    plane = ControlPlane.objects.create(
+        name="No Creds", cloud_provider=cloud_provider, api_credentials={}
+    )
+    success, message = plane.test_connection()
+    assert success is False
+    assert "No API credentials" in str(message)
+
+
+def test_service_definition_str(service_definition):
+    assert str(service_definition) == "Redis Standard"
+
+
+def test_service_definition_control_planes(service_definition, control_plane_crd):
+    assert control_plane_crd.control_plane in service_definition.control_planes
+
+
+def test_control_plane_crd_str(control_plane_crd):
+    result = str(control_plane_crd)
+    assert "Redis" in result and "Exoscale" in result and "Geneva" in result
+
+
+@pytest.mark.parametrize(
+    "prop,expected",
+    [("group", "vshn.appcat.vshn.io"), ("version", "v1"), ("kind", "VSHNRedis")],
+)
+def test_control_plane_crd_api_properties(control_plane_crd, prop, expected):
+    assert getattr(control_plane_crd, prop) == expected
+
+
+def test_service_offering_str(service_offering):
+    result = str(service_offering)
+    assert "Redis" in result and "Exoscale" in result
+
+
+def test_service_offering_control_planes(service_offering, control_plane_crd):
+    assert control_plane_crd.control_plane in service_offering.control_planes
+
+
+def test_generate_resource_name_consistent(organization, service):
+    name1 = ServiceInstance.generate_resource_name(organization, "My Instance", service)
+    name2 = ServiceInstance.generate_resource_name(organization, "My Instance", service)
+    assert name1 == name2
+
+
+def test_generate_resource_name_different_inputs(organization, service):
+    name_a = ServiceInstance.generate_resource_name(organization, "Instance A", service)
+    name_b = ServiceInstance.generate_resource_name(organization, "Instance B", service)
+    assert name_a != name_b
+
+
+def test_generate_resource_name_attempt_changes_hash(organization, service):
+    name0 = ServiceInstance.generate_resource_name(
+        organization, "X", service, attempt=0
+    )
+    name1 = ServiceInstance.generate_resource_name(
+        organization, "X", service, attempt=1
+    )
+    assert name0 != name1
+
+
+def test_generate_resource_name_format(organization, service, settings):
+    name = ServiceInstance.generate_resource_name(organization, "Test", service)
+    assert name.startswith(f"{settings.SERVALA_INSTANCE_NAME_PREFIX}-")
+    assert len(name.split("-")[-1]) == 8
+
+
+@pytest.mark.parametrize(
+    "display_name", ["My Instance", "MY INSTANCE", "  My Instance  "]
+)
+def test_generate_resource_name_normalizes_display_name(
+    organization, service, display_name
+):
+    canonical = ServiceInstance.generate_resource_name(
+        organization, "my instance", service
+    )
+    assert (
+        ServiceInstance.generate_resource_name(organization, display_name, service)
+        == canonical
+    )
+
+
+@pytest.mark.parametrize("spec_data", [None, {}])
+def test_prepare_spec_data_handles_empty(spec_data):
+    assert ServiceInstance._prepare_spec_data(spec_data) == {}
+
+
+def test_prepare_spec_data_prunes_empty_values():
+    assert ServiceInstance._prepare_spec_data({"keep": "v", "rm": None}) == {
+        "keep": "v"
+    }
+
+
+def test_apply_compute_plan_to_spec(compute_plan_assignment):
+    result = ServiceInstance._apply_compute_plan_to_spec({}, compute_plan_assignment)
+    assert result["parameters"]["size"]["memory"] == "2Gi"
+    assert result["parameters"]["size"]["cpu"] == "1000m"
+    assert result["parameters"]["size"]["requests"]["memory"] == "1Gi"
+    assert result["parameters"]["size"]["requests"]["cpu"] == "500m"
+    assert result["parameters"]["service"]["serviceLevel"] == "besteffort"
+
+
+def test_apply_compute_plan_to_spec_none_assignment():
+    spec = {"existing": "value"}
+    assert ServiceInstance._apply_compute_plan_to_spec(spec, None) == {
+        "existing": "value"
+    }
+
+
+def test_apply_compute_plan_preserves_existing(compute_plan_assignment):
+    spec = {"parameters": {"custom": "setting", "service": {"other": "config"}}}
+    result = ServiceInstance._apply_compute_plan_to_spec(spec, compute_plan_assignment)
+    assert result["parameters"]["custom"] == "setting"
+    assert result["parameters"]["service"]["other"] == "config"
+
+
+def test_build_billing_annotations_display_name():
+    cp = MagicMock(storage_plan_odoo_product_id=None, storage_plan_odoo_unit_id=None)
+    annotations = ServiceInstance._build_billing_annotations(
+        compute_plan_assignment=None, control_plane=cp, display_name="My Service"
+    )
+    assert annotations["servala.com/displayName"] == "My Service"
+
+
+def test_build_billing_annotations_no_display_name():
+    cp = MagicMock(storage_plan_odoo_product_id=None, storage_plan_odoo_unit_id=None)
+    annotations = ServiceInstance._build_billing_annotations(
+        compute_plan_assignment=None, control_plane=cp, display_name=None
+    )
+    assert "servala.com/displayName" not in annotations
+
+
+def test_build_billing_annotations_compute_plan(compute_plan_assignment):
+    cp = MagicMock(storage_plan_odoo_product_id=None, storage_plan_odoo_unit_id=None)
+    annotations = ServiceInstance._build_billing_annotations(
+        compute_plan_assignment=compute_plan_assignment, control_plane=cp
+    )
+    assert annotations["servala.com/erp_product_id_resource"] == "test-product-id"
+    assert annotations["servala.com/erp_unit_id_resource"] == "test-unit-id"
+
+
+def test_build_billing_annotations_storage_plan(control_plane_with_storage):
+    annotations = ServiceInstance._build_billing_annotations(
+        compute_plan_assignment=None, control_plane=control_plane_with_storage
+    )
+    assert annotations["servala.com/erp_product_id_storage"] == "storage-product-123"
+    assert annotations["servala.com/erp_unit_id_storage"] == "storage-unit-456"
+
+
+@pytest.mark.parametrize(
+    "error_msg,expected_has_list,expected_errors",
+    [
+        ("", False, None),
+        (None, False, None),
+        ("Something went wrong", False, None),
+        ("Error: [single error]", False, None),
+        ("Validation failed: [e1, e2, e3]", True, ["e1", "e2", "e3"]),
+    ],
+)
+def test_format_kubernetes_error(error_msg, expected_has_list, expected_errors):
+    result = ServiceInstance._format_kubernetes_error(error_msg)
+    assert result["has_list"] == expected_has_list
+    assert result["errors"] == expected_errors
+
+
+def test_format_kubernetes_error_strips_quotes():
+    result = ServiceInstance._format_kubernetes_error("Errors: [\"quoted\", 'single']")
+    assert "quoted" in result["errors"]
+    assert "single" in result["errors"]
+
+
+def test_safe_format_error_non_dict():
+    assert ServiceInstance._safe_format_error("plain string") == "plain string"
+
+
+def test_safe_format_error_escapes_html():
+    error_data = {"message": "<script>alert('xss')</script>", "has_list": False}
+    result = ServiceInstance._safe_format_error(error_data)
+    assert "<script>" not in result
+    assert "&lt;script&gt;" in result
+
+
+def test_safe_format_error_formats_list():
+    error_data = {"message": "Failed", "errors": ["e1", "e2"], "has_list": True}
+    result = str(ServiceInstance._safe_format_error(error_data))
+    assert "Failed" in result and "<ul>" in result and "<li>e1</li>" in result
+
+
+@pytest.mark.parametrize(
+    "spec,expected",
+    [
+        ({}, None),
+        ({"parameters": {"service": {}}}, None),
+        ({"parameters": {"service": {"fqdn": "app.example.com"}}}, "app.example.com"),
+        (
+            {"parameters": {"service": {"fqdn": ["first.com", "second.com"]}}},
+            "first.com",
+        ),
+        ({"parameters": {"service": {"fqdn": 12345}}}, None),
+    ],
+)
+def test_fqdn_url(service_instance, mocker, spec, expected):
+    mocker.patch.object(
+        ServiceInstance, "spec", new_callable=mocker.PropertyMock, return_value=spec
+    )
+    assert service_instance.fqdn_url == expected
+
+
+def test_clear_kubernetes_caches(service_instance):
+    service_instance.__dict__["kubernetes_object"] = {"cached": True}
+    service_instance.__dict__["spec"] = {"cached": True}
+    service_instance._clear_kubernetes_caches()
+    assert "kubernetes_object" not in service_instance.__dict__
+    assert "spec" not in service_instance.__dict__