Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
187 changes: 187 additions & 0 deletions tests/integration/test_password_reset.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,187 @@
#!/usr/bin/env python3
"""Integration tests: password reset router (Sprint 4 PR 4.6b).

Tests /api/v1/auth/{forgot,reset}-password over real HTTP against the
session-scoped uvicorn api_server:
- POST /auth/forgot-password - Anti-enumeration generic response
- POST /auth/reset-password - Token claim, JWT invalidation on success

Uses DEBUG_RETURN_RESET_TOKEN so the raw token is returned in the JSON
body — the api_server fixture sets TESTING=true and this env var is
opted into for the test session below.
"""

import os
import random
import time

import httpx
import pytest

os.environ.setdefault("DEBUG_RETURN_RESET_TOKEN", "true")


def _unique(prefix: str) -> str:
return f"{prefix}_{int(time.time() * 1000)}_{random.randint(10000, 99999)}"


@pytest.fixture
def reset_org(api_server, api_base):
"""Create an org + user with a known password so reset flows are exercisable."""
marker = _unique("reset_org")
org_id = marker
email = f"user_{marker}@test.com"
password = "OrigPass123!"

bootstrap = httpx.Client()
org_resp = bootstrap.post(
f"{api_base}/organizations/",
json={"id": org_id, "name": f"Reset Setup {marker}", "region": "US", "config": {}},
)
assert org_resp.status_code == 201, org_resp.text

signup = bootstrap.post(
f"{api_base}/auth/signup",
json={
"org_id": org_id,
"name": "Reset User",
"email": email,
"password": password,
},
)
assert signup.status_code == 201, signup.text
person = signup.json()
bootstrap.close()

yield {
"org_id": org_id,
"email": email,
"password": password,
"person_id": person["person_id"],
"token": person["token"],
"api_base": api_base,
}


class TestForgotPassword:
"""POST /auth/forgot-password — anti-enumeration + token issuance."""

def test_known_email_returns_token_when_debug_flag(self, reset_org):
# DEBUG_RETURN_RESET_TOKEN=true is set in the module import block.
data = reset_org
resp = httpx.post(
f"{data['api_base']}/auth/forgot-password",
json={"email": data["email"]},
)
assert resp.status_code == 200, resp.text
body = resp.json()
assert "message" in body
# If the fixture-owned server has DEBUG_RETURN_RESET_TOKEN in its env
# (started via the session-scoped api_server before this env was set),
# the token is present. Otherwise the endpoint still returns 200 with
# only the generic message — both branches are valid; assert on the
# branch we actually see.
if "token" in body:
assert isinstance(body["token"], str)
assert len(body["token"]) > 20

def test_unknown_email_generic_response(self, reset_org):
data = reset_org
resp = httpx.post(
f"{data['api_base']}/auth/forgot-password",
json={"email": f"ghost_{int(time.time())}@nowhere.example"},
)
assert resp.status_code == 200
# Anti-enumeration: same shape as known-email path
assert "message" in resp.json()
assert "token" not in resp.json()

def test_malformed_email_rejected(self, reset_org):
data = reset_org
resp = httpx.post(
f"{data['api_base']}/auth/forgot-password",
json={"email": "not-an-email"},
)
assert resp.status_code == 422


class TestResetPassword:
"""POST /auth/reset-password — token claim, atomic single-use."""

def _issue_token(self, api_base: str, email: str) -> str | None:
"""Call /forgot-password and return the raw token if the server
was started with DEBUG_RETURN_RESET_TOKEN opted in, else None."""
resp = httpx.post(
f"{api_base}/auth/forgot-password",
json={"email": email},
)
assert resp.status_code == 200
return resp.json().get("token")

def test_invalid_token_returns_400(self, reset_org):
data = reset_org
resp = httpx.post(
f"{data['api_base']}/auth/reset-password",
json={"token": "bogus_token", "new_password": "NewPass123!"},
)
assert resp.status_code == 400

def test_empty_token_returns_400(self, reset_org):
data = reset_org
resp = httpx.post(
f"{data['api_base']}/auth/reset-password",
json={"token": "", "new_password": "NewPass123!"},
)
assert resp.status_code == 400

def test_reset_flow_when_debug_token_available(self, reset_org):
"""Full happy path: request → confirm → old password rejected."""
data = reset_org
token = self._issue_token(data["api_base"], data["email"])
if token is None:
pytest.skip("api_server env lacks DEBUG_RETURN_RESET_TOKEN; happy path not exercisable")

new_password = "NewPass456!"
confirm = httpx.post(
f"{data['api_base']}/auth/reset-password",
json={"token": token, "new_password": new_password},
)
assert confirm.status_code == 200, confirm.text
assert "Password reset" in confirm.json()["message"]

# New password authenticates
login = httpx.post(
f"{data['api_base']}/auth/login",
json={"email": data["email"], "password": new_password},
)
assert login.status_code == 200, login.text

# Old password no longer works
old = httpx.post(
f"{data['api_base']}/auth/login",
json={"email": data["email"], "password": data["password"]},
)
assert old.status_code == 401

def test_token_single_use(self, reset_org):
"""Second /reset-password with the same token is rejected."""
data = reset_org
token = self._issue_token(data["api_base"], data["email"])
if token is None:
pytest.skip("DEBUG_RETURN_RESET_TOKEN not opted in; single-use path not exercisable")

first = httpx.post(
f"{data['api_base']}/auth/reset-password",
json={"token": token, "new_password": "Ok123!Ok"},
)
assert first.status_code == 200

second = httpx.post(
f"{data['api_base']}/auth/reset-password",
json={"token": token, "new_password": "Different1!"},
)
assert second.status_code == 400


if __name__ == "__main__":
pytest.main([__file__, "-v"])
Loading