Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
120 changes: 119 additions & 1 deletion python/tests/test_console_config_compare.py
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,124 @@ def test_all_match_and_no_private_output(self):
for private in ['PRIVATE_NAME_SENTINEL', 'a' * 32, 'b' * 32, 'c' * 32, 'demo-console']:
self.assertNotIn(private, output)

def test_empty_flags_match_omitted_null_or_empty_api_flags(self):
for rendered in (compare.MISSING, []):
for setting in (compare.MISSING, None, []):
for runtime in (compare.MISSING, None, []):
with self.subTest(rendered=rendered, setting=setting, runtime=runtime):
self.config, self.live = fixture()
if rendered is not compare.MISSING:
self.config['compatibility_flags'] = rendered
for source, value in [(self.live['settings'], setting), (self.live['runtime'], runtime)]:
source.pop('compatibility_flags', None)
if value is not compare.MISSING:
source['compatibility_flags'] = value
self.assertTrue(all(row['status'] == 'match' for row in self.rows()), self.rows())

def test_nonempty_or_invalid_flags_still_block_real_differences(self):
for scope in ('settings', 'runtime'):
for actual in (compare.MISSING, None, [], ['different_flag'], {}, False):
with self.subTest(scope=scope, actual=actual):
self.config, self.live = fixture()
self.config['compatibility_flags'] = ['nodejs_compat']
for source in (self.live['settings'], self.live['runtime']):
source['compatibility_flags'] = ['nodejs_compat']
self.assertTrue(all(row['status'] == 'match' for row in self.rows()))
self.live[scope].pop('compatibility_flags')
if actual is not compare.MISSING:
self.live[scope]['compatibility_flags'] = actual
row = next(row for row in self.rows() if row['item'] == scope + '.compatibility_flags')
self.assertNotEqual(row['status'], 'match')
if actual is compare.MISSING:
self.assertEqual(row['reason'], 'missing_setting' if scope == 'settings' else 'missing_runtime_setting')
for actual in (['nodejs_compat'], {}, False, ''):
self.config, self.live = fixture()
self.live[scope]['compatibility_flags'] = actual
self.assertBlocked()
for rendered in (None, {}, '', False, [None], ['']):
self.config, self.live = fixture()
self.config['compatibility_flags'] = rendered
self.live['settings']['compatibility_flags'] = rendered
self.live['runtime']['compatibility_flags'] = rendered
self.assertBlocked()
for scope in ('settings', 'runtime'):
self.config, self.live = fixture()
self.live[scope].pop('compatibility_date')
self.assertBlocked()

def test_no_placement_empty_tags_and_absent_observability_match(self):
self.config['placement'] = {'mode': 'off'}
self.live['settings'].update(placement={}, tags=[])
self.live['settings'].pop('observability')
self.live['runtime'].pop('compatibility_flags')
self.assertTrue(all(row['status'] == 'match' for row in self.rows()), self.rows())
self.live['settings'].pop('placement')
self.assertTrue(all(row['status'] == 'match' for row in self.rows()))

def test_optional_settings_absence_and_real_live_only_values(self):
for key, value in [('placement', {'mode': 'smart'}), ('tags', ['synthetic-tag']),
('observability', {'enabled': True})]:
with self.subTest(key=key):
self.config, self.live = fixture()
self.live['settings'].pop(key, None)
self.assertTrue(all(row['status'] == 'match' for row in self.rows()))
self.live['settings'][key] = value
row = next(row for row in self.rows() if row['item'] == 'settings.' + key)
self.assertEqual(row['status'], 'unreadable')
self.assertEqual(row['reason'], 'undeclared_effective_setting')
if key == 'tags':
# Pinned Wrangler ignores top-level tags. Never present a
# fictitious declaration as an effective configuration.
self.config[key] = copy.deepcopy(value)
self.assertBlocked()
continue
self.config[key] = copy.deepcopy(value)
self.assertTrue(all(row['status'] == 'match' for row in self.rows()))
self.live['settings'].pop(key)
self.assertBlocked()
self.live['settings'][key] = None
self.assertBlocked()
self.config, self.live = fixture()
self.config['placement'] = {'mode': 'off'}
self.live['settings']['placement'] = {'mode': 'smart'}
self.assertBlocked()
self.config['placement']['hint'] = 'synthetic-region'
self.live['settings']['placement'] = {}
self.assertBlocked()
# Empty provider objects must not erase an explicit logging declaration.
self.config, self.live = fixture()
self.config['observability'] = {'enabled': False}
self.live['settings'].pop('observability')
self.assertBlocked()
for tags in ([], ['synthetic-tag']):
self.config, self.live = fixture()
self.config['tags'] = tags
self.live['settings']['tags'] = tags
self.assertBlocked()

def test_unsupported_fields_emit_only_sanitized_names_and_remain_blocked(self):
cases = [('settings', 'unsupported_live_setting'), ('config', 'unsupported_config_key'),
('subdomain', 'unsupported_subdomain_setting'), ('runtime', 'unmapped_runtime_setting')]
for scope, reason in cases:
for name in ('annotations', 'A_0.key-name', 'x' * 64, '', 'x' * 65,
'bad/name', 'bad=PRIVATE_TOKEN', 'bad\nPRIVATE_TOKEN', '非法字段'):
with self.subTest(scope=scope, name=name):
self.config, self.live = fixture()
target = self.config if scope == 'config' else self.live[scope]
target[name] = {'token': 'PRIVATE_TOKEN_VALUE', 'account_id': 'a' * 32,
'nested_secret': ['PRIVATE_SECRET_VALUE']}
rows = self.rows()
row = next(row for row in rows if row.get('reason') == reason)
self.assertEqual(row['status'], 'unreadable')
valid = name in ('annotations', 'A_0.key-name', 'x' * 64)
self.assertEqual(row['field'], name if valid else '<unsupported-field>')
output = json.dumps(rows, ensure_ascii=False)
for private in ('PRIVATE_TOKEN', 'PRIVATE_SECRET', 'nested_secret', 'account_id',
'PRIVATE_NAME_SENTINEL', 'a' * 32):
self.assertNotIn(private, output)
if not valid and name:
self.assertNotIn(name, output)

def test_each_binding_value_change_or_removal_blocks(self):
for index, field in [(0, 'text'), (1, 'namespace_id'), (2, 'class_name'), (3, 'name')]:
with self.subTest(index=index):
Expand Down Expand Up @@ -144,7 +262,7 @@ def test_settings_cron_routes_and_inventory_gaps_block(self):
_, self.live = fixture()
mutation(self.live)
self.assertBlocked()
self.assertNotIn('PRIVATE_', json.dumps(self.rows()))
self.assertNotIn('PRIVATE_VALUE', json.dumps(self.rows()))

def test_unknown_local_configuration_is_not_ignored(self):
self.config['unsafe'] = {'metadata': {'PRIVATE_KEY': 'PRIVATE_VALUE'}}
Expand Down
64 changes: 45 additions & 19 deletions scripts/console_config_compare.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@
'name', 'main', 'account_id', 'compatibility_date', 'compatibility_flags',
'workers_dev', 'preview_urls', 'vars', 'kv_namespaces', 'durable_objects',
'migrations', 'triggers', 'observability', 'limits', 'placement', 'logpush',
'tail_consumers', 'tags', 'usage_model',
'tail_consumers', 'usage_model',
}
SETTING_KEYS = {
'compatibility_date', 'compatibility_flags', 'usage_model', 'limits',
Expand Down Expand Up @@ -185,15 +185,17 @@ def snapshot(reader, name):
def compare(config, live):
rows = []

def row(item, expected=MISSING, actual=MISSING, reason=None):
def row(item, expected=MISSING, actual=MISSING, reason=None, *, field=MISSING):
status = 'unreadable' if reason else ('match' if expected == actual else 'mismatch')
rows.append({'item': item, 'status': status,
'rendered': 'absent' if expected is MISSING else '<redacted>',
'live': 'absent' if actual is MISSING else '<redacted>',
**({'reason': reason} if reason else {})})
**({'reason': reason} if reason else {}),
**({'field': field if isinstance(field, str) and re.fullmatch(r'[A-Za-z0-9_.-]{1,64}', field)
else '<unsupported-field>'} if field is not MISSING else {})})

for index, _ in enumerate(sorted(set(config) - KNOWN_CONFIG)):
row(f'config.unsupported[{index}]', reason='unsupported_config_key')
for index, key in enumerate(sorted(set(config) - KNOWN_CONFIG)):
row(f'config.unsupported[{index}]', reason='unsupported_config_key', field=key)
# These identify the request target and build entrypoint, not deployed code.
row('target.account', config.get('account_id'), os.environ.get('CLOUDFLARE_ACCOUNT_ID'))
if config.get('main') != 'worker.js':
Expand Down Expand Up @@ -271,18 +273,41 @@ def row(item, expected=MISSING, actual=MISSING, reason=None):
row('bindings.' + name, desired[name], MISSING)
for index, _ in enumerate(secret_names):
row(f'secrets.unmatched[{index}]', reason='secret_inventory_disagrees')
# Missing required fields are unknown, not empty defaults.
for key in ('compatibility_date', 'compatibility_flags'):
expected = config.get(key, [] if key == 'compatibility_flags' else MISSING)
row('settings.' + key, expected, settings.get(key, MISSING),
'missing_setting' if key not in settings else None)
row('runtime.' + key, settings.get(key, MISSING), runtime.get(key, MISSING),
'missing_runtime_setting' if key not in runtime else None)
# A date is required. Only empty flags have an equivalent omitted/null form
# in API responses; never invent a flag when the rendered list is nonempty.
row('settings.compatibility_date', config.get('compatibility_date', MISSING),
settings.get('compatibility_date', MISSING),
'missing_setting' if 'compatibility_date' not in settings else None)
row('runtime.compatibility_date', settings.get('compatibility_date', MISSING),
runtime.get('compatibility_date', MISSING),
'missing_runtime_setting' if 'compatibility_date' not in runtime else None)
flags = config.get('compatibility_flags', [])
if not isinstance(flags, list) or any(not isinstance(flag, str) or not flag for flag in flags):
row('config.compatibility_flags', reason='unsupported_compatibility_flags')
for scope, source, reason in [('settings', settings, 'missing_setting'),
('runtime', runtime, 'missing_runtime_setting')]:
actual = source.get('compatibility_flags', MISSING)
if flags == [] and (actual is MISSING or actual is None or actual == []):
row(scope + '.compatibility_flags', [], [])
else:
row(scope + '.compatibility_flags', flags, actual, reason if actual is MISSING else None)
# Known implicit upload behavior; other provider defaults are NOT guessed.
expected_settings = {'observability': {'enabled': False}}
for key in SETTING_KEYS - {'compatibility_date', 'compatibility_flags'}:
expected = config.get(key, expected_settings.get(key, MISSING))
expected = config.get(key, MISSING)
actual = settings.get(key, MISSING)
# Pinned Wrangler's parseConfigPlacement omits placement for mode=off.
# The API can return that no-placement state as an empty object.
if key == 'placement' and expected == {'mode': 'off'}:
expected = {} if actual is not MISSING else MISSING
# An omitted observability config disables logging. Pinned Wrangler
# documents this as a no-op when it is already absent/disabled on live.
if key == 'observability' and expected is MISSING and actual == {'enabled': False}:
expected = actual
# Pinned Wrangler has no top-level tags option. With no configured
# environments, its service/environment tag transform preserves [].
# Nonempty live-only tags still require review rather than inference.
if key == 'tags' and expected is MISSING and actual == []:
expected = []
# Pinned ordinary deploy omits these settings when not configured:
# logpush/tail consumers are omitted from PATCH, usage_model is never
# uploaded. No secret sync or special release mode runs here.
Expand All @@ -292,8 +317,8 @@ def row(item, expected=MISSING, actual=MISSING, reason=None):
row('settings.' + key, expected, actual, 'undeclared_effective_setting')
else:
row('settings.' + key, expected, actual)
for index, _ in enumerate(sorted(set(settings) - SETTING_KEYS - {'bindings'})):
row(f'settings.unsupported[{index}]', reason='unsupported_live_setting')
for index, key in enumerate(sorted(set(settings) - SETTING_KEYS - {'bindings'})):
row(f'settings.unsupported[{index}]', reason='unsupported_live_setting', field=key)
migrations = config.get('migrations', [])
if len(migrations) != 1 or set(migrations[0]) != {'tag', 'new_sqlite_classes'}:
row('migrations', reason='unsupported_migrations')
Expand All @@ -318,7 +343,8 @@ def row(item, expected=MISSING, actual=MISSING, reason=None):
for key in sorted(set(runtime) - {'compatibility_date', 'compatibility_flags', 'migration_tag', 'exports'}):
label = key if key in SETTING_KEYS else 'unsupported'
row('runtime.' + label, settings.get(key, MISSING), runtime[key],
'unmapped_runtime_setting' if key not in SETTING_KEYS or key not in settings else None)
'unmapped_runtime_setting' if key not in SETTING_KEYS or key not in settings else None,
field=key if label == 'unsupported' else MISSING)
subdomain = live.get('subdomain')
if not isinstance(subdomain, dict):
row('subdomain', reason='missing_subdomain')
Expand All @@ -332,8 +358,8 @@ def row(item, expected=MISSING, actual=MISSING, reason=None):
expected = actual if retained else config.get(local, MISSING)
row('subdomain.' + key + ('.retained' if retained else ''), expected, actual,
'unreadable_subdomain_setting' if not isinstance(actual, bool) else None)
for index, _ in enumerate(set(subdomain) - {'enabled', 'previews_enabled'}):
row(f'subdomain.unsupported[{index}]', reason='unsupported_subdomain_setting')
for index, key in enumerate(sorted(set(subdomain) - {'enabled', 'previews_enabled'})):
row(f'subdomain.unsupported[{index}]', reason='unsupported_subdomain_setting', field=key)
schedules = live.get('schedules')
crons = schedules.get('schedules') if isinstance(schedules, dict) else None
if (not isinstance(crons, list) or any(not isinstance(c, dict) or not isinstance(c.get('cron'), str)
Expand Down
19 changes: 19 additions & 0 deletions web/strategy-switch-console/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -341,6 +341,25 @@ HTTP errors use `http`; unexpected exceptions retain
`request_or_schema_unreadable`. Reasons never contain response bodies, URLs,
provider messages, identifiers or exception text.

Unsupported settings remain unreadable and include only a sanitized `field`
name: ASCII letters, digits, `_`, `.`, or `-`, with 1–64 characters. Other names
become `<unsupported-field>`; values and nested content are never emitted.
This diagnostic does not make the field supported.

An omitted or empty rendered `compatibility_flags` list matches an omitted,
null or empty list in the settings and active runtime responses. Nonempty flags
still require an exact match; invalid shapes and missing required dates block
comparison.
The template declares placement mode `off`. Pinned Wrangler
omits placement from the upload for `off` without a hint, matching an absent or
empty live placement object; smart/targeted placement is not inferred. Pinned
Wrangler has no top-level `tags` option: such declarations fail closed. With no
configured environments, its tag transform preserves an empty live tag list;
nonempty undeclared tags still block comparison. An
omitted observability config matches absent logging or exactly
`{"enabled": false}`. Wrangler's implicit disable is a no-op for either state;
enabled logging or additional observability fields still block comparison.

Undeclared provider settings are reported as unknown unless the pinned Wrangler
version has a verified preservation rule; absence is not an invented default.
An omitted `preview_urls` retains the readable live boolean, rather than copying
Expand Down
5 changes: 5 additions & 0 deletions web/strategy-switch-console/wrangler.toml.example
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,11 @@ name = "qsl-strategy-switch-console"
main = "worker.js"
compatibility_date = "2026-06-08"
workers_dev = true
# Preserve the no-placement state (Wrangler omits placement for mode=off).
placement = { mode = "off" }
# No environment grouping; Wrangler preserves the empty live tag inventory.
# Top-level tags is not a supported Wrangler option.
# Keep observability omitted so logging remains disabled.

# Set these with `wrangler secret put ...` instead of committing values:
# - GITHUB_CLIENT_ID
Expand Down
Loading