tests/test_harvest_parser.py
Rama main · réplica de lectura del repositorio original.
"""
Tests for `src.pipeline.harvest_parser`.
Acceptance:
- the extract "Die Quelle bietet keine Datenblätter, sondern nur Marketing"
is classified as polarity="meta" (a statement about the source), not
discarded.
- the extract "Quelle X erwähnt B300-Spezifikationen nicht" is
classified as polarity="negative".
- no phrase list: texts that contain typical "no information" phrases
but carry a [POSITIV] marker are let through as positive.
"""
import unittest
from src.pipeline.harvest_parser import (
count_by_polarity,
meta_only,
parse_harvest_response_with_polarity,
positive_only,
)
class TestHarvestParserPolarity(unittest.TestCase):
def test_basic_positive_extract(self):
response = """[F1] [POSITIV] Fakt: Die DGX B300 verbraucht typisch 14kW
Kontext: Datasheet, Power-Sektion
Verlässlichkeit: hoch"""
extracts = parse_harvest_response_with_polarity(
response, "https://nvidia.com/x", "DGX Datasheet",
)
self.assertEqual(len(extracts), 1)
self.assertEqual(extracts[0].question_id, "F1")
self.assertEqual(extracts[0].polarity, "positive")
self.assertIn("14kW", extracts[0].fact)
self.assertEqual(extracts[0].reliability, "high") # normalised code
def test_negative_extract(self):
"""NEGATIV marker → polarity='negative'."""
response = """[F2] [NEGATIV] Fakt: Quelle X erwähnt B300-Spezifikationen nicht
Kontext: Allgemeine NVIDIA-Übersicht
Verlässlichkeit: hoch"""
extracts = parse_harvest_response_with_polarity(response, "u", "t")
self.assertEqual(len(extracts), 1)
self.assertEqual(extracts[0].polarity, "negative")
def test_meta_extract(self):
"""META marker for statements about the source.
'Die Quelle bietet keine Datenblätter, sondern nur Marketing' —
that is a useful meta finding, not a negative extract.
"""
response = """[F1] [META] Fakt: Die Quelle bietet keine Datenblätter, sondern nur Marketing
Kontext: NVIDIA-Marketing-Seite
Verlässlichkeit: hoch"""
extracts = parse_harvest_response_with_polarity(response, "u", "t")
self.assertEqual(len(extracts), 1)
self.assertEqual(extracts[0].polarity, "meta")
def test_mixed_polarities_in_one_response(self):
"""Realistic case: three polarities in one source."""
response = """[F1] [POSITIV] Fakt: TDP 14kW typisch
Verlässlichkeit: hoch
[F2] [NEGATIV] Fakt: Quelle erwähnt MTBF nicht
Verlässlichkeit: hoch
[F3] [META] Fakt: Die Quelle ist eine Marketing-Folie, kein Datasheet
Verlässlichkeit: hoch"""
extracts = parse_harvest_response_with_polarity(response, "u", "t")
counts = count_by_polarity(extracts)
self.assertEqual(counts, {"positive": 1, "negative": 1, "meta": 1})
def test_missing_marker_defaults_to_positive(self):
"""No marker → POSITIVE (default).
Important so that harvest responses without markers are not
discarded as 'unknown' by mistake.
"""
response = """[F1] Fakt: Etwas ohne Marker
Verlässlichkeit: hoch"""
extracts = parse_harvest_response_with_polarity(response, "u", "t")
self.assertEqual(len(extracts), 1)
self.assertEqual(extracts[0].polarity, "positive")
def test_blacklist_phrase_with_positive_marker_kept(self):
"""A POSITIVE extract may contain phrases like "keine".
A phrase list would discard this extract because it contains the
word 'keine'. With the marker approach it stays — it IS a positive
statement about the source.
"""
response = """[F1] [POSITIV] Fakt: Die offizielle Spec-Tabelle zeigt: keine Unterstützung für PCIe Gen6, dafür Gen5 mit 128 Lanes
Verlässlichkeit: hoch"""
extracts = parse_harvest_response_with_polarity(response, "u", "t")
self.assertEqual(len(extracts), 1)
self.assertEqual(extracts[0].polarity, "positive")
# content kept completely
self.assertIn("keine Unterstützung", extracts[0].fact)
def test_multiline_fact(self):
"""A fact spans several lines → it is concatenated."""
response = """[F1] [POSITIV] Fakt: Die DGX B300 verbraucht 14kW typisch
und 16kW bei voller Last — laut offiziellem Datasheet von NVIDIA.
Kontext: Power-Sektion
Verlässlichkeit: hoch"""
extracts = parse_harvest_response_with_polarity(response, "u", "t")
self.assertEqual(len(extracts), 1)
self.assertIn("14kW", extracts[0].fact)
self.assertIn("16kW", extracts[0].fact)
def test_count_by_polarity(self):
from src.pipeline.models import SourceExtract
extracts = [
SourceExtract("u", "t", "F1", "x", polarity="positive"),
SourceExtract("u", "t", "F1", "x", polarity="positive"),
SourceExtract("u", "t", "F2", "x", polarity="negative"),
SourceExtract("u", "t", "F3", "x", polarity="meta"),
]
self.assertEqual(
count_by_polarity(extracts),
{"positive": 2, "negative": 1, "meta": 1},
)
def test_positive_only_filter(self):
from src.pipeline.models import SourceExtract
extracts = [
SourceExtract("u", "t", "F1", "a", polarity="positive"),
SourceExtract("u", "t", "F1", "b", polarity="negative"),
SourceExtract("u", "t", "F1", "c", polarity="meta"),
]
pos = positive_only(extracts)
self.assertEqual(len(pos), 1)
self.assertEqual(pos[0].fact, "a")
def test_meta_only_filter(self):
from src.pipeline.models import SourceExtract
extracts = [
SourceExtract("u", "t", "F1", "a", polarity="positive"),
SourceExtract("u", "t", "F1", "b", polarity="meta"),
]
m = meta_only(extracts)
self.assertEqual(len(m), 1)
self.assertEqual(m[0].fact, "b")
def test_invalid_marker_treated_as_positive(self):
"""Marker value not in {POSITIV,NEGATIV,META} → treated like no marker."""
response = """[F1] [BLA] Fakt: Etwas
Verlässlichkeit: hoch"""
extracts = parse_harvest_response_with_polarity(response, "u", "t")
# Marker not recognised → the ID line is parsed as an ordinary fact,
# polarity stays at the default
self.assertEqual(len(extracts), 1)
self.assertEqual(extracts[0].polarity, "positive")
if __name__ == "__main__":
unittest.main()
def test_english_markers_and_labels_are_parsed():
"""The prompt asks for English markers and labels; German ones stay accepted."""
from src.pipeline.harvest_parser import parse_harvest_response_with_polarity as parse_harvest_response
response = """[F1] [POSITIVE] Fact: The DGX B300 draws 14 kW
Context: datasheet
Reliability: high
[F2] [NEGATIVE] Fact: The source does not cover pricing
Context: -
Reliability: medium
[F3] [META] Fact: The page is marketing material
Context: -
Reliability: low"""
ex = parse_harvest_response(response, "https://example.org", "Example")
assert [e.polarity for e in ex] == ["positive", "negative", "meta"]
assert ex[0].fact.startswith("The DGX B300") and "[POSITIVE]" not in ex[0].fact
assert ex[0].reliability in ("high", "hoch")