"""Fase A tests: routes, upload+sync (idempotent, status-behoud), seed, alerts.""" import io import json import os from datetime import date, timedelta from openpyxl import Workbook from conftest import SEED_PATH, login import agenda_module from agenda_module import (get_all_items, get_upcoming_items, load_agenda_seed, parse_agenda_xlsx, render_agenda_alerts_html, sync_agenda_from_items) def make_xlsx(rows, headers=('Datum', 'Titel', 'Type', 'Locatie', 'Omschrijving', 'T/M')): """Genereer een test-xlsx in-memory met flexibele kolomnamen.""" wb = Workbook() ws = wb.active ws.append(list(headers)) for r in rows: ws.append(list(r)) buf = io.BytesIO() wb.save(buf) buf.seek(0) return buf # ── 1. Routes geregistreerd ─────────────────────────────────────────────────── def test_agenda_page_empty(client): rv = client.get('/agenda') assert rv.status_code == 200 assert b'Agenda' in rv.data assert b'Nog geen agenda-items' in rv.data # lege staat def test_items_and_api_endpoints(client, db): today = date.today().isoformat() conn = db.get_db() conn.execute("INSERT INTO agenda_items (datum, titel) VALUES (?,?)", (today, 'Test item')) conn.commit() conn.close() rv = client.get('/agenda/items') assert rv.status_code == 200 data = rv.get_json() assert data['count'] == 1 assert data['items'][0]['titel'] == 'Test item' assert data['items'][0]['status'] == 'open' # default rv = client.get('/agenda/api/items?days=14') assert rv.status_code == 200 data = rv.get_json() assert data['count'] == 1 assert data['items'][0]['badge'] == 'vandaag' assert data['days'] == 14 def test_api_days_filter(client, db): later = (date.today() + timedelta(days=45)).isoformat() conn = db.get_db() conn.execute("INSERT INTO agenda_items (datum, titel) VALUES (?,?)", (later, 'Ver weg')) conn.commit() conn.close() assert client.get('/agenda/api/items?days=14').get_json()['count'] == 0 assert client.get('/agenda/api/items?days=60').get_json()['count'] == 1 # ── 2. Upload: parsing + admin-only + rommeldetectie ───────────────────────── def test_upload_requires_admin(client): buf = make_xlsx([['2026-12-01', 'Test', '', '', '', '']]) rv = client.post('/agenda/upload', data={'file': (buf, 'agenda.xlsx')}, follow_redirects=False, content_type='multipart/form-data') assert rv.status_code == 403 # geen sessie -> geen upload def test_upload_rejects_non_xlsx(client): login(client) rv = client.post('/agenda/upload', data={'file': (io.BytesIO(b'hello'), 'agenda.csv')}, content_type='multipart/form-data') assert rv.status_code == 400 def test_upload_and_sync(db, client): login(client) buf = make_xlsx([ (date(2026, 9, 21), 'Kabinetsvergadering GMT', 'kabinetsvergadering', 'Utrecht', 'Aftrap Alzheimer', ''), ('13-10-2026', 'Inspiratieavond', 'evenement', '', '', ''), # string-datum DD-MM-YYYY ('2026-11-21', 'Najaarsconventie', 'conventie', 'Zaandam', '', ''), # string-datum ISO ('geen datum', 'Rommelrij', '', '', '', ''), # rommel -> overgeslagen ('2026-12-05', '', '', '', '', ''), # geen titel -> overgeslagen ]) rv = client.post('/agenda/upload', data={'file': (buf, 'jaaragenda.xlsx')}, content_type='multipart/form-data', follow_redirects=True) assert rv.status_code == 200 items = get_all_items(db) assert len(items) == 3 # 3 geldig, 2 rommel overgeslagen titels = {i['titel'] for i in items} assert titels == {'Kabinetsvergadering GMT', 'Inspiratieavond', 'Najaarsconventie'} by_titel = {i['titel']: i for i in items} assert by_titel['Kabinetsvergadering GMT']['datum'] == '2026-09-21' # echte date-cel assert by_titel['Inspiratieavond']['datum'] == '2026-10-13' # dd-mm-yyyy geparsed assert all(i['status'] == 'open' for i in items) # sync-rapport in de redirect-flash assert b'toegevoegd' in rv.data def test_upload_header_flexible_and_offset(db, client): """Header met afwijkende namen + titelrij erboven moet ook werken.""" login(client) wb = Workbook() ws = wb.active ws.append(['Jaarprogramma Districtsgouverneur 2026-2027']) # titel-rij ws.append(['Wanneer', 'Activiteit', 'Soort', 'Waar', 'Details', '']) ws.append(['2026-10-13', 'Ledenontmoetingsavond', 'evenement', 'Almere', 'Inspiratie', '']) buf = io.BytesIO(); wb.save(buf); buf.seek(0) rv = client.post('/agenda/upload', data={'file': (buf, 'a.xlsx')}, content_type='multipart/form-data', follow_redirects=True) assert rv.status_code == 200 items = get_all_items(db) assert len(items) == 1 assert items[0]['titel'] == 'Ledenontmoetingsavond' assert items[0]['datum'] == '2026-10-13' assert items[0]['locatie'] == 'Almere' # ── 3. Idempotente sync: status-behoud op datum+titel-match ─────────────────── def test_sync_preserves_status_and_replaces(db, client): login(client) items_v1 = [ {'datum': '2026-09-21', 'datum_eind': '', 'titel': 'Kabinetsvergadering GMT', 'type': 'kabinetsvergadering', 'locatie': 'Utrecht', 'beschrijving': 'v1', 'status': 'open'}, {'datum': '2026-11-16', 'datum_eind': '', 'titel': 'Kabinetsvergadering GLT', 'type': 'kabinetsvergadering', 'locatie': '', 'beschrijving': '', 'status': 'open'}, ] rep = sync_agenda_from_items(db, items_v1) assert rep['toegevoegd'] == 2 and rep['verwijderd'] == 0 # status handmatig op 'voorbereid' zetten (zoals Fase C straks doet) conn = db.get_db() conn.execute("UPDATE agenda_items SET status='voorbereid' WHERE titel='Kabinetsvergadering GMT'") conn.commit(); conn.close() # v2: locatie gewijzigd, GLT weg, nieuw item erbij items_v2 = [ {'datum': '2026-09-21', 'datum_eind': '', 'titel': 'Kabinetsvergadering GMT', 'type': 'kabinetsvergadering', 'locatie': 'Amersfoort', 'beschrijving': 'v2', 'status': 'open'}, {'datum': '2026-11-21', 'datum_eind': '', 'titel': 'Najaarsconventie', 'type': 'conventie', 'locatie': 'Zaandam', 'beschrijving': '', 'status': 'open'}, ] rep2 = sync_agenda_from_items(db, items_v2) assert rep2['bijgewerkt'] == 1 assert rep2['toegevoegd'] == 1 assert rep2['verwijderd'] == 1 # GLT vervallen assert rep2['status_behouden'] == 1 items = {i['titel']: i for i in get_all_items(db)} assert set(items) == {'Kabinetsvergadering GMT', 'Najaarsconventie'} assert items['Kabinetsvergadering GMT']['status'] == 'voorbereid' # BEHOUDEN assert items['Kabinetsvergadering GMT']['locatie'] == 'Amersfoort' # wél bijgewerkt assert items['Kabinetsvergadering GMT']['beschrijving'] == 'v2' # derde keer zelfde set: volledig idempotent rep3 = sync_agenda_from_items(db, items_v2) assert rep3['toegevoegd'] == 0 and rep3['verwijderd'] == 0 assert len(get_all_items(db)) == 2 # ── 4. Seed-lader ───────────────────────────────────────────────────────────── def test_seed_load(db): assert os.path.exists(SEED_PATH) rep = load_agenda_seed(db, SEED_PATH) assert rep['toegevoegd'] == 6 items = get_all_items(db) assert len(items) == 6 by_titel = {i['titel']: i for i in items} assert by_titel['Nationale najaarsconventie MD110']['datum'] == '2026-11-21' assert by_titel['Start marketingcampagne ledenwerving 110 AN']['datum_eind'] == '2026-09-30' assert by_titel['Kabinetsvergadering GMT & Wereld Alzheimerdag']['beschrijving'].startswith('Aftrap') assert all(i['status'] == 'open' for i in items) # opnieuw seeden over bestaande set: geen duplicaten, statussen behouden conn = db.get_db() conn.execute("UPDATE agenda_items SET status='klaar' WHERE titel LIKE 'Kabinetsvergadering%'") conn.commit(); conn.close() rep2 = load_agenda_seed(db, SEED_PATH) assert rep2['toegevoegd'] == 0 items = get_all_items(db) assert len(items) == 6 # nog steeds 6, geen dubbelen assert sum(1 for i in items if i['status'] == 'klaar') == 2 # statussen bewaard # ── 5. Alerts / dashboard-voorloper Fase B ──────────────────────────────────── def test_upcoming_items_badges(db): conn = db.get_db() now = date.today() for delta, titel in [(1, 'T-1 item'), (3, 'T-3 item'), (5, 'T-5 item'), (10, 'T-10 item'), (25, 'Buiten horizon')]: conn.execute("INSERT INTO agenda_items (datum, titel) VALUES (?,?)", ((now + timedelta(days=delta)).isoformat(), titel)) # lopende campagne: begon eerder, eindigt binnenkort conn.execute("INSERT INTO agenda_items (datum, datum_eind, titel) VALUES (?,?,?)", ((now - timedelta(days=2)).isoformat(), (now + timedelta(days=5)).isoformat(), 'Lopende campagne')) conn.commit(); conn.close() up = get_upcoming_items(db, days=14) titels = {i['titel']: i for i in up} assert 'Buiten horizon' not in titels assert titels['T-1 item']['badge'] == 'T-2' assert titels['T-3 item']['badge'] == 'T-7' # 3 > 2 dus T-7 assert titels['T-5 item']['badge'] == 'T-7' assert titels['T-10 item']['badge'] == 'T-14' assert titels['Lopende campagne']['badge'] == 'lopend' def test_alerts_html_rendering(db): html = render_agenda_alerts_html(db, BASE_PATH='/lgcc', days=14) assert html == '' # lege agenda -> geen card conn = db.get_db() conn.execute("INSERT INTO agenda_items (datum, titel, type, locatie) VALUES (?,?,?,?)", ((date.today() + timedelta(days=5)).isoformat(), 'Clubbezoek Lions Almere', 'clubbezoek', 'Almere')) conn.commit(); conn.close() html = render_agenda_alerts_html(db, BASE_PATH='/lgcc', days=14) assert 'Komende agendapunten' in html assert 'Clubbezoek Lions Almere' in html assert 'T-7' in html # oranje badge assert '#F76707' in html or '#E8590C' in html # oranje kleur assert '/lgcc/agenda' in html # link naar volledige agenda # ── 6. Parser los (unit) ───────────────────────────────────────────────────── def test_parse_invalid_file(): items, fouten = parse_agenda_xlsx(io.BytesIO(b'geen excel')) assert items == [] def test_migration_idempotent(db): """init_agenda_db draaide al 2x in de fixture zonder fout — controleer tabelstructuur.""" conn = db.get_db() cols = [r['name'] for r in conn.execute("PRAGMA table_info(agenda_items)").fetchall()] conn.close() assert cols == ['id', 'datum', 'datum_eind', 'titel', 'type', 'locatie', 'beschrijving', 'status', 'aangemaakt', 'bijgewerkt']