718 lines
44 KiB
Python
718 lines
44 KiB
Python
|
from datetime import date, datetime, time, timedelta
|
||
|
|
from pathlib import Path
|
||
|
|
from uuid import uuid4
|
||
|
|
import os
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from fastapi import HTTPException
|
||
|
|
|
||
|
|
from app.modules.planner.backend import work, calendar, capacity
|
||
|
|
|
||
|
|
|
||
|
|
def test_readiness_respects_waiting_pause_future_start_and_dependencies():
|
||
|
|
now=datetime(2026,10,17,10,tzinfo=work.TZ)
|
||
|
|
assert work.ready({'status':'åben','work_status':'ready'},now)
|
||
|
|
for extra in ({'status':'afventer'},{'work_status':'paused'},
|
||
|
|
{'start_date':date(2026,10,18)}, {'snoozed_until':now+timedelta(hours=1)},
|
||
|
|
{'deferred_until':datetime(2026,10,17,11)},
|
||
|
|
{'deferred_until_case_id':12,'deferred_until_status':'afsluttet','deferred_case_status':'åben'}):
|
||
|
|
assert not work.ready({'status':'åben','work_status':'ready',**extra},now)
|
||
|
|
assert work.ready({'deferred_until_case_id':12,'deferred_until_status':'afsluttet, lukket','deferred_case_status':'lukket'},now)
|
||
|
|
|
||
|
|
|
||
|
|
def test_remaining_estimate_does_not_duplicate_unallocated_case_estimate():
|
||
|
|
assert work.remaining({'case_estimated_minutes':720,'actual_minutes':60}) is None
|
||
|
|
assert work.remaining({'estimated_minutes':120,'actual_minutes':90})==30
|
||
|
|
assert work.remaining({'estimated_minutes':120,'actual_minutes':180})==0
|
||
|
|
assert work.remaining({'remaining_minutes':45,'estimated_minutes':120,'actual_minutes':180})==45
|
||
|
|
|
||
|
|
|
||
|
|
def test_capacity_unions_busy_intervals_and_ignores_nights():
|
||
|
|
start=datetime(2026,10,19,tzinfo=work.TZ)
|
||
|
|
schedules=[{'weekday':0,'start_time':time(8),'end_time':time(16),'break_minutes':0}]
|
||
|
|
occupied=[(start.replace(hour=9),start.replace(hour=12)),(start.replace(hour=10),start.replace(hour=13)),
|
||
|
|
(start.replace(hour=20),start.replace(hour=22))]
|
||
|
|
assert capacity.working_capacity(schedules,occupied,start,start+timedelta(days=1))=={
|
||
|
|
'normal_minutes':480,'busy_minutes':240,'free_minutes':240}
|
||
|
|
|
||
|
|
|
||
|
|
def test_absence_only_registers_scheduled_days_and_partial_hours():
|
||
|
|
schedules=[{'weekday':d,'start_time':time(8),'end_time':time(16),'break_minutes':30} for d in range(5)]
|
||
|
|
period={'starts_at':datetime(2026,10,16,12,tzinfo=work.TZ),'ends_at':datetime(2026,10,19,12,tzinfo=work.TZ)}
|
||
|
|
days=calendar.absence_work_days(period,schedules)
|
||
|
|
assert [r[0] for r in days]==[date(2026,10,16),date(2026,10,19)]
|
||
|
|
assert [r[3] for r in days]==[240,240]
|
||
|
|
assert calendar.absence_work_minutes({'starts_at':datetime(2026,10,19,8,tzinfo=work.TZ),
|
||
|
|
'ends_at':datetime(2026,10,19,16,tzinfo=work.TZ)},schedules)==450
|
||
|
|
|
||
|
|
|
||
|
|
def test_calendar_rejects_naive_or_reversed_dates():
|
||
|
|
with pytest.raises(ValueError):
|
||
|
|
calendar.EventInput(title='Test',user_ids=[1],starts_at=datetime(2026,10,17),ends_at=datetime(2026,10,18))
|
||
|
|
with pytest.raises(ValueError):
|
||
|
|
calendar.AbsenceInput(user_id=1,kind='vacation',starts_at=datetime(2026,10,18,tzinfo=work.TZ),ends_at=datetime(2026,10,17,tzinfo=work.TZ))
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def planning_db():
|
||
|
|
if os.environ.get('BMC_PLANNING_DB_TESTS')!='1':
|
||
|
|
pytest.skip('Opt in to a dedicated local planning test database')
|
||
|
|
import psycopg2
|
||
|
|
from psycopg2.extensions import parse_dsn, make_dsn
|
||
|
|
from psycopg2.pool import ThreadedConnectionPool
|
||
|
|
from app.core import database
|
||
|
|
from app.core.config import settings
|
||
|
|
target=os.environ.get('BMC_PLANNING_TEST_DATABASE','')
|
||
|
|
if not target.startswith('bmc_planning_test_'):
|
||
|
|
pytest.fail('A dedicated bmc_planning_test_* database is required')
|
||
|
|
values=parse_dsn(settings.DATABASE_URL)
|
||
|
|
values.update(host='127.0.0.1',port='5433',dbname=target)
|
||
|
|
original=database.connection_pool
|
||
|
|
pool=ThreadedConnectionPool(1,6,dsn=make_dsn(**values))
|
||
|
|
database.connection_pool=pool
|
||
|
|
try:
|
||
|
|
from app.core.database import execute_query, execute_query_single, atomic_transaction
|
||
|
|
with atomic_transaction():
|
||
|
|
# Only the dedicated disposable database may have test schema adjusted.
|
||
|
|
execute_query('DROP INDEX IF EXISTS planner_actual_absence',fetch=False)
|
||
|
|
execute_query(Path('migrations/1070_unified_planning.sql').read_text(),fetch=False)
|
||
|
|
execute_query(Path('migrations/1071_case_confidentiality.sql').read_text(),fetch=False)
|
||
|
|
execute_query(Path('migrations/1072_absence_reports.sql').read_text(),fetch=False)
|
||
|
|
suffix=uuid4().hex[:10]
|
||
|
|
users=[]
|
||
|
|
for i in range(3):
|
||
|
|
users.append(execute_query_single("INSERT INTO users(username,email,password_hash,is_active) VALUES(%s,%s,'disabled-test-account',true) RETURNING user_id",(f'plan_{suffix}_{i}',f'plan_{suffix}_{i}@example.invalid'))['user_id'])
|
||
|
|
actor={'id':users[0],'is_superadmin':True,'permissions':[]}
|
||
|
|
cases=[]
|
||
|
|
for i in range(3):
|
||
|
|
cases.append(execute_query_single("INSERT INTO sag_sager(titel,created_by_user_id,status,type) VALUES(%s,%s,'åben','ticket') RETURNING id",(f'Planning test {suffix} {i}',users[0]))['id'])
|
||
|
|
yield actor,users,cases
|
||
|
|
finally:
|
||
|
|
database.connection_pool=original
|
||
|
|
pool.closeall()
|
||
|
|
|
||
|
|
|
||
|
|
def test_db_independent_queues_stale_versions_and_transaction_rollback(planning_db):
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
from app.core.database import execute_query_single
|
||
|
|
for uid in users:
|
||
|
|
for cid in cases:
|
||
|
|
work.add_assignee(cid,work.AssignmentInput(user_id=uid),actor)
|
||
|
|
a=work.get_queue(users[0],100,0,actor)
|
||
|
|
b=work.get_queue(users[1],100,0,actor)
|
||
|
|
work.change_queue(users[0],cases[2],work.QueueChange(version=a['version'],before_sag_id=cases[0]),actor)
|
||
|
|
assert [r['sag_id'] for r in work.get_queue(users[0],100,0,actor)['items']]==[cases[2],cases[0],cases[1]]
|
||
|
|
assert [r['sag_id'] for r in work.get_queue(users[1],100,0,actor)['items']]==[r['sag_id'] for r in b['items']]
|
||
|
|
with pytest.raises(HTTPException) as error:
|
||
|
|
work.change_queue(users[0],cases[0],work.QueueChange(version=a['version'],pinned=True),actor)
|
||
|
|
assert error.value.status_code==409
|
||
|
|
assert execute_query_single('SELECT pinned FROM sag_case_assignees WHERE sag_id=%s AND user_id=%s',(cases[0],users[0]))['pinned'] is False
|
||
|
|
|
||
|
|
|
||
|
|
def test_db_shared_event_preserves_participant_ids_and_registers_absence_once(planning_db):
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
from app.core.database import execute_query,execute_query_single
|
||
|
|
start=datetime(2026,10,19,10,tzinfo=work.TZ)
|
||
|
|
payload=calendar.EventInput(title='Shared',sag_id=cases[0],user_ids=users,starts_at=start,ends_at=start+timedelta(hours=1))
|
||
|
|
event=calendar.create_event(payload,actor)['event']
|
||
|
|
before=execute_query('SELECT id FROM planner_allocations WHERE event_id=%s ORDER BY id',(event['id'],))
|
||
|
|
moved=payload.model_copy(update={'version':event['version'],'starts_at':start+timedelta(hours=2),'ends_at':start+timedelta(hours=3)})
|
||
|
|
event=calendar.update_event(event['id'],moved,actor)['event']
|
||
|
|
visible=calendar.schedule(start,start+timedelta(days=1),[users[0]],actor)
|
||
|
|
assert visible['bookings'][0]['participants']==sorted(users)
|
||
|
|
after=execute_query('SELECT id,starts_at FROM planner_allocations WHERE event_id=%s ORDER BY id',(event['id'],))
|
||
|
|
assert [r['id'] for r in before]==[r['id'] for r in after]
|
||
|
|
assert all(r['starts_at']==moved.starts_at for r in after)
|
||
|
|
assert len(work.get_queue(users[1],100,0,actor)['items'])==1
|
||
|
|
calendar.cancel_event(event['id'],event['version'],actor)
|
||
|
|
assert not execute_query('SELECT id FROM planner_allocations WHERE event_id=%s AND deleted_at IS NULL',(event['id'],))
|
||
|
|
for weekday in (0,1):
|
||
|
|
execute_query('INSERT INTO planner_work_schedules(user_id,weekday,break_minutes) VALUES(%s,%s,0)',(users[0],weekday),fetch=False)
|
||
|
|
absence=calendar.create_absence(calendar.AbsenceInput(user_id=users[0],kind='vacation',starts_at=datetime(2026,10,12,8,tzinfo=work.TZ),ends_at=datetime(2026,10,13,16,tzinfo=work.TZ)),actor)['absence']
|
||
|
|
one=calendar.register_absence(absence['id'],absence['version'],actor)
|
||
|
|
two=calendar.register_absence(absence['id'],absence['version'],actor)
|
||
|
|
assert one['time_entry_ids']==two['time_entry_ids']
|
||
|
|
assert len(one['time_entry_ids'])==2
|
||
|
|
assert two['already_registered']
|
||
|
|
rows=execute_query('SELECT billable,fakturerbar_tid_min FROM tmodule_times WHERE planner_absence_id=%s',(absence['id'],))
|
||
|
|
assert all(not r['billable'] and r['fakturerbar_tid_min']==0 for r in rows)
|
||
|
|
|
||
|
|
|
||
|
|
def test_db_confidentiality_hides_queue_calendar_api_and_allows_group(planning_db):
|
||
|
|
import asyncio
|
||
|
|
from fastapi import FastAPI
|
||
|
|
from fastapi.testclient import TestClient
|
||
|
|
from app.core import case_visibility as policy
|
||
|
|
from app.modules.sag.backend import visibility
|
||
|
|
from app.core.database import execute_query,execute_query_single
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
reader={'id':users[1],'permissions':['planner.view','planner.edit_own','cases.view']}
|
||
|
|
work.add_assignee(cases[0],work.AssignmentInput(user_id=users[1]),actor)
|
||
|
|
start=datetime(2026,11,2,10,tzinfo=work.TZ)
|
||
|
|
calendar.create_event(calendar.EventInput(title='Confidential title',description='Private note',sag_id=cases[0],user_ids=[users[1]],starts_at=start,ends_at=start+timedelta(hours=1)),actor)
|
||
|
|
locked=visibility.update_visibility(cases[0],visibility.VisibilityInput(restricted=True,version=0,user_ids=[users[0]]),actor)
|
||
|
|
assert cases[0] in policy.hidden_case_ids(reader)
|
||
|
|
assert not work.get_queue(users[1],100,0,reader)['items']
|
||
|
|
busy=calendar.schedule(start,start+timedelta(days=1),[users[1]],reader)['bookings'][0]
|
||
|
|
assert busy['title']=='Optaget' and busy['private']
|
||
|
|
assert not {'sag_id','description','event_id'} & busy.keys()
|
||
|
|
with pytest.raises(HTTPException) as error:
|
||
|
|
visibility.get_visibility(cases[0],reader)
|
||
|
|
assert error.value.status_code==404
|
||
|
|
app=FastAPI()
|
||
|
|
@app.middleware('http')
|
||
|
|
async def privacy(request,call_next):
|
||
|
|
return await policy.protect_request(request,call_next,reader)
|
||
|
|
@app.get('/api/v1/search')
|
||
|
|
def search():
|
||
|
|
return {'items':[{'id':cases[0],'title':'Confidential title','url':f'/sag/{cases[0]}'}]}
|
||
|
|
@app.get('/sag/{sag_id}')
|
||
|
|
def details(sag_id:int):
|
||
|
|
return {'titel':'Confidential title'}
|
||
|
|
calls=[]
|
||
|
|
@app.post('/api/v1/test-write')
|
||
|
|
def write():
|
||
|
|
calls.append(True)
|
||
|
|
return {'ok':True}
|
||
|
|
with TestClient(app) as client:
|
||
|
|
assert client.get('/api/v1/search').json()=={'items':[]}
|
||
|
|
assert client.get(f'/sag/{cases[0]}').status_code==404
|
||
|
|
assert client.post('/api/v1/test-write',json={'sag_id':cases[0]}).status_code==404
|
||
|
|
assert not calls
|
||
|
|
group=execute_query_single('INSERT INTO groups(name) VALUES(%s) RETURNING id',(f'privacy_{uuid4().hex}',))['id']
|
||
|
|
execute_query('INSERT INTO user_groups(user_id,group_id) VALUES(%s,%s)',(users[1],group),fetch=False)
|
||
|
|
updated=visibility.update_visibility(cases[0],visibility.VisibilityInput(restricted=True,version=locked['version'],group_ids=[group]),actor)
|
||
|
|
assert cases[0] not in policy.hidden_case_ids(reader)
|
||
|
|
assert work.get_queue(users[1],100,0,reader)['items'][0]['sag_id']==cases[0]
|
||
|
|
# Unlocked/relocked cases keep a monotonic version (no stale version reuse).
|
||
|
|
unlocked=visibility.update_visibility(cases[0],visibility.VisibilityInput(restricted=False,version=updated['version']),actor)
|
||
|
|
assert unlocked['version']>updated['version']
|
||
|
|
|
||
|
|
|
||
|
|
def test_private_calendar_and_nested_notifications_are_scrubbed():
|
||
|
|
from app.core.case_visibility import filter_data
|
||
|
|
data={'sections':{'urgent':{'list':[{'id':174,'title':'Secret'}]}},
|
||
|
|
'notifications':[{'sag_id':174,'case_title':'Secret','message':'Private'}],
|
||
|
|
'bookings':[{'sag_id':174,'id':9,'case_title':'Secret','customer_name':'Secret customer',
|
||
|
|
'starts_at':'2026-10-17T10:00:00Z','ends_at':'2026-10-17T11:00:00Z'}]}
|
||
|
|
result=filter_data(data,{174})
|
||
|
|
assert result['sections']['urgent']['list']==[]
|
||
|
|
assert result['notifications']==[]
|
||
|
|
assert result['bookings'][0]['title']=='Optaget'
|
||
|
|
assert 'Secret' not in str(result)
|
||
|
|
|
||
|
|
|
||
|
|
def test_db_case_start_clock_and_stale_case_update(planning_db):
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
row=work.case_planning(cases[0],actor)
|
||
|
|
result=work.update_case_planning(cases[0],work.CasePlanningInput(updated_at=row['updated_at'],
|
||
|
|
start_date=date(2026,10,20),start_time=time(13),estimated_minutes=720),actor)
|
||
|
|
assert result['start_date']==date(2026,10,20) and result['start_time']==time(13)
|
||
|
|
with pytest.raises(HTTPException) as error:
|
||
|
|
work.update_case_planning(cases[0],work.CasePlanningInput(updated_at=row['updated_at'],estimated_minutes=30),actor)
|
||
|
|
assert error.value.status_code==409
|
||
|
|
|
||
|
|
|
||
|
|
def test_db_atomic_individual_distribution_rolls_back_on_stale_queue(planning_db):
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
from app.core.database import execute_query_single
|
||
|
|
for uid in users[:2]:
|
||
|
|
work.add_assignee(cases[0],work.AssignmentInput(user_id=uid,estimated_minutes=120),actor)
|
||
|
|
plan=work.case_planning(cases[0],actor)
|
||
|
|
versions={r['user_id']:r['queue_version'] for r in plan['assignees']}
|
||
|
|
work.change_queue(users[1],cases[0],work.QueueChange(version=versions[users[1]],pinned=True),actor)
|
||
|
|
payload=work.CasePlanningInput(updated_at=plan['updated_at'],estimated_minutes=600,
|
||
|
|
assignee_estimates=[work.AssigneeEstimateInput(user_id=uid,queue_version=versions[uid],estimated_minutes=200) for uid in users[:2]],
|
||
|
|
add_assignee=work.AssignmentInput(user_id=users[2],estimated_minutes=200))
|
||
|
|
with pytest.raises(HTTPException) as error:
|
||
|
|
work.update_case_planning(cases[0],payload,actor)
|
||
|
|
assert error.value.status_code==409
|
||
|
|
after=work.case_planning(cases[0],actor)
|
||
|
|
assert after['estimated_minutes']==plan['estimated_minutes']
|
||
|
|
assert len(after['assignees'])==2
|
||
|
|
assert all(r['estimated_minutes']==120 for r in after['assignees'])
|
||
|
|
for item in payload.assignee_estimates:
|
||
|
|
item.queue_version=work.get_queue(item.user_id,100,0,actor)['version']
|
||
|
|
work.update_case_planning(cases[0],payload,actor)
|
||
|
|
assert len(work.case_planning(cases[0],actor)['assignees'])==3
|
||
|
|
assert execute_query_single('SELECT COUNT(*) AS n FROM planner_allocation_history WHERE sag_id=%s AND action=\'case_plan_update\'',(cases[0],))['n']==1
|
||
|
|
|
||
|
|
|
||
|
|
def test_db_concurrent_queue_and_booking_updates_have_one_winner(planning_db):
|
||
|
|
from concurrent.futures import ThreadPoolExecutor
|
||
|
|
from threading import Barrier
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
work.add_assignee(cases[0],work.AssignmentInput(user_id=users[0]),actor)
|
||
|
|
version=work.get_queue(users[0],100,0,actor)['version']
|
||
|
|
barrier=Barrier(2)
|
||
|
|
def queue_change(value):
|
||
|
|
barrier.wait(timeout=5)
|
||
|
|
try:
|
||
|
|
work.change_queue(users[0],cases[0],work.QueueChange(version=version,estimated_minutes=value),actor)
|
||
|
|
return 200
|
||
|
|
except HTTPException as exc:
|
||
|
|
return exc.status_code
|
||
|
|
with ThreadPoolExecutor(2) as executor:
|
||
|
|
assert sorted(executor.map(queue_change,[60,120]))==[200,409]
|
||
|
|
start=datetime(2026,11,9,10,tzinfo=work.TZ)
|
||
|
|
payload=calendar.EventInput(title='Concurrent',user_ids=[users[0]],starts_at=start,ends_at=start+timedelta(hours=1))
|
||
|
|
event=calendar.create_event(payload,actor)['event']
|
||
|
|
barrier=Barrier(2)
|
||
|
|
def event_change(offset):
|
||
|
|
barrier.wait(timeout=5)
|
||
|
|
try:
|
||
|
|
calendar.update_event(event['id'],payload.model_copy(update={'version':event['version'],
|
||
|
|
'starts_at':start+timedelta(hours=offset),'ends_at':start+timedelta(hours=offset+1)}),actor)
|
||
|
|
return 200
|
||
|
|
except HTTPException as exc:
|
||
|
|
return exc.status_code
|
||
|
|
with ThreadPoolExecutor(2) as executor:
|
||
|
|
assert sorted(executor.map(event_change,[2,4]))==[200,409]
|
||
|
|
|
||
|
|
|
||
|
|
def test_db_deadline_scheduler_deduplicates_and_uses_existing_reminders(planning_db):
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
from app.core.database import execute_query,execute_query_single
|
||
|
|
from app.modules.planner.backend.checks import run_planning_checks
|
||
|
|
work.add_assignee(cases[0],work.AssignmentInput(user_id=users[0]),actor)
|
||
|
|
execute_query("UPDATE sag_sager SET deadline=(NOW() AT TIME ZONE 'UTC')+INTERVAL '2 hours' WHERE id=%s",(cases[0],),fetch=False)
|
||
|
|
run_planning_checks()
|
||
|
|
run_planning_checks()
|
||
|
|
result=execute_query_single("SELECT COUNT(*) AS n,BOOL_AND(notify_frontend AND NOT notify_email AND NOT notify_mattermost) AS local_only FROM sag_reminders WHERE sag_id=%s AND event_type='deadline'",(cases[0],))
|
||
|
|
assert result['n']==1 and result['local_only']
|
||
|
|
|
||
|
|
|
||
|
|
def test_db_transfer_failure_restores_source_and_responsible(planning_db,monkeypatch):
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
from app.core.database import execute_query,execute_query_single
|
||
|
|
execute_query('UPDATE sag_sager SET ansvarlig_bruger_id=%s WHERE id=%s',(users[0],cases[0]),fetch=False)
|
||
|
|
work.add_assignee(cases[0],work.AssignmentInput(user_id=users[0],estimated_minutes=180),actor)
|
||
|
|
versions=[work.get_queue(uid,100,0,actor)['version'] for uid in users[:2]]
|
||
|
|
original=work.audit
|
||
|
|
def fail(*args,**kwargs):
|
||
|
|
raise RuntimeError('Synthetic audit failure')
|
||
|
|
monkeypatch.setattr(work,'audit',fail)
|
||
|
|
payload=work.TransferInput(source_user_id=users[0],target_user_id=users[1],source_version=versions[0],target_version=versions[1])
|
||
|
|
with pytest.raises(RuntimeError):
|
||
|
|
work.transfer(cases[0],payload,actor)
|
||
|
|
assert [r['user_id'] for r in execute_query('SELECT user_id FROM sag_case_assignees WHERE sag_id=%s',(cases[0],))]==[users[0]]
|
||
|
|
assert [work.get_queue(uid,100,0,actor)['version'] for uid in users[:2]]==versions
|
||
|
|
monkeypatch.setattr(work,'audit',original)
|
||
|
|
work.transfer(cases[0],payload,actor)
|
||
|
|
assert execute_query_single('SELECT ansvarlig_bruger_id FROM sag_sager WHERE id=%s',(cases[0],))['ansvarlig_bruger_id']==users[0]
|
||
|
|
assert [r['user_id'] for r in execute_query('SELECT user_id FROM sag_case_assignees WHERE sag_id=%s',(cases[0],))]==[users[1]]
|
||
|
|
|
||
|
|
|
||
|
|
def test_db_absence_reasons_are_masked_but_period_editing_is_allowed(planning_db):
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
from app.core.database import execute_query
|
||
|
|
from app.core.case_visibility import scrub_absence_time
|
||
|
|
reader={'id':users[1],'permissions':['planner.view','planner.view_team','planner.edit_team','planner.absence_team','cases.view']}
|
||
|
|
execute_query('INSERT INTO planner_team_scope(editor_user_id,target_user_id) VALUES(%s,%s)',(users[1],users[0]),fetch=False)
|
||
|
|
start=datetime(2026,11,16,8,tzinfo=work.TZ)
|
||
|
|
item=calendar.create_absence(calendar.AbsenceInput(user_id=users[0],kind='sickness',note='Private medical note',starts_at=start,ends_at=start+timedelta(hours=8)),actor)['absence']
|
||
|
|
view=calendar.schedule(start,start+timedelta(days=1),[users[0]],reader)['absences'][0]
|
||
|
|
assert view['kind']=='unavailable' and not view['note']
|
||
|
|
result=calendar.update_absence(item['id'],calendar.AbsenceInput(user_id=users[0],version=item['version'],starts_at=start+timedelta(hours=1),ends_at=start+timedelta(hours=8)),reader)
|
||
|
|
assert result['absence']['kind']=='unavailable'
|
||
|
|
redacted=scrub_absence_time({'planner_absence_id':item['id'],'planner_activity_code':'sickness','medarbejder_id':users[0],'description':'Sygdom'},users[1],False)
|
||
|
|
assert redacted['description']=='Fravær' and redacted['planner_activity_code'] is None
|
||
|
|
|
||
|
|
|
||
|
|
def test_work_windows_respect_danish_clock_changes():
|
||
|
|
from datetime import timezone
|
||
|
|
schedules=[{'user_id':1,'weekday':6,'start_time':time(1),'end_time':time(4),'break_minutes':0}]
|
||
|
|
# On the winter transition 01:00-04:00 spans four real hours.
|
||
|
|
start=datetime(2026,10,25,1,tzinfo=work.TZ)
|
||
|
|
end=datetime(2026,10,25,4,tzinfo=work.TZ)
|
||
|
|
assert (end.astimezone(timezone.utc)-start.astimezone(timezone.utc)).total_seconds()==14400
|
||
|
|
assert not calendar.outside_work_hours(1,start,end,schedules)
|
||
|
|
assert capacity.working_capacity(schedules,[],start,end)['normal_minutes']==240
|
||
|
|
assert calendar.absence_work_minutes({'starts_at':start,'ends_at':end},schedules)==240
|
||
|
|
|
||
|
|
|
||
|
|
def test_db_parent_summary_deduplicates_shared_descendants(planning_db):
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
from app.core.database import execute_query
|
||
|
|
for parent,child in [(cases[0],cases[1]),(cases[0],cases[2]),(cases[1],cases[2])]:
|
||
|
|
execute_query("INSERT INTO sag_relationer(kilde_sag_id,målsag_id,relationstype) VALUES(%s,%s,'undersag')",(parent,child),fetch=False)
|
||
|
|
for cid in cases:
|
||
|
|
execute_query('UPDATE sag_sager SET estimated_minutes=120 WHERE id=%s',(cid,),fetch=False)
|
||
|
|
execute_query("INSERT INTO tmodule_times(sag_id,medarbejder_id,description,original_hours,faktisk_tid_min,status) VALUES(%s,%s,'Synthetic time',1,60,'pending')",(cid,users[0]),fetch=False)
|
||
|
|
total=capacity.case_summary(cases[0],True,actor,'all')
|
||
|
|
children=capacity.case_summary(cases[0],True,actor,'children')
|
||
|
|
own=capacity.case_summary(cases[0],False,actor,'own')
|
||
|
|
assert total['case_count']==3 and total['actual_minutes']==180 and total['remaining_minutes']==180
|
||
|
|
assert children['case_count']==2 and children['actual_minutes']==120
|
||
|
|
assert own['case_count']==1 and own['actual_minutes']==60
|
||
|
|
|
||
|
|
|
||
|
|
def test_db_next_task_warns_and_does_not_create_time(planning_db):
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
from app.core.database import execute_query_single
|
||
|
|
for cid,estimate in zip(cases,[180,5,None]):
|
||
|
|
work.add_assignee(cid,work.AssignmentInput(user_id=users[0],estimated_minutes=estimate),actor)
|
||
|
|
start=datetime.now(work.TZ)+timedelta(minutes=20)
|
||
|
|
count=execute_query_single('SELECT COUNT(*) AS n FROM tmodule_times')['n']
|
||
|
|
calendar.create_event(calendar.EventInput(title='Fixed meeting',user_ids=[users[0]],starts_at=start,ends_at=start+timedelta(hours=1)),actor)
|
||
|
|
result=work.next_task(actor)
|
||
|
|
assert result['task']['sag_id']==cases[0] and result['warning']
|
||
|
|
assert result['shorter_task']['sag_id']==cases[1]
|
||
|
|
assert execute_query_single('SELECT COUNT(*) AS n FROM tmodule_times')['n']==count
|
||
|
|
|
||
|
|
|
||
|
|
def test_db_queue_paginates_and_searches_two_thousand_cases(planning_db):
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
from app.core.database import execute_query,execute_query_single
|
||
|
|
from time import monotonic
|
||
|
|
with work.atomic_transaction():
|
||
|
|
execute_query('''WITH new_cases AS (
|
||
|
|
INSERT INTO sag_sager(titel,created_by_user_id,status,type)
|
||
|
|
SELECT 'Large queue '||i,%s,'åben','ticket' FROM generate_series(1,2000) i RETURNING id)
|
||
|
|
INSERT INTO sag_case_assignees(sag_id,user_id) SELECT id,%s FROM new_cases''',(users[0],users[0]),fetch=False)
|
||
|
|
started=monotonic()
|
||
|
|
first=work.get_queue(users[0],100,0,actor)
|
||
|
|
second=work.get_queue(users[0],100,100,actor)
|
||
|
|
found=work.get_queue(users[0],100,0,actor,'Large queue 2000')
|
||
|
|
assert len(first['items'])==len(second['items'])==100
|
||
|
|
assert not {r['sag_id'] for r in first['items']} & {r['sag_id'] for r in second['items']}
|
||
|
|
assert len(found['items'])==1 and found['items'][0]['titel']=='Large queue 2000'
|
||
|
|
assert monotonic()-started<10 # A generous bound catches accidental per-row round trips.
|
||
|
|
|
||
|
|
|
||
|
|
def test_db_existing_calendar_feed_reads_shared_event_once(planning_db):
|
||
|
|
from types import SimpleNamespace
|
||
|
|
from app.modules.calendar.backend.router import _planning_feed_events,_build_ical_response
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
from app.core.database import execute_query
|
||
|
|
execute_query('UPDATE users SET is_superadmin=TRUE WHERE user_id=%s',(users[0],),fetch=False)
|
||
|
|
start=datetime(2026,11,23,10,tzinfo=work.TZ)
|
||
|
|
calendar.create_event(calendar.EventInput(title='Shared feed event',sag_id=cases[0],user_ids=users,
|
||
|
|
starts_at=start,ends_at=start+timedelta(hours=1)),actor)
|
||
|
|
request=SimpleNamespace(state=SimpleNamespace(user_id=users[0]))
|
||
|
|
events=_planning_feed_events(request,start,start+timedelta(days=1),users[0],False,None,{'planner_booking'})
|
||
|
|
selected=[e for e in events if e.get('sag_id')==cases[0]]
|
||
|
|
assert len(selected)==1
|
||
|
|
assert _build_ical_response(selected,start).body.count(b'BEGIN:VEVENT')==1
|
||
|
|
|
||
|
|
|
||
|
|
def test_db_unassigned_search_paging_and_assignment(planning_db):
|
||
|
|
from app.core.database import execute_query
|
||
|
|
from app.modules.sag.backend import visibility
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
marker=f'Unassigned-{cases[0]}'
|
||
|
|
for cid in cases:
|
||
|
|
execute_query('UPDATE sag_sager SET titel=%s,ansvarlig_bruger_id=%s WHERE id=%s',
|
||
|
|
(marker+' '+str(cid),users[0],cid),fetch=False)
|
||
|
|
first=capacity.unassigned(2,0,actor,marker)
|
||
|
|
assert len(first['items'])==2 and first['has_more']
|
||
|
|
assert all(row['responsible_name'] for row in first['items'])
|
||
|
|
second=capacity.unassigned(2,2,actor,marker)
|
||
|
|
assert len(second['items'])==1 and not second['has_more']
|
||
|
|
work.add_assignee(cases[0],work.AssignmentInput(user_id=users[1],estimated_minutes=60),actor)
|
||
|
|
remaining=capacity.unassigned(25,0,actor,marker)
|
||
|
|
assert {row['id'] for row in remaining['items']}==set(cases[1:])
|
||
|
|
assert work.get_queue(users[1],100,0,actor)['items'][0]['estimated_minutes']==60
|
||
|
|
visibility.update_visibility(cases[1],visibility.VisibilityInput(restricted=True,user_ids=[users[0]],version=0),actor)
|
||
|
|
reader={'id':users[2],'permissions':['cases.view','planner.view','planner.allocate']}
|
||
|
|
hidden=capacity.unassigned(25,0,reader,marker)
|
||
|
|
assert [row['id'] for row in hidden['items']]==[cases[2]]
|
||
|
|
|
||
|
|
|
||
|
|
def test_db_drag_dates_preserve_clock_and_detect_stale_case(planning_db):
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
original=work.case_planning(cases[0],actor)
|
||
|
|
initial=work.update_case_planning(cases[0],work.CasePlanningInput(updated_at=original['updated_at'],
|
||
|
|
start_date=date(2026,10,23),start_time=time(9),deadline=datetime(2026,10,23,13,30,tzinfo=work.TZ)),actor)
|
||
|
|
moved=work.update_case_planning(cases[0],work.CasePlanningInput(updated_at=initial['updated_at'],
|
||
|
|
deadline_date=date(2026,11,2)),actor)
|
||
|
|
assert moved['deadline']==datetime(2026,11,2,13,30,tzinfo=work.TZ)
|
||
|
|
assert moved['start_date']==date(2026,10,23) and moved['start_time']==time(9)
|
||
|
|
with pytest.raises(HTTPException) as error:
|
||
|
|
work.update_case_planning(cases[0],work.CasePlanningInput(updated_at=initial['updated_at'],start_date=date(2026,11,1)),actor)
|
||
|
|
assert error.value.status_code==409
|
||
|
|
reader={'id':users[1],'permissions':['cases.view','planner.view']}
|
||
|
|
with pytest.raises(HTTPException) as error:
|
||
|
|
work.update_case_planning(cases[0],work.CasePlanningInput(updated_at=moved['updated_at'],deadline_date=date(2026,11,3)),reader)
|
||
|
|
assert error.value.status_code==403
|
||
|
|
next_plan=work.case_planning(cases[1],actor)
|
||
|
|
new=work.update_case_planning(cases[1],work.CasePlanningInput(updated_at=next_plan['updated_at'],
|
||
|
|
start_date=date(2026,11,2),deadline_date=date(2026,11,3)),actor)
|
||
|
|
assert new['deadline'].hour==17
|
||
|
|
schedule=calendar.schedule(datetime(2026,11,1,tzinfo=work.TZ),datetime(2026,11,5,tzinfo=work.TZ),[users[0]],actor,True)
|
||
|
|
marker=next(m for m in schedule['milestones'] if m['sag_id']==cases[1] and m['kind']=='case_start')
|
||
|
|
assert marker['allDay'] and marker['case_updated_at']==new['updated_at']
|
||
|
|
|
||
|
|
|
||
|
|
def test_db_card_keeps_multiple_allocated_time_slots(planning_db):
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
start=datetime(2026,11,16,10,tzinfo=work.TZ)
|
||
|
|
payload=calendar.EventInput(sag_id=cases[0],title='Multiple slots',user_ids=[users[0]],starts_at=start,ends_at=start+timedelta(hours=1))
|
||
|
|
first=calendar.create_event(payload,actor)['event']
|
||
|
|
second=calendar.create_event(payload.model_copy(update={'starts_at':start+timedelta(hours=2),'ends_at':start+timedelta(hours=3)}),actor)['event']
|
||
|
|
row=next(r for r in work.get_queue(users[0],100,0,actor)['items'] if r['sag_id']==cases[0])
|
||
|
|
assert row['future_planned_minutes']==120
|
||
|
|
assert {s['event_id'] for s in row['planned_slots']}=={first['id'],second['id']}
|
||
|
|
calendar.update_event(first['id'],payload.model_copy(update={'version':first['version'],
|
||
|
|
'starts_at':start+timedelta(hours=3),'ends_at':start+timedelta(hours=4)}),actor)
|
||
|
|
after=next(r for r in work.get_queue(users[0],100,0,actor)['items'] if r['sag_id']==cases[0])
|
||
|
|
assert len(after['planned_slots'])==2
|
||
|
|
original=next(s for s in row['planned_slots'] if s['event_id']==second['id'])
|
||
|
|
preserved=next(s for s in after['planned_slots'] if s['event_id']==second['id'])
|
||
|
|
assert original==preserved
|
||
|
|
|
||
|
|
|
||
|
|
def test_db_pause_resume_reuses_the_same_timer(planning_db):
|
||
|
|
import asyncio
|
||
|
|
from app.core.database import execute_query,execute_query_single
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
work.add_assignee(cases[0],work.AssignmentInput(user_id=users[0]),actor)
|
||
|
|
timer=execute_query_single('''INSERT INTO tmodule_times(sag_id,medarbejder_id,description,original_hours,
|
||
|
|
worked_date,status,billable,aktiv_timer,start_tid,entry_status)
|
||
|
|
VALUES(%s,%s,'Synthetic timer',0.01,CURRENT_DATE,'pending',false,true,NOW(),'kladde') RETURNING id''',(cases[0],users[0]))['id']
|
||
|
|
queue=work.get_queue(users[0],100,0,actor)
|
||
|
|
paused=asyncio.run(work.pause_work(cases[0],work.WorkAction(version=queue['version'],timer_action=True),actor))
|
||
|
|
assert paused['item']['work_status']=='paused'
|
||
|
|
assert execute_query_single('SELECT aktiv_timer FROM tmodule_times WHERE id=%s',(timer,))['aktiv_timer'] is False
|
||
|
|
resumed=asyncio.run(work.resume_work(cases[0],work.WorkAction(version=paused['version'],timer_action=True),actor))
|
||
|
|
assert resumed['item']['work_status']=='working'
|
||
|
|
assert resumed['timer']['id']==timer
|
||
|
|
|
||
|
|
|
||
|
|
def test_db_capacity_counts_booked_work_once_and_flags_absence_overlap(planning_db):
|
||
|
|
from app.core.database import execute_query,execute_query_single
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
start=datetime(2026,11,2,8,tzinfo=work.TZ)
|
||
|
|
execute_query('INSERT INTO planner_work_schedules(user_id,weekday,start_time,end_time,break_minutes) VALUES(%s,0,%s,%s,0)',(users[0],time(8),time(16)),fetch=False)
|
||
|
|
work.add_assignee(cases[0],work.AssignmentInput(user_id=users[0]),actor)
|
||
|
|
queue=work.get_queue(users[0],100,0,actor)
|
||
|
|
work.change_queue(users[0],cases[0],work.QueueChange(version=queue['version'],estimated_minutes=600),actor)
|
||
|
|
calendar.create_event(calendar.EventInput(title='Capacity test',sag_id=cases[0],user_ids=[users[0]],starts_at=start+timedelta(hours=2),ends_at=start+timedelta(hours=4)),actor)
|
||
|
|
calendar.create_absence(calendar.AbsenceInput(user_id=users[0],kind='vacation',starts_at=start+timedelta(hours=1),ends_at=start+timedelta(hours=3)),actor)
|
||
|
|
row=capacity.capacity(start,start+timedelta(hours=8),[users[0]],actor)['items'][0]
|
||
|
|
assert row['normal_minutes']==480 and row['busy_minutes']==180 and row['free_minutes']==300
|
||
|
|
assert row['queue_minutes']==480 and row['overload_minutes']==180
|
||
|
|
assert calendar.schedule(start,start+timedelta(hours=8),[users[0]],actor)['bookings'][0]['has_conflict'] is True
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize('action', ['transfer','remove'])
|
||
|
|
def test_db_transfer_offers_booking_choice_and_preserves_shared_participants(planning_db, action):
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
from app.core.database import execute_query,execute_query_single
|
||
|
|
start=datetime(2026,11,9,9,tzinfo=work.TZ)
|
||
|
|
for offset,participants in ((0,[users[0]]),(2,[users[0],users[2]])):
|
||
|
|
calendar.create_event(calendar.EventInput(title='Transfer reservation',sag_id=cases[0],
|
||
|
|
user_ids=participants,starts_at=start+timedelta(hours=offset),
|
||
|
|
ends_at=start+timedelta(hours=offset+1),confirm_conflicts=True),actor)
|
||
|
|
execute_query('UPDATE sag_sager SET ansvarlig_bruger_id=%s WHERE id=%s',(users[0],cases[0]),fetch=False)
|
||
|
|
old=execute_query('SELECT * FROM planner_allocations WHERE sag_id=%s AND user_id=%s ORDER BY id',(cases[0],users[0]))
|
||
|
|
body=dict(source_user_id=users[0],target_user_id=users[1],
|
||
|
|
source_version=work.get_queue(users[0],100,0,actor)['version'],
|
||
|
|
target_version=work.get_queue(users[1],100,0,actor)['version'])
|
||
|
|
with pytest.raises(HTTPException) as choice:
|
||
|
|
work.transfer(cases[0],work.TransferInput(**body),actor)
|
||
|
|
assert choice.value.status_code==409
|
||
|
|
assert choice.value.detail['code']=='future_bookings_choice'
|
||
|
|
assert choice.value.detail['booking_count']==2
|
||
|
|
payload=work.TransferInput(**body,future_bookings_action=action,
|
||
|
|
future_bookings=choice.value.detail['bookings'],confirm_conflicts=True)
|
||
|
|
work.transfer(cases[0],payload,actor)
|
||
|
|
changed=execute_query('SELECT * FROM planner_allocations WHERE id=ANY(%s) ORDER BY id',([r['id'] for r in old],))
|
||
|
|
for before,after in zip(old,changed):
|
||
|
|
assert after['starts_at']==before['starts_at'] and after['ends_at']==before['ends_at']
|
||
|
|
assert after['version']==before['version']+1
|
||
|
|
assert (after['user_id']==users[1] and after['deleted_at'] is None) if action=='transfer' else after['deleted_at'] is not None
|
||
|
|
assert execute_query_single('SELECT COUNT(*) AS n FROM planner_allocations WHERE sag_id=%s AND user_id=%s AND deleted_at IS NULL',(cases[0],users[2]))['n']==1
|
||
|
|
assert execute_query_single('SELECT ansvarlig_bruger_id FROM sag_sager WHERE id=%s',(cases[0],))['ansvarlig_bruger_id']==users[0]
|
||
|
|
assert {r['user_id'] for r in execute_query('SELECT user_id FROM sag_case_assignees WHERE sag_id=%s',(cases[0],))}=={users[1],users[2]}
|
||
|
|
|
||
|
|
|
||
|
|
def test_db_transfer_bookings_stale_conflict_and_audit_rollback(planning_db,monkeypatch):
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
from app.core.database import execute_query,execute_query_single
|
||
|
|
start=datetime(2026,11,9,9,tzinfo=work.TZ)
|
||
|
|
calendar.create_event(calendar.EventInput(title='Move',sag_id=cases[0],user_ids=[users[0]],
|
||
|
|
starts_at=start,ends_at=start+timedelta(hours=1),confirm_conflicts=True),actor)
|
||
|
|
calendar.create_event(calendar.EventInput(title='Busy',user_ids=[users[1]],
|
||
|
|
starts_at=start,ends_at=start+timedelta(hours=1),confirm_conflicts=True),actor)
|
||
|
|
body=dict(source_user_id=users[0],target_user_id=users[1],
|
||
|
|
source_version=work.get_queue(users[0],100,0,actor)['version'],
|
||
|
|
target_version=work.get_queue(users[1],100,0,actor)['version'])
|
||
|
|
with pytest.raises(HTTPException) as choice:
|
||
|
|
work.transfer(cases[0],work.TransferInput(**body),actor)
|
||
|
|
snapshot=choice.value.detail['bookings']
|
||
|
|
with pytest.raises(HTTPException) as conflict:
|
||
|
|
work.transfer(cases[0],work.TransferInput(**body,future_bookings_action='transfer',future_bookings=snapshot),actor)
|
||
|
|
assert conflict.value.detail['confirm_conflicts']
|
||
|
|
original_audit=work.audit
|
||
|
|
def fail(*args,**kwargs):
|
||
|
|
raise RuntimeError('Audit rollback')
|
||
|
|
monkeypatch.setattr(work,'audit',fail)
|
||
|
|
payload=work.TransferInput(**body,future_bookings_action='transfer',future_bookings=snapshot,confirm_conflicts=True)
|
||
|
|
with pytest.raises(RuntimeError):
|
||
|
|
work.transfer(cases[0],payload,actor)
|
||
|
|
row=execute_query_single('SELECT * FROM planner_allocations WHERE id=%s',(snapshot[0]['id'],))
|
||
|
|
assert row['user_id']==users[0] and row['version']==snapshot[0]['version']
|
||
|
|
assert execute_query_single('SELECT version FROM planner_events WHERE id=%s',(row['event_id'],))['version']==snapshot[0]['event_version']
|
||
|
|
assert work.get_queue(users[0],100,0,actor)['version']==body['source_version']
|
||
|
|
assert not execute_query_single('SELECT 1 FROM sag_case_assignees WHERE sag_id=%s AND user_id=%s',(cases[0],users[1]))
|
||
|
|
monkeypatch.setattr(work,'audit',original_audit)
|
||
|
|
execute_query('UPDATE planner_events SET version=version+1 WHERE id=%s',(row['event_id'],),fetch=False)
|
||
|
|
with pytest.raises(HTTPException) as stale:
|
||
|
|
work.transfer(cases[0],payload,actor)
|
||
|
|
assert stale.value.status_code==409 and 'ændret' in stale.value.detail
|
||
|
|
|
||
|
|
|
||
|
|
def test_absence_whole_days_are_visible_and_follow_copenhagen_dst():
|
||
|
|
placeholder=datetime(2026,10,1,tzinfo=work.TZ)
|
||
|
|
row=calendar.AbsenceInput(user_id=1,kind='vacation',starts_at=placeholder,
|
||
|
|
ends_at=placeholder+timedelta(hours=1),whole_day_start=date(2026,10,24),whole_day_end=date(2026,10,26))
|
||
|
|
assert row.starts_at.utcoffset()==timedelta(hours=2)
|
||
|
|
assert row.ends_at.utcoffset()==timedelta(hours=1)
|
||
|
|
safe=calendar.visible_absence(dict(id=1,user_id=1,kind='vacation',note='Private',
|
||
|
|
starts_at=row.starts_at,ends_at=row.ends_at),{'id':1})
|
||
|
|
assert safe['all_day'] and safe['display_start']=='2026-10-24' and safe['display_end']=='2026-10-27'
|
||
|
|
safe=calendar.visible_absence(dict(id=1,user_id=1,kind='vacation',note='Private',
|
||
|
|
starts_at=row.starts_at.replace(hour=8),ends_at=row.ends_at.replace(hour=16)),{'id':2})
|
||
|
|
assert safe['kind']=='unavailable' and safe['note'] is None and safe['all_day']
|
||
|
|
assert safe['display_end']=='2026-10-28'
|
||
|
|
|
||
|
|
|
||
|
|
def test_db_absence_report_permissions_scope_masking_overlap_and_pagination(planning_db):
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
from app.core.database import execute_query
|
||
|
|
from datetime import timezone
|
||
|
|
monday=datetime(2026,11,9,tzinfo=work.TZ)
|
||
|
|
for uid in users:
|
||
|
|
execute_query('DELETE FROM planner_work_schedules WHERE user_id=%s',(uid,),fetch=False)
|
||
|
|
for weekday in range(5):
|
||
|
|
execute_query("INSERT INTO planner_work_schedules(user_id,weekday,start_time,end_time,break_minutes) VALUES(%s,%s,'08:00','16:00',30)",(uid,weekday),fetch=False)
|
||
|
|
# Two overlapping periods must count as two rows, but only one day in totals.
|
||
|
|
for kind,start,end in [('vacation',monday,monday+timedelta(days=1)),
|
||
|
|
('sickness',monday.replace(hour=9),monday.replace(hour=12))]:
|
||
|
|
calendar.create_absence(calendar.AbsenceInput(user_id=users[1],kind=kind,
|
||
|
|
starts_at=start,ends_at=end,note='Never export this'),actor)
|
||
|
|
calendar.create_absence(calendar.AbsenceInput(user_id=users[2],kind='vacation',
|
||
|
|
starts_at=monday,ends_at=monday+timedelta(days=1)),actor)
|
||
|
|
reader={'id':users[0],'permissions':['planner.view','cases.view','planner.view_team'],
|
||
|
|
'_planner_team_scope_ids':{users[1]}}
|
||
|
|
with pytest.raises(HTTPException) as denied:
|
||
|
|
calendar.absence_report(monday.date(),monday.date(),None,50,0,reader)
|
||
|
|
assert denied.value.status_code==403
|
||
|
|
reader['permissions'].append('planner.absence_report')
|
||
|
|
result=calendar.absence_report(monday.date(),monday.date(),None,1,0,reader)
|
||
|
|
assert result['total']==2 and len(result['items'])==1
|
||
|
|
assert {r['id'] for r in result['employees']}=={users[0],users[1]}
|
||
|
|
assert result['totals']['planned_minutes']==450 and result['totals']['planned_days']==1
|
||
|
|
assert result['items'][0]['kind']=='unavailable' and result['items'][0]['kind_label']=='Fravær'
|
||
|
|
assert 'Never export this' not in str(result)
|
||
|
|
assert 'note' not in result['items'][0]
|
||
|
|
assert len(result['type_summary'])==1
|
||
|
|
masked=result['type_summary'][0]
|
||
|
|
assert masked['kind']=='unavailable' and masked['kind_label']=='Fravær'
|
||
|
|
assert masked['planned_minutes']==450 and masked['periods']==2
|
||
|
|
assert {r['user_id'] for r in masked['employees']}=={users[1]}
|
||
|
|
next_page=calendar.absence_report(monday.date(),monday.date(),None,1,1,reader)
|
||
|
|
assert next_page['items'][0]['id']!=result['items'][0]['id']
|
||
|
|
with pytest.raises(HTTPException) as denied:
|
||
|
|
calendar.absence_report(monday.date(),monday.date(),users[2],50,0,reader)
|
||
|
|
assert denied.value.status_code==403
|
||
|
|
reader['permissions'].append('planner.absence_reason')
|
||
|
|
result=calendar.absence_report(monday.date(),monday.date(),users[1],50,0,reader)
|
||
|
|
assert {r['kind_label'] for r in result['items']}=={'Ferie','Sygdom'}
|
||
|
|
grouped=calendar.absence_report(monday.date(),monday.date(),users[1],1,0,reader,group_by='type')
|
||
|
|
assert [r['kind_label'] for r in grouped['type_summary']]==['Ferie','Sygdom']
|
||
|
|
assert grouped['type_summary'][0]['planned_minutes']==450
|
||
|
|
assert grouped['type_summary'][1]['planned_minutes']==180
|
||
|
|
assert grouped['items'][0]['kind']=='vacation'
|
||
|
|
next_group=calendar.absence_report(monday.date(),monday.date(),users[1],1,1,reader,group_by='type')
|
||
|
|
assert next_group['items'][0]['kind']=='sickness'
|
||
|
|
assert next_group['type_summary']==grouped['type_summary']
|
||
|
|
# A range ending before the periods must contain no rows.
|
||
|
|
result=calendar.absence_report(date(2026,11,1),date(2026,11,8),users[1],50,0,reader)
|
||
|
|
assert result['total']==0 and result['totals']['planned_minutes']==0
|
||
|
|
|
||
|
|
|
||
|
|
def test_db_absence_report_keeps_actual_time_separate_and_filters_worked_dates(planning_db):
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
from app.core.database import execute_query
|
||
|
|
for weekday in (0,1):
|
||
|
|
execute_query("INSERT INTO planner_work_schedules(user_id,weekday,start_time,end_time,break_minutes) VALUES(%s,%s,'08:00','16:00',30)",(users[0],weekday),fetch=False)
|
||
|
|
start=datetime(2020,1,6,tzinfo=work.TZ)
|
||
|
|
absence=calendar.create_absence(calendar.AbsenceInput(user_id=users[0],kind='vacation',
|
||
|
|
starts_at=start,ends_at=start+timedelta(days=2)),actor)['absence']
|
||
|
|
before=calendar.absence_report(date(2020,1,7),date(2020,1,7),users[0],50,0,actor)
|
||
|
|
assert before['totals']=={'planned_minutes':450,'planned_days':1,'actual_minutes':0}
|
||
|
|
assert not before['items'][0]['registered']
|
||
|
|
calendar.register_absence(absence['id'],absence['version'],actor)
|
||
|
|
after=calendar.absence_report(date(2020,1,7),date(2020,1,7),users[0],50,0,actor)
|
||
|
|
assert after['totals']=={'planned_minutes':450,'planned_days':1,'actual_minutes':450}
|
||
|
|
assert after['items'][0]['registered']
|
||
|
|
assert after['total']==1
|
||
|
|
|
||
|
|
|
||
|
|
def test_db_team_calendar_exposes_absence_periods_with_masked_reasons(planning_db):
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
from app.modules.planner.backend.router import period
|
||
|
|
start=datetime(2026,11,9,tzinfo=work.TZ)
|
||
|
|
for uid in users:
|
||
|
|
calendar.create_absence(calendar.AbsenceInput(user_id=uid,kind='vacation',
|
||
|
|
starts_at=start,ends_at=start+timedelta(days=2),note='Private absence note'),actor)
|
||
|
|
reader={'id':users[0],'permissions':['planner.view','planner.view_team','cases.view'],
|
||
|
|
'_planner_team_scope_ids':{users[1]}}
|
||
|
|
result=period(start,start+timedelta(days=7),users[:2],reader)
|
||
|
|
assert len(result['absences'])==2
|
||
|
|
assert {r['user_id'] for r in result['absences']}==set(users[:2])
|
||
|
|
assert next(r for r in result['absences'] if r['user_id']==users[0])['kind_label']=='Ferie'
|
||
|
|
safe=next(r for r in result['absences'] if r['user_id']==users[1])
|
||
|
|
assert safe['kind_label']=='Fravær' and safe['kind']=='unavailable'
|
||
|
|
assert 'Private absence note' not in str(result['absences'])
|
||
|
|
assert all('note' not in r for r in result['absences'])
|
||
|
|
reader['permissions'].append('planner.absence_reason')
|
||
|
|
assert all(r['kind_label']=='Ferie' for r in period(start,start+timedelta(days=7),users[:2],reader)['absences'])
|
||
|
|
assert not period(start+timedelta(days=2),start+timedelta(days=3),users[:2],reader)['absences']
|
||
|
|
with pytest.raises(HTTPException) as denied:
|
||
|
|
period(start,start+timedelta(days=7),[users[2]],reader)
|
||
|
|
assert denied.value.status_code==403
|
||
|
|
|
||
|
|
|
||
|
|
def test_db_duplicate_absence_type_rejected_for_create_and_edit(planning_db):
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
from app.core.database import execute_query_single
|
||
|
|
start=datetime(2026,11,9,8,tzinfo=work.TZ)
|
||
|
|
def payload(a,b,uid=users[0],kind='vacation',version=None):
|
||
|
|
return calendar.AbsenceInput(user_id=uid,kind=kind,starts_at=a,ends_at=b,version=version)
|
||
|
|
first=calendar.create_absence(payload(start,start+timedelta(hours=8)),actor)['absence']
|
||
|
|
for a,b in [(start,start+timedelta(hours=8)),(start+timedelta(hours=1),start+timedelta(hours=9)),
|
||
|
|
(start-timedelta(hours=1),start+timedelta(hours=1))]:
|
||
|
|
with pytest.raises(HTTPException) as denied:
|
||
|
|
calendar.create_absence(payload(a,b),actor)
|
||
|
|
assert denied.value.status_code==409 and denied.value.detail['code']=='absence_overlap'
|
||
|
|
# Touching periods and another employee's holiday remain valid.
|
||
|
|
next_period=calendar.create_absence(payload(start+timedelta(hours=8),start+timedelta(hours=10)),actor)['absence']
|
||
|
|
calendar.create_absence(payload(start,start+timedelta(hours=8),users[1]),actor)
|
||
|
|
calendar.update_absence(first['id'],payload(start,start+timedelta(hours=7),version=first['version']),actor)
|
||
|
|
with pytest.raises(HTTPException) as denied:
|
||
|
|
calendar.update_absence(next_period['id'],payload(start+timedelta(hours=1),start+timedelta(hours=9),version=next_period['version']),actor)
|
||
|
|
assert denied.value.status_code==409
|
||
|
|
assert execute_query_single('SELECT starts_at FROM planner_absences WHERE id=%s',(next_period['id'],))['starts_at']==next_period['starts_at']
|
||
|
|
calendar.cancel_absence(next_period['id'],next_period['version'],actor)
|
||
|
|
calendar.create_absence(payload(start+timedelta(hours=7),start+timedelta(hours=10)),actor)
|
||
|
|
|
||
|
|
|
||
|
|
def test_db_concurrent_duplicate_holidays_have_one_winner(planning_db):
|
||
|
|
from concurrent.futures import ThreadPoolExecutor
|
||
|
|
from threading import Barrier
|
||
|
|
actor,users,cases=planning_db
|
||
|
|
start=datetime(2026,11,9,8,tzinfo=work.TZ)
|
||
|
|
barrier=Barrier(2)
|
||
|
|
def create():
|
||
|
|
barrier.wait(timeout=10)
|
||
|
|
try:
|
||
|
|
calendar.create_absence(calendar.AbsenceInput(user_id=users[0],kind='vacation',starts_at=start,ends_at=start+timedelta(hours=8)),actor)
|
||
|
|
return 201
|
||
|
|
except HTTPException as error:
|
||
|
|
return error.status_code
|
||
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
||
|
|
assert sorted(pool.map(lambda _:create(),range(2)))==[201,409]
|