tests/test_prospection_cli.py

# tests/test_prospection_cli.py
"""tests/test_prospection_cli.py — export .eml et surface de la CLI."""
from __future__ import annotations

import email
from pathlib import Path

import pytest

from prospection import cli
from prospection.enrich.base import hash_email
from prospection.outreach import compliance, eml
from prospection.store import prospects_db as db


@pytest.fixture(autouse=True)
def tmp_db(tmp_path, monkeypatch):
    monkeypatch.setattr(db, "DB_PATH", tmp_path / "prospects.db")
    monkeypatch.setenv("PROSPECTION_SENDER_NAME", "Vincent Durey")
    monkeypatch.setenv("PROSPECTION_SENDER_EMAIL", "vincent@example.fr")
    monkeypatch.setenv("PROSPECTION_OPTOUT_EMAIL", "stop@example.fr")
    db.init_db()


def test_render_eml_produces_a_parsable_message():
    raw = eml.render_eml("Sujet", "Corps du message", "marie@x.fr")
    parsed = email.message_from_bytes(raw)
    assert parsed["Subject"] == "Sujet"
    assert parsed["To"] == "marie@x.fr"
    assert parsed["From"] == "Vincent Durey <vincent@example.fr>"


def test_render_eml_attaches_the_pdf_as_a_second_part(tmp_path):
    pdf = tmp_path / "brief.pdf"
    pdf.write_bytes(b"%PDF-1.4 fake")
    parsed = email.message_from_bytes(eml.render_eml("S", "B", "m@x.fr", attachment=pdf))
    names = [part.get_filename() for part in parsed.walk()]
    assert "brief.pdf" in names


def test_render_eml_has_no_tracking_pixel_and_no_html_part():
    parsed = email.message_from_bytes(eml.render_eml("S", "B", "m@x.fr"))
    types = [part.get_content_type() for part in parsed.walk()]
    assert "text/html" not in types


def test_cli_rejects_an_unknown_command(capsys):
    with pytest.raises(SystemExit):
        cli.main(["inexistant"])


def test_cli_report_runs_on_an_empty_database(capsys):
    assert cli.main(["report"]) == 0
    assert "Tier 1" in capsys.readouterr().out


def test_cli_optout_add_records_the_hash_and_is_idempotent():
    assert cli.main(["optout", "add", "marie@x.fr"]) == 0
    assert cli.main(["optout", "add", "marie@x.fr"]) == 0
    with db.connect() as conn:
        assert conn.execute("SELECT COUNT(*) c FROM suppression").fetchone()["c"] == 1


def _approved_touch(email, **contact_overrides):
    """Compte + contact + touche approuvée, prêts à l'export."""
    account_id = db.upsert_account({"no_amf": "GP-1", "nom": "B", "statut": "vivant",
                                    "independante": 1})
    row = {"account_id": account_id, "prenom": "Marie", "nom": "Durand",
           "titre": "Gérante", "seniorite": "c-suite", "email": email,
           "email_sha256": hash_email(email) if email else None,
           "email_status": "probable", "telephone": None, "linkedin_url": None,
           "source": "pattern", "confidence": "probable",
           "retrieved_at": "2026-08-29T09:00:00", "is_generic": 0}
    row.update(contact_overrides)
    contact_id = db.upsert_contact(row)
    touch_id = db.upsert_touch({"contact_id": contact_id, "channel": "email",
                                "template_id": "email_1", "subject": "S", "body": "B",
                                "status": "approved"})
    return account_id, contact_id, touch_id


def test_cli_campaign_export_refuses_a_contact_who_opted_out(tmp_path):
    """La dernière barrière avant qu'un fichier quitte la machine. Sans ce test,
    supprimer entièrement l'appel à `assert_sendable` ne ferait échouer aucun
    test de la suite."""
    _approved_touch("marie.durand@x.fr")
    compliance.record_optout("marie.durand@x.fr")
    assert cli.main(["campaign", "export", "--out", str(tmp_path)]) == 0
    assert list(tmp_path.glob("*.eml")) == []


def test_cli_campaign_export_writes_an_eml_for_a_clean_contact(tmp_path):
    """Le pendant du test précédent : sans lui, une barrière qui refuse tout
    passerait pour une barrière qui fonctionne."""
    _approved_touch("marie.durand@x.fr")
    assert cli.main(["campaign", "export", "--out", str(tmp_path)]) == 0
    assert len(list(tmp_path.glob("*.eml"))) == 1


def test_cli_campaign_export_skips_a_contact_with_no_address(tmp_path):
    """Un contact issu de LinkedIn n'a pas d'adresse et n'est pas générique :
    rien en amont ne l'écarte, et `EmailMessage` écrirait « To: None »."""
    _approved_touch(None)
    assert cli.main(["campaign", "export", "--out", str(tmp_path)]) == 0
    assert list(tmp_path.glob("*.eml")) == []


def test_render_eml_refuses_an_empty_recipient():
    with pytest.raises(ValueError):
        eml.render_eml("S", "B", "")


def test_cli_campaign_export_refuses_a_draft_that_is_not_approved(tmp_path):
    account_id = db.upsert_account({"no_amf": "GP-1", "nom": "B", "statut": "vivant",
                                    "independante": 1})
    contact_id = db.upsert_contact({"account_id": account_id, "email": "m@x.fr",
                                    "email_sha256": "d" * 64, "email_status": "probable",
                                    "source": "pattern", "confidence": "probable",
                                    "retrieved_at": "2026-08-29T09:00:00", "is_generic": 0})
    db.upsert_touch({"contact_id": contact_id, "channel": "email", "template_id": "email_1",
                     "subject": "S", "body": "B", "status": "draft"})
    assert cli.main(["campaign", "export", "--out", str(tmp_path)]) == 0
    assert list(tmp_path.glob("*.eml")) == []


def test_contact_rank_puts_the_decision_maker_first():
    manager = {"titre": "Gérante fondatrice", "email_status": "probable"}
    research = {"titre": "Head of Research", "email_status": "verified"}
    compliance_officer = {"titre": "RCCI", "email_status": "verified"}
    ranked = sorted([compliance_officer, research, manager], key=cli._contact_rank)
    assert [c["titre"] for c in ranked] == ["Gérante fondatrice", "Head of Research", "RCCI"]


def test_contact_rank_breaks_ties_on_address_quality():
    """À rang égal, l'adresse vérifiée passe devant la probable : la règle « un
    seul contact » rend ce départage décisif."""
    probable = {"titre": "Head of Research", "email_status": "probable"}
    verified = {"titre": "Head of Research", "email_status": "verified"}
    assert sorted([probable, verified], key=cli._contact_rank)[0] is verified


def _tier1_ready_account(no_amf: str = "GP-1") -> int:
    """Un compte que `score --final` promeut réellement en Tier 1.

    Sans poche actions, sans position confirmée et sans contact nommé, un compte
    finit en Tier 0 quoi qu'il arrive : un test du veto de démo construit sur une
    coquille passerait sans rien vérifier.
    """
    account_id = db.upsert_account({
        "no_amf": no_amf, "nom": "Boutique Gestion SAS", "statut": "vivant",
        "independante": 1, "effectif": 18, "type_activite": "Actions",
        "dirigeants": ["Marie Durand"]})
    holding_id = db.upsert_holding({
        "account_id": account_id, "fund_name": "Fonds Actions Internationales",
        "raw_label": "Microsoft Corp", "ticker": "MSFT", "weight_pct": 5.4,
        "as_of": "2026-07-31", "source_url": "https://x.fr/r.pdf",
        "confidence": "probable", "retrieved_at": "2026-08-29T09:00:00"})
    db.confirm_holding(holding_id, "2026-08-29T09:00:00")
    db.upsert_contact({
        "account_id": account_id, "prenom": "Marie", "nom": "Durand",
        "titre": "Gérante fondatrice", "seniorite": "c-suite",
        "email": "marie.durand@x.fr", "email_sha256": hash_email("marie.durand@x.fr"),
        "email_status": "probable", "source": "pattern", "confidence": "probable",
        "retrieved_at": "2026-08-29T09:00:00", "is_generic": 0})
    return account_id


def test_best_unconfirmed_holding_is_the_heaviest_line_nobody_confirmed():
    """`universe_inferred` signifie exactement qu'aucune position n'est
    confirmée. Sans cet accesseur, `campaign draft` passait `confirmed[0]` —
    donc `None` — et le prompt présentait la chaîne « None » comme un fait."""
    account_id = db.upsert_account({"no_amf": "GP-2", "nom": "B2", "statut": "vivant",
                                    "independante": 1})
    for label, weight in (("Apple Inc", 2.0), ("Microsoft Corp", 7.5)):
        db.upsert_holding({"account_id": account_id, "fund_name": "Fonds A",
                           "raw_label": label, "ticker": None, "weight_pct": weight,
                           "as_of": "2026-07-31", "source_url": "https://x.fr/r.pdf",
                           "confidence": "probable",
                           "retrieved_at": "2026-08-29T09:00:00"})
    heaviest = db.upsert_holding({"account_id": account_id, "fund_name": "Fonds A",
                                  "raw_label": "Nvidia Corp", "ticker": "NVDA",
                                  "weight_pct": 9.9, "as_of": "2026-07-31",
                                  "source_url": "https://x.fr/r.pdf",
                                  "confidence": "probable",
                                  "retrieved_at": "2026-08-29T09:00:00"})
    db.confirm_holding(heaviest, "2026-08-29T09:00:00")
    best = db.best_unconfirmed_holding(account_id)
    assert best["raw_label"] == "Microsoft Corp"


def test_a_rescore_promotes_an_account_with_nothing_against_it():
    """Le témoin du test suivant : sans lui, un veto qui refuse tout passerait
    pour un veto qui fonctionne."""
    account_id = _tier1_ready_account()
    assert cli.main(["score", "--final"]) == 0
    assert db.get_account_by_id(account_id)["tier"] == 1


def test_a_blocked_demo_survives_a_rescore():
    """`assign_final` ne sait rien de la démo : sans persistance, le rescore
    suivant annulerait le constat qu'aucun brief n'est livrable."""
    account_id = _tier1_ready_account()
    db.block_demo(account_id, "2026-08-29T09:00:00")
    assert cli.main(["score", "--final"]) == 0
    assert db.get_account_by_id(account_id)["tier"] == 2


def test_build_demo_blocks_the_account_when_no_brief_ships(tmp_path):
    """La marque doit être posée par la démo elle-même, pas seulement lisible."""
    from prospection.demo import build_demo as demo_module

    account_id = _tier1_ready_account()
    assert demo_module.build_demo(
        account_id, tmp_path,
        brief_runner=lambda ticker: {"status": "PARTIAL", "evidence_coverage": {}},
        pdf_builder=lambda brief, ticker: b"%PDF") is None
    assert db.get_account_by_id(account_id)["demo_blocked_at"]
    assert cli.main(["score", "--final"]) == 0
    assert db.get_account_by_id(account_id)["tier"] == 2