feat: add portal app config sync commands

This commit is contained in:
Md Bayazid Bostame
2026-03-29 01:33:50 +01:00
parent 48afccbca3
commit 5697f42306
5 changed files with 340 additions and 0 deletions

View File

@@ -0,0 +1,162 @@
from __future__ import annotations
import json
from pathlib import Path
from django.core.serializers.json import DjangoJSONEncoder
from django.db import transaction
from django.utils import timezone
from .app_registry import APP_DEFINITIONS, ensure_portal_app_configs, normalize_portal_app_sort_orders
from .models import PortalAppConfig
EXPORT_FIELDS = (
'key',
'section',
'sort_order',
'is_enabled',
'visible_to_super_admin',
'visible_to_admin',
'visible_to_it_staff',
'visible_to_staff',
'title_override',
'title_override_en',
'description_override',
'description_override_en',
'action_label_override',
'action_label_override_en',
)
FORMAT_VERSION = 1
def _valid_keys() -> set[str]:
return {definition.key for definition in APP_DEFINITIONS}
def _valid_sections() -> set[str]:
return {key for key, _label in PortalAppConfig.SECTION_CHOICES}
def export_portal_app_config_payload() -> dict[str, object]:
ensure_portal_app_configs()
rows = list(
PortalAppConfig.objects.order_by('section', 'sort_order', 'key').values(*EXPORT_FIELDS)
)
return {
'format': 'portal_app_config',
'version': FORMAT_VERSION,
'generated_at': timezone.now(),
'items': rows,
}
def write_portal_app_config_export(path: str | Path) -> Path:
payload = export_portal_app_config_payload()
output_path = Path(path)
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(json.dumps(payload, indent=2, cls=DjangoJSONEncoder) + '\n', encoding='utf-8')
return output_path
def load_portal_app_config_payload(path: str | Path) -> dict[str, object]:
payload = json.loads(Path(path).read_text(encoding='utf-8'))
if payload.get('format') != 'portal_app_config':
raise ValueError('Unsupported payload format.')
if payload.get('version') != FORMAT_VERSION:
raise ValueError(f"Unsupported payload version: {payload.get('version')}")
items = payload.get('items')
if not isinstance(items, list):
raise ValueError('Payload items must be a list.')
return payload
def validate_portal_app_config_payload(payload: dict[str, object]) -> list[dict[str, object]]:
valid_keys = _valid_keys()
valid_sections = _valid_sections()
items = payload['items']
seen_keys: set[str] = set()
normalized_items: list[dict[str, object]] = []
for item in items:
if not isinstance(item, dict):
raise ValueError('Each item must be an object.')
key = item.get('key')
if not isinstance(key, str) or not key.strip():
raise ValueError('Each item must contain a non-empty key.')
if key not in valid_keys:
raise ValueError(f'Unknown app key: {key}')
if key in seen_keys:
raise ValueError(f'Duplicate app key: {key}')
seen_keys.add(key)
section = item.get('section')
if section not in valid_sections:
raise ValueError(f'Invalid section for {key}: {section}')
sort_order = item.get('sort_order')
if not isinstance(sort_order, int) or sort_order < 0:
raise ValueError(f'Invalid sort_order for {key}: {sort_order}')
normalized: dict[str, object] = {'key': key, 'section': section, 'sort_order': sort_order}
for field in EXPORT_FIELDS:
if field in {'key', 'section', 'sort_order'}:
continue
value = item.get(field, False if field.startswith('visible_to_') or field == 'is_enabled' else '')
if field == 'is_enabled' or field.startswith('visible_to_'):
if not isinstance(value, bool):
raise ValueError(f'Invalid boolean for {key}: {field}')
else:
if value is None:
value = ''
if not isinstance(value, str):
raise ValueError(f'Invalid string for {key}: {field}')
normalized[field] = value
normalized_items.append(normalized)
missing = valid_keys.difference(seen_keys)
if missing:
raise ValueError(f'Missing app keys: {", ".join(sorted(missing))}')
return normalized_items
@transaction.atomic
def import_portal_app_config_payload(payload: dict[str, object], dry_run: bool = False) -> dict[str, int]:
ensure_portal_app_configs()
items = validate_portal_app_config_payload(payload)
updated = 0
created = 0
existing_map = {config.key: config for config in PortalAppConfig.objects.all()}
for item in items:
key = str(item['key'])
config = existing_map.get(key)
if config is None:
config = PortalAppConfig(key=key)
created += 1
changed = False
for field in EXPORT_FIELDS:
value = item[field]
if getattr(config, field) != value:
setattr(config, field, value)
changed = True
if config.pk is None:
changed = True
if changed:
updated += 1
if not dry_run:
config.save()
if not dry_run:
normalize_portal_app_sort_orders()
else:
transaction.set_rollback(True)
return {
'created': created,
'updated': updated,
'total': len(items),
'dry_run': 1 if dry_run else 0,
}

View File

@@ -0,0 +1,22 @@
from django.core.management.base import BaseCommand
from workflows.app_registry_sync import export_portal_app_config_payload, write_portal_app_config_export
class Command(BaseCommand):
help = 'Export PortalAppConfig to JSON for environment sync.'
def add_arguments(self, parser):
parser.add_argument('--output', help='Write the export to a file instead of stdout.')
def handle(self, *args, **options):
output = options.get('output')
if output:
path = write_portal_app_config_export(output)
self.stdout.write(self.style.SUCCESS(f'Exported PortalAppConfig to {path}'))
return
payload = export_portal_app_config_payload()
import json
self.stdout.write(json.dumps(payload, indent=2, default=str))

View File

@@ -0,0 +1,25 @@
from django.core.management.base import BaseCommand, CommandError
from workflows.app_registry_sync import import_portal_app_config_payload, load_portal_app_config_payload
class Command(BaseCommand):
help = 'Import PortalAppConfig JSON for environment sync.'
def add_arguments(self, parser):
parser.add_argument('input', help='Path to a previously exported PortalAppConfig JSON file.')
parser.add_argument('--dry-run', action='store_true', help='Validate and report changes without saving.')
def handle(self, *args, **options):
try:
payload = load_portal_app_config_payload(options['input'])
summary = import_portal_app_config_payload(payload, dry_run=options['dry_run'])
except (OSError, ValueError) as exc:
raise CommandError(str(exc)) from exc
mode = 'Dry run' if options['dry_run'] else 'Import'
self.stdout.write(
self.style.SUCCESS(
f"{mode} complete. total={summary['total']} updated={summary['updated']} created={summary['created']}"
)
)

View File

@@ -244,6 +244,12 @@ docker compose exec -T web django-admin compilemessages</code></pre>
<li>The landing page now renders from registry data instead of hardcoded cards.</li>
<li>Security remains code-based: app visibility/order is configurable, but access still depends on role capabilities in <code>roles.py</code>.</li>
<li>Management UI: <code>/admin-tools/apps/</code> for <code>Platform Owner</code>.</li>
<li>Environment sync is explicit, not automatic. Use:
<pre><code>docker compose exec -T web python manage.py export_portal_app_config --output /tmp/portal-app-config.json
docker compose exec -T web python manage.py import_portal_app_config /tmp/portal-app-config.json --dry-run
docker compose exec -T web python manage.py import_portal_app_config /tmp/portal-app-config.json</code></pre>
</li>
<li>This is the correct path when local and server app ordering differ.</li>
</ul>
<h2 id="trial">12c) Trial Lifecycle</h2>

View File

@@ -0,0 +1,125 @@
import json
from pathlib import Path
from tempfile import TemporaryDirectory
from django.core.management import call_command
from django.core.management.base import CommandError
from django.test import TestCase
from workflows.app_registry import ensure_portal_app_configs
from workflows.models import PortalAppConfig
class PortalAppConfigSyncTests(TestCase):
def setUp(self):
ensure_portal_app_configs()
def test_export_writes_expected_payload(self):
with TemporaryDirectory() as tmpdir:
output_path = Path(tmpdir) / 'portal-app-config.json'
call_command('export_portal_app_config', '--output', str(output_path))
payload = json.loads(output_path.read_text(encoding='utf-8'))
self.assertEqual(payload['format'], 'portal_app_config')
self.assertEqual(payload['version'], 1)
self.assertEqual(len(payload['items']), PortalAppConfig.objects.count())
self.assertIn('generated_at', payload)
def test_import_applies_runtime_ordering_changes(self):
payload = {
'format': 'portal_app_config',
'version': 1,
'items': list(PortalAppConfig.objects.order_by('section', 'sort_order', 'key').values(
'key',
'section',
'sort_order',
'is_enabled',
'visible_to_super_admin',
'visible_to_admin',
'visible_to_it_staff',
'visible_to_staff',
'title_override',
'title_override_en',
'description_override',
'description_override_en',
'action_label_override',
'action_label_override_en',
)),
}
for item in payload['items']:
if item['key'] == 'branding':
item['sort_order'] = 0
elif item['key'] == 'company_config':
item['sort_order'] = 1
item['title_override'] = 'Company Setup'
with TemporaryDirectory() as tmpdir:
input_path = Path(tmpdir) / 'portal-app-config.json'
input_path.write_text(json.dumps(payload), encoding='utf-8')
call_command('import_portal_app_config', str(input_path))
branding = PortalAppConfig.objects.get(key='branding')
company_config = PortalAppConfig.objects.get(key='company_config')
self.assertEqual(branding.sort_order, 0)
self.assertEqual(company_config.title_override, 'Company Setup')
def test_import_dry_run_does_not_persist(self):
payload = {
'format': 'portal_app_config',
'version': 1,
'items': list(PortalAppConfig.objects.order_by('section', 'sort_order', 'key').values(
'key',
'section',
'sort_order',
'is_enabled',
'visible_to_super_admin',
'visible_to_admin',
'visible_to_it_staff',
'visible_to_staff',
'title_override',
'title_override_en',
'description_override',
'description_override_en',
'action_label_override',
'action_label_override_en',
)),
}
for item in payload['items']:
if item['key'] == 'branding':
item['title_override'] = 'Dry Run Only'
with TemporaryDirectory() as tmpdir:
input_path = Path(tmpdir) / 'portal-app-config.json'
input_path.write_text(json.dumps(payload), encoding='utf-8')
call_command('import_portal_app_config', str(input_path), '--dry-run')
self.assertEqual(PortalAppConfig.objects.get(key='branding').title_override, '')
def test_import_rejects_unknown_keys(self):
payload = {
'format': 'portal_app_config',
'version': 1,
'items': [
{
'key': 'unknown_app',
'section': 'platform',
'sort_order': 0,
'is_enabled': True,
'visible_to_super_admin': True,
'visible_to_admin': True,
'visible_to_it_staff': False,
'visible_to_staff': False,
'title_override': '',
'title_override_en': '',
'description_override': '',
'description_override_en': '',
'action_label_override': '',
'action_label_override_en': '',
}
],
}
with TemporaryDirectory() as tmpdir:
input_path = Path(tmpdir) / 'portal-app-config.json'
input_path.write_text(json.dumps(payload), encoding='utf-8')
with self.assertRaises(CommandError):
call_command('import_portal_app_config', str(input_path))