328 lines
10 KiB
Python
328 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import date
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from pipelines.web import auth, main
|
|
from pipelines.web.repository import (
|
|
ChartSeries,
|
|
LegislatorOption,
|
|
RadarSeries,
|
|
RankingResult,
|
|
RankingRow,
|
|
TimePoint,
|
|
)
|
|
|
|
|
|
def test_healthz() -> None:
|
|
client = TestClient(main.app)
|
|
response = client.get("/healthz")
|
|
assert response.status_code == 200
|
|
assert response.text == "ok"
|
|
|
|
|
|
def test_public_home_page_renders() -> None:
|
|
client = TestClient(main.app)
|
|
response = client.get("/")
|
|
assert response.status_code == 200
|
|
assert "Invite-only access" in response.text
|
|
assert "Sign in" in response.text
|
|
|
|
|
|
def test_dashboard_redirects_to_login() -> None:
|
|
client = TestClient(main.app)
|
|
response = client.get("/dashboard?issues=Health", follow_redirects=False)
|
|
assert response.status_code == 303
|
|
assert response.headers["location"].endswith(
|
|
"/login?next=%2Fdashboard%3Fissues%3DHealth"
|
|
)
|
|
|
|
|
|
def test_other_protected_routes_redirect_when_unauthenticated() -> None:
|
|
client = TestClient(main.app)
|
|
for path in ["/legislators", "/compare", "/admin"]:
|
|
response = client.get(path, follow_redirects=False)
|
|
assert response.status_code == 303
|
|
assert response.headers["location"].endswith(f"/login?next={path.replace('/', '%2F', 1)}")
|
|
|
|
|
|
def test_login_redirects_to_workos(monkeypatch) -> None:
|
|
monkeypatch.setattr(main.auth, "get_current_session", lambda request: None)
|
|
monkeypatch.setattr(
|
|
main.auth,
|
|
"build_authorization_url",
|
|
lambda next_path: f"https://auth.example/login?state={next_path}",
|
|
)
|
|
|
|
client = TestClient(main.app)
|
|
response = client.get("/login?next=/compare", follow_redirects=False)
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "https://auth.example/login?state=/compare"
|
|
|
|
|
|
def test_login_redirects_authenticated_user(monkeypatch) -> None:
|
|
monkeypatch.setattr(main.auth, "get_current_session", lambda request: _viewer_session())
|
|
|
|
client = TestClient(main.app)
|
|
response = client.get("/login?next=/compare", follow_redirects=False)
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/compare"
|
|
|
|
|
|
def test_callback_sets_session_cookie(monkeypatch) -> None:
|
|
monkeypatch.setattr(
|
|
main.auth,
|
|
"exchange_code",
|
|
lambda request: auth.CallbackResult(
|
|
sealed_session="sealed-session-value", next_path="/dashboard"
|
|
),
|
|
)
|
|
monkeypatch.setattr(main.auth, "get_auth_config", _fake_auth_config)
|
|
|
|
client = TestClient(main.app)
|
|
response = client.get("/callback?code=abc&state=/dashboard", follow_redirects=False)
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/dashboard"
|
|
assert "workos_session=sealed-session-value" in response.headers["set-cookie"]
|
|
|
|
|
|
def test_callback_failure_redirects_home_and_clears_cookie(monkeypatch) -> None:
|
|
def raise_exchange_error(request):
|
|
raise RuntimeError("bad code")
|
|
|
|
monkeypatch.setattr(main.auth, "exchange_code", raise_exchange_error)
|
|
|
|
client = TestClient(main.app)
|
|
response = client.get("/callback?code=bad", follow_redirects=False)
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "/?auth_error=1"
|
|
assert "workos_session=" in response.headers["set-cookie"]
|
|
|
|
|
|
def test_logout_redirects_to_workos_and_clears_cookie(monkeypatch) -> None:
|
|
monkeypatch.setattr(
|
|
main.auth,
|
|
"get_logout_url",
|
|
lambda request: "https://auth.example/logout",
|
|
)
|
|
|
|
client = TestClient(main.app)
|
|
response = client.post("/logout", follow_redirects=False)
|
|
assert response.status_code == 303
|
|
assert response.headers["location"] == "https://auth.example/logout"
|
|
assert "workos_session=" in response.headers["set-cookie"]
|
|
|
|
|
|
def test_dashboard_route_renders_with_stubbed_repository(monkeypatch) -> None:
|
|
_patch_authenticated_dashboard(monkeypatch, current_user=_viewer_session())
|
|
|
|
client = TestClient(main.app)
|
|
response = client.get("/dashboard?issues=Health&chamber=senate")
|
|
assert response.status_code == 200
|
|
assert "Legislative accountability" in response.text
|
|
assert "Most supportive" in response.text
|
|
assert "viewer@nornsight.test" in response.text
|
|
assert "/admin" not in response.text
|
|
assert '/partials/dashboard?issues=Health&chamber=house' in response.text
|
|
assert "/partials/dashboarddashboard?" not in response.text
|
|
|
|
|
|
def test_admin_route_forbids_viewer(monkeypatch) -> None:
|
|
monkeypatch.setattr(main.auth, "get_current_session", lambda request: _viewer_session())
|
|
|
|
client = TestClient(main.app)
|
|
response = client.get("/admin")
|
|
assert response.status_code == 403
|
|
assert response.json()["detail"] == "Admin access required."
|
|
|
|
|
|
def test_admin_route_renders_for_admin(monkeypatch) -> None:
|
|
monkeypatch.setattr(main.auth, "get_current_session", lambda request: _admin_session())
|
|
monkeypatch.setattr(main.auth, "get_auth_config", _fake_auth_config)
|
|
|
|
client = TestClient(main.app)
|
|
response = client.get("/admin")
|
|
assert response.status_code == 200
|
|
assert "Admin settings" in response.text
|
|
assert "admin@nornsight.test" in response.text
|
|
assert "org_test_123" in response.text
|
|
|
|
|
|
def test_compare_page_renders_for_authenticated_user(monkeypatch) -> None:
|
|
monkeypatch.setattr(main.auth, "get_current_session", lambda request: _viewer_session())
|
|
_patch_compare_page_data(monkeypatch)
|
|
|
|
client = TestClient(main.app)
|
|
response = client.get("/compare")
|
|
assert response.status_code == 200
|
|
assert "Compare legislators" in response.text
|
|
assert "Sanders, B." in response.text
|
|
|
|
|
|
def _viewer_session() -> auth.AuthSession:
|
|
return auth.AuthSession(
|
|
user_id="user_viewer",
|
|
email="viewer@nornsight.test",
|
|
first_name="Viewer",
|
|
last_name="User",
|
|
role_slugs={"viewer"},
|
|
organization_id="org_test_123",
|
|
raw_user=None,
|
|
raw_session=None,
|
|
)
|
|
|
|
|
|
def _admin_session() -> auth.AuthSession:
|
|
return auth.AuthSession(
|
|
user_id="user_admin",
|
|
email="admin@nornsight.test",
|
|
first_name="Admin",
|
|
last_name="User",
|
|
role_slugs={"admin", "viewer"},
|
|
organization_id="org_test_123",
|
|
raw_user=None,
|
|
raw_session=None,
|
|
)
|
|
|
|
|
|
def _fake_auth_config() -> auth.AuthConfig:
|
|
return auth.AuthConfig(
|
|
api_key="sk_test",
|
|
client_id="client_test",
|
|
cookie_password="x" * 32,
|
|
redirect_uri="http://localhost:8000/callback",
|
|
logout_redirect_uri="http://localhost:8000/",
|
|
session_cookie_name="workos_session",
|
|
organization_id="org_test_123",
|
|
)
|
|
|
|
|
|
def _patch_authenticated_dashboard(monkeypatch, *, current_user: auth.AuthSession) -> None:
|
|
monkeypatch.setattr(main.auth, "get_current_session", lambda request: current_user)
|
|
|
|
class DummySession:
|
|
pass
|
|
|
|
class DummyScope:
|
|
def __enter__(self):
|
|
return DummySession()
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
rankings = RankingResult(
|
|
supportive=[
|
|
RankingRow(
|
|
legislator_id=1,
|
|
display_name="Sanders, B.",
|
|
party="I",
|
|
state="VT",
|
|
chamber="senate",
|
|
score=78.0,
|
|
supportive=7,
|
|
opposed=2,
|
|
)
|
|
],
|
|
opposed=[
|
|
RankingRow(
|
|
legislator_id=2,
|
|
display_name="Cruz, T.",
|
|
party="R",
|
|
state="TX",
|
|
chamber="senate",
|
|
score=22.0,
|
|
supportive=2,
|
|
opposed=7,
|
|
)
|
|
],
|
|
)
|
|
history = [
|
|
ChartSeries(
|
|
legislator_id=1,
|
|
label="Sanders, B.",
|
|
party="I",
|
|
state="VT",
|
|
points=[TimePoint(year=2024, score=74.0), TimePoint(year=2025, score=78.0)],
|
|
)
|
|
]
|
|
|
|
monkeypatch.setattr(main, "session_scope", lambda: DummyScope())
|
|
monkeypatch.setattr(main.repository, "latest_congress", lambda session: 119)
|
|
monkeypatch.setattr(main.repository, "has_scores", lambda session: True)
|
|
monkeypatch.setattr(main.repository, "latest_score_year", lambda session: 2026)
|
|
monkeypatch.setattr(
|
|
main.repository, "latest_vote_date", lambda session, congress: date(2026, 1, 15)
|
|
)
|
|
monkeypatch.setattr(
|
|
main.repository,
|
|
"issue_suggestions",
|
|
lambda session, congress=None, limit=12: ["Health", "Taxation"],
|
|
)
|
|
monkeypatch.setattr(
|
|
main.repository,
|
|
"get_rankings",
|
|
lambda session, *, issues, chamber, congress: rankings,
|
|
)
|
|
monkeypatch.setattr(
|
|
main.repository,
|
|
"get_score_history",
|
|
lambda session, *, issues, chamber, congress, legislator_ids: history,
|
|
)
|
|
|
|
|
|
def _patch_compare_page_data(monkeypatch) -> None:
|
|
class DummySession:
|
|
pass
|
|
|
|
class DummyScope:
|
|
def __enter__(self):
|
|
return DummySession()
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
legislator = LegislatorOption(
|
|
legislator_id=1,
|
|
display_name="Sanders, B.",
|
|
party="I",
|
|
state="VT",
|
|
chamber="senate",
|
|
)
|
|
topics = ["Health", "Taxation", "Energy"]
|
|
series = [
|
|
RadarSeries(
|
|
legislator=legislator,
|
|
average_score=77.0,
|
|
scores_by_topic={"Health": 82.0, "Taxation": 71.0, "Energy": 78.0},
|
|
)
|
|
]
|
|
|
|
monkeypatch.setattr(main, "session_scope", lambda: DummyScope())
|
|
monkeypatch.setattr(
|
|
main.repository,
|
|
"get_compare_defaults",
|
|
lambda session: ([1], topics),
|
|
)
|
|
monkeypatch.setattr(
|
|
main.repository,
|
|
"get_legislator_options",
|
|
lambda session, selected_legislators: [legislator],
|
|
)
|
|
monkeypatch.setattr(
|
|
main.repository,
|
|
"get_compare_radar_series",
|
|
lambda session, *, legislator_ids, topics: series,
|
|
)
|
|
monkeypatch.setattr(
|
|
main.repository,
|
|
"search_legislators",
|
|
lambda session, query=None, limit=12: [legislator],
|
|
)
|
|
monkeypatch.setattr(
|
|
main.repository,
|
|
"issue_suggestions",
|
|
lambda session, congress=None, limit=12: topics,
|
|
)
|