refactor(#27): IMP-27 K5 catalog loader + _get_block_by_id cleanup

Consolidate three duplicated catalog readers and two _get_block_by_id
implementations behind a single shared module (src/catalog.py) that owns
file-read + mtime cache. All caller signatures and return contracts
remain byte-identical.

Units:
- u1 NEW src/catalog.py (76 lines): load_root_catalog / load_blocks /
  get_block_by_id / get_catalog_mtime as the sole file-read +
  mtime-cache owner.
- u2 src/block_reference.py: _load_catalog delegates to load_blocks
  (list[dict] preserved); _get_block_by_id (no-arg) delegates to
  catalog.get_block_by_id. Module-level _catalog_cache removed.
- u3 src/block_selector.py: load_catalog delegates to load_root_catalog
  (root dict preserved); _get_block_by_id (catalog-injected sig
  preserved) delegates to catalog.get_block_by_id. Module-level
  _catalog_cache / _catalog_mtime / CATALOG_PATH removed.
- u4 src/renderer.py: _load_catalog_map and
  _load_catalog_map_with_variants consume catalog.load_blocks; renderer
  projection caches kept local but keyed via
  catalog.get_catalog_mtime(). Per-projection invalidation keys
  (_CATALOG_MAP_MTIME / _CATALOG_VARIANT_MAP_MTIME) introduced. import
  yaml, CATALOG_PATH, legacy _CATALOG_MTIME removed.
- tests NEW tests/test_catalog_shared_loader.py (421 lines, 23 cases):
  shared loader + 3 wrappers covering single file-read, contract
  preservation, signature preservation, shared cache, private state
  absence, mtime invalidation propagation to renderer projections.

Verification:
- pytest tests/test_catalog_shared_loader.py -v: 23/23 PASS in 0.13s.
- pytest tests/ -q --ignore=tests/matching: 365/365 PASS in 38.10s.
- src/fit_verifier.py, src/space_allocator.py, src/pipeline.py and
  templates/catalog.yaml unchanged (git diff empty).

Out of scope:
- catalog.yaml schema/path unchanged.
- Catalog direct-read call sites in fit_verifier / space_allocator /
  pipeline left for a separate follow-up axis.
- Phase Z 22-step runtime, frame_selection, light_edit/restructure
  flows untouched.

Refs: IMP-27 (gitea #27), INSIGHT-MAP §5 K5, PHASE-Q-AUDIT §2.10
This commit is contained in:
2026-05-20 19:31:26 +09:00
parent 2896bb691c
commit 909bf75edc
5 changed files with 550 additions and 101 deletions

View File

@@ -20,9 +20,10 @@ import re
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
import yaml
from jinja2 import Environment, FileSystemLoader from jinja2 import Environment, FileSystemLoader
from src import catalog as _catalog_mod
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# 템플릿 디렉토리 # 템플릿 디렉토리
@@ -101,32 +102,18 @@ RELATION_CATEGORY_MAP: dict[str, list[str]] = {
# ══════════════════════════════════════ # ══════════════════════════════════════
# 카탈로그 로딩 (mtime 캐싱) # 카탈로그 로딩 (IMP-27: src.catalog 공유 로더 위임)
# ══════════════════════════════════════ # ══════════════════════════════════════
_catalog_cache: dict[str, Any] = {"data": None, "mtime": 0}
def _load_catalog() -> list[dict]: def _load_catalog() -> list[dict]:
"""catalog.yaml 로드 (mtime 캐싱).""" """catalog.yaml blocks list (IMP-27: shared loader delegation)."""
path = TEMPLATES_DIR / "catalog.yaml" return _catalog_mod.load_blocks()
mtime = path.stat().st_mtime
if _catalog_cache["data"] is not None and _catalog_cache["mtime"] == mtime:
return _catalog_cache["data"]
data = yaml.safe_load(path.read_text(encoding="utf-8"))
blocks = data.get("blocks", [])
_catalog_cache["data"] = blocks
_catalog_cache["mtime"] = mtime
return blocks
def _get_block_by_id(block_id: str) -> dict | None: def _get_block_by_id(block_id: str) -> dict | None:
"""블록 ID로 카탈로그 엔트리 조회.""" """블록 ID로 카탈로그 엔트리 조회 (IMP-27: shared loader delegation)."""
for b in _load_catalog(): return _catalog_mod.get_block_by_id(block_id)
if b["id"] == block_id:
return b
return None
# ══════════════════════════════════════ # ══════════════════════════════════════

View File

@@ -5,24 +5,18 @@ AI에게 불가능한 선택지를 주지 않는다 (Beautiful.ai 원칙).
주요 함수: 주요 함수:
- select_block_candidates(): topic + 컨테이너 → 물리적으로 가능한 후보 2-4개 - select_block_candidates(): topic + 컨테이너 → 물리적으로 가능한 후보 2-4개
- load_catalog(): catalog.yaml 로딩 + 캐싱 - load_catalog(): catalog.yaml 로딩 + 캐싱 (IMP-27: src.catalog 공유 로더 위임)
""" """
from __future__ import annotations from __future__ import annotations
import logging import logging
from pathlib import Path
from typing import Any from typing import Any
import yaml from src import catalog as _catalog_mod
from src.space_allocator import ContainerSpec, HEIGHT_COST_ORDER from src.space_allocator import ContainerSpec, HEIGHT_COST_ORDER
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
CATALOG_PATH = Path("templates/catalog.yaml")
_catalog_cache: dict | None = None
_catalog_mtime: float = 0.0
# ────────────────────────────────────── # ──────────────────────────────────────
# relation_type → 블록 카테고리 매핑 (Napkin.ai 방식) # relation_type → 블록 카테고리 매핑 (Napkin.ai 방식)
@@ -52,35 +46,16 @@ BLOCKS_FORCING_FORMAT_CHANGE = {
# ────────────────────────────────────── # ──────────────────────────────────────
# catalog.yaml 로딩 (mtime 캐시) # catalog.yaml 로딩 (IMP-27: src.catalog 공유 로더 위임)
# ────────────────────────────────────── # ──────────────────────────────────────
def load_catalog() -> dict: def load_catalog() -> dict:
"""catalog.yaml을 로딩한다. mtime 기반 캐싱.""" """catalog.yaml root dict (IMP-27: shared loader delegation)."""
global _catalog_cache, _catalog_mtime return _catalog_mod.load_root_catalog()
if not CATALOG_PATH.exists():
logger.error(f"catalog.yaml 미발견: {CATALOG_PATH}")
return {"blocks": []}
current_mtime = CATALOG_PATH.stat().st_mtime
if _catalog_cache is not None and current_mtime == _catalog_mtime:
return _catalog_cache
with open(CATALOG_PATH, encoding="utf-8") as f:
_catalog_cache = yaml.safe_load(f)
_catalog_mtime = current_mtime
block_count = len(_catalog_cache.get("blocks", []))
logger.info(f"[Q-2] catalog.yaml 로딩: {block_count}개 블록")
return _catalog_cache
def _get_block_by_id(block_id: str, catalog: dict) -> dict | None: def _get_block_by_id(block_id: str, catalog: dict) -> dict | None:
"""catalog에서 블록 ID로 검색.""" """catalog-injected 블록 ID 조회 (IMP-27: shared loader delegation)."""
for block in catalog.get("blocks", []): return _catalog_mod.get_block_by_id(block_id, catalog)
if block.get("id") == block_id:
return block
return None
# ────────────────────────────────────── # ──────────────────────────────────────

76
src/catalog.py Normal file
View File

@@ -0,0 +1,76 @@
"""IMP-27: Shared catalog.yaml loader (single file-read + mtime cache).
Phase Q evolution 중 block_reference, block_selector, renderer 가 각각 templates/
catalog.yaml 을 읽고 mtime 캐시하던 중복을 한 곳으로 통합한다. call-site
signature 는 그대로 유지되며, 각 wrapper 는 본 모듈의 결과를 자신이 약속하는
형태(list[dict] / root dict / id→path projection)로 변환만 수행한다.
Functions:
load_root_catalog() -> dict : raw catalog dict (matches block_selector contract)
load_blocks() -> list[dict] : root_catalog.get("blocks", []) projection
get_block_by_id(block_id, catalog=None) -> dict | None
get_catalog_mtime() -> float : current cached mtime (renderer projection key)
"""
from __future__ import annotations
import logging
from pathlib import Path
import yaml
logger = logging.getLogger(__name__)
CATALOG_PATH = Path(__file__).parent.parent / "templates" / "catalog.yaml"
_catalog_cache: dict | None = None
_catalog_mtime: float = 0.0
def load_root_catalog() -> dict:
"""Load templates/catalog.yaml as root dict, with mtime caching.
Missing file → logs warning and returns ``{"blocks": []}`` (matches the
pre-IMP-27 behavior of block_selector.load_catalog and renderer._load_catalog_map).
"""
global _catalog_cache, _catalog_mtime
if not CATALOG_PATH.exists():
logger.warning(f"catalog.yaml 미발견: {CATALOG_PATH}")
return {"blocks": []}
current_mtime = CATALOG_PATH.stat().st_mtime
if _catalog_cache is not None and current_mtime == _catalog_mtime:
return _catalog_cache
with open(CATALOG_PATH, encoding="utf-8") as f:
_catalog_cache = yaml.safe_load(f)
_catalog_mtime = current_mtime
block_count = len((_catalog_cache or {}).get("blocks", []))
logger.info(f"[catalog] load: {block_count} blocks")
return _catalog_cache
def load_blocks() -> list[dict]:
"""Return blocks list (= root_catalog.get('blocks', []))."""
return load_root_catalog().get("blocks", [])
def get_block_by_id(block_id: str, catalog: dict | None = None) -> dict | None:
"""Locate a block entry by id.
``catalog=None`` → uses shared loader. caller-supplied catalog dict is
accepted as-is so the existing block_selector contract (catalog-injected)
keeps working unchanged.
"""
if catalog is None:
catalog = load_root_catalog()
for block in catalog.get("blocks", []):
if block.get("id") == block_id:
return block
return None
def get_catalog_mtime() -> float:
"""Current cached mtime (renderer projection caches key off this)."""
return _catalog_mtime

View File

@@ -13,86 +13,76 @@ from collections import OrderedDict
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
import yaml
from jinja2 import Environment, FileSystemLoader from jinja2 import Environment, FileSystemLoader
from src import catalog as _catalog_mod
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
TEMPLATES_DIR = Path(__file__).parent.parent / "templates" TEMPLATES_DIR = Path(__file__).parent.parent / "templates"
STATIC_DIR = Path(__file__).parent.parent / "static" STATIC_DIR = Path(__file__).parent.parent / "static"
CATALOG_PATH = TEMPLATES_DIR / "catalog.yaml"
# 카테고리 검색 순서 # 카테고리 검색 순서
BLOCK_CATEGORIES = ["headers", "cards", "tables", "visuals", "emphasis", "media"] BLOCK_CATEGORIES = ["headers", "cards", "tables", "visuals", "emphasis", "media"]
# catalog.yaml에서 id → template 경로 매핑 로드 (BF-10: mtime 체크로 자동 갱신) # id → template 경로 매핑 (IMP-27: src.catalog 공유 로더 위임, renderer-local projection cache)
_CATALOG_MAP: dict[str, str] | None = None _CATALOG_MAP: dict[str, str] | None = None
_CATALOG_MTIME: float = 0.0 _CATALOG_MAP_MTIME: float = 0.0
# Phase R: variant별 template 경로 캐시 (renderer-local projection)
_CATALOG_VARIANT_MAP: dict[str, str] | None = None
_CATALOG_VARIANT_MAP_MTIME: float = 0.0
def _load_catalog_map() -> dict[str, str]: def _load_catalog_map() -> dict[str, str]:
"""catalog.yaml에서 블록 id → template 경로 매핑을 로드한다. """블록 id → template 경로 projection (IMP-27: src.catalog 공유 로더 위임).
파일 수정시간(mtime)을 확인하여, 변경 시에만 재로드한다. catalog 파일 읽기와 mtime 캐싱은 ``src.catalog`` 가 단독 소유. 본 함수는
그 결과를 ``id → template`` 형태로 변환한 renderer-local projection 캐시만
유지하며, projection 무효화는 ``src.catalog.get_catalog_mtime()`` 키잉.
""" """
global _CATALOG_MAP, _CATALOG_MTIME global _CATALOG_MAP, _CATALOG_MAP_MTIME
current_mtime = CATALOG_PATH.stat().st_mtime if CATALOG_PATH.exists() else 0.0 blocks = _catalog_mod.load_blocks()
current_mtime = _catalog_mod.get_catalog_mtime()
if _CATALOG_MAP is not None and _CATALOG_MTIME == current_mtime: if _CATALOG_MAP is not None and _CATALOG_MAP_MTIME == current_mtime:
return _CATALOG_MAP # 파일 변경 없음 → 캐시 재사용 return _CATALOG_MAP
# 변경 감지 또는 첫 로드 → 새로 읽기 _CATALOG_MAP_MTIME = current_mtime
_CATALOG_MTIME = current_mtime
_CATALOG_MAP = {} _CATALOG_MAP = {}
if CATALOG_PATH.exists(): for block in blocks:
try: block_id = block.get("id", "")
with open(CATALOG_PATH, encoding="utf-8") as f: template = block.get("template", "")
catalog = yaml.safe_load(f) if block_id and template:
for block in catalog.get("blocks", []): _CATALOG_MAP[block_id] = template
block_id = block.get("id", "") logger.info(f"catalog.yaml 로드: {len(_CATALOG_MAP)}개 블록 매핑")
template = block.get("template", "")
if block_id and template:
_CATALOG_MAP[block_id] = template
logger.info(f"catalog.yaml 로드: {len(_CATALOG_MAP)}개 블록 매핑")
except Exception as e:
logger.warning(f"catalog.yaml 로드 실패: {e}")
else:
logger.warning(f"catalog.yaml 미발견: {CATALOG_PATH}")
return _CATALOG_MAP return _CATALOG_MAP
# Phase R: variant별 template 경로 캐시
_CATALOG_VARIANT_MAP: dict[str, str] | None = None
def _load_catalog_map_with_variants() -> dict[str, str]: def _load_catalog_map_with_variants() -> dict[str, str]:
"""catalog.yaml에서 variant별 template 경로 매핑을 로드한다. """variant별 template 경로 projection (IMP-27: src.catalog 공유 로더 위임).
키: "block_id--variant_id" → 값: template 경로 키: "block_id--variant_id" → 값: template 경로.
""" """
global _CATALOG_VARIANT_MAP global _CATALOG_VARIANT_MAP, _CATALOG_VARIANT_MAP_MTIME
# _load_catalog_map이 이미 캐시 관리하므로 같은 mtime 사용 blocks = _catalog_mod.load_blocks()
_load_catalog_map() # 캐시 갱신 보장 current_mtime = _catalog_mod.get_catalog_mtime()
if _CATALOG_VARIANT_MAP is not None and _CATALOG_MTIME == (CATALOG_PATH.stat().st_mtime if CATALOG_PATH.exists() else 0.0): if _CATALOG_VARIANT_MAP is not None and _CATALOG_VARIANT_MAP_MTIME == current_mtime:
return _CATALOG_VARIANT_MAP return _CATALOG_VARIANT_MAP
_CATALOG_VARIANT_MAP_MTIME = current_mtime
_CATALOG_VARIANT_MAP = {} _CATALOG_VARIANT_MAP = {}
if CATALOG_PATH.exists(): for block in blocks:
try: block_id = block.get("id", "")
with open(CATALOG_PATH, encoding="utf-8") as f: for variant in block.get("variants", []):
catalog = yaml.safe_load(f) vid = variant.get("id", "default")
for block in catalog.get("blocks", []): vtemplate = variant.get("template", "")
block_id = block.get("id", "") if vid != "default" and vtemplate:
for variant in block.get("variants", []): _CATALOG_VARIANT_MAP[f"{block_id}--{vid}"] = vtemplate
vid = variant.get("id", "default")
vtemplate = variant.get("template", "")
if vid != "default" and vtemplate:
_CATALOG_VARIANT_MAP[f"{block_id}--{vid}"] = vtemplate
except Exception as e:
logger.warning(f"catalog variant 로드 실패: {e}")
return _CATALOG_VARIANT_MAP return _CATALOG_VARIANT_MAP

View File

@@ -0,0 +1,421 @@
"""IMP-27: Shared catalog loader tests (u1).
Validates that ``src.catalog`` provides a single file-read + mtime cache and
that its four public functions honor the documented contracts.
Note: ``templates/catalog.yaml`` was deleted in cc2f434 (legacy block library
cleanup). The shared loader still preserves the loader contract for any
remaining Phase Q call sites; tests use a fixture catalog via monkeypatch so
they are independent of the deleted production file.
Delegation tests for block_reference / block_selector / renderer wrappers are
added in u2 / u3 / u4.
"""
from __future__ import annotations
from pathlib import Path
import pytest
import yaml
FIXTURE_CATALOG = {
"blocks": [
{
"id": "fixture-block-a",
"category": "emphasis",
"template": "blocks/emphasis/fixture-block-a.html",
},
{
"id": "fixture-block-b",
"category": "cards",
"template": "blocks/cards/fixture-block-b.html",
"variants": [
{"id": "default", "template": "blocks/cards/fixture-block-b.html"},
{"id": "compact", "template": "blocks/cards/fixture-block-b--compact.html"},
],
},
]
}
def _reset_catalog_module():
import src.catalog as catalog_mod
catalog_mod._catalog_cache = None
catalog_mod._catalog_mtime = 0.0
def _reset_renderer_projection_cache():
"""IMP-27 u4: clear renderer-local projection caches between tests."""
import src.renderer as renderer_mod
renderer_mod._CATALOG_MAP = None
renderer_mod._CATALOG_MAP_MTIME = 0.0
renderer_mod._CATALOG_VARIANT_MAP = None
renderer_mod._CATALOG_VARIANT_MAP_MTIME = 0.0
@pytest.fixture
def fixture_catalog_path(tmp_path, monkeypatch):
"""Point src.catalog at a tmp catalog.yaml fixture and reset its cache."""
import src.catalog as catalog_mod
fixture_path = tmp_path / "catalog.yaml"
fixture_path.write_text(yaml.safe_dump(FIXTURE_CATALOG), encoding="utf-8")
monkeypatch.setattr(catalog_mod, "CATALOG_PATH", fixture_path)
_reset_catalog_module()
_reset_renderer_projection_cache()
return fixture_path
def test_load_root_catalog_returns_root_dict(fixture_catalog_path):
from src import catalog
root = catalog.load_root_catalog()
assert isinstance(root, dict)
assert "blocks" in root
assert len(root["blocks"]) == 2
def test_load_blocks_returns_list_of_block_dicts(fixture_catalog_path):
from src import catalog
blocks = catalog.load_blocks()
assert isinstance(blocks, list)
assert len(blocks) == 2
assert all(isinstance(b, dict) for b in blocks)
assert all("id" in b for b in blocks)
def test_get_block_by_id_without_catalog_arg(fixture_catalog_path):
from src import catalog
found = catalog.get_block_by_id("fixture-block-a")
assert found is not None
assert found["id"] == "fixture-block-a"
assert found["category"] == "emphasis"
def test_get_block_by_id_with_catalog_arg_preserves_block_selector_contract(fixture_catalog_path):
from src import catalog
root = catalog.load_root_catalog()
found = catalog.get_block_by_id("fixture-block-b", catalog=root)
assert found is not None
assert found["id"] == "fixture-block-b"
def test_get_block_by_id_unknown_returns_none(fixture_catalog_path):
from src import catalog
assert catalog.get_block_by_id("__nonexistent_block_id__") is None
def test_mtime_cache_does_single_file_read(fixture_catalog_path, monkeypatch):
import src.catalog as catalog_mod
read_count = {"n": 0}
real_safe_load = catalog_mod.yaml.safe_load
def counting_safe_load(stream):
read_count["n"] += 1
return real_safe_load(stream)
monkeypatch.setattr(catalog_mod.yaml, "safe_load", counting_safe_load)
catalog_mod.load_root_catalog()
catalog_mod.load_root_catalog()
catalog_mod.load_blocks()
catalog_mod.get_block_by_id("fixture-block-a")
assert read_count["n"] == 1, (
f"Expected exactly one yaml.safe_load call on cold cache, "
f"got {read_count['n']}"
)
def test_mtime_change_triggers_reload(fixture_catalog_path, monkeypatch):
import os
import src.catalog as catalog_mod
read_count = {"n": 0}
real_safe_load = catalog_mod.yaml.safe_load
def counting_safe_load(stream):
read_count["n"] += 1
return real_safe_load(stream)
monkeypatch.setattr(catalog_mod.yaml, "safe_load", counting_safe_load)
catalog_mod.load_root_catalog()
assert read_count["n"] == 1
# Forcibly advance file mtime and verify cache invalidates.
original_mtime = fixture_catalog_path.stat().st_mtime
os.utime(fixture_catalog_path, (original_mtime + 10, original_mtime + 10))
catalog_mod.load_root_catalog()
assert read_count["n"] == 2
def test_get_catalog_mtime_matches_file_after_load(fixture_catalog_path):
from src import catalog
catalog.load_root_catalog()
actual = fixture_catalog_path.stat().st_mtime
assert catalog.get_catalog_mtime() == actual
def test_get_catalog_mtime_is_zero_before_first_load(monkeypatch, tmp_path):
import src.catalog as catalog_mod
monkeypatch.setattr(catalog_mod, "CATALOG_PATH", tmp_path / "unused.yaml")
_reset_catalog_module()
assert catalog_mod.get_catalog_mtime() == 0.0
def test_load_root_catalog_missing_file_returns_empty_blocks(monkeypatch, tmp_path):
import src.catalog as catalog_mod
missing_path = tmp_path / "no_such_catalog.yaml"
monkeypatch.setattr(catalog_mod, "CATALOG_PATH", missing_path)
_reset_catalog_module()
root = catalog_mod.load_root_catalog()
assert root == {"blocks": []}
assert catalog_mod.load_blocks() == []
# ──────────────────────────────────────────────────────────
# u2: block_reference delegation tests
# ──────────────────────────────────────────────────────────
def test_block_reference_load_catalog_returns_list_of_blocks(fixture_catalog_path):
"""block_reference._load_catalog preserves list[dict] contract via delegation."""
from src import block_reference
blocks = block_reference._load_catalog()
assert isinstance(blocks, list)
assert len(blocks) == 2
assert all(isinstance(b, dict) for b in blocks)
assert {b["id"] for b in blocks} == {"fixture-block-a", "fixture-block-b"}
def test_block_reference_get_block_by_id_no_arg_signature(fixture_catalog_path):
"""block_reference._get_block_by_id preserves no-catalog-argument contract."""
from src import block_reference
found = block_reference._get_block_by_id("fixture-block-a")
assert found is not None
assert found["id"] == "fixture-block-a"
assert found["category"] == "emphasis"
assert block_reference._get_block_by_id("__nonexistent__") is None
def test_block_reference_shares_cache_with_shared_loader(fixture_catalog_path, monkeypatch):
"""block_reference wrappers must hit the shared mtime cache, not a private copy."""
import src.catalog as catalog_mod
from src import block_reference
read_count = {"n": 0}
real_safe_load = catalog_mod.yaml.safe_load
def counting_safe_load(stream):
read_count["n"] += 1
return real_safe_load(stream)
monkeypatch.setattr(catalog_mod.yaml, "safe_load", counting_safe_load)
block_reference._load_catalog()
block_reference._get_block_by_id("fixture-block-a")
catalog_mod.load_root_catalog()
assert read_count["n"] == 1, (
f"block_reference wrappers should share the catalog cache; "
f"got {read_count['n']} reads"
)
def test_block_reference_has_no_private_catalog_cache(fixture_catalog_path):
"""IMP-27 u2 guard: block_reference must not retain a module-level _catalog_cache."""
from src import block_reference
assert not hasattr(block_reference, "_catalog_cache"), (
"block_reference._catalog_cache must be removed by u2 — "
"loader is delegated to src.catalog"
)
# ──────────────────────────────────────────────────────────
# u3: block_selector delegation tests
# ──────────────────────────────────────────────────────────
def test_block_selector_load_catalog_returns_root_dict(fixture_catalog_path):
"""block_selector.load_catalog preserves root-dict contract via delegation."""
from src import block_selector
root = block_selector.load_catalog()
assert isinstance(root, dict)
assert "blocks" in root
assert isinstance(root["blocks"], list)
assert {b["id"] for b in root["blocks"]} == {"fixture-block-a", "fixture-block-b"}
def test_block_selector_get_block_by_id_catalog_injected_signature(fixture_catalog_path):
"""block_selector._get_block_by_id preserves catalog-injected signature."""
from src import block_selector
catalog_dict = block_selector.load_catalog()
found = block_selector._get_block_by_id("fixture-block-b", catalog_dict)
assert found is not None
assert found["id"] == "fixture-block-b"
assert found["category"] == "cards"
assert block_selector._get_block_by_id("__nonexistent__", catalog_dict) is None
def test_block_selector_shares_cache_with_shared_loader(fixture_catalog_path, monkeypatch):
"""block_selector wrappers must hit the shared mtime cache, not a private copy."""
import src.catalog as catalog_mod
from src import block_selector
read_count = {"n": 0}
real_safe_load = catalog_mod.yaml.safe_load
def counting_safe_load(stream):
read_count["n"] += 1
return real_safe_load(stream)
monkeypatch.setattr(catalog_mod.yaml, "safe_load", counting_safe_load)
block_selector.load_catalog()
block_selector._get_block_by_id("fixture-block-a", block_selector.load_catalog())
catalog_mod.load_root_catalog()
assert read_count["n"] == 1, (
f"block_selector wrappers should share the catalog cache; "
f"got {read_count['n']} reads"
)
def test_block_selector_has_no_private_catalog_cache(fixture_catalog_path):
"""IMP-27 u3 guard: block_selector must not retain module-level cache/mtime/CATALOG_PATH."""
from src import block_selector
assert not hasattr(block_selector, "_catalog_cache"), (
"block_selector._catalog_cache must be removed by u3 — "
"loader is delegated to src.catalog"
)
assert not hasattr(block_selector, "_catalog_mtime"), (
"block_selector._catalog_mtime must be removed by u3"
)
assert not hasattr(block_selector, "CATALOG_PATH"), (
"block_selector.CATALOG_PATH must be removed by u3 — "
"path lives in src.catalog only"
)
# ──────────────────────────────────────────────────────────
# u4: renderer delegation tests
# ──────────────────────────────────────────────────────────
def test_renderer_load_catalog_map_returns_id_to_template_dict(fixture_catalog_path):
"""renderer._load_catalog_map preserves id → template projection contract."""
from src import renderer
mapping = renderer._load_catalog_map()
assert isinstance(mapping, dict)
assert mapping["fixture-block-a"] == "blocks/emphasis/fixture-block-a.html"
assert mapping["fixture-block-b"] == "blocks/cards/fixture-block-b.html"
def test_renderer_load_catalog_map_with_variants_returns_compound_key_dict(fixture_catalog_path):
"""renderer._load_catalog_map_with_variants preserves 'id--variant' → template projection."""
from src import renderer
vmap = renderer._load_catalog_map_with_variants()
assert isinstance(vmap, dict)
# 'default' variants must be excluded (preserves pre-IMP-27 behavior).
assert "fixture-block-b--default" not in vmap
# Non-default variants must be present with the compound key.
assert vmap["fixture-block-b--compact"] == "blocks/cards/fixture-block-b--compact.html"
def test_renderer_shares_cache_with_shared_loader(fixture_catalog_path, monkeypatch):
"""renderer projections must read through src.catalog, never opening the file directly."""
import src.catalog as catalog_mod
from src import renderer
read_count = {"n": 0}
real_safe_load = catalog_mod.yaml.safe_load
def counting_safe_load(stream):
read_count["n"] += 1
return real_safe_load(stream)
monkeypatch.setattr(catalog_mod.yaml, "safe_load", counting_safe_load)
renderer._load_catalog_map()
renderer._load_catalog_map_with_variants()
catalog_mod.load_root_catalog()
assert read_count["n"] == 1, (
f"renderer projections should share the catalog cache; "
f"got {read_count['n']} reads"
)
def test_renderer_projection_invalidates_when_shared_mtime_changes(
fixture_catalog_path, monkeypatch
):
"""IMP-27 u4 contract: renderer projection cache keyed off src.catalog.get_catalog_mtime."""
import os
import src.catalog as catalog_mod
from src import renderer
first_map = renderer._load_catalog_map()
first_id_to_path = dict(first_map)
# Rewrite the fixture file with a different block id and bump mtime so the
# shared cache (and therefore the renderer projection) must rebuild.
new_catalog = {
"blocks": [
{
"id": "fixture-block-c",
"category": "headers",
"template": "blocks/headers/fixture-block-c.html",
},
]
}
fixture_catalog_path.write_text(yaml.safe_dump(new_catalog), encoding="utf-8")
original_mtime = fixture_catalog_path.stat().st_mtime
os.utime(fixture_catalog_path, (original_mtime + 10, original_mtime + 10))
# Force src.catalog to drop its in-memory cache so the next call re-reads.
catalog_mod._catalog_cache = None
catalog_mod._catalog_mtime = 0.0
second_map = renderer._load_catalog_map()
assert "fixture-block-c" in second_map
assert "fixture-block-a" not in second_map
assert first_id_to_path != second_map
def test_renderer_has_no_private_catalog_path_or_yaml(fixture_catalog_path):
"""IMP-27 u4 guard: renderer must not retain its own file-read state."""
from src import renderer
assert not hasattr(renderer, "CATALOG_PATH"), (
"renderer.CATALOG_PATH must be removed by u4 — path lives in src.catalog only"
)
assert not hasattr(renderer, "yaml"), (
"renderer.yaml import must be removed by u4 — file-read is delegated to src.catalog"
)
assert not hasattr(renderer, "_CATALOG_MTIME"), (
"renderer._CATALOG_MTIME (legacy single mtime) must be removed by u4 — "
"projection caches now key off src.catalog.get_catalog_mtime() via "
"_CATALOG_MAP_MTIME / _CATALOG_VARIANT_MAP_MTIME"
)