Ten31-Portal/backend/tests/test_lp_polish.py

"""0.2.32 LP-facing behavior: 8-char password floor, default investor password,
enable-investor-logins conversion, and the documents "New" badge watermark."""

from datetime import datetime, timedelta

from sqlmodel import select

from ten31portal import config
from ten31portal.auth import hash_password, verify_password
from ten31portal.models import (
    Document, Entity, EntityAccess, EntityType, User, UserRole,
)
from tests.conftest import make_user


def _login(client, username, password):
    return client.post("/api/auth/login", json={"login": username, "password": password})


def test_change_password_requires_eight_chars(auth_client):
    resp = auth_client.post(
        "/api/auth/change-password",
        json={"current_password": "password123", "new_password": "short7c"},
    )
    assert resp.status_code == 400
    assert "8 characters" in resp.json()["detail"]

    resp = auth_client.post(
        "/api/auth/change-password",
        json={"current_password": "password123", "new_password": "longenough8"},
    )
    assert resp.status_code == 200


def test_import_created_member_gets_default_password(auth_client, session):
    entity = Entity(name="Test Fund", type=EntityType.fund)
    session.add(entity)
    session.commit()

    resp = auth_client.post(
        "/api/import/capital-accounts/commit",
        json={
            "entity_id": entity.id,
            "as_of_date": "2026-03-31",
            "investors": [{
                "action": "create",
                "name": "New LP",
                "username": "newlp",
                "value_dollars": 100_000,
            }],
        },
    )
    assert resp.status_code == 200, resp.text

    user = session.exec(select(User).where(User.username == "newlp")).one()
    assert user.login_enabled is True
    assert verify_password(config.DEFAULT_INVESTOR_PASSWORD, user.password_hash)

    # And the account can actually sign in with it.
    resp = _login(auth_client, "newlp", config.DEFAULT_INVESTOR_PASSWORD)
    assert resp.status_code == 200, resp.text


def test_enable_investor_logins_converts_only_no_login_accounts(session):
    from ten31portal.cli import enable_investor_logins  # imported here: cli pulls in argparse setup

    no_login = make_user(session, username="dormant", role=UserRole.investor,
                         login_enabled=False, name="Dormant LP")
    has_login = make_user(session, username="active-lp", role=UserRole.investor,
                          password="theirownpw", name="Active LP")
    linked = make_user(session, username="linked", role=UserRole.investor,
                       login_enabled=False, primary_account_id=has_login.id, name="Linked Name")
    old_active_hash = has_login.password_hash

    # Run the conversion against the test engine (CLI normally uses the real one).
    import ten31portal.cli as cli_mod
    orig_engine, orig_migrate = cli_mod.engine, cli_mod.run_migrations
    cli_mod.engine = session.get_bind()
    cli_mod.run_migrations = lambda: None
    try:
        enable_investor_logins(None)
    finally:
        cli_mod.engine, cli_mod.run_migrations = orig_engine, orig_migrate

    session.refresh(no_login)
    session.refresh(has_login)
    session.refresh(linked)
    assert no_login.login_enabled is True
    assert verify_password(config.DEFAULT_INVESTOR_PASSWORD, no_login.password_hash)
    # Working logins and linked secondary names are untouched.
    assert has_login.password_hash == old_active_hash
    assert linked.login_enabled is False


def test_documents_new_badge(client, session):
    entity = Entity(name="Badge Fund", type=EntityType.fund)
    session.add(entity)
    session.commit()
    lp = make_user(session, username="lp", role=UserRole.investor, password="password123")
    session.add(EntityAccess(user_id=lp.id, entity_id=entity.id))

    old_doc = Document(
        entity_id=entity.id, category="statement", title="Old statement",
        original_filename="old.pdf", content_type="application/pdf", size_bytes=1,
        storage_path="x-old", created_at=datetime.utcnow() - timedelta(days=30),
    )
    session.add(old_doc)
    session.commit()

    assert _login(client, "lp", "password123").status_code == 200

    # First ever visit: nothing badged (no watermark yet), watermark gets set.
    docs = client.get("/api/documents").json()
    assert [d["is_new"] for d in docs] == [False]

    # A doc uploaded after that visit is badged next time; pretend the visit was yesterday.
    session.refresh(lp)
    lp.docs_seen_at = datetime.utcnow() - timedelta(days=1)
    session.add(lp)
    new_doc = Document(
        entity_id=entity.id, category="k1", title="Fresh K-1",
        original_filename="k1.pdf", content_type="application/pdf", size_bytes=1,
        storage_path="x-new",
    )
    session.add(new_doc)
    session.commit()

    docs = client.get("/api/documents").json()
    flags = {d["title"]: d["is_new"] for d in docs}
    assert flags == {"Fresh K-1": True, "Old statement": False}

    # Within the same visit (watermark just advanced) the badge computation stays stable.
    docs = client.get("/api/documents").json()
    assert all(d["is_new"] is False for d in docs)  # watermark now newer than both docs