hyungi_document_server/tests/eid/test_eid_append_only_pg.py

"""eid_* append-only 구조강제 + 파생뷰 PG 통합 테스트 (W3, review #1).

설계 불변식 #8 의 load-bearing 부분 = DB 강제. 단일 owner role pkm 이라 REVOKE 무효 +
migration 검증기가 plpgsql BEGIN 거부 → RULE(DO INSTEAD NOTHING) + NOT NULL 스탬프로 강제.
순수함수 테스트(test_compose/test_weakness_compute)로는 검증 불가한 'DB 가 실제로 막는가'를 본다.

★ 실행 환경: Postgres(Docker 스택, migrations 301-305 적용 후) 필요 — MacBook 로컬엔 PG 없어
   hard-fail(skip 아님). test_worker_jobs_smoke.py 와 동일 idiom(_worker_pool_helpers).
   staging(devsbx/개발서버 배포 후)에서 가동. 트랜잭션 rollback 으로 테스트 행 오염 0.
"""

from __future__ import annotations

import os
import sys
from pathlib import Path

import pytest
import pytest_asyncio

sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "app"))
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))  # tests/ (helpers)

from sqlalchemy import text  # noqa: E402
from sqlalchemy.exc import IntegrityError  # noqa: E402
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine  # noqa: E402

from _worker_pool_helpers import ensure_user, get_database_url  # noqa: E402

_VALID_INSERT = (
    "INSERT INTO eid_study_weakness "
    "(user_id, weaknesses, habit_signals, trend_label, actor, source_generated_at) "
    "VALUES (:u, '[]'::jsonb, '{}'::jsonb, '악화', 'eid', now()) RETURNING id"
)


@pytest_asyncio.fixture
async def uid():
    return await ensure_user("test-eid-append-only")


@pytest.mark.asyncio
async def test_unstamped_insert_rejected(uid):
    """actor 스탬프 누락 INSERT → NOT NULL 위반 (owner 도 적용 — 스탬프 없는 행 거부)."""
    engine = create_async_engine(get_database_url())
    sm = async_sessionmaker(engine, expire_on_commit=False)
    try:
        with pytest.raises(IntegrityError):
            async with sm() as s:
                await s.execute(
                    text(
                        "INSERT INTO eid_study_weakness "
                        "(user_id, weaknesses, habit_signals, trend_label, source_generated_at) "
                        "VALUES (:u, '[]'::jsonb, '{}'::jsonb, '정체', now())"  # actor 누락
                    ),
                    {"u": uid},
                )
                await s.commit()
    finally:
        await engine.dispose()


@pytest.mark.asyncio
async def test_update_and_delete_are_no_op(uid):
    """RULE DO INSTEAD NOTHING — owner pkm 의 UPDATE/DELETE 도 행을 못 바꾼다(append-only)."""
    engine = create_async_engine(get_database_url())
    sm = async_sessionmaker(engine, expire_on_commit=False)
    try:
        async with sm() as s:
            wid = (await s.execute(text(_VALID_INSERT), {"u": uid})).scalar_one()
            await s.flush()  # 같은 트랜잭션 내 가시 (commit 안 함 → 끝에 rollback 으로 오염 0)

            await s.execute(
                text("UPDATE eid_study_weakness SET trend_label='개선' WHERE id=:i"), {"i": wid}
            )
            tl = (
                await s.execute(text("SELECT trend_label FROM eid_study_weakness WHERE id=:i"), {"i": wid})
            ).scalar_one()
            assert tl == "악화", "UPDATE 가 값을 바꿈 — RULE 미적용(append-only 깨짐)"

            await s.execute(text("DELETE FROM eid_study_weakness WHERE id=:i"), {"i": wid})
            cnt = (
                await s.execute(text("SELECT count(*) FROM eid_study_weakness WHERE id=:i"), {"i": wid})
            ).scalar_one()
            assert cnt == 1, "DELETE 가 행을 지움 — RULE 미적용(append-only 깨짐)"

            await s.rollback()  # 테스트 행 폐기
    finally:
        await engine.dispose()


@pytest.mark.asyncio
async def test_schedule_views_queryable(uid):
    """v_schedule_today / v_schedule_defer_pattern 정의 유효성 smoke (enum 리터럴·LATERAL·date_trunc).

    뷰가 invalid 면 CREATE 시점 또는 SELECT 시점에 에러 → 쿼리 성공 = DDL 유효.
    """
    engine = create_async_engine(get_database_url())
    sm = async_sessionmaker(engine, expire_on_commit=False)
    try:
        async with sm() as s:
            await s.execute(text("SELECT * FROM v_schedule_today LIMIT 1"))
            await s.execute(text("SELECT * FROM v_schedule_defer_pattern LIMIT 1"))
    finally:
        await engine.dispose()