Initial public upload from Projekte/StudyAgent

c72a1e8 verified about 1 month ago

7.24 kB

"""Basis-Tests fuer StudyAgent V1 — kritische Endpunkte."""

import os
import tempfile

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Ensure no API key so we get deterministic fallback steps
os.environ.pop("MINIMAX_API_KEY", None)

from database import Base, get_db  # noqa: E402
from main import app, _utcnow, is_panic_mode, is_overdue, _should_auto_compress  # noqa: E402
from models import Deadline, MicroStep, MicroStepStatus, Subject  # noqa: E402

from datetime import timedelta  # noqa: E402


@pytest.fixture()
def db_session():
    """In-memory SQLite DB per test."""
    tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
    tmp.close()
    engine = create_engine(f"sqlite:///{tmp.name}", connect_args={"check_same_thread": False})
    Base.metadata.create_all(bind=engine)
    TestSession = sessionmaker(bind=engine)
    session = TestSession()
    try:
        yield session
    finally:
        session.close()
        os.unlink(tmp.name)


@pytest.fixture()
def client(db_session):
    """TestClient with overridden DB dependency."""
    def _override():
        try:
            yield db_session
        finally:
            pass
    app.dependency_overrides[get_db] = _override
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()


@pytest.fixture()
def seeded_db(db_session):
    """DB with one subject, two deadlines (overdue + future), and steps."""
    subj = Subject(name="Mathe", semester="3")
    db_session.add(subj)
    db_session.commit()
    db_session.refresh(subj)

    # Overdue deadline
    dl_past = Deadline(
        subject_id=subj.id,
        title="Klausur Lineare Algebra",
        due_date=_utcnow() - timedelta(hours=48),
    )
    # Future deadline (within 48h = panic)
    dl_future = Deadline(
        subject_id=subj.id,
        title="Klausur Analysis",
        due_date=_utcnow() + timedelta(hours=24),
    )
    db_session.add_all([dl_past, dl_future])
    db_session.commit()
    db_session.refresh(dl_past)
    db_session.refresh(dl_future)

    # Steps for both
    for dl in [dl_past, dl_future]:
        for i in range(3):
            db_session.add(MicroStep(
                deadline_id=dl.id,
                title=f"Step {i+1} fuer {dl.title}",
                order_num=i,
                duration_minutes=5,
            ))
    db_session.commit()
    return {"subject": subj, "dl_past": dl_past, "dl_future": dl_future}


# ── Dashboard ──

def test_dashboard_returns_200(client, seeded_db):
    resp = client.get("/")
    assert resp.status_code == 200
    assert "Mathe" in resp.text


def test_dashboard_shows_overdue_badge(client, seeded_db):
    resp = client.get("/")
    assert "Ueberfaellig" in resp.text


# ── WIP-Limit ──

def test_wip_limit_allows_second_subject(client, seeded_db):
    resp = client.post("/subjects", data={"name": "Physik", "semester": "2"})
    assert resp.status_code == 200
    assert "Physik" in resp.text


def test_wip_limit_rejects_third_subject(client, seeded_db):
    client.post("/subjects", data={"name": "Physik", "semester": "2"})
    resp = client.post("/subjects", data={"name": "Chemie", "semester": "1"})
    assert resp.status_code == 200
    assert "Parkiere" in resp.text
    assert "Chemie" not in resp.text


# ── Step Transitions ──

def test_complete_step(client, seeded_db, db_session):
    step = db_session.query(MicroStep).filter(MicroStep.status == MicroStepStatus.pending).first()
    resp = client.post(f"/steps/{step.id}/complete")
    assert resp.status_code == 200
    db_session.refresh(step)
    assert step.status == MicroStepStatus.completed


def test_skip_step(client, seeded_db, db_session):
    step = db_session.query(MicroStep).filter(MicroStep.status == MicroStepStatus.pending).first()
    resp = client.post(f"/steps/{step.id}/skip")
    assert resp.status_code == 200
    db_session.refresh(step)
    assert step.status == MicroStepStatus.skipped
    assert "Kein Stress" in resp.text


# ── Panic Mode / Overdue Detection ──

def test_is_panic_mode_future_close(seeded_db):
    assert is_panic_mode(seeded_db["dl_future"]) is True  # 24h < 48h


def test_is_panic_mode_overdue(seeded_db):
    assert is_panic_mode(seeded_db["dl_past"]) is True  # negative = still panic


def test_is_overdue(seeded_db):
    assert is_overdue(seeded_db["dl_past"]) is True
    assert is_overdue(seeded_db["dl_future"]) is False


# ── Auto-Compress ──

def test_auto_compress_overdue_with_pending_steps(seeded_db):
    """Overdue deadline with pending steps must trigger compression."""
    assert _should_auto_compress(seeded_db["dl_past"], pending_count=3) is True


def test_auto_compress_overdue_no_pending_steps(seeded_db):
    """Overdue deadline with zero pending steps must NOT trigger compression."""
    assert _should_auto_compress(seeded_db["dl_past"], pending_count=0) is False


# ── Plan View ──

def test_plan_view_overdue(client, seeded_db):
    resp = client.get(f"/deadlines/{seeded_db['dl_past'].id}/plan")
    assert resp.status_code == 200
    assert "Ueberfaellig" in resp.text


def test_plan_view_future(client, seeded_db):
    resp = client.get(f"/deadlines/{seeded_db['dl_future'].id}/plan")
    assert resp.status_code == 200
    assert "Endspurt" in resp.text


# ── Replan ──

def test_replan_removes_pending_keeps_completed(client, seeded_db, db_session):
    """Replan deletes pending+skipped steps, keeps completed, adds new ones."""
    dl = seeded_db["dl_future"]

    # Complete one step, skip another — only 1 pending should remain
    steps = db_session.query(MicroStep).filter(
        MicroStep.deadline_id == dl.id,
    ).order_by(MicroStep.order_num).all()
    steps[0].status = MicroStepStatus.completed
    steps[1].status = MicroStepStatus.skipped
    db_session.commit()

    resp = client.post(f"/deadlines/{dl.id}/replan")
    assert resp.status_code == 200
    assert "Plan angepasst" in resp.text

    # Reload steps from DB
    db_session.expire_all()
    remaining = db_session.query(MicroStep).filter(
        MicroStep.deadline_id == dl.id,
    ).order_by(MicroStep.order_num).all()

    # 1 completed survives + new fallback steps (3 panic or 4 normal)
    completed = [s for s in remaining if s.status == MicroStepStatus.completed]
    pending = [s for s in remaining if s.status == MicroStepStatus.pending]
    assert len(completed) == 1
    assert len(pending) >= 3  # fallback generates 3-4 steps

    # New steps order_num starts after completed count
    assert pending[0].order_num == 1  # 1 completed = offset 1


def test_replan_overdue_deadline(client, seeded_db, db_session):
    """Replan works for overdue deadlines (negative hours_until_due)."""
    dl = seeded_db["dl_past"]  # 48h in the past
    resp = client.post(f"/deadlines/{dl.id}/replan")
    assert resp.status_code == 200
    assert "Plan angepasst" in resp.text


def test_replan_nonexistent_deadline(client):
    """Replan returns buddy warning for missing deadline."""
    resp = client.post("/deadlines/99999/replan")
    assert resp.status_code == 200
    assert "nicht gefunden" in resp.text

	"""Basis-Tests fuer StudyAgent V1 — kritische Endpunkte."""

	import os
	import tempfile

	import pytest
	from fastapi.testclient import TestClient
	from sqlalchemy import create_engine
	from sqlalchemy.orm import sessionmaker

	# Ensure no API key so we get deterministic fallback steps
	os.environ.pop("MINIMAX_API_KEY", None)

	from database import Base, get_db # noqa: E402
	from main import app, _utcnow, is_panic_mode, is_overdue, _should_auto_compress # noqa: E402
	from models import Deadline, MicroStep, MicroStepStatus, Subject # noqa: E402

	from datetime import timedelta # noqa: E402


	@pytest.fixture()
	def db_session():
	"""In-memory SQLite DB per test."""
	tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
	tmp.close()
	engine = create_engine(f"sqlite:///{tmp.name}", connect_args={"check_same_thread": False})
	Base.metadata.create_all(bind=engine)
	TestSession = sessionmaker(bind=engine)
	session = TestSession()
	try:
	yield session
	finally:
	session.close()
	os.unlink(tmp.name)


	@pytest.fixture()
	def client(db_session):
	"""TestClient with overridden DB dependency."""
	def _override():
	try:
	yield db_session
	finally:
	pass
	app.dependency_overrides[get_db] = _override
	with TestClient(app) as c:
	yield c
	app.dependency_overrides.clear()


	@pytest.fixture()
	def seeded_db(db_session):
	"""DB with one subject, two deadlines (overdue + future), and steps."""
	subj = Subject(name="Mathe", semester="3")
	db_session.add(subj)
	db_session.commit()
	db_session.refresh(subj)

	# Overdue deadline
	dl_past = Deadline(
	subject_id=subj.id,
	title="Klausur Lineare Algebra",
	due_date=_utcnow() - timedelta(hours=48),
	)
	# Future deadline (within 48h = panic)
	dl_future = Deadline(
	subject_id=subj.id,
	title="Klausur Analysis",
	due_date=_utcnow() + timedelta(hours=24),
	)
	db_session.add_all([dl_past, dl_future])
	db_session.commit()
	db_session.refresh(dl_past)
	db_session.refresh(dl_future)

	# Steps for both
	for dl in [dl_past, dl_future]:
	for i in range(3):
	db_session.add(MicroStep(
	deadline_id=dl.id,
	title=f"Step {i+1} fuer {dl.title}",
	order_num=i,
	duration_minutes=5,
	))
	db_session.commit()
	return {"subject": subj, "dl_past": dl_past, "dl_future": dl_future}


	# ── Dashboard ──

	def test_dashboard_returns_200(client, seeded_db):
	resp = client.get("/")
	assert resp.status_code == 200
	assert "Mathe" in resp.text


	def test_dashboard_shows_overdue_badge(client, seeded_db):
	resp = client.get("/")
	assert "Ueberfaellig" in resp.text


	# ── WIP-Limit ──

	def test_wip_limit_allows_second_subject(client, seeded_db):
	resp = client.post("/subjects", data={"name": "Physik", "semester": "2"})
	assert resp.status_code == 200
	assert "Physik" in resp.text


	def test_wip_limit_rejects_third_subject(client, seeded_db):
	client.post("/subjects", data={"name": "Physik", "semester": "2"})
	resp = client.post("/subjects", data={"name": "Chemie", "semester": "1"})
	assert resp.status_code == 200
	assert "Parkiere" in resp.text
	assert "Chemie" not in resp.text


	# ── Step Transitions ──

	def test_complete_step(client, seeded_db, db_session):
	step = db_session.query(MicroStep).filter(MicroStep.status == MicroStepStatus.pending).first()
	resp = client.post(f"/steps/{step.id}/complete")
	assert resp.status_code == 200
	db_session.refresh(step)
	assert step.status == MicroStepStatus.completed


	def test_skip_step(client, seeded_db, db_session):
	step = db_session.query(MicroStep).filter(MicroStep.status == MicroStepStatus.pending).first()
	resp = client.post(f"/steps/{step.id}/skip")
	assert resp.status_code == 200
	db_session.refresh(step)
	assert step.status == MicroStepStatus.skipped
	assert "Kein Stress" in resp.text


	# ── Panic Mode / Overdue Detection ──

	def test_is_panic_mode_future_close(seeded_db):
	assert is_panic_mode(seeded_db["dl_future"]) is True # 24h < 48h


	def test_is_panic_mode_overdue(seeded_db):
	assert is_panic_mode(seeded_db["dl_past"]) is True # negative = still panic


	def test_is_overdue(seeded_db):
	assert is_overdue(seeded_db["dl_past"]) is True
	assert is_overdue(seeded_db["dl_future"]) is False


	# ── Auto-Compress ──

	def test_auto_compress_overdue_with_pending_steps(seeded_db):
	"""Overdue deadline with pending steps must trigger compression."""
	assert _should_auto_compress(seeded_db["dl_past"], pending_count=3) is True


	def test_auto_compress_overdue_no_pending_steps(seeded_db):
	"""Overdue deadline with zero pending steps must NOT trigger compression."""
	assert _should_auto_compress(seeded_db["dl_past"], pending_count=0) is False


	# ── Plan View ──

	def test_plan_view_overdue(client, seeded_db):
	resp = client.get(f"/deadlines/{seeded_db['dl_past'].id}/plan")
	assert resp.status_code == 200
	assert "Ueberfaellig" in resp.text


	def test_plan_view_future(client, seeded_db):
	resp = client.get(f"/deadlines/{seeded_db['dl_future'].id}/plan")
	assert resp.status_code == 200
	assert "Endspurt" in resp.text


	# ── Replan ──

	def test_replan_removes_pending_keeps_completed(client, seeded_db, db_session):
	"""Replan deletes pending+skipped steps, keeps completed, adds new ones."""
	dl = seeded_db["dl_future"]

	# Complete one step, skip another — only 1 pending should remain
	steps = db_session.query(MicroStep).filter(
	MicroStep.deadline_id == dl.id,
	).order_by(MicroStep.order_num).all()
	steps[0].status = MicroStepStatus.completed
	steps[1].status = MicroStepStatus.skipped
	db_session.commit()

	resp = client.post(f"/deadlines/{dl.id}/replan")
	assert resp.status_code == 200
	assert "Plan angepasst" in resp.text

	# Reload steps from DB
	db_session.expire_all()
	remaining = db_session.query(MicroStep).filter(
	MicroStep.deadline_id == dl.id,
	).order_by(MicroStep.order_num).all()

	# 1 completed survives + new fallback steps (3 panic or 4 normal)
	completed = [s for s in remaining if s.status == MicroStepStatus.completed]
	pending = [s for s in remaining if s.status == MicroStepStatus.pending]
	assert len(completed) == 1
	assert len(pending) >= 3 # fallback generates 3-4 steps

	# New steps order_num starts after completed count
	assert pending[0].order_num == 1 # 1 completed = offset 1


	def test_replan_overdue_deadline(client, seeded_db, db_session):
	"""Replan works for overdue deadlines (negative hours_until_due)."""
	dl = seeded_db["dl_past"] # 48h in the past
	resp = client.post(f"/deadlines/{dl.id}/replan")
	assert resp.status_code == 200
	assert "Plan angepasst" in resp.text


	def test_replan_nonexistent_deadline(client):
	"""Replan returns buddy warning for missing deadline."""
	resp = client.post("/deadlines/99999/replan")
	assert resp.status_code == 200
	assert "nicht gefunden" in resp.text