Files
ServerKit/backend/tests/test_template_magic_vars.py
Juan DenisandClaude Opus 4.8 f47d0e1428 feat: Phase 1 foundation — status aggregator, API scopes, onboarding, template magic vars
Four foundational enhancements toward the deploy-code developer experience:

- Container status aggregator (C7): new ContainerStatusService applies a
  deterministic priority hierarchy (degraded > restarting > running:unhealthy >
  starting > running:healthy > exited > unknown) over an app's containers,
  cached with a short TTL. Exposed at /api/v1/status and pushed over a new
  container_status Socket.IO channel; Applications page now renders an
  aggregated ContainerStatusPill with sub-status tooltip.

- API token scopes + sensitive data masking (C10): require_scope() decorator
  enforces fine-grained scopes for API-key requests (pass-through for JWT),
  a canonical SCOPES catalog at /api/v1/api-keys/scopes, and a reusable
  sensitive_data_filter (mask_sensitive) honoring secrets:read. Frontend gains
  a grouped ApiKeyScopesModal picker and per-key scope pills.

- Server onboarding state machine (C8): new onboarding lifecycle
  (pending -> validating -> installing_prerequisites -> installing_docker ->
  pairing_agent -> ready/failed) with ServerOnboardingService, a
  server_onboarding_logs table, servers.onboarding_* columns, REST endpoints,
  and an OnboardingWizard surfaced on the server detail overview.

- Declarative template catalog (C5): documented catalog schema and
  magic-variable resolution (SERVICE_PASSWORD_/USER_/FQDN_/URL_/BASE64_*)
  layered onto the existing YAML template loader without changing behavior for
  templates that don't use them; GET /api/v1/templates/catalog/schema.

Migration 029 adds the onboarding table + columns. 60 new tests pass; frontend
builds clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-23 18:23:47 -04:00

213 lines
7.8 KiB
Python

"""Proving tests for declarative-catalog magic variables and catalog validation.
Covers ``TemplateService.resolve_magic_variables`` (token detection, stable
generation, each token form, dict/list shapes, and no-op when no tokens are
present) plus ``TemplateService.validate_catalog_entry``. These are pure /
unit-testable: no Docker, no network.
"""
import base64
import re
import pytest
from app.services.template_service import TemplateService as TS
# ---------------------------------------------------------------------------
# Token detection + stable generation
# ---------------------------------------------------------------------------
def test_password_token_detected_and_generated():
text = 'MYSQL_ROOT_PASSWORD=${SERVICE_PASSWORD_DB}'
out, generated = TS.resolve_magic_variables(text)
assert 'SERVICE_PASSWORD_DB' in generated
value = generated['SERVICE_PASSWORD_DB']
# Strong password, default length, alnum (compose/shell-safe).
assert len(value) == TS.MAGIC_PASSWORD_LENGTH
assert value.isalnum()
assert out == f'MYSQL_ROOT_PASSWORD={value}'
assert '${SERVICE_PASSWORD_DB}' not in out
def test_same_token_resolves_to_one_stable_value():
text = 'A=${SERVICE_PASSWORD_DB} B=${SERVICE_PASSWORD_DB}'
out, generated = TS.resolve_magic_variables(text)
# Generated exactly once for the unique token.
assert list(generated.keys()) == ['SERVICE_PASSWORD_DB']
value = generated['SERVICE_PASSWORD_DB']
assert out == f'A={value} B={value}'
def test_different_names_get_different_values():
text = '${SERVICE_PASSWORD_DB} ${SERVICE_PASSWORD_CACHE}'
_, generated = TS.resolve_magic_variables(text)
assert set(generated) == {'SERVICE_PASSWORD_DB', 'SERVICE_PASSWORD_CACHE'}
assert generated['SERVICE_PASSWORD_DB'] != generated['SERVICE_PASSWORD_CACHE']
def test_user_token_form():
_, generated = TS.resolve_magic_variables('${SERVICE_USER_APP}')
user = generated['SERVICE_USER_APP']
assert user.startswith('svc_app_')
# safe identifier characters only
assert re.fullmatch(r'svc_app_[0-9a-f]+', user)
def test_base64_token_form():
_, generated = TS.resolve_magic_variables('SECRET=${SERVICE_BASE64_KEY}')
val = generated['SERVICE_BASE64_KEY']
# Decodes cleanly as base64.
decoded = base64.b64decode(val)
assert len(decoded) > 0
def test_fqdn_and_url_use_context():
text = 'host=${SERVICE_FQDN_WEB} url=${SERVICE_URL_WEB}'
out, generated = TS.resolve_magic_variables(
text, context={'fqdn': 'app.example.com', 'scheme': 'https'}
)
assert generated['SERVICE_FQDN_WEB'] == 'app.example.com'
assert generated['SERVICE_URL_WEB'] == 'https://app.example.com'
assert out == 'host=app.example.com url=https://app.example.com'
def test_fqdn_url_fall_back_to_placeholder_without_context():
text = '${SERVICE_FQDN_WEB} ${SERVICE_URL_WEB}'
out, generated = TS.resolve_magic_variables(text, context={'app_name': 'myapp'})
# No FQDN known -> documented placeholder (app_name), http scheme default.
assert generated['SERVICE_FQDN_WEB'] == 'myapp'
assert generated['SERVICE_URL_WEB'] == 'http://myapp'
assert out == 'myapp http://myapp'
# ---------------------------------------------------------------------------
# Shapes: dicts / lists (a compose section) are walked and substituted in place
# ---------------------------------------------------------------------------
def test_resolves_inside_nested_dict_and_list():
compose = {
'services': {
'db': {
'environment': [
'MYSQL_PASSWORD=${SERVICE_PASSWORD_DB}',
'MYSQL_USER=${SERVICE_USER_DB}',
],
},
},
}
out, generated = TS.resolve_magic_variables(compose)
assert set(generated) == {'SERVICE_PASSWORD_DB', 'SERVICE_USER_DB'}
env = out['services']['db']['environment']
assert env[0] == f"MYSQL_PASSWORD={generated['SERVICE_PASSWORD_DB']}"
assert env[1] == f"MYSQL_USER={generated['SERVICE_USER_DB']}"
# ---------------------------------------------------------------------------
# No-op behavior: templates without magic tokens are unchanged
# ---------------------------------------------------------------------------
def test_noop_when_no_magic_tokens_present():
text = 'image: nginx\nports:\n - "${HTTP_PORT}:80"'
out, generated = TS.resolve_magic_variables(text)
# The plain ${HTTP_PORT} is NOT a magic token; nothing changes.
assert generated == {}
assert out == text
def test_non_service_dollar_vars_untouched():
out, generated = TS.resolve_magic_variables('${APP_NAME} ${DB_PASSWORD}')
assert generated == {}
assert out == '${APP_NAME} ${DB_PASSWORD}'
def test_collect_magic_variables_for_template_scans_all_sections():
template = {
'compose': {'services': {'app': {'environment': ['P=${SERVICE_PASSWORD_X}']}}},
'files': [{'path': '/etc/app.conf', 'content': 'user=${SERVICE_USER_X}'}],
'scripts': {'post_install': 'echo ${SERVICE_BASE64_X}'},
}
generated = TS.collect_magic_variables(template)
assert set(generated) == {'SERVICE_PASSWORD_X', 'SERVICE_USER_X', 'SERVICE_BASE64_X'}
def test_collect_magic_variables_empty_for_plain_template():
template = {
'compose': {'services': {'app': {'image': 'nginx', 'ports': ['${HTTP_PORT}:80']}}},
}
assert TS.collect_magic_variables(template) == {}
# ---------------------------------------------------------------------------
# validate_catalog_entry
# ---------------------------------------------------------------------------
def test_validate_catalog_entry_accepts_minimal_valid():
entry = {
'id': 'my-service',
'name': 'My Service',
'version': '1.0',
'description': 'A service',
'compose': {'services': {'app': {'image': 'nginx'}}},
}
result = TS.validate_catalog_entry(entry)
assert result['valid'] is True
assert result['errors'] == []
def test_validate_catalog_entry_flags_missing_required_fields():
result = TS.validate_catalog_entry({'name': 'X'})
assert result['valid'] is False
# missing version/description and missing compose/dockerfile
assert any('version' in e for e in result['errors'])
def test_validate_catalog_entry_rejects_bad_id_slug():
entry = {
'id': 'Bad ID!',
'name': 'X', 'version': '1', 'description': 'd',
'compose': {'services': {'app': {'image': 'nginx'}}},
}
result = TS.validate_catalog_entry(entry)
assert result['valid'] is False
assert any("'id'" in e for e in result['errors'])
def test_validate_catalog_entry_warns_unknown_var_type():
entry = {
'id': 'svc', 'name': 'X', 'version': '1', 'description': 'd',
'compose': {'services': {'app': {'image': 'nginx'}}},
'variables': [{'name': 'FOO', 'type': 'mystery'}],
}
result = TS.validate_catalog_entry(entry)
assert result['valid'] is True # warnings are non-fatal
assert any('mystery' in w for w in result['warnings'])
def test_validate_catalog_entry_warns_on_malformed_magic_token():
entry = {
'id': 'svc', 'name': 'X', 'version': '1', 'description': 'd',
'compose': {'services': {'app': {
'image': 'nginx',
'environment': ['BAD=${SERVICE_BOGUS_X}'], # unknown kind
}}},
}
result = TS.validate_catalog_entry(entry)
assert any('SERVICE_BOGUS_X' in w for w in result['warnings'])
def test_validate_catalog_entry_accepts_well_formed_magic_tokens():
entry = {
'id': 'svc', 'name': 'X', 'version': '1', 'description': 'd',
'compose': {'services': {'app': {
'image': 'nginx',
'environment': ['PW=${SERVICE_PASSWORD_DB}'],
}}},
}
result = TS.validate_catalog_entry(entry)
assert result['valid'] is True
assert result['warnings'] == []