Skip to content

Commit

Permalink
✨(organization) add metadata update command
Browse files Browse the repository at this point in the history
This allows to update the Organization metadata with default values.
  • Loading branch information
qbey committed Mar 10, 2025
1 parent 492766f commit a23cde8
Show file tree
Hide file tree
Showing 10 changed files with 327 additions and 0 deletions.
1 change: 1 addition & 0 deletions src/backend/core/management/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
"""Management commands for core app."""
1 change: 1 addition & 0 deletions src/backend/core/management/commands/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
"""Management commands for core app."""
35 changes: 35 additions & 0 deletions src/backend/core/management/commands/fill_organization_metadata.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
"""
Management command overriding the "createsuperuser" command to allow creating users
with their email and no username.
"""

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))
1 change: 1 addition & 0 deletions src/backend/core/tests/management_commands/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
"""Test for management commands for core app."""
Original file line number Diff line number Diff line change
@@ -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
1 change: 1 addition & 0 deletions src/backend/core/tests/utils/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
"""Tests for the utils module."""
162 changes: 162 additions & 0 deletions src/backend/core/tests/utils/test_json_schema.py
Original file line number Diff line number Diff line change
@@ -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
32 changes: 32 additions & 0 deletions src/backend/core/utils/json_schema.py
Original file line number Diff line number Diff line change
@@ -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

0 comments on commit a23cde8

Please sign in to comment.