|
17 | 17 |
|
18 | 18 | import json |
19 | 19 | import re |
| 20 | +from collections.abc import Iterator |
20 | 21 | from typing import Any |
21 | 22 |
|
22 | 23 | import pytest |
| 24 | +import yaml |
23 | 25 | from pydantic import ValidationError |
24 | 26 | from sqlalchemy_celery_beat.models import Period |
25 | 27 |
|
| 28 | +from app import BASE_DIR |
26 | 29 | from app.core.celery.models import IntervalSchedule |
27 | 30 | from app.core.settings_override.registry import ( |
28 | 31 | is_hot_reloadable, |
29 | 32 | materialize_override_value, |
30 | 33 | ) |
31 | 34 | from app.sep.apps.framework.schema import EXECUTION_HOST_LABEL |
32 | | -from app.sep.snippets.config import SnippetsSettings, SnippetSudoOption |
| 35 | +from app.sep.snippets.config import ( |
| 36 | + SnippetFilter, |
| 37 | + SnippetFilterType, |
| 38 | + SnippetsSettings, |
| 39 | + SnippetSudoOption, |
| 40 | +) |
33 | 41 | from app.sep.snippets.models.snippet import BaseSnippet, SUDO_INPUT_NAME |
34 | 42 |
|
35 | 43 | EXECUTOR_HOSTS = frozenset({("host1", "host1")}) |
@@ -355,3 +363,44 @@ def test_invalid_override_rejected(self, bad: Any) -> None: |
355 | 363 | materialize_override_value( |
356 | 364 | SnippetsSettings, "SYNC_INTERVAL", field_info, bad |
357 | 365 | ) |
| 366 | + |
| 367 | + |
| 368 | +class TestShippedSyncFilterConfig: |
| 369 | + """Guard the shipped ``settings.yaml`` snippet sync filter declaration. |
| 370 | +
|
| 371 | + ``FILTER_EXTENSIONS`` was never a ``SnippetsSettings`` field, so the shipped |
| 372 | + key was silently dropped by ``extra="ignore"`` and snippet sync ran |
| 373 | + unfiltered. These tests pin the live ``SYNC_FILTER`` key and keep the dead |
| 374 | + key from creeping back in. |
| 375 | + """ |
| 376 | + |
| 377 | + @staticmethod |
| 378 | + def _shipped_settings() -> dict[str, Any]: |
| 379 | + """Load the repository's tracked ``settings.yaml``.""" |
| 380 | + return yaml.safe_load((BASE_DIR / "settings.yaml").read_text()) |
| 381 | + |
| 382 | + @staticmethod |
| 383 | + def _iter_keys(node: Any) -> Iterator[str]: |
| 384 | + """Yield every mapping key found anywhere in a parsed YAML document.""" |
| 385 | + if isinstance(node, dict): |
| 386 | + for key, value in node.items(): |
| 387 | + yield key |
| 388 | + yield from TestShippedSyncFilterConfig._iter_keys(value) |
| 389 | + elif isinstance(node, list): |
| 390 | + for item in node: |
| 391 | + yield from TestShippedSyncFilterConfig._iter_keys(item) |
| 392 | + |
| 393 | + def test_no_dead_filter_extensions_key(self): |
| 394 | + """Assert no section of the shipped config declares the dead key.""" |
| 395 | + assert "FILTER_EXTENSIONS" not in set(self._iter_keys(self._shipped_settings())) |
| 396 | + |
| 397 | + def test_sync_filter_declares_shell_scripts_only(self): |
| 398 | + """Assert the shipped config restricts sync to ``.sh`` via the live field.""" |
| 399 | + snippets = self._shipped_settings()["default"]["SEP"]["SNIPPETS"] |
| 400 | + assert snippets["SYNC_FILTER"] == [".sh"] |
| 401 | + |
| 402 | + def test_shipped_sync_filter_parses_to_extension_filter(self): |
| 403 | + """Verify the shipped value validates into an extension ``SnippetFilter``.""" |
| 404 | + snippets = self._shipped_settings()["default"]["SEP"]["SNIPPETS"] |
| 405 | + parsed = SnippetsSettings(SYNC_FILTER=snippets["SYNC_FILTER"]) |
| 406 | + assert {SnippetFilter(".sh", SnippetFilterType.EXTENSION)} == parsed.SYNC_FILTER |
0 commit comments