(organization) add metadata update command

This allows to update the Organization metadata with default values.
This commit is contained in:
Quentin BEY
2025-03-10 16:41:21 +01:00
committed by BEY Quentin
parent 7ce5b28af4
commit f64a592648
10 changed files with 326 additions and 0 deletions

View File

@@ -0,0 +1 @@
"""Management commands for core app."""

View File

@@ -0,0 +1 @@
"""Management commands for core app."""

View File

@@ -0,0 +1,34 @@
"""
Management command for filling organization metadata with default values.
"""
from django.core.management.base import BaseCommand
from core import models
from core.utils.json_schema import generate_default_from_schema
class Command(BaseCommand):
"""Management command to fill organization metadata with default values."""
help = "Fill organization metadata with default values"
def handle(self, *args, **options):
"""Fill organizations metadata missing values with default values."""
organization_metadata_schema = models.get_organization_metadata_schema()
if not organization_metadata_schema:
message = "No organization metadata schema defined."
self.stdout.write(self.style.ERROR(message))
return
default_metadata = generate_default_from_schema(organization_metadata_schema)
for organization in models.Organization.objects.all():
organization.metadata = {**default_metadata, **organization.metadata}
# Save the organization with the updated metadata
# We don't use bulk update because we want to trigger the clean method
organization.save(update_fields=["metadata", "updated_at"])
message = "Organization metadata filled with default values."
self.stdout.write(self.style.SUCCESS(message))

View File

@@ -0,0 +1 @@
"""Test for management commands for core app."""

View File

@@ -0,0 +1,94 @@
"""Tests for the fill_organization_metadata management command."""
from io import StringIO
from unittest.mock import patch
from django.core.management import call_command
import pytest
from core import factories
pytestmark = pytest.mark.django_db
@pytest.fixture(name="command_output")
def command_output_fixture():
"""Capture command output."""
out = StringIO()
return out
@pytest.mark.django_db
def test_fill_organization_metadata_no_schema(command_output):
"""Test command behavior when no schema is available."""
organization_1 = factories.OrganizationFactory(
name="Org with empty metadata",
metadata={},
with_registration_id=True,
)
organization_2 = factories.OrganizationFactory(
name="Org with partial metadata",
metadata={"existing_key": "existing_value"},
with_registration_id=True,
)
# Mock the schema function to return None (no schema)
with patch("core.models.get_organization_metadata_schema") as mock_get_schema:
mock_get_schema.return_value = None
# Call the command
call_command("fill_organization_metadata", stdout=command_output)
# Check the command output
assert "No organization metadata schema defined" in command_output.getvalue()
organization_1.refresh_from_db()
assert organization_1.metadata == {}
organization_2.refresh_from_db()
assert organization_2.metadata == {"existing_key": "existing_value"}
@pytest.mark.django_db
@pytest.mark.parametrize(
"existing_metadata,expected_result",
[
({}, {"field1": "default_value"}), # Empty metadata gets defaults
({"field1": "custom"}, {"field1": "custom"}), # Existing values preserved
(
{"other_field": "value"},
{"other_field": "value", "field1": "default_value"},
), # Mixed case
],
)
def test_metadata_merging_scenarios(existing_metadata, expected_result):
"""Test various metadata merging scenarios."""
# Create a simple schema with one field
simple_schema = {
"type": "object",
"properties": {
"field1": {"type": "string", "default": "default_value"},
},
}
# Create an organization with the specified metadata
organization = factories.OrganizationFactory(
name="Test organization",
metadata=existing_metadata,
with_registration_id=True,
)
# Mock the schema function to return our simple schema
with patch("core.models.get_organization_metadata_schema") as mock_get_schema:
mock_get_schema.return_value = simple_schema
# Call the command
call_command("fill_organization_metadata")
# Refresh from DB and check
organization.refresh_from_db()
# Check that the metadata has been merged correctly
for key, value in expected_result.items():
assert organization.metadata[key] == value

View File

@@ -0,0 +1 @@
"""Tests for the utils module."""

View File

@@ -0,0 +1,162 @@
"""Tests for the JSON schema `generate_default_from_schema` utility functions."""
import pytest
from core.utils.json_schema import generate_default_from_schema
@pytest.mark.parametrize(
"schema,expected",
[
# Test empty schema
({}, {}),
# Test schema with no properties
({"type": "object"}, {}),
# Test basic property types
(
{
"properties": {
"name": {"type": "string"},
"age": {"type": "integer"},
"active": {"type": "boolean"},
"data": {"type": "null"},
}
},
{"name": "", "age": None, "active": False, "data": None},
),
# Test default values
(
{
"properties": {
"name": {"type": "string", "default": "John Doe"},
"age": {"type": "integer", "default": 30},
"active": {"type": "boolean", "default": True},
}
},
{
"name": "John Doe",
"age": 30,
"active": True,
},
),
# Test array type
(
{"properties": {"items": {"type": "array"}, "tags": {"type": "array"}}},
{"items": [], "tags": []},
),
# Test nested object
(
{
"properties": {
"user": {
"type": "object",
"properties": {
"name": {"type": "string"},
"details": {
"type": "object",
"properties": {
"age": {"type": "integer"},
"active": {"type": "boolean", "default": True},
},
},
},
}
}
},
{"user": {"name": "", "details": {"age": None, "active": True}}},
),
# Test complex schema with multiple types and nesting
(
{
"properties": {
"id": {"type": "string", "default": "user-123"},
"profile": {
"type": "object",
"properties": {
"firstName": {"type": "string"},
"lastName": {"type": "string"},
"age": {"type": "number", "default": 25},
},
},
"roles": {"type": "array"},
"settings": {
"type": "object",
"properties": {
"notifications": {"type": "boolean", "default": False},
"theme": {"type": "string", "default": "light"},
},
},
"unknown": {"type": "something-else"},
}
},
{
"id": "user-123",
"profile": {"firstName": "", "lastName": "", "age": 25},
"roles": [],
"settings": {"notifications": False, "theme": "light"},
"unknown": None,
},
),
],
)
def test_generate_default_from_schema(schema, expected):
"""Test the generate_default_from_schema function with various schema inputs."""
result = generate_default_from_schema(schema)
assert result == expected
def test_with_invalid_inputs():
"""Test the function with invalid inputs to ensure it handles them gracefully."""
# pylint: disable=use-implicit-booleaness-not-comparison
# None input
assert generate_default_from_schema(None) == {}
# Invalid schema type
assert generate_default_from_schema([]) == {}
assert generate_default_from_schema("not-a-schema") == {}
# Empty properties
assert generate_default_from_schema({"properties": {}}) == {}
def test_complex_nested_arrays():
"""Test handling of complex schemas with nested arrays and objects."""
schema = {
"properties": {
"users": {
"type": "array",
},
"config": {
"type": "object",
"properties": {
"features": {
"type": "object",
"properties": {
"enabledFlags": {"type": "array"},
"limits": {
"type": "object",
"properties": {
"maxUsers": {"type": "integer", "default": 10},
"maxStorage": {"type": "integer", "default": 5120},
},
},
},
}
},
},
}
}
expected = {
"users": [],
"config": {
"features": {
"enabledFlags": [],
"limits": {"maxUsers": 10, "maxStorage": 5120},
}
},
}
result = generate_default_from_schema(schema)
assert result == expected

View File

@@ -0,0 +1,32 @@
"""Useful functions for working with JSON schemas"""
def generate_default_from_schema(schema: dict) -> dict:
"""
Generate default values based on a JSON schema
"""
if not schema or "properties" not in schema:
return {}
result = {}
for prop_name, prop_schema in schema.get("properties", {}).items():
prop_type = prop_schema.get("type")
match prop_type:
case "object" if "properties" in prop_schema:
result[prop_name] = generate_default_from_schema(prop_schema)
case "array":
result[prop_name] = []
case "string":
result[prop_name] = prop_schema.get("default", "")
case "number" | "integer":
result[prop_name] = prop_schema.get("default", None)
case "boolean":
result[prop_name] = prop_schema.get("default", False)
case "null":
result[prop_name] = None
case _:
result[prop_name] = None
return result