510 lines
No EOL
20 KiB
Python
510 lines
No EOL
20 KiB
Python
"""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() |