o2o-negosium-original/negodata/backend/tests/test_supplier_account.py

"""협력사 채팅(협상) 계정 관리 — supplier.supplier_users 발급/조회/비번재설정/활성상태.

계정 정본은 루트 backend 공유 테이블이므로 negodata 는 협력사(회사 스코프) 소유권을 확인한 뒤
발급(협력사당 1개, 비번 자동생성·1회 반환)/재설정/비활성만 수행한다. 재설정·비활성 시 토큰이 지워져
기존 채팅 로그인이 즉시 끊기는 것까지 확인한다.
"""
import uuid
from datetime import datetime, timedelta

from sqlalchemy import text


async def _create_supplier(client, headers, name="채팅협력사", code=None):
    body = {"name": name, "code": code or f"CHAT-{uuid.uuid4().hex[:8]}", "manager_name": "김담당", "manager_email": "m@x.co"}
    res = (await client.post("/v1/supplier/create", json=body, headers=headers)).json()
    return res["supplier"]["supplier_id"]


async def _seed_token(db_engine, su_id):
    async with db_engine.begin() as conn:
        await conn.execute(
            text(
                "INSERT INTO supplier_user_tokens (sut_id, su_id, type, token, issued_at, expired_at) "
                "VALUES (:sut, :su, 1, '{}'::jsonb, :ia, :ea)"
            ),
            {"sut": uuid.uuid4(), "su": uuid.UUID(su_id), "ia": datetime.utcnow(), "ea": datetime.utcnow() + timedelta(days=1)},
        )


async def _token_count(db_engine, su_id) -> int:
    async with db_engine.begin() as conn:
        row = await conn.execute(text("SELECT count(*) FROM supplier_user_tokens WHERE su_id = :su"), {"su": uuid.UUID(su_id)})
        return int(row.scalar() or 0)


async def _password_hash(db_engine, su_id) -> str:
    async with db_engine.begin() as conn:
        row = await conn.execute(text("SELECT password FROM supplier_users WHERE su_id = :su"), {"su": uuid.UUID(su_id)})
        return row.scalar()


async def test_supplier_account_create_and_get(client, auth_headers):
    """검증: 계정 발급(커스텀 로그인 ID) 후 단건 조회·협력사 상세·목록 응답 확인.
    기대결과: initial_password 1회 반환(XXXX-XXXX), 조회 account.login_id 일치, 상세/목록에 account_login_id 노출."""
    h = await auth_headers("caA")
    sid = await _create_supplier(client, h)
    login_id = f"chat{uuid.uuid4().hex[:8]}"

    created = (await client.post(f"/v1/supplier/{sid}/account/create", json={"login_id": login_id}, headers=h)).json()
    assert created["result"]["success"] is True
    assert created["account"]["login_id"] == login_id
    assert created["account"]["status"] == 1
    pw = created["initial_password"]
    assert len(pw) == 9 and pw[4] == "-"

    got = (await client.get(f"/v1/supplier/{sid}/account", headers=h)).json()
    assert got["account"]["login_id"] == login_id
    assert "initial_password" not in got  # 평문 비번은 발급 응답에서만

    detail = (await client.get(f"/v1/supplier/{sid}", headers=h)).json()
    assert detail["supplier"]["account_login_id"] == login_id
    assert detail["supplier"]["account_status"] == 1
    lst = (await client.get("/v1/supplier/list", headers=h)).json()
    assert {s["supplier_id"]: s.get("account_login_id") for s in lst["suppliers"]}[sid] == login_id


async def test_supplier_account_get_before_create_is_null(client, auth_headers):
    """검증: 발급 전 채팅 계정 단건 조회.
    기대결과: success=True 에 account 없음(미발급은 에러가 아니라 null)."""
    h = await auth_headers("caN")
    sid = await _create_supplier(client, h)
    got = (await client.get(f"/v1/supplier/{sid}/account", headers=h)).json()
    assert got["result"]["success"] is True
    assert got.get("account") is None


async def test_supplier_account_one_per_supplier(client, auth_headers):
    """검증: 이미 계정이 있는 협력사에 재발급 시도.
    기대결과: code=1404(SUPPLIER_ACCOUNT_ALREADY_EXISTS)."""
    h = await auth_headers("caO")
    sid = await _create_supplier(client, h)
    assert (await client.post(f"/v1/supplier/{sid}/account/create", json={"login_id": f"one{uuid.uuid4().hex[:8]}"}, headers=h)).json()["result"]["success"] is True
    dup = (await client.post(f"/v1/supplier/{sid}/account/create", json={"login_id": f"two{uuid.uuid4().hex[:8]}"}, headers=h)).json()
    assert dup["result"]["code"] == 1404


async def test_supplier_account_login_id_duplicate(client, auth_headers):
    """검증: 다른 협력사에서 이미 쓰는 로그인 ID 로 발급 시도.
    기대결과: code=1405(SUPPLIER_ACCOUNT_LOGIN_ID_DUPLICATE) — 로그인 ID 는 전역 유일."""
    h = await auth_headers("caD")
    login_id = f"dup{uuid.uuid4().hex[:8]}"
    s1 = await _create_supplier(client, h, name="협력사1")
    s2 = await _create_supplier(client, h, name="협력사2")
    assert (await client.post(f"/v1/supplier/{s1}/account/create", json={"login_id": login_id}, headers=h)).json()["result"]["success"] is True
    dup = (await client.post(f"/v1/supplier/{s2}/account/create", json={"login_id": login_id}, headers=h)).json()
    assert dup["result"]["code"] == 1405


async def test_supplier_account_reset_password(client, auth_headers, db_engine):
    """검증: 비밀번호 재설정 — 새 평문 1회 반환, DB 해시 교체, 로그인 토큰 삭제.
    기대결과: new_password 반환, password 해시가 발급 때와 달라짐, supplier_user_tokens 0건."""
    h = await auth_headers("caR")
    sid = await _create_supplier(client, h)
    created = (await client.post(f"/v1/supplier/{sid}/account/create", json={"login_id": f"rst{uuid.uuid4().hex[:8]}"}, headers=h)).json()
    su_id = created["account"]["su_id"]
    before = await _password_hash(db_engine, su_id)
    await _seed_token(db_engine, su_id)

    reset = (await client.post(f"/v1/supplier/{sid}/account/reset-password", json={}, headers=h)).json()
    assert reset["result"]["success"] is True
    assert reset["new_password"] and reset["new_password"] != created["initial_password"]
    assert await _password_hash(db_engine, su_id) != before
    assert await _token_count(db_engine, su_id) == 0


async def test_supplier_account_reset_password_custom(client, auth_headers, db_engine):
    """검증: 비밀번호 재설정에 커스텀 값을 지정(password 필드).
    기대결과: new_password 가 지정값 그대로, DB 해시도 교체됨(지정값 미저장·해시만)."""
    h = await auth_headers("caRC")
    sid = await _create_supplier(client, h)
    created = (await client.post(f"/v1/supplier/{sid}/account/create", json={"login_id": f"cst{uuid.uuid4().hex[:8]}"}, headers=h)).json()
    su_id = created["account"]["su_id"]
    before = await _password_hash(db_engine, su_id)

    reset = (await client.post(f"/v1/supplier/{sid}/account/reset-password", json={"password": "Custom-Pw9"}, headers=h)).json()
    assert reset["result"]["success"] is True
    assert reset["new_password"] == "Custom-Pw9"
    after = await _password_hash(db_engine, su_id)
    assert after != before and after != "Custom-Pw9"


async def test_supplier_account_deactivate_and_activate(client, auth_headers, db_engine):
    """검증: 비활성화(토큰 삭제 동반) 후 재활성화.
    기대결과: 비활성 시 status=2·토큰 0건, 활성 복귀 시 status=1."""
    h = await auth_headers("caS")
    sid = await _create_supplier(client, h)
    created = (await client.post(f"/v1/supplier/{sid}/account/create", json={"login_id": f"sts{uuid.uuid4().hex[:8]}"}, headers=h)).json()
    su_id = created["account"]["su_id"]
    await _seed_token(db_engine, su_id)

    off = (await client.patch(f"/v1/supplier/{sid}/account/status", json={"status": 2}, headers=h)).json()
    assert off["account"]["status"] == 2
    assert await _token_count(db_engine, su_id) == 0

    on = (await client.patch(f"/v1/supplier/{sid}/account/status", json={"status": 1}, headers=h)).json()
    assert on["account"]["status"] == 1


async def test_supplier_account_reset_without_account(client, auth_headers):
    """검증: 미발급 협력사에 비밀번호 재설정 시도.
    기대결과: code=1403(SUPPLIER_ACCOUNT_NOT_FOUND)."""
    h = await auth_headers("caX")
    sid = await _create_supplier(client, h)
    res = (await client.post(f"/v1/supplier/{sid}/account/reset-password", json={}, headers=h)).json()
    assert res["result"]["code"] == 1403


async def test_supplier_account_hidden_across_company(client, auth_headers, other_company_id):
    """검증: 회사A 협력사의 채팅 계정을 회사B 유저가 조회/발급 시도.
    기대결과: 둘 다 code=1400(SUPPLIER_NOT_FOUND) — 협력사 자체가 없는 것처럼 막힘."""
    ha = await auth_headers("caCA")
    sid = await _create_supplier(client, ha)
    hb = await auth_headers("caCB", other_company_id)
    assert (await client.get(f"/v1/supplier/{sid}/account", headers=hb)).json()["result"]["code"] == 1400
    assert (await client.post(f"/v1/supplier/{sid}/account/create", json={"login_id": "hack"}, headers=hb)).json()["result"]["code"] == 1400