Cachy / backend /tests /test_structuring.py
Vatxzz's picture
refactor: remove HuggingFace as LLM backend; keep Cerebras→Groq chain
ef9f7d7
Raw
History Blame Contribute Delete
6.14 kB
"""Structuring validation + fallback (docs/04). No Gemini key in tests, so the
real call is never made; we exercise validation paths directly and via structure()."""
from app.models.card import ContentType, PrimaryActionKind
from app.pipeline import structuring
def test_no_key_falls_back_to_paragraph():
sc = structuring.structure(
bundle="CAPTION:\nTRANSCRIPT: Mix flour and water.\nON-SCREEN TEXT:\nSOURCE: x",
transcript="Mix flour and water.",
caption="",
)
assert sc.base.one_liner # always non-empty
assert sc.base.tldr
assert any(b["type"] == "paragraph" for b in sc.blocks)
def test_valid_gemini_json_is_accepted():
raw = (
'{"base": {"one_liner": "3 ways to cut AWS bill", "tldr": "Cut costs.",'
' "content_type": "tip", "type_confidence": 0.8},'
' "blocks": [{"type": "bullet_list", "items": ["a", "b"]}]}'
)
sc = structuring._validate(raw, bundle="", transcript="", caption="")
assert sc.base.content_type == ContentType.TIP
assert sc.base.one_liner == "3 ways to cut AWS bill"
assert sc.blocks[0]["type"] == "bullet_list"
assert sc.primary_action.kind == PrimaryActionKind.REMINDER
def test_bad_json_falls_back():
sc = structuring._validate(
"not json at all", bundle="", transcript="some words here", caption=""
)
assert sc.base.one_liner
assert any(b["type"] == "paragraph" for b in sc.blocks)
def test_fenced_json_is_stripped():
raw = '```json\n{"base": {"one_liner": "x", "tldr": "y"}, "blocks": [{"type":"paragraph","text":"z"}]}\n```'
sc = structuring._validate(raw, bundle="", transcript="", caption="")
assert sc.base.one_liner == "x"
assert sc.blocks[0]["text"] == "z"
def test_one_liner_synthesized_when_omitted():
raw = '{"base": {"content_type": "other"}, "blocks": [{"type":"paragraph","text":"body"}]}'
sc = structuring._validate(raw, bundle="", transcript="A useful fact. More.", caption="")
assert sc.base.one_liner
assert sc.base.tldr
def test_no_llm_key_falls_back_to_paragraph():
# No CEREBRAS_API_KEY or GROQ_API_KEY in tests -> dispatch must not
# crash and must degrade to the paragraph fallback.
sc = structuring.structure(
bundle="TRANSCRIPT: Mix flour and water.",
transcript="Mix flour and water.",
caption="",
)
assert sc.base.one_liner
assert any(b["type"] == "paragraph" for b in sc.blocks)
def test_primary_action_mapping():
raw_recipe = '{"base":{"one_liner":"a","tldr":"b","content_type":"recipe"},"blocks":[{"type":"paragraph","text":"x"}]}'
sc = structuring._validate(raw_recipe, bundle="", transcript="", caption="")
assert sc.primary_action.kind == PrimaryActionKind.SHOPPING_LIST
def test_artifacts_extracted_and_validated():
raw = (
'{"base":{"one_liner":"5 books","tldr":"reading list","content_type":"tip"},'
'"blocks":[{"type":"bullet_list","items":["a"]}],'
'"artifacts":['
'{"type":"book","title":"Atomic Habits","creator":"James Clear","year":2018},'
'{"type":"bogus","title":"Dune"},'
'{"title":""},'
'"not a dict",'
'{"type":"book","title":"atomic habits"}]}' # dup of first (case-insensitive)
)
sc = structuring._validate(raw, bundle="", transcript="", caption="")
titles = [(a.type.value, a.title) for a in sc.artifacts]
assert ("book", "Atomic Habits") in titles
assert ("other", "Dune") in titles # bad type coerced to other
assert len(sc.artifacts) == 2 # empty-title, non-dict, and dup dropped
def test_artifacts_kept_even_when_blocks_empty():
raw = '{"base":{"one_liner":"x","tldr":"y"},"blocks":[],"artifacts":[{"type":"movie","title":"Inception"}]}'
sc = structuring._validate(raw, bundle="", transcript="body text", caption="")
assert any(b["type"] == "paragraph" for b in sc.blocks) # fallback body
assert [a.title for a in sc.artifacts] == ["Inception"]
def test_no_artifacts_is_normal():
raw = '{"base":{"one_liner":"x","tldr":"y","content_type":"tip"},"blocks":[{"type":"paragraph","text":"z"}]}'
sc = structuring._validate(raw, bundle="", transcript="", caption="")
assert sc.artifacts == []
# ---------------------------------------------------------------------------
# _coerce_tags
# ---------------------------------------------------------------------------
def test_tags_coerced_lowercase_deduped():
from app.pipeline.structuring import _coerce_tags
result = _coerce_tags(["AWS", "cost-optimization", "AWS", "cloud"])
assert result == ["aws", "cost-optimization", "cloud"]
def test_tags_non_string_entries_dropped():
from app.pipeline.structuring import _coerce_tags
result = _coerce_tags(["valid", 42, None, True, "also-valid"])
assert result == ["valid", "also-valid"]
def test_tags_capped_at_six():
from app.pipeline.structuring import _coerce_tags
result = _coerce_tags([f"tag{i}" for i in range(10)])
assert len(result) == 6
def test_tags_over_40_chars_dropped():
from app.pipeline.structuring import _coerce_tags
long_tag = "a" * 41
result = _coerce_tags([long_tag, "short"])
assert result == ["short"]
def test_tags_empty_list_accepted():
from app.pipeline.structuring import _coerce_tags
assert _coerce_tags([]) == []
def test_tags_non_list_returns_empty():
from app.pipeline.structuring import _coerce_tags
assert _coerce_tags("not-a-list") == []
assert _coerce_tags(None) == []
def test_tags_present_in_validated_card():
raw = (
'{"base":{"one_liner":"x","tldr":"y","content_type":"tip",'
'"tags":["Python","python","machine learning"]},'
'"blocks":[{"type":"paragraph","text":"z"}]}'
)
sc = structuring._validate(raw, bundle="", transcript="", caption="")
assert sc.base.tags == ["python", "machine learning"] # deduped
def test_tags_absent_defaults_to_empty():
raw = '{"base":{"one_liner":"x","tldr":"y"},"blocks":[{"type":"paragraph","text":"z"}]}'
sc = structuring._validate(raw, bundle="", transcript="", caption="")
assert sc.base.tags == []