"""Tests for reply proposal generation and writer integration.""" from __future__ import annotations import logging from datetime import datetime import pytest from agent_samochodowy.config import Settings from agent_samochodowy.dgred.client import DgredClient from agent_samochodowy.dgred.reply_proposal import ( TEMPLATES, generate_reply_proposal, ) from agent_samochodowy.dgred.writer import LeadWriter from agent_samochodowy.models import MatchResult, Offer, PartQuery, Post # --------------------------------------------------------------------------- # Fixtures / helpers # --------------------------------------------------------------------------- def _post(pid: str = "1", author: str = "Jan Kowalski") -> Post: return Post( post_id=pid, group="TIR części", author_name=author, author_profile_url=f"https://facebook.com/{author.replace(' ', '')}", text="Szukam pompy wody do Volvo FH13, numer 20744939", timestamp=datetime(2025, 1, 15, 10, 30), post_url=f"https://facebook.com/groups/tir/posts/{pid}", ) def _query(**overrides) -> PartQuery: defaults = dict( is_part_request=True, raw_text="Szukam pompy wody do Volvo FH13, numer 20744939", brand="VOLVO", model="FH13", category="pompa wody", oe_numbers=["20744939"], reason="Zapytanie o pompę wody z numerem OE", ) defaults.update(overrides) return PartQuery(**defaults) def _offer(**overrides) -> Offer: defaults = dict( offer_id="1", platform="allegro", url="https://allegro.pl/1", title="Pompa wody VOLVO FH13 20744939", price=450.0, currency="PLN", oe_numbers=["20744939"], ) defaults.update(overrides) return Offer(**defaults) def _matched_result(**overrides) -> MatchResult: defaults = dict( confidence=0.95, method="oe", offers=[_offer()], matched=True, ) defaults.update(overrides) return MatchResult(**defaults) def _settings() -> Settings: return Settings(dgred_api_key="test", dry_run=True) # --------------------------------------------------------------------------- # generate_reply_proposal unit tests # --------------------------------------------------------------------------- class TestGenerateReplyProposal: def test_basic_fields(self): proposal = generate_reply_proposal( _post(), _query(), _matched_result(), template_index=0, ) assert "reply_text" in proposal assert "offer_url" in proposal assert "author_profile" in proposal assert "template_id" in proposal def test_contains_part_name(self): proposal = generate_reply_proposal( _post(), _query(), _matched_result(), template_index=0, ) assert "pompa wody" in proposal["reply_text"] def test_contains_vehicle(self): proposal = generate_reply_proposal( _post(), _query(), _matched_result(), template_index=0, ) assert "VOLVO" in proposal["reply_text"] assert "FH13" in proposal["reply_text"] def test_contains_offer_url(self): proposal = generate_reply_proposal( _post(), _query(), _matched_result(), template_index=0, ) assert "https://allegro.pl/1" in proposal["reply_text"] assert proposal["offer_url"] == "https://allegro.pl/1" def test_contains_price(self): proposal = generate_reply_proposal( _post(), _query(), _matched_result(), template_index=0, ) assert "450 PLN" in proposal["reply_text"] def test_author_profile(self): proposal = generate_reply_proposal( _post(), _query(), _matched_result(), template_index=0, ) assert proposal["author_profile"] == "https://facebook.com/JanKowalski" def test_all_templates_render(self): """Every template variant must render without error.""" for i in range(len(TEMPLATES)): proposal = generate_reply_proposal( _post(), _query(), _matched_result(), template_index=i, ) assert proposal["template_id"] == i assert len(proposal["reply_text"]) > 20 def test_template_index_wraps(self): proposal = generate_reply_proposal( _post(), _query(), _matched_result(), template_index=len(TEMPLATES) + 2, ) assert proposal["template_id"] == 2 def test_random_template_when_none(self): """Without explicit template_index, a random one is selected.""" indices = set() for _ in range(100): proposal = generate_reply_proposal( _post(), _query(), _matched_result(), ) indices.add(proposal["template_id"]) assert len(indices) > 1 # at least 2 different templates def test_missing_price(self): offer = _offer(price=None) result = _matched_result(offers=[offer]) proposal = generate_reply_proposal( _post(), _query(), result, template_index=0, ) assert "do uzgodnienia" in proposal["reply_text"] def test_decimal_price(self): offer = _offer(price=199.99) result = _matched_result(offers=[offer]) proposal = generate_reply_proposal( _post(), _query(), result, template_index=0, ) assert "199.99 PLN" in proposal["reply_text"] def test_missing_category_uses_offer_title(self): query = _query(category=None) proposal = generate_reply_proposal( _post(), query, _matched_result(), template_index=0, ) # Should fall back to offer title snippet assert "Pompa wody VOLVO FH13" in proposal["reply_text"] def test_missing_brand_model(self): query = _query(brand=None, model=None) proposal = generate_reply_proposal( _post(), query, _matched_result(), template_index=0, ) assert "Twojego pojazdu" in proposal["reply_text"] def test_vehicle_with_year(self): query = _query(year="2018") proposal = generate_reply_proposal( _post(), query, _matched_result(), template_index=0, ) assert "2018" in proposal["reply_text"] def test_no_offers_returns_empty_url(self): result = _matched_result(offers=[]) proposal = generate_reply_proposal( _post(), _query(), result, template_index=0, ) assert proposal["offer_url"] == "" # --------------------------------------------------------------------------- # Writer integration: proposal saved only for status 67 # --------------------------------------------------------------------------- class TestWriterProposalIntegration: def test_proposal_saved_for_dopasowano(self, caplog): """Status 67 (Dopasowano) → proposal saved to custom fields.""" settings = _settings() dgred = DgredClient(settings) writer = LeadWriter(dgred, settings) with caplog.at_level(logging.INFO): writer.write_lead(_post(), _query(), _matched_result()) # Should see proposal log assert any("Reply proposal saved" in r.message for r in caplog.records) # Should see PUT with propozycja_odpowiedzi put_calls = [ r for r in caplog.records if "PUT" in r.message and "propozycja_odpowiedzi" in r.message ] assert len(put_calls) == 1 dgred.close() def test_no_proposal_for_review(self, caplog): """Status 68 (Do weryfikacji) → no proposal.""" settings = _settings() dgred = DgredClient(settings) writer = LeadWriter(dgred, settings) result = MatchResult( confidence=0.60, method="atrybuty", offers=[_offer()], matched=True, ) with caplog.at_level(logging.INFO): writer.write_lead(_post(), _query(), result) assert not any("Reply proposal saved" in r.message for r in caplog.records) dgred.close() def test_no_proposal_for_no_match(self, caplog): """Status 69 (Brak dopasowania) → no proposal.""" settings = _settings() dgred = DgredClient(settings) writer = LeadWriter(dgred, settings) result = MatchResult(confidence=0.0, method="brak", matched=False) with caplog.at_level(logging.INFO): writer.write_lead(_post(), _query(), result) assert not any("Reply proposal saved" in r.message for r in caplog.records) dgred.close() def test_proposal_fields_in_put(self, caplog): """Verify the PUT payload contains all three custom fields.""" settings = _settings() dgred = DgredClient(settings) writer = LeadWriter(dgred, settings) with caplog.at_level(logging.INFO): writer.write_lead(_post(), _query(), _matched_result()) put_msgs = [ r.message for r in caplog.records if "PUT" in r.message and "propozycja_odpowiedzi" in r.message ] assert len(put_msgs) == 1 msg = put_msgs[0] assert "link_oferty" in msg assert "link_profilu" in msg dgred.close() def test_no_proposal_when_matched_but_no_offers(self, caplog): """Edge case: confidence high but offers list empty → no proposal.""" settings = _settings() dgred = DgredClient(settings) writer = LeadWriter(dgred, settings) result = MatchResult(confidence=0.95, method="oe", offers=[], matched=True) with caplog.at_level(logging.INFO): writer.write_lead(_post(), _query(), result) assert not any("Reply proposal saved" in r.message for r in caplog.records) dgred.close()