bmc_hub/tests/test_work_desk.py

import asyncio
from datetime import date, datetime, timedelta
from pathlib import Path
from uuid import uuid4

import pytest
from fastapi import HTTPException
from app.modules.planner.backend import desk, work, calendar
from test_unified_planning import planning_db


def row(**values):
    return {'sag_id': 12, 'status': 'åben', 'work_status': 'ready', 'estimated_minutes': 120,
            'actual_minutes': 30, 'future_planned_minutes': 30, 'planned_slots': [], **values}


def test_daily_selection_keeps_unfinished_work_and_future_appointments():
    day = date(2026, 10, 21)
    now = datetime(2026, 10, 21, 8, tzinfo=work.TZ)
    assert desk.daily_task(row(start_date=day-timedelta(days=10)), day, now)['in_day']
    assert not desk.daily_task(row(start_date=day+timedelta(days=1)), day, now)['in_day']
    assert not desk.daily_task(row(work_status='done'), day, now)['in_day']
    task = desk.daily_task(row(start_date=day, start_time=datetime.strptime('14:00','%H:%M').time()), day, now)
    assert task['in_day'] and not task['ready']


def test_missing_time_is_individual_and_counts_future_bookings_once():
    day = date(2026, 10, 21)
    now = datetime(2026, 10, 21, 8, tzinfo=work.TZ)
    task = desk.daily_task(row(), day, now)
    assert task['remaining_work_minutes'] == 90 and task['missing_minutes'] == 60
    unknown = desk.daily_task(row(estimated_minutes=None, case_estimated_minutes=600), day, now)
    assert unknown['remaining_work_minutes'] is None and unknown['missing_minutes'] is None


def test_today_slots_are_clipped_at_midnight_and_deadline_precedes_ready_backlog():
    day = date(2026, 10, 21)
    now = datetime(2026, 10, 21, 8, tzinfo=work.TZ)
    overnight = desk.daily_task(row(planned_slots=[{'starts_at':'2026-10-20T23:00:00+02:00','ends_at':'2026-10-21T01:00:00+02:00'}]), day, now)
    assert overnight['planned_today_minutes'] == 60
    overdue = desk.daily_task(row(deadline=now-timedelta(hours=1)), day, now)
    assert overdue['_rank'] < desk.daily_task(row(), day, now)['_rank']
    blocked = desk.daily_task(row(work_status='waiting'), day, now)
    assert blocked['blocked_reason'] == 'Afventer'
    too_late = desk.daily_task(row(deadline=now.replace(hour=16), planned_slots=[
        {'starts_at':'2026-10-22T09:00:00+02:00','ends_at':'2026-10-22T10:00:00+02:00'}]), day, now)
    assert too_late['booking_after_deadline']
    assert 'tid reserveret efter deadline' in too_late['priority_reason']


def test_meeting_registration_requires_aware_positive_period_and_unique_people():
    with pytest.raises(ValueError):
        desk.MeetingAttendance(user_id=1,starts_at=datetime(2026,10,20),ends_at=datetime(2026,10,21))
    with pytest.raises(ValueError):
        desk.MeetingAttendance(user_id=1,starts_at=datetime(2026,10,20,tzinfo=work.TZ),ends_at=datetime(2026,10,19,tzinfo=work.TZ))


def test_min_queue_uses_desk_and_can_restore_legacy_view(monkeypatch):
    from starlette.requests import Request
    from app.modules.planner.frontend import views
    monkeypatch.setattr(views.templates,'TemplateResponse',lambda name,context:name)
    monkeypatch.setenv('PLANNING_UI_ENABLED','true')
    monkeypatch.delenv('WORK_DESK_ENABLED',raising=False)
    def view(query):
        return asyncio.run(views.planning(Request({'type':'http','path':'/planning','query_string':query.encode(),'headers':[]}),{'id':1}))
    assert view('view=queue').endswith('/desk.html')
    assert view('view=tasks').endswith('/work.html')
    monkeypatch.setenv('WORK_DESK_ENABLED','false')
    assert view('view=queue').endswith('/work.html')


@pytest.fixture
def desk_db(planning_db):
    from app.core.database import execute_query
    execute_query(Path('migrations/1073_work_desk.sql').read_text(), fetch=False)
    execute_query(Path('migrations/1073_work_desk.sql').read_text(), fetch=False)
    return planning_db


def test_db_desk_order_estimate_contacts_and_seen(desk_db):
    from app.core.database import execute_query, execute_query_single
    actor, users, cases = desk_db
    for cid in cases:
        work.add_assignee(cid, work.AssignmentInput(user_id=actor['id']), actor)
    today = datetime.now(work.TZ).date()
    execute_query('UPDATE sag_sager SET start_date=%s,deadline=%s WHERE id=%s',
                  (today,datetime.now(work.TZ)-timedelta(days=1),cases[1]),fetch=False)
    contact=execute_query_single("INSERT INTO contacts(first_name,last_name,email) VALUES('Desk','Test','desk@example.invalid') RETURNING id")
    execute_query('INSERT INTO sag_kontakter(sag_id,contact_id,is_primary) VALUES(%s,%s,true)',(cases[1],contact['id']),fetch=False)
    initial=desk.desk(user=actor)
    assert initial['items'][0]['sag_id']==cases[1]
    assert initial['items'][0]['contact']['name']=='Desk Test'
    assert all(r['is_new'] for r in initial['items'])
    manual=desk.desk_order(desk.DeskOrder(version=initial['version'],ids=list(reversed(cases))),actor)
    assert [r['sag_id'] for r in desk.desk(user=actor)['items']]==list(reversed(cases))
    with pytest.raises(HTTPException) as stale:
        desk.desk_order(desk.DeskOrder(version=initial['version'],ids=cases),actor)
    assert stale.value.status_code==409
    updated=work.change_queue(actor['id'],cases[0],work.QueueChange(version=manual['version'],estimated_minutes=75),actor)
    assert next(r for r in desk.desk(user=actor)['items'] if r['sag_id']==cases[0])['remaining_work_minutes']==75
    desk.desk_order(desk.DeskOrder(version=updated['version'],mode='recommended'),actor)
    assert desk.desk(user=actor)['items'][0]['sag_id']==cases[1]
    desk.desk_seen(actor)
    assert not any(r['is_new'] for r in desk.desk(user=actor)['items'])


def test_db_meeting_actual_time_replay_and_permission_are_atomic(desk_db):
    from app.core.database import execute_query_single
    actor, users, cases = desk_db
    start=datetime.now(work.TZ)-timedelta(days=1,hours=2)
    end=start+timedelta(hours=1)
    meeting=calendar.create_event(calendar.EventInput(title='Desk meeting',kind='meeting',user_ids=users[:2],starts_at=start,ends_at=end,confirm_conflicts=True),actor)['event']
    attendance=desk.MeetingRegistration(version=meeting['version'],attendees=[desk.MeetingAttendance(user_id=users[0],starts_at=start,ends_at=end),desk.MeetingAttendance(user_id=users[1],starts_at=start+timedelta(minutes=30),ends_at=end)])
    first=desk.register_meeting(meeting['id'],attendance,actor)
    assert desk.register_meeting(meeting['id'],attendance,actor)==first
    count=execute_query_single('SELECT COUNT(*) AS n,SUM(faktisk_tid_min) AS minutes FROM tmodule_times WHERE planner_event_id=%s',(meeting['id'],))
    assert count=={'n':2,'minutes':90}
    changed=attendance.model_copy(deep=True)
    changed.attendees[0].ends_at-=timedelta(minutes=15)
    with pytest.raises(HTTPException) as duplicate:
        desk.register_meeting(meeting['id'],changed,actor)
    assert duplicate.value.status_code==409
    assert execute_query_single('SELECT COUNT(*) AS n FROM tmodule_times WHERE planner_event_id=%s',(meeting['id'],))['n']==2
    outsider={'id':users[2],'permissions':['planner.view','cases.view']}
    with pytest.raises(HTTPException) as denied:
        desk.meeting_attendance(meeting['id'],outsider)
    assert denied.value.status_code==403


def test_db_standalone_timer_has_no_case_and_rejects_second_active_timer(desk_db):
    actor, users, _ = desk_db
    from app.core.database import execute_query_single
    first=asyncio.run(desk.start_desk_timer(desk.DeskTimer(description='Desk standalone'),actor))
    saved=execute_query_single('SELECT sag_id,billable,medarbejder_id,aktiv_timer FROM tmodule_times WHERE id=%s',(first['entry']['id'],))
    assert saved=={'sag_id':None,'billable':False,'medarbejder_id':actor['id'],'aktiv_timer':True}
    with pytest.raises(HTTPException) as duplicate:
        asyncio.run(desk.start_desk_timer(desk.DeskTimer(description='Duplicate'),actor))
    assert duplicate.value.status_code==409


def test_db_meeting_rejects_future_actual_time(desk_db):
    actor, users, _ = desk_db
    start=datetime.now(work.TZ)+timedelta(days=1)
    meeting=calendar.create_event(calendar.EventInput(title='Future meeting',kind='meeting',user_ids=[users[0]],starts_at=start,ends_at=start+timedelta(hours=1),confirm_conflicts=True),actor)['event']
    with pytest.raises(HTTPException) as future:
        desk.register_meeting(meeting['id'],desk.MeetingRegistration(version=meeting['version'],attendees=[desk.MeetingAttendance(user_id=users[0],starts_at=start,ends_at=start+timedelta(hours=1))]),actor)
    assert future.value.status_code==400


def test_db_case_time_retry_does_not_duplicate_entry(desk_db):
    actor, _, cases = desk_db
    from app.core.database import execute_query_single
    payload=desk.DeskTime(request_id=uuid4(),sag_id=cases[0],description='Desk case time',
                         starts_at=datetime.now(work.TZ)-timedelta(minutes=20),minutes=15)
    first=asyncio.run(desk.register_case_time(payload,actor))
    assert asyncio.run(desk.register_case_time(payload,actor))==first
    assert execute_query_single('SELECT COUNT(*) AS n FROM tmodule_times WHERE sag_id=%s AND description=%s',
                                (cases[0],payload.description))['n']==1
    with pytest.raises(HTTPException) as changed:
        asyncio.run(desk.register_case_time(payload.model_copy(update={'minutes':30}),actor))
    assert changed.value.status_code==409


def test_db_today_reservations_do_not_double_count_and_keep_simultaneous_events(desk_db):
    actor, users, cases = desk_db
    start=datetime.now(work.TZ).replace(hour=14,minute=0,second=0,microsecond=0)
    end=start+timedelta(hours=1)
    for _ in range(2):
        calendar.create_event(calendar.EventInput(title='Two reservations',sag_id=cases[0],user_ids=[users[0]],
                              starts_at=start,ends_at=end,confirm_conflicts=True),actor)
    result=desk.desk(user=actor)
    task=next(r for r in result['items'] if r['sag_id']==cases[0])
    assert len(task['today_slots'])==2
    assert task['planned_today_minutes']==120
    before=result['stored_order_ids']
    manual=desk.desk_order(desk.DeskOrder(version=result['version'],ids=list(reversed(before))),actor)
    desk.desk_order(desk.DeskOrder(version=manual['version'],mode='recommended',ids=before),actor)
    assert desk.desk(user=actor)['stored_order_ids']==before