"""Tests for v1 batch scoring, today digest, interview prep, seed demo, and SMTP transport.""" from __future__ import annotations from datetime import datetime, timedelta, timezone import psycopg import pytest from fastapi.testclient import TestClient from app.config import DATABASE_URL from app.db import repo_app, repo_profile from app.transport import ( ClipboardTransport, SmtpTransport, get_transport, is_smtp_configured, reset_transport, set_transport, ) @pytest.fixture() def client(): from app.main import app return TestClient(app) # --- Batch Scoring --- class TestBatchScoring: def test_batch_score_multiple(self, client): """Batch score multiple applications.""" ids = [] for i in range(3): resp = client.post( "/api/postings", json={"url": f"https://example.com/batch/{i}"}, ) ids.append(resp.json()["id"]) resp = client.post( "/api/scoring/batch", json={"application_ids": ids}, ) assert resp.status_code == 200 data = resp.json() assert len(data["results"]) == 3 for r in data["results"]: assert "score" in r assert "red_flags" in r assert isinstance(r["red_flags"], list) def test_batch_score_empty_list(self, client): """Empty application_ids list -> 200 with empty results.""" resp = client.post( "/api/scoring/batch", json={"application_ids": []}, ) assert resp.status_code == 200 assert resp.json()["results"] == [] def test_batch_score_nonexistent_app_skipped(self, client): """Nonexistent application IDs are silently skipped.""" resp = client.post( "/api/scoring/batch", json={"application_ids": ["00000000-0000-0000-0000-000000000000"]}, ) assert resp.status_code == 200 assert resp.json()["results"] == [] def test_batch_score_includes_red_flags(self, client): """Batch score results include red_flags field.""" resp = client.post( "/api/postings", json={"url": "https://example.com/redflag/test"}, ) app_id = resp.json()["id"] resp = client.post( "/api/scoring/batch", json={"application_ids": [app_id]}, ) assert resp.status_code == 200 result = resp.json()["results"][0] assert "red_flags" in result assert isinstance(result["red_flags"], list) def test_batch_score_updates_application_state(self, client): """After batch scoring, application state should be 'scored'.""" resp = client.post( "/api/postings", json={"url": "https://example.com/state/test"}, ) app_id = resp.json()["id"] client.post( "/api/scoring/batch", json={"application_ids": [app_id]}, ) apps = client.get("/api/applications").json() scored = [a for a in apps if a["id"] == app_id] assert len(scored) == 1 assert scored[0]["state"] == "scored" assert scored[0]["score"] is not None # --- Today --- class TestToday: def test_today_empty(self, client): """Today endpoint with no data returns empty digest and zero pending.""" resp = client.get("/api/today") assert resp.status_code == 200 data = resp.json() assert data["digest"] == [] assert data["nudges"] == [] assert data["pending_approvals"] == 0 def test_today_with_scored_applications(self, client): """Today digest includes scored applications ordered by score desc.""" # Create and score two applications for url in ["https://example.com/today/1", "https://example.com/today/2"]: client.post("/api/postings", json={"url": url}) apps = client.get("/api/applications").json() client.post( "/api/scoring/batch", json={"application_ids": [a["id"] for a in apps]}, ) resp = client.get("/api/today") assert resp.status_code == 200 digest = resp.json()["digest"] assert len(digest) == 2 # Should be ordered by score desc assert digest[0]["score"] >= digest[1]["score"] def test_today_nudge_for_backdated_sent(self, client): """A sent application backdated 8 days should appear in nudges.""" # Create posting + application resp = client.post("/api/postings", json={"url": "https://example.com/nudge/1"}) app_id = resp.json()["id"] # Score it, then move through states to sent client.post("/api/scoring/batch", json={"application_ids": [app_id]}) client.post(f"/api/applications/{app_id}/transition", json={"to": "approved"}) client.post(f"/api/applications/{app_id}/transition", json={"to": "drafting"}) # Directly set state to sent (bypassing guard for test) with psycopg.connect(DATABASE_URL) as conn: conn.execute( "UPDATE application SET state = 'sent', last_activity_at = %s WHERE id = %s", (datetime.now(timezone.utc) - timedelta(days=8), app_id), ) conn.commit() resp = client.get("/api/today") assert resp.status_code == 200 nudges = resp.json()["nudges"] assert len(nudges) == 1 assert nudges[0]["application_id"] == app_id assert nudges[0]["days_since_sent"] >= 8 def test_today_nudge_snoozed_excluded(self, client): """A snoozed application should not appear in nudges.""" resp = client.post("/api/postings", json={"url": "https://example.com/nudge/2"}) app_id = resp.json()["id"] # Score and set to sent with backdated activity client.post("/api/scoring/batch", json={"application_ids": [app_id]}) with psycopg.connect(DATABASE_URL) as conn: conn.execute( "UPDATE application SET state = 'sent', last_activity_at = %s WHERE id = %s", (datetime.now(timezone.utc) - timedelta(days=10), app_id), ) conn.commit() # Verify it shows up first resp = client.get("/api/today") assert len(resp.json()["nudges"]) == 1 # Snooze it for a future date future = (datetime.now(timezone.utc) + timedelta(days=7)).date() with psycopg.connect(DATABASE_URL) as conn: conn.execute( "UPDATE application SET follow_up_snoozed_until = %s WHERE id = %s", (future, app_id), ) conn.commit() resp = client.get("/api/today") assert len(resp.json()["nudges"]) == 0 def test_today_pending_approvals_count(self, client): """Today endpoint counts pending (unconfirmed, unexpired) approvals.""" # Create posting + application + artifact resp = client.post("/api/postings", json={"url": "https://example.com/today/3"}) app_id = resp.json()["id"] client.post( f"/api/applications/{app_id}/artifacts", json={"kind": "email", "content": "test content"}, ) artifacts = client.get(f"/api/applications/{app_id}/artifacts").json() artifact_id = artifacts[0]["id"] # Create approval (not confirmed) client.post( f"/api/applications/{app_id}/approvals", json={"action": "send_email", "artifact_id": artifact_id}, ) resp = client.get("/api/today") assert resp.json()["pending_approvals"] == 1 # --- Interview Prep --- class TestInterviewPrep: def test_interview_prep_creates_artifact(self, client): """POST /applications/{id}/interview-prep creates an artifact.""" resp = client.post("/api/postings", json={"url": "https://example.com/iprep/1"}) app_id = resp.json()["id"] resp = client.post(f"/api/applications/{app_id}/interview-prep") assert resp.status_code == 200 data = resp.json() assert "artifact_id" in data assert "content" in data assert len(data["content"]) > 0 assert "Q1" in data["content"] def test_interview_prep_artifact_in_list(self, client): """Interview prep artifact appears in GET artifacts list.""" resp = client.post("/api/postings", json={"url": "https://example.com/iprep/2"}) app_id = resp.json()["id"] resp = client.post(f"/api/applications/{app_id}/interview-prep") artifact_id = resp.json()["artifact_id"] resp = client.get(f"/api/applications/{app_id}/artifacts") artifacts = resp.json() assert any(a["id"] == artifact_id for a in artifacts) def test_interview_prep_sets_artifact_id_on_app(self, client): """interview_prep_artifact_id is set on the application.""" resp = client.post("/api/postings", json={"url": "https://example.com/iprep/3"}) app_id = resp.json()["id"] resp = client.post(f"/api/applications/{app_id}/interview-prep") artifact_id = resp.json()["artifact_id"] apps = client.get("/api/applications").json() app_row = [a for a in apps if a["id"] == app_id][0] assert app_row["interview_prep_artifact_id"] == artifact_id def test_interview_prep_404_nonexistent_app(self, client): """404 for nonexistent application.""" resp = client.post( "/api/applications/00000000-0000-0000-0000-000000000000/interview-prep" ) assert resp.status_code == 404 def test_interview_prep_creates_telemetry(self, client): """Interview prep creates a task_run entry.""" resp = client.post("/api/postings", json={"url": "https://example.com/iprep/4"}) app_id = resp.json()["id"] client.post(f"/api/applications/{app_id}/interview-prep") resp = client.get("/api/telemetry/tasks") tasks = resp.json() assert any(t["task"] == "interview_prep" for t in tasks) # --- Seed Demo --- class TestSeedDemo: def test_seed_demo_creates_data(self, client): """POST /concierge/seed-demo creates profile, postings, applications, and extended data.""" resp = client.post("/api/concierge/seed-demo") assert resp.status_code == 200 data = resp.json() assert data["profile"] == "Demo Demosson" assert data["sections"] == 4 # 6 standalone + 3 agency + 2 deadline + 1 redflag + 1 interviewing = 13 postings assert data["postings"] == 13 # 6 standalone + 3 agency + 2 deadline + 1 redflag + 1 interviewing = 13 applications assert data["applications"] == 13 assert data["clusters"] >= 1 assert data["deadlines"] >= 2 assert data["suggestions"] == 2 assert data["notifications"] == 3 assert data["task_runs"] == 6 assert data["cv_artifacts"] >= 1 def test_seed_demo_idempotent(self, client): """Running seed-demo twice returns the same counts.""" resp1 = client.post("/api/concierge/seed-demo") assert resp1.status_code == 200 resp2 = client.post("/api/concierge/seed-demo") assert resp2.status_code == 200 assert resp2.json()["postings"] == resp1.json()["postings"] assert resp2.json()["applications"] == resp1.json()["applications"] assert resp2.json()["sections"] == resp1.json()["sections"] assert resp2.json()["clusters"] == resp1.json()["clusters"] assert resp2.json()["deadlines"] == resp1.json()["deadlines"] assert resp2.json()["suggestions"] == resp1.json()["suggestions"] assert resp2.json()["notifications"] == resp1.json()["notifications"] assert resp2.json()["task_runs"] == resp1.json()["task_runs"] assert resp2.json()["cv_artifacts"] == resp1.json()["cv_artifacts"] def test_seed_demo_has_nudge_candidate(self, client): """After seeding, /today should show a nudge for the backdated sent app.""" client.post("/api/concierge/seed-demo") resp = client.get("/api/today") nudges = resp.json()["nudges"] assert len(nudges) >= 1 def test_seed_demo_has_scored_digest(self, client): """After seeding, /today digest should have scored applications.""" client.post("/api/concierge/seed-demo") resp = client.get("/api/today") digest = resp.json()["digest"] assert len(digest) >= 1 # --- WS1: Extended seed demo tests (8 new) --- def test_seed_demo_agency_cluster_present(self, client): """Seed creates a 3-posting agency cluster with the same cluster_id.""" client.post("/api/concierge/seed-demo") postings = client.get("/api/postings").json() agency_names = {"Aderanto AB", "Wise IT", "TechTalent Nord"} agency_postings = [p for p in postings if p.get("company") in agency_names] assert len(agency_postings) == 3 cluster_ids = {p["cluster_id"] for p in agency_postings if p.get("cluster_id")} assert len(cluster_ids) == 1, f"Expected 1 cluster_id, got {cluster_ids}" def test_seed_demo_deadlines_populated(self, client): """Seed creates at least 2 postings with apply_by in the next 7 days.""" client.post("/api/concierge/seed-demo") resp = client.get("/api/today") assert resp.status_code == 200 deadlines = resp.json().get("deadlines", []) assert len(deadlines) >= 2 for d in deadlines: assert d["apply_by"] is not None def test_seed_demo_red_flag_rationale(self, client): """Seed creates an application with red_flags in its score_rationale.""" client.post("/api/concierge/seed-demo") apps = client.get("/api/applications").json() red_flag_apps = [ a for a in apps if a.get("score_rationale") and isinstance(a["score_rationale"], dict) and "red_flags" in a["score_rationale"] ] assert len(red_flag_apps) >= 1 red_flags = red_flag_apps[0]["score_rationale"]["red_flags"] assert isinstance(red_flags, list) assert any("unpaid trial" in str(rf).lower() for rf in red_flags) def test_seed_demo_has_interviewing_application(self, client): """Seed creates at least one application in interviewing state.""" client.post("/api/concierge/seed-demo") apps = client.get("/api/applications").json() interviewing = [a for a in apps if a["state"] == "interviewing"] assert len(interviewing) >= 1 def test_seed_demo_has_cover_letter_artifact(self, client): """Seed creates a cover_letter artifact (origin user_drafted) with Swedish text on the approved app.""" client.post("/api/concierge/seed-demo") apps = client.get("/api/applications").json() for a in apps: artifacts = client.get(f"/api/applications/{a['id']}/artifacts").json() for art in artifacts: if art["kind"] == "cover_letter" and art["origin"] == "user_drafted": return assert False, "No user_drafted cover_letter artifact found" def test_seed_demo_suggestions_present(self, client): """Seed creates 2 pending email_suggestion rows with expected classifications.""" client.post("/api/concierge/seed-demo") suggestions = client.get("/api/suggestions").json() assert len(suggestions) == 2 classifications = {s["classification"] for s in suggestions} assert "interview_invite" in classifications assert "question" in classifications # Verify the interview_invite comes from recruiter@festina-demo.se interview_suggestion = [s for s in suggestions if s["classification"] == "interview_invite"][0] assert interview_suggestion["mailbox_from"] == "recruiter@festina-demo.se" def test_seed_demo_notification_log_present(self, client): """Seed creates 3 notification_log rows: 2 delivered, 1 webhook failed.""" client.post("/api/concierge/seed-demo") resp = client.get("/api/notifications/log") assert resp.status_code == 200 logs = resp.json() assert len(logs) == 3 # At least one delivered (daily_digest or email_suggestion) delivered = [l for l in logs if l["delivered"] is True] assert len(delivered) >= 2 # At least one webhook failed with error text failed = [l for l in logs if l["delivered"] is False] assert len(failed) >= 1 assert failed[0]["error"] is not None assert len(failed[0]["error"]) > 0 def test_seed_demo_task_run_telemetry_variance(self, client): """Seed creates 6 task_run rows across multiple providers and models.""" client.post("/api/concierge/seed-demo") resp = client.get("/api/telemetry/tasks") assert resp.status_code == 200 tasks = resp.json() assert len(tasks) == 6 providers = {t["provider"] for t in tasks} models = {t["model"] for t in tasks} assert len(providers) >= 3, f"Expected >= 3 providers, got {providers}" assert len(models) >= 4, f"Expected >= 4 models, got {models}" # Verify cost variance for CostDisplay costs = [t["cost_usd"] for t in tasks if t["cost_usd"] is not None] assert len(costs) >= 2 assert max(costs) > min(costs) # --- SMTP Transport --- class TestSmtpTransportSelection: def test_clipboard_when_no_smtp(self, monkeypatch): """Without SMTP_HOST, transport is ClipboardTransport.""" monkeypatch.delenv("SMTP_HOST", raising=False) reset_transport() t = get_transport() assert isinstance(t, ClipboardTransport) def test_smtp_when_configured(self, monkeypatch): """With SMTP_HOST set, transport is SmtpTransport.""" monkeypatch.setenv("SMTP_HOST", "smtp.example.com") monkeypatch.setenv("SMTP_PORT", "587") monkeypatch.setenv("SMTP_USER", "user@example.com") monkeypatch.setenv("SMTP_PASS", "pass") monkeypatch.setenv("SMTP_FROM", "from@example.com") reset_transport() t = get_transport() assert isinstance(t, SmtpTransport) assert t.host == "smtp.example.com" assert t.port == 587 reset_transport() def test_smtp_ssl_on_465(self, monkeypatch): """SMTP port 465 triggers SSL.""" monkeypatch.setenv("SMTP_HOST", "smtp.example.com") monkeypatch.setenv("SMTP_PORT", "465") reset_transport() t = get_transport() assert isinstance(t, SmtpTransport) assert t.port == 465 reset_transport() def test_clipboard_send_success(self): """ClipboardTransport.send returns success with payload.""" t = ClipboardTransport() payload = {"to": "test@example.com", "subject": "Hi", "body": "Hello"} result = t.send(payload) assert result["success"] is True assert result["transport"] == "clipboard" assert result["payload"] == payload def test_set_transport_override(self): """set_transport overrides the default.""" custom = ClipboardTransport() set_transport(custom) assert get_transport() is custom reset_transport() def test_is_smtp_configured_false(self, monkeypatch): """is_smtp_configured returns False when no SMTP_HOST.""" monkeypatch.delenv("SMTP_HOST", raising=False) assert not is_smtp_configured() def test_is_smtp_configured_true(self, monkeypatch): """is_smtp_configured returns True when SMTP_HOST is set.""" monkeypatch.setenv("SMTP_HOST", "smtp.example.com") assert is_smtp_configured() # --- Scheduler --- class TestScheduler: def test_scheduler_disabled_by_default(self, monkeypatch): """SCHEDULER_ENABLED defaults to false.""" from app.scheduler import is_scheduler_enabled monkeypatch.delenv("SCHEDULER_ENABLED", raising=False) assert not is_scheduler_enabled() def test_scheduler_enabled_when_true(self, monkeypatch): """SCHEDULER_ENABLED=true enables scheduler.""" from app.scheduler import is_scheduler_enabled monkeypatch.setenv("SCHEDULER_ENABLED", "true") assert is_scheduler_enabled() def test_start_scheduler_noop_when_disabled(self, monkeypatch): """start_scheduler does nothing when disabled.""" from app import scheduler monkeypatch.setenv("SCHEDULER_ENABLED", "false") scheduler.start_scheduler() # should not raise scheduler.stop_scheduler()