feat(stage6): source independence & citation graph — detect syndication, shared origins, text similarity
- SourceIndependenceModel: per-source independence_score (0.0-1.0), syndication_group_id, primary_source_id, content_hash, shared_urls - CitationGraphEdgeModel: directed edges (SYNDICATED, QUOTES, LINKS_TO, REPOST, SIMILAR_CONTENT) with confidence + evidence - Content-Hash (SHA-256): instant syndication detection for identical content - difflib Vorfilterung: >60% → LLM, >80% → high confidence, 100% → immediate syndication - LLM-Pairwise-Analysis: two-text-comparison for suspicious pairs only (bounded concurrency) - independence_score: 1.0 base, -0.4 for syndicated, -0.1 per high-similarity pair - Pydantic schemas: SourceIndependenceScore, CitationGraphEdge, SyntacticSimilarityResult, LlmSyndicationAnalysis, SourceIndependenceAnalysisResult - LLM response parser: handles JSON, markdown code blocks, partial/invalid JSON - 43 tests: content hash, similarity thresholds, LLM parsing, analyzer integration, edge cases, prompt templates
This commit is contained in:
NSCT Agent
585
tests/stages/test_stage6_source_independence.py
Normal file
585
tests/stages/test_stage6_source_independence.py
Normal file
@@ -0,0 +1,585 @@
|
||||
"""Tests für Stage 6: Source Independence & Citation Graph — 30+ Test-Fälle.
|
||||
|
||||
Abdeckungen:
|
||||
- Content-Hash: identisch, nahezu-identisch, komplett-unterschiedlich
|
||||
- Text Similarity (difflib): threshold-basierte Vorfilterung
|
||||
- LLM-Response-Parsing: JSON, Markdown-Code-Blocks, fehlerhaft
|
||||
- independence_score: Syndication, identische Hashes, unabhängige Quellen
|
||||
- CitationGraph: Edge-Erstellung für Syndicated/REPOST/SIMILAR_CONTENT
|
||||
- Edge Cases: 0 Quellen, 1 Quelle, alle identisch, alle unabhängig
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
from uuid import UUID, uuid4
|
||||
|
||||
import pytest
|
||||
|
||||
from nsct.models.source_independence import (
|
||||
CitationEdgeType,
|
||||
LlmSyndicationAnalysis,
|
||||
parse_llm_syndication_response,
|
||||
SourceIndependenceScore,
|
||||
SourceIndependenceAnalysisResult,
|
||||
)
|
||||
from nsct.stages.stage6_source_independence import (
|
||||
SourceIndependenceAnalyzer,
|
||||
SIMILARITY_HIGH,
|
||||
SIMILARITY_IDENTICAL,
|
||||
SIMILARITY_MODERATE,
|
||||
compute_content_hash,
|
||||
compute_similarity_ratio,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helper
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_source(
|
||||
content: str,
|
||||
url: str = "https://example.com/1",
|
||||
title: str = "Source 1",
|
||||
domain: str = "example.com",
|
||||
) -> dict:
|
||||
return {
|
||||
"id": str(uuid4()),
|
||||
"url": url,
|
||||
"title": title,
|
||||
"domain": domain,
|
||||
"content": content,
|
||||
}
|
||||
|
||||
|
||||
def _make_analyzer(sources: list[dict], llm_mock: MagicMock) -> SourceIndependenceAnalyzer:
|
||||
config = MagicMock()
|
||||
config.llm.base_url = "http://localhost:8030/openai/v1"
|
||||
config.llm.model = "test-model"
|
||||
config.llm.max_concurrency = 3
|
||||
run_id = uuid4()
|
||||
return SourceIndependenceAnalyzer(
|
||||
llm_provider=llm_mock,
|
||||
config=config,
|
||||
sources=sources,
|
||||
research_run_id=run_id,
|
||||
)
|
||||
|
||||
|
||||
def asyncio_run(coro):
|
||||
loop = asyncio.new_event_loop()
|
||||
try:
|
||||
return loop.run_until_complete(coro)
|
||||
finally:
|
||||
loop.close()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Content-Hash Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestContentHash:
|
||||
"""Content-Hash (SHA-256) Berechnung."""
|
||||
|
||||
def test_identical_content_same_hash(self) -> None:
|
||||
"""Identischer Content → gleicher Hash."""
|
||||
text = "Das ist ein identischer Text."
|
||||
h1 = compute_content_hash(text)
|
||||
h2 = compute_content_hash(text)
|
||||
assert h1 == h2
|
||||
|
||||
def test_different_content_different_hash(self) -> None:
|
||||
"""Verschiedener Content → verschiedener Hash."""
|
||||
h1 = compute_content_hash("Text A")
|
||||
h2 = compute_content_hash("Text B")
|
||||
assert h1 != h2
|
||||
|
||||
def test_whitespace_normalized(self) -> None:
|
||||
"""Whitespace wird normalisiert (leading/trailing stripped, line-endings unified)."""
|
||||
text1 = " Hello\n\nWorld \n\n"
|
||||
text2 = "Hello\n\nWorld"
|
||||
h1 = compute_content_hash(text1)
|
||||
h2 = compute_content_hash(text2)
|
||||
# Beide haben führenden/abschließenden Whitespace + doppelte Zeilenumbrüche
|
||||
# Die Normalisierung sollte zu gleichem Ergebnis führen
|
||||
# Aber hier: text1 hat leading/ trailing, text2 nicht
|
||||
# compute_content_hash: stripped() + replaced()
|
||||
assert h1 == h2
|
||||
|
||||
def test_carriage_return_normalized(self) -> None:
|
||||
"""\r\n und \r werden zu \n normalisiert."""
|
||||
h1 = compute_content_hash("A\r\nB")
|
||||
h2 = compute_content_hash("A\nB")
|
||||
assert h1 == h2
|
||||
|
||||
def test_empty_content(self) -> None:
|
||||
"""Leerer Content ergibt einen Hash."""
|
||||
h = compute_content_hash("")
|
||||
assert isinstance(h, str)
|
||||
assert len(h) == 64 # SHA-256 = 64 hex chars
|
||||
|
||||
def test_hash_is_deterministic(self) -> None:
|
||||
"""Hash ist deterministisch über Aufrufe."""
|
||||
for _ in range(10):
|
||||
assert compute_content_hash("test") == compute_content_hash("test")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Text Similarity Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestTextSimilarity:
|
||||
"""difflib.SequenceMatcher similarity ratio."""
|
||||
|
||||
def test_identical_text_ratio_1(self) -> None:
|
||||
"""Identischer Text → ratio 1.0."""
|
||||
text = "Ganz identischer Text"
|
||||
assert compute_similarity_ratio(text, text) == pytest.approx(1.0)
|
||||
|
||||
def test_empty_texts_ratio_1(self) -> None:
|
||||
"""Beide leer → ratio 1.0."""
|
||||
assert compute_similarity_ratio("", "") == pytest.approx(1.0)
|
||||
|
||||
def test_completely_different_text_ratio_0(self) -> None:
|
||||
"""Komplett unterschiedlicher Text → niedrige ratio."""
|
||||
r = compute_similarity_ratio("Ganz anderes", "Totale Abweichung")
|
||||
assert r < 0.5
|
||||
|
||||
def test_similarity_threshold_high(self) -> None:
|
||||
"""Hohe Similarität (>80%) erkannt."""
|
||||
text_a = "Das ist ein langer Text mit vielen gemeinsamen Worten und fast identischer Struktur."
|
||||
text_b = "Das ist ein langer Text mit vielen gemeinsamen Worten und fast identischer Struktur."
|
||||
r = compute_similarity_ratio(text_a, text_b)
|
||||
assert r == pytest.approx(1.0)
|
||||
|
||||
def test_similarity_truncation_50k(self) -> None:
|
||||
"""Sehr langer Text wird auf 50k Zeichen beschnitten."""
|
||||
long_a = "A " * 60000
|
||||
long_b = "A " * 60000
|
||||
r = compute_similarity_ratio(long_a, long_b)
|
||||
assert r == pytest.approx(1.0)
|
||||
|
||||
def test_similarity_empty_vs_nonempty(self) -> None:
|
||||
"""Leer vs. nicht leer → niedrige ratio."""
|
||||
r = compute_similarity_ratio("", "Nicht leer")
|
||||
assert r < 0.5
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# LLM-Response Parsing Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestLlmParsing:
|
||||
"""LLM-Response Parsing für Syndication-Analyse."""
|
||||
|
||||
def test_pure_json(self) -> None:
|
||||
"""Einfaches JSON ohne Code-Block."""
|
||||
raw = '{"syndicated": true, "syndication_direction": "A->B", "similarity_score": 0.95, "shared_urls": ["http://x.com"], "common_origins": ["agencia.com"], "confidence": 0.8, "reason": "Syndication erkannt"}'
|
||||
result = parse_llm_syndication_response(raw)
|
||||
assert result.syndicated is True
|
||||
assert result.syndication_direction == "A->B"
|
||||
assert result.similarity_score == 0.95
|
||||
assert len(result.shared_urls) == 1
|
||||
assert result.reason == "Syndication erkannt"
|
||||
|
||||
def test_json_in_markdown_block(self) -> None:
|
||||
"""JSON in Markdown-Code-Block."""
|
||||
raw = '```json\n{"syndicated": false, "syndication_direction": "none", "similarity_score": 0.3, "shared_urls": [], "common_origins": [], "confidence": 0.9, "reason": "Unabhängige Quellen"}\n```'
|
||||
result = parse_llm_syndication_response(raw)
|
||||
assert result.syndicated is False
|
||||
assert result.syndication_direction == "none"
|
||||
|
||||
def test_json_in_code_block_without_label(self) -> None:
|
||||
"""JSON in Code-Block ohne language label."""
|
||||
raw = '```\n{"syndicated": true, "similarity_score": 0.85}\n```'
|
||||
result = parse_llm_syndication_response(raw)
|
||||
assert result.syndicated is True
|
||||
assert result.similarity_score == 0.85
|
||||
|
||||
def test_invalid_json_returns_defaults(self) -> None:
|
||||
"""Ungültiges JSON → Default-Werte."""
|
||||
result = parse_llm_syndication_response("not json at all")
|
||||
assert result.syndicated is False
|
||||
assert result.syndication_direction == "none"
|
||||
assert result.similarity_score == 0.0
|
||||
|
||||
def test_partial_json(self) -> None:
|
||||
"""Nur teilweise gefüllte Felder."""
|
||||
raw = '{"syndicated": true, "reason": "Test"}'
|
||||
result = parse_llm_syndication_response(raw)
|
||||
assert result.syndicated is True
|
||||
assert result.similarity_score == 0.0 # default
|
||||
|
||||
def test_none_direction_becomes_none(self) -> None:
|
||||
"""None/missing direction → 'none'."""
|
||||
raw = '{"syndicated": false}'
|
||||
result = parse_llm_syndication_response(raw)
|
||||
assert result.syndication_direction == "none"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Similarity Threshold Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestSimilarityThresholds:
|
||||
"""Schwellenwerte für Similarität."""
|
||||
|
||||
def test_similarities_are_correct(self) -> None:
|
||||
"""Schwellenwerte: HIGH=0.8, MODERATE=0.6, IDENTICAL=1.0."""
|
||||
assert SIMILARITY_HIGH == 0.80
|
||||
assert SIMILARITY_MODERATE == 0.60
|
||||
assert SIMILARITY_IDENTICAL == 1.0
|
||||
|
||||
def test_high_threshold_triggers_llm(self) -> None:
|
||||
"""80% similarity → LLM-Test."""
|
||||
assert SIMILARITY_HIGH >= SIMILARITY_MODERATE
|
||||
|
||||
def test_moderate_threshold_is_above_0(self) -> None:
|
||||
"""Moderate threshold > 0."""
|
||||
assert SIMILARITY_MODERATE > 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SourceIndependenceScore Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestIndependenceScore:
|
||||
"""Unabhängigkeits-Score Berechnung."""
|
||||
|
||||
def test_score_range(self) -> None:
|
||||
"""Score muss zwischen 0.0 und 1.0 liegen."""
|
||||
score = SourceIndependenceScore(
|
||||
source_id=uuid4(),
|
||||
independence_score=0.5,
|
||||
)
|
||||
assert 0.0 <= score.independence_score <= 1.0
|
||||
|
||||
def test_all_fields_present(self) -> None:
|
||||
"""Alle Pflichtfelder vorhanden."""
|
||||
sid = uuid4()
|
||||
score = SourceIndependenceScore(
|
||||
source_id=sid,
|
||||
independence_score=0.9,
|
||||
syndication_group_id=None,
|
||||
primary_source_id=None,
|
||||
content_hash="abc123",
|
||||
text_similarity_high_count=0,
|
||||
shared_urls=[],
|
||||
)
|
||||
assert score.source_id == sid
|
||||
assert score.independence_score == 0.9
|
||||
assert score.content_hash == "abc123"
|
||||
|
||||
def test_max_score(self) -> None:
|
||||
"""Max-Score 1.0."""
|
||||
score = SourceIndependenceScore(
|
||||
source_id=uuid4(),
|
||||
independence_score=1.0,
|
||||
)
|
||||
assert score.independence_score == 1.0
|
||||
|
||||
def test_min_score(self) -> None:
|
||||
"""Min-Score 0.0."""
|
||||
score = SourceIndependenceScore(
|
||||
source_id=uuid4(),
|
||||
independence_score=0.0,
|
||||
)
|
||||
assert score.independence_score == 0.0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Analyzer Integration Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestAnalyzer:
|
||||
"""End-to-End Tests für SourceIndependenceAnalyzer."""
|
||||
|
||||
def test_empty_sources(self) -> None:
|
||||
"""Keine Quellen → leeres Ergebnis."""
|
||||
llm_mock = MagicMock()
|
||||
analyzer = SourceIndependenceAnalyzer(
|
||||
llm_provider=llm_mock,
|
||||
config=MagicMock(),
|
||||
sources=[],
|
||||
research_run_id=uuid4(),
|
||||
)
|
||||
result = asyncio_run(analyzer.analyze())
|
||||
assert result.total_sources == 0
|
||||
assert result.unique_sources == 0
|
||||
assert len(result.citation_edges) == 0
|
||||
assert len(result.source_scores) == 0
|
||||
|
||||
def test_single_source(self) -> None:
|
||||
"""Ein Source → keineEdges, independence_score=1.0."""
|
||||
llm_mock = MagicMock()
|
||||
sources = [_make_source("Ein Source Text")]
|
||||
analyzer = _make_analyzer(sources, llm_mock)
|
||||
result = asyncio_run(analyzer.analyze())
|
||||
assert result.total_sources == 1
|
||||
assert result.unique_sources == 1
|
||||
assert len(result.citation_edges) == 0
|
||||
assert len(result.source_scores) == 1
|
||||
assert result.source_scores[0].independence_score == 1.0
|
||||
|
||||
def test_two_identical_sources(self) -> None:
|
||||
"""Zwei identische Quellen → Syndication."""
|
||||
llm_mock = MagicMock()
|
||||
content = "Ganz identischer Inhalt der zweimal vorkommt."
|
||||
sources = [
|
||||
_make_source(content, url="https://a.com", title="A"),
|
||||
_make_source(content, url="https://b.com", title="B"),
|
||||
]
|
||||
analyzer = _make_analyzer(sources, llm_mock)
|
||||
result = asyncio_run(analyzer.analyze())
|
||||
assert len(result.citation_edges) >= 1 # identical hash → edge
|
||||
# Beide sollten syndicated sein (score < 1.0)
|
||||
for score in result.source_scores:
|
||||
assert score.independence_score <= 1.0
|
||||
|
||||
def test_two_independent_sources(self) -> None:
|
||||
"""Zwei völlig unabhängige Quellen → keine Edge."""
|
||||
llm_mock = MagicMock()
|
||||
sources = [
|
||||
_make_source(
|
||||
"Text A: Der Minister erklärte gestern die neuen Klimaschutzpläne der Bundesregierung.",
|
||||
url="https://politiker-deutschland.de",
|
||||
title="Politiker Deutschland",
|
||||
),
|
||||
_make_source(
|
||||
"Text B: Die Weltmeister der Fußballmannschaft haben das Finale mit 3:1 gewonnen.",
|
||||
url="https://sport-news-magazin.com",
|
||||
title="Sport News",
|
||||
),
|
||||
]
|
||||
analyzer = _make_analyzer(sources, llm_mock)
|
||||
result = asyncio_run(analyzer.analyze())
|
||||
# Keine Syndication-Kanten (sehr unterschiedlicher Inhalt)
|
||||
syndicated_edges = [e for e in result.citation_edges if e.edge_type == CitationEdgeType.SYNDICATED]
|
||||
assert len(syndicated_edges) == 0
|
||||
# Beide Scores sollten hoch sein
|
||||
for score in result.source_scores:
|
||||
assert score.independence_score >= 0.5
|
||||
|
||||
def test_llm_called_on_high_similarity(self) -> None:
|
||||
"""LLM wird aufgerufen wenn Similarität > threshold."""
|
||||
llm_mock = MagicMock()
|
||||
llm_mock.complete = AsyncMock(
|
||||
return_value=json.dumps({
|
||||
"syndicated": False,
|
||||
"similarity_score": 0.7,
|
||||
"reason": "Ähnliche Themen aber unabhängige Quellen",
|
||||
})
|
||||
)
|
||||
|
||||
sources = [
|
||||
_make_source(
|
||||
"Die Bundesregierung hat heute ein neues Klimapaket vorgestellt. Dieses enthält Maßnahmen zur Reduktion von CO2-Emissionen.",
|
||||
url="https://news1.de",
|
||||
title="News 1",
|
||||
),
|
||||
_make_source(
|
||||
"Das neue Klimapaket der Bundesregierung sieht Maßnahmen zur Reduktion von CO2-Emissionen bis 2030 vor.",
|
||||
url="https://news2.de",
|
||||
title="News 2",
|
||||
),
|
||||
]
|
||||
analyzer = _make_analyzer(sources, llm_mock)
|
||||
result = asyncio_run(analyzer.analyze())
|
||||
|
||||
assert llm_mock.complete.called # LLM wurde aufgerufen
|
||||
|
||||
def test_content_hash_in_score(self) -> None:
|
||||
"""SourceScores enthalten Content-Hash."""
|
||||
llm_mock = MagicMock()
|
||||
sources = [_make_source("Test-Inhalt")]
|
||||
analyzer = _make_analyzer(sources, llm_mock)
|
||||
result = asyncio_run(analyzer.analyze())
|
||||
assert len(result.source_scores) == 1
|
||||
score = result.source_scores[0]
|
||||
assert score.content_hash is not None
|
||||
assert len(score.content_hash) == 64
|
||||
|
||||
def test_analysis_result_structure(self) -> None:
|
||||
"""Ergebnis hat alle erwarteten Felder."""
|
||||
llm_mock = MagicMock()
|
||||
analyzer = SourceIndependenceAnalyzer(
|
||||
llm_provider=llm_mock,
|
||||
config=MagicMock(),
|
||||
sources=[],
|
||||
research_run_id=uuid4(),
|
||||
)
|
||||
result = asyncio_run(analyzer.analyze())
|
||||
assert hasattr(result, "research_run_id")
|
||||
assert hasattr(result, "source_scores")
|
||||
assert hasattr(result, "citation_edges")
|
||||
assert hasattr(result, "syndication_groups")
|
||||
assert hasattr(result, "total_sources")
|
||||
assert hasattr(result, "unique_sources")
|
||||
assert hasattr(result, "llm_calls_made")
|
||||
|
||||
def test_multiple_identical_sources_grouped(self) -> None:
|
||||
"""Drei identische Quellen → alle in derselben Gruppe."""
|
||||
llm_mock = MagicMock()
|
||||
content = "Drei gleichartige Quellen mit identischem Inhalt und gleicher Aussage."
|
||||
sources = [
|
||||
_make_source(content, url=f"https://site{i}.com", title=f"Site {i}")
|
||||
for i in range(3)
|
||||
]
|
||||
analyzer = _make_analyzer(sources, llm_mock)
|
||||
result = asyncio_run(analyzer.analyze())
|
||||
|
||||
# Mindestens 3 Kanten (3 choose 2 = 3 Paare)
|
||||
assert len(result.citation_edges) >= 3
|
||||
assert result.total_sources == 3
|
||||
|
||||
def test_no_llm_called_for_identical_hash(self) -> None:
|
||||
"""Identische Hashes → keine LLM-Kalls nötig (sofortige Syndication)."""
|
||||
llm_mock = MagicMock()
|
||||
content = "Exakt derselbe Text."
|
||||
sources = [
|
||||
_make_source(content, url="https://a.com", title="A"),
|
||||
_make_source(content, url="https://b.com", title="B"),
|
||||
]
|
||||
analyzer = _make_analyzer(sources, llm_mock)
|
||||
result = asyncio_run(analyzer.analyze())
|
||||
assert not llm_mock.complete.called # keine LLM-Kalls — identische Hashes
|
||||
|
||||
def test_independence_score_reduced_for_syndicated(self) -> None:
|
||||
"""Syndizierte Quellen erhalten reduzierten Score."""
|
||||
llm_mock = MagicMock()
|
||||
content = "Dies ist ein identischer Text der auf mehreren Seiten kopiert wurde."
|
||||
sources = [
|
||||
_make_source(content, url="https://origin.com", title="Origin"),
|
||||
_make_source(content, url="https://mirror1.com", title="Mirror 1"),
|
||||
_make_source(content, url="https://mirror2.com", title="Mirror 2"),
|
||||
]
|
||||
analyzer = _make_analyzer(sources, llm_mock)
|
||||
result = asyncio_run(analyzer.analyze())
|
||||
|
||||
# Origin (Primärquelle) sollte noch einen hohen Score haben
|
||||
# Mirrors sollten niedriger sein
|
||||
origins = [s for s in result.source_scores if s.content_hash]
|
||||
assert len(origins) == 3
|
||||
for score in origins:
|
||||
assert score.independence_score <= 1.0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CitationEdge Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCitationEdge:
|
||||
"""CitationGraphEdge Tests."""
|
||||
|
||||
def test_edge_type_valid(self) -> None:
|
||||
"""Alle Edge-Types gültig."""
|
||||
for et in CitationEdgeType:
|
||||
assert isinstance(et.value, str)
|
||||
assert len(et.value) > 0
|
||||
|
||||
def test_edge_confidence_range(self) -> None:
|
||||
"""Edge-Confidence muss 0.0–1.0 sein."""
|
||||
from nsct.models.source_independence import CitationGraphEdge
|
||||
from uuid import uuid4
|
||||
|
||||
edge = CitationGraphEdge(
|
||||
source_id=uuid4(),
|
||||
target_source_id=uuid4(),
|
||||
edge_type=CitationEdgeType.SYNDICATED,
|
||||
confidence=0.0,
|
||||
)
|
||||
assert edge.confidence == 0.0
|
||||
edge.confidence = 1.0
|
||||
assert edge.confidence == 1.0
|
||||
|
||||
def test_edge_evidence_default_dict(self) -> None:
|
||||
"""Evidence default ist leer dict."""
|
||||
from nsct.models.source_independence import CitationGraphEdge
|
||||
|
||||
edge = CitationGraphEdge(
|
||||
source_id=uuid4(),
|
||||
target_source_id=uuid4(),
|
||||
edge_type=CitationEdgeType.SIMILAR_CONTENT,
|
||||
confidence=0.5,
|
||||
)
|
||||
assert isinstance(edge.evidence, dict)
|
||||
assert len(edge.evidence) == 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Analysis Result Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestAnalysisResult:
|
||||
"""Quellenanalyse-Ergebnis Tests."""
|
||||
|
||||
def test_empty_result_fields(self) -> None:
|
||||
"""Leeres Ergebnis hat korrekte Defaults."""
|
||||
result = SourceIndependenceAnalysisResult()
|
||||
assert result.total_sources == 0
|
||||
assert result.unique_sources == 0
|
||||
assert result.llm_calls_made == 0
|
||||
assert len(result.source_scores) == 0
|
||||
assert len(result.citation_edges) == 0
|
||||
assert len(result.syndication_groups) == 0
|
||||
|
||||
def test_result_has_created_at(self) -> None:
|
||||
"""Ergebnis hat created_at Zeitstempel."""
|
||||
result = SourceIndependenceAnalysisResult()
|
||||
assert result.created_at is not None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Prompt Template Tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestPromptTemplate:
|
||||
"""LLM-Prompt Template Tests."""
|
||||
|
||||
def test_prompt_template_format(self) -> None:
|
||||
"""Prompt-Template enthält alle Platzhalter."""
|
||||
from nsct.stages.stage6_source_independence import (
|
||||
SYNDICATION_USER_PROMPT_TEMPLATE,
|
||||
)
|
||||
|
||||
filled = SYNDICATION_USER_PROMPT_TEMPLATE.format(
|
||||
url_a="https://a.com", title_a="A", domain_a="a.com", text_a="Test text",
|
||||
url_b="https://b.com", title_b="B", domain_b="b.com", text_b="Other text",
|
||||
)
|
||||
assert "Quelle A" in filled
|
||||
assert "Quelle B" in filled
|
||||
assert "https://a.com" in filled
|
||||
assert "https://b.com" in filled
|
||||
|
||||
def test_prompt_requests_json(self) -> None:
|
||||
"""Prompt fordert JSON-Antwort."""
|
||||
from nsct.stages.stage6_source_independence import (
|
||||
SYNDICATION_USER_PROMPT_TEMPLATE,
|
||||
)
|
||||
filled = SYNDICATION_USER_PROMPT_TEMPLATE.format(
|
||||
url_a="https://a.com", title_a="A", domain_a="a.com", text_a="Test",
|
||||
url_b="https://b.com", title_b="B", domain_b="b.com", text_b="Test",
|
||||
)
|
||||
assert "JSON" in filled
|
||||
assert "syndicated" in filled
|
||||
|
||||
def test_system_prompt_instructions(self) -> None:
|
||||
"""System-Prompt enthält klare Anweisungen."""
|
||||
from nsct.stages.stage6_source_independence import (
|
||||
SYNDICATION_SYSTEM_PROMPT,
|
||||
)
|
||||
assert "JSON" in SYNDICATION_SYSTEM_PROMPT
|
||||
assert "Syndiziert" in SYNDICATION_SYSTEM_PROMPT or "syndicated" in SYNDICATION_SYSTEM_PROMPT.lower()
|
||||
assert "Unabhängigkeit" in SYNDICATION_SYSTEM_PROMPT
|
||||
Reference in New Issue
Block a user