|
|
@@ -1,15 +1,16 @@
|
|
|
"""Dashboard and exact drill-down contracts on the shared production SQL projection."""
|
|
|
-from datetime import datetime
|
|
|
-from collections import namedtuple
|
|
|
-from decimal import Decimal
|
|
|
+
|
|
|
import json
|
|
|
import re
|
|
|
import unittest
|
|
|
+from collections import namedtuple
|
|
|
+from datetime import datetime
|
|
|
+from decimal import Decimal
|
|
|
from unittest.mock import patch
|
|
|
-from urllib.parse import parse_qs, urlsplit, urlencode
|
|
|
+from urllib.parse import parse_qs, urlencode, urlsplit
|
|
|
|
|
|
-from app.dashboard_segment import parse_segment, pack_segment, segment_sql
|
|
|
-from app.forderung_dashboard import build_dashboard, BERLIN, color
|
|
|
+from app.dashboard_segment import pack_segment, parse_segment, segment_sql
|
|
|
+from app.forderung_dashboard import BERLIN, build_dashboard, color
|
|
|
from app.navigation import list_return_url, normalize_page_url
|
|
|
from tests import test_verursacher as fixture
|
|
|
|
|
|
@@ -21,12 +22,17 @@ class DashboardTests(unittest.TestCase):
|
|
|
self.addCleanup(self.fixture.doCleanups)
|
|
|
self.client, self.db = self.fixture.client, self.fixture.db
|
|
|
from app.jinja_templates import templates
|
|
|
- sql = templates.env.get_template('forderungen/liste/queries/forderungen_liste.sql').render()
|
|
|
- needed = set(re.findall(r'(?:\[F\]|F)\.\[([^]]+)\]', sql))
|
|
|
- existing = {row[1] for row in self.db.connection.execute('PRAGMA dbo.table_info(Forderungen_Kopf)')}
|
|
|
+
|
|
|
+ sql = templates.env.get_template("forderungen/liste/queries/forderungen_liste.sql").render()
|
|
|
+ needed = set(re.findall(r"(?:\[F\]|F)\.\[([^]]+)\]", sql))
|
|
|
+ existing = {row[1] for row in self.db.connection.execute("PRAGMA dbo.table_info(Forderungen_Kopf)")}
|
|
|
for name in needed - existing:
|
|
|
- kind = 'INTEGER' if name == 'Beleg_Tage_offen' else 'REAL' if name in {'Beleg_Betrag','Kunde_offen_gesamt'} else 'TEXT'
|
|
|
- self.db.connection.execute(f'ALTER TABLE dbo.Forderungen_Kopf ADD [{name}] {kind}')
|
|
|
+ kind = (
|
|
|
+ "INTEGER"
|
|
|
+ if name == "Beleg_Tage_offen"
|
|
|
+ else "REAL" if name in {"Beleg_Betrag", "Kunde_offen_gesamt"} else "TEXT"
|
|
|
+ )
|
|
|
+ self.db.connection.execute(f"ALTER TABLE dbo.Forderungen_Kopf ADD [{name}] {kind}")
|
|
|
self.db.connection.executescript("""
|
|
|
ALTER TABLE dbo.Forderungen_Details ADD Rechtsanwalt TEXT;
|
|
|
ALTER TABLE dbo.Forderungen_Details ADD Reklamation TEXT;
|
|
|
@@ -35,7 +41,7 @@ class DashboardTests(unittest.TestCase):
|
|
|
CREATE TABLE Garantie2 (Client_DB TEXT, Beleg_Nr TEXT, Antrag_Art TEXT, Antrag_Nr TEXT, Antrag_Datum TEXT, Antrag_Benutzer_ID TEXT, Antrag_Status TEXT, Letzte_Beantragung TEXT);
|
|
|
UPDATE dbo.Forderungen_Kopf SET Standort_Name='ULM', Hauptbetrieb_Name='AHR',
|
|
|
Staffel_Text='> 12 Wochen', Beleg_Tage_offen=100, Rechnung_Gutschrift='Rechnung',
|
|
|
- Beleg_Faellig_Datum='2026-11-02', Beleg_Betrag=Beleg_offen, Kunde_offen_gesamt=100;
|
|
|
+ Beleg_Faellig_Datum='2026-11-02', Beleg_Betrag=Beleg_Offen, Kunde_offen_gesamt=100;
|
|
|
UPDATE dbo.Forderungen_Kopf SET Hauptbetrieb_ID='2', Hauptbetrieb_Name='AAM' WHERE Client_DB='2';
|
|
|
UPDATE dbo.Forderungen_Kopf SET Staffel_Text='< 4 Wochen', Beleg_Tage_offen=10 WHERE Beleg_Nr='R1';
|
|
|
INSERT INTO dbo.Forderungen_Details (Client_DB,Beleg_Nr,Verursacher_Benutzer_ID,Wiedervorlage) VALUES
|
|
|
@@ -46,250 +52,315 @@ class DashboardTests(unittest.TestCase):
|
|
|
INSERT INTO dbo.Benutzer_Rechte VALUES ('tester','1','10','Service');
|
|
|
INSERT INTO dbo.Benutzer_Rechte VALUES ('tester','1','10','');
|
|
|
""")
|
|
|
- columns = [row[1] for row in self.db.connection.execute('PRAGMA dbo.table_info(Forderungen_Kopf)')]
|
|
|
- select = ','.join("'CR'" if name == 'Beleg_Nr' else '-40' if name == 'Beleg_offen' else '['+name+']' for name in columns)
|
|
|
- self.db.connection.execute('INSERT INTO dbo.Forderungen_Kopf SELECT '+select+" FROM dbo.Forderungen_Kopf WHERE Client_DB='1' AND Beleg_Nr='NV'")
|
|
|
- self.db.connection.execute("UPDATE dbo.Forderungen_Kopf SET Rechnung_Gutschrift='Gutschrift', Abteilung='', Staffel_Text=NULL, Verursacher_Benutzer_ID=NULL, Verursacher_Text=NULL, Beleg_Faellig_Datum=NULL WHERE Beleg_Nr='CR'")
|
|
|
+ columns = [row[1] for row in self.db.connection.execute("PRAGMA dbo.table_info(Forderungen_Kopf)")]
|
|
|
+ select = ",".join(
|
|
|
+ "'CR'" if name == "Beleg_Nr" else "-40" if name == "Beleg_Offen" else "[" + name + "]" for name in columns
|
|
|
+ )
|
|
|
+ self.db.connection.execute(
|
|
|
+ "INSERT INTO dbo.Forderungen_Kopf SELECT "
|
|
|
+ + select
|
|
|
+ + " FROM dbo.Forderungen_Kopf WHERE Client_DB='1' AND Beleg_Nr='NV'"
|
|
|
+ )
|
|
|
+ self.db.connection.execute(
|
|
|
+ "UPDATE dbo.Forderungen_Kopf SET Rechnung_Gutschrift='Gutschrift', Abteilung='', Staffel_Text=NULL, Verursacher_Benutzer_ID=NULL, Verursacher_Text=NULL, Beleg_Faellig_Datum=NULL WHERE Beleg_Nr='CR'"
|
|
|
+ )
|
|
|
self.db.connection.commit()
|
|
|
execute = self.db.execute
|
|
|
+
|
|
|
def dialect(query, values=None):
|
|
|
query = str(query)
|
|
|
- for column in ['Kommentar_Fibu', 'Kommentar_Abteilung']:
|
|
|
- query = query.replace(f"left(isnull([FK].[{column}], ''), 50)", f"substr(ifnull([FK].[{column}], ''), 1, 50)")
|
|
|
- query = query.replace('isnull(', 'ifnull(').replace('dateadd(day, 4, getdate())', "'2026-11-06'").replace('getdate()', "'2026-11-02'")
|
|
|
- if query.strip().startswith('SELECT F.* FROM'):
|
|
|
+ for column in ["Kommentar_Fibu", "Kommentar_Abteilung"]:
|
|
|
+ query = query.replace(
|
|
|
+ f"left(isnull([FK].[{column}], ''), 50)", f"substr(ifnull([FK].[{column}], ''), 1, 50)"
|
|
|
+ )
|
|
|
+ query = (
|
|
|
+ query.replace("isnull(", "ifnull(")
|
|
|
+ .replace("dateadd(day, 4, getdate())", "'2026-11-06'")
|
|
|
+ .replace("getdate()", "'2026-11-02'")
|
|
|
+ )
|
|
|
+ if query.strip().startswith("SELECT F.* FROM"):
|
|
|
self.db.calls.append((query, values))
|
|
|
cursor = self.db.connection.execute(query, values or {})
|
|
|
columns = [item[0] for item in cursor.description]
|
|
|
- record = namedtuple('DashboardInvoice', columns)
|
|
|
+ record = namedtuple("DashboardInvoice", columns)
|
|
|
rows = []
|
|
|
for raw in cursor.fetchall():
|
|
|
row = dict(zip(columns, raw))
|
|
|
- for name in ['Beleg_Datum','Beleg_Faellig_Datum','Wiedervorlage']:
|
|
|
- if row.get(name): row[name] = datetime.fromisoformat(row[name])
|
|
|
+ for name in ["Beleg_Datum", "Beleg_Faellig_Datum", "Wiedervorlage"]:
|
|
|
+ if row.get(name):
|
|
|
+ row[name] = datetime.fromisoformat(row[name])
|
|
|
rows.append(record(**row))
|
|
|
+
|
|
|
class SQLRows:
|
|
|
- def keys(self): return columns
|
|
|
- def fetchall(self): return rows
|
|
|
+ def keys(self):
|
|
|
+ return columns
|
|
|
+
|
|
|
+ def fetchall(self):
|
|
|
+ return rows
|
|
|
+
|
|
|
return SQLRows()
|
|
|
return execute(query, values)
|
|
|
- adapter = patch.object(self.db, 'execute', side_effect=dialect)
|
|
|
- adapter.start(); self.addCleanup(adapter.stop)
|
|
|
+
|
|
|
+ adapter = patch.object(self.db, "execute", side_effect=dialect)
|
|
|
+ adapter.start()
|
|
|
+ self.addCleanup(adapter.stop)
|
|
|
self.now = datetime(2026, 11, 2, 10, tzinfo=BERLIN)
|
|
|
|
|
|
def data(self):
|
|
|
- return build_dashboard(self.db, 'tester', self.now)
|
|
|
+ return build_dashboard(self.db, "tester", self.now)
|
|
|
|
|
|
def test_totals_rights_duplicate_roles_and_duplicate_join_rows(self):
|
|
|
data = self.data()
|
|
|
- self.assertEqual(data['count'], 5)
|
|
|
- self.assertEqual(data['amount'], '200.00')
|
|
|
- self.assertEqual(data['amount_display'], '200,00')
|
|
|
- for groups in data['views'].values():
|
|
|
- self.assertEqual(sum(group['count'] for group in groups), 5)
|
|
|
- self.assertEqual(sum(Decimal(group['amount']) for group in groups), Decimal(200))
|
|
|
+ self.assertEqual(data["count"], 5)
|
|
|
+ self.assertEqual(data["amount"], "200.00")
|
|
|
+ self.assertEqual(data["amount_display"], "200,00")
|
|
|
+ for groups in data["views"].values():
|
|
|
+ self.assertEqual(sum(group["count"] for group in groups), 5)
|
|
|
+ self.assertEqual(sum(Decimal(group["amount"]) for group in groups), Decimal(200))
|
|
|
sql, values = self.db.calls[-1]
|
|
|
- self.assertIn('WHERE EXISTS', sql)
|
|
|
- self.assertEqual(values['auth_username'], 'tester')
|
|
|
- self.assertTrue(all('AAM' not in group['label'] for group in data['views']['Standort']))
|
|
|
+ self.assertIn("WHERE EXISTS", sql)
|
|
|
+ self.assertEqual(values["auth_username"], "tester")
|
|
|
+ self.assertTrue(all("AAM" not in group["label"] for group in data["views"]["Standort"]))
|
|
|
|
|
|
def test_only_invoices_filters_all_aggregates_and_drilldowns(self):
|
|
|
- data = build_dashboard(self.db, 'tester', self.now, only_invoices=True)
|
|
|
- self.assertEqual(data['count'], 4)
|
|
|
- self.assertEqual(data['amount'], '240.00')
|
|
|
- self.assertEqual(self.data()['count'], 5)
|
|
|
- for groups in data['views'].values():
|
|
|
- self.assertEqual(sum(group['count'] for group in groups), 4)
|
|
|
+ data = build_dashboard(self.db, "tester", self.now, only_invoices=True)
|
|
|
+ self.assertEqual(data["count"], 4)
|
|
|
+ self.assertEqual(data["amount"], "240.00")
|
|
|
+ self.assertEqual(self.data()["count"], 5)
|
|
|
+ for groups in data["views"].values():
|
|
|
+ self.assertEqual(sum(group["count"] for group in groups), 4)
|
|
|
for group in groups:
|
|
|
- for item in [group, *group['segments']]:
|
|
|
- raw = parse_qs(urlsplit(item['url']).query)['dashboard_segment'][0]
|
|
|
- self.assertTrue(parse_segment(raw)['only_invoices'])
|
|
|
- response = self.client.get(item['url'])
|
|
|
- self.assertIn('id="list-count">'+str(item['count']), response.text)
|
|
|
- self.assertIn('Nur Rechnungen', response.text)
|
|
|
- self.db.connection.execute("UPDATE dbo.Forderungen_Kopf SET Rechnung_Gutschrift='Rechnungskorrektur' WHERE Beleg_Nr='CR'")
|
|
|
- self.db.connection.execute("UPDATE dbo.Forderungen_Kopf SET Beleg_offen=-100 WHERE Client_DB='1' AND Beleg_Nr='R1'")
|
|
|
- negative = build_dashboard(self.db, 'tester', self.now, only_invoices=True)
|
|
|
- self.assertEqual(negative['count'], 4)
|
|
|
- self.assertEqual(negative['amount'], '40.00')
|
|
|
+ for item in [group, *group["segments"]]:
|
|
|
+ raw = parse_qs(urlsplit(item["url"]).query)["dashboard_segment"][0]
|
|
|
+ self.assertTrue(parse_segment(raw)["only_invoices"])
|
|
|
+ response = self.client.get(item["url"])
|
|
|
+ self.assertIn('id="list-count">' + str(item["count"]), response.text)
|
|
|
+ self.assertIn("Nur Rechnungen", response.text)
|
|
|
+ self.db.connection.execute(
|
|
|
+ "UPDATE dbo.Forderungen_Kopf SET Rechnung_Gutschrift='Rechnungskorrektur' WHERE Beleg_Nr='CR'"
|
|
|
+ )
|
|
|
+ self.db.connection.execute(
|
|
|
+ "UPDATE dbo.Forderungen_Kopf SET Beleg_Offen=-100 WHERE Client_DB='1' AND Beleg_Nr='R1'"
|
|
|
+ )
|
|
|
+ negative = build_dashboard(self.db, "tester", self.now, only_invoices=True)
|
|
|
+ self.assertEqual(negative["count"], 4)
|
|
|
+ self.assertEqual(negative["amount"], "40.00")
|
|
|
|
|
|
def test_invoice_checkbox_url_state_and_strict_segment_flag(self):
|
|
|
- response = self.client.get('/app/forderungen/dashboard?page=Verursacher&metric=anzahl&only_invoices=true')
|
|
|
+ response = self.client.get("/app/forderungen/dashboard?page=Verursacher&metric=anzahl&only_invoices=true")
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
self.assertIn('name="only_invoices" value="true" checked', response.text)
|
|
|
self.assertIn('name="page" value="Verursacher"', response.text)
|
|
|
self.assertIn('name="metric" value="anzahl"', response.text)
|
|
|
- self.assertIn('240,00', response.text)
|
|
|
- unchecked = self.client.get('/app/forderungen/dashboard')
|
|
|
+ self.assertIn("240,00", response.text)
|
|
|
+ unchecked = self.client.get("/app/forderungen/dashboard")
|
|
|
self.assertNotIn('name="only_invoices" value="true" checked', unchecked.text)
|
|
|
- self.assertIn('200,00', unchecked.text)
|
|
|
+ self.assertIn("200,00", unchecked.text)
|
|
|
self.assertIn('data-dashboard-view="Standort" data-dashboard-metric=', unchecked.text)
|
|
|
- for invalid in ['true', 1, None]:
|
|
|
+ for invalid in ["true", 1, None]:
|
|
|
with self.assertRaises(ValueError):
|
|
|
- parse_segment(pack_segment({'view':'Altersstaffel','key':None,'only_invoices':invalid}))
|
|
|
+ parse_segment(pack_segment({"view": "Altersstaffel", "key": None, "only_invoices": invalid}))
|
|
|
|
|
|
def test_all_segments_round_trip_to_exact_list_count_and_amount(self):
|
|
|
- for view, groups in self.data()['views'].items():
|
|
|
+ for view, groups in self.data()["views"].items():
|
|
|
for group in groups:
|
|
|
- for segment in group['segments']:
|
|
|
- with self.subTest(view=view, label=group['label'], department=segment['department']):
|
|
|
- response = self.client.get(segment['url'])
|
|
|
+ for segment in group["segments"]:
|
|
|
+ with self.subTest(view=view, label=group["label"], department=segment["department"]):
|
|
|
+ response = self.client.get(segment["url"])
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
count = int(re.search(r'id="list-count">(\d+)', response.text)[1])
|
|
|
- self.assertEqual(count, segment['count'])
|
|
|
+ self.assertEqual(count, segment["count"])
|
|
|
self.assertIn('name="dashboard_segment"', response.text)
|
|
|
- self.assertIn('Dashboardfilter entfernen', response.text)
|
|
|
- self.assertIn('Aktive Filter (1)', response.text)
|
|
|
+ self.assertIn("Dashboardfilter entfernen", response.text)
|
|
|
+ self.assertIn("Aktive Filter (1)", response.text)
|
|
|
from app.jinja_templates import number_format
|
|
|
- self.assertIn(number_format(Decimal(segment['amount']))+' €', response.text)
|
|
|
+
|
|
|
+ self.assertIn(number_format(Decimal(segment["amount"])) + " €", response.text)
|
|
|
|
|
|
def test_follow_up_boundaries_and_null_group(self):
|
|
|
- cases = [('2026-11-01', 'overdue'), ('2026-11-02', 'soon'), ('2026-11-05 23:30:00', 'soon'),
|
|
|
- ('2026-11-06', 'not_due'), ('2026-11-16 23:30:00', 'not_due'), ('2026-11-17', 'deferred')]
|
|
|
+ cases = [
|
|
|
+ ("2026-11-01", "overdue"),
|
|
|
+ ("2026-11-02", "soon"),
|
|
|
+ ("2026-11-05 23:30:00", "soon"),
|
|
|
+ ("2026-11-06", "not_due"),
|
|
|
+ ("2026-11-16 23:30:00", "not_due"),
|
|
|
+ ("2026-11-17", "deferred"),
|
|
|
+ ]
|
|
|
for value, expected in cases:
|
|
|
with self.subTest(date=value):
|
|
|
- self.db.connection.execute("UPDATE dbo.Forderungen_Details SET Wiedervorlage=? WHERE Beleg_Nr='R1'", (value,))
|
|
|
- data = build_dashboard(self.db, 'tester', self.now, stack='wiedervorlage')
|
|
|
- group = next(group for group in data['views']['Altersstaffel'] if group['label']=='< 4 Wochen')
|
|
|
- self.assertEqual(group['segments'][0]['id'], expected)
|
|
|
- response = self.client.get(group['segments'][0]['url'])
|
|
|
+ self.db.connection.execute(
|
|
|
+ "UPDATE dbo.Forderungen_Details SET Wiedervorlage=? WHERE Beleg_Nr='R1'", (value,)
|
|
|
+ )
|
|
|
+ data = build_dashboard(self.db, "tester", self.now, stack="wiedervorlage")
|
|
|
+ group = next(group for group in data["views"]["Altersstaffel"] if group["label"] == "< 4 Wochen")
|
|
|
+ self.assertEqual(group["segments"][0]["id"], expected)
|
|
|
+ response = self.client.get(group["segments"][0]["url"])
|
|
|
self.assertIn('id="list-count">1', response.text)
|
|
|
- ids = [item['id'] for item in data['stack_items']]
|
|
|
- self.assertEqual(ids, ['overdue','soon','not_due','deferred','undated'])
|
|
|
- self.assertNotIn('Wiedervorlage', data['views'])
|
|
|
+ ids = [item["id"] for item in data["stack_items"]]
|
|
|
+ self.assertEqual(ids, ["overdue", "soon", "not_due", "deferred", "undated"])
|
|
|
+ self.assertNotIn("Wiedervorlage", data["views"])
|
|
|
|
|
|
def test_follow_up_falls_back_to_document_due_date(self):
|
|
|
self.db.connection.execute("UPDATE dbo.Forderungen_Details SET Wiedervorlage=NULL WHERE Beleg_Nr='R1'")
|
|
|
- self.db.connection.execute("UPDATE dbo.Forderungen_Kopf SET Beleg_Faellig_Datum='2026-11-17' WHERE Client_DB='1' AND Beleg_Nr='R1'")
|
|
|
- data = build_dashboard(self.db, 'tester', self.now, stack='wiedervorlage')
|
|
|
- group = next(group for group in data['views']['Altersstaffel'] if group['label']=='< 4 Wochen')
|
|
|
- self.assertEqual(group['segments'][0]['id'], 'deferred')
|
|
|
- self.assertIn('id="list-count">1', self.client.get(group['segments'][0]['url']).text)
|
|
|
+ self.db.connection.execute(
|
|
|
+ "UPDATE dbo.Forderungen_Kopf SET Beleg_Faellig_Datum='2026-11-17' WHERE Client_DB='1' AND Beleg_Nr='R1'"
|
|
|
+ )
|
|
|
+ data = build_dashboard(self.db, "tester", self.now, stack="wiedervorlage")
|
|
|
+ group = next(group for group in data["views"]["Altersstaffel"] if group["label"] == "< 4 Wochen")
|
|
|
+ self.assertEqual(group["segments"][0]["id"], "deferred")
|
|
|
+ self.assertIn('id="list-count">1', self.client.get(group["segments"][0]["url"]).text)
|
|
|
|
|
|
def test_follow_up_year_boundary_and_berlin_calendar_day(self):
|
|
|
self.db.connection.execute("UPDATE dbo.Forderungen_Details SET Wiedervorlage='2027-01-01' WHERE Beleg_Nr='R1'")
|
|
|
- now = datetime.fromisoformat('2026-12-28T23:30:00+00:00')
|
|
|
- groups = build_dashboard(self.db, 'tester', now, stack='wiedervorlage')['views']['Altersstaffel']
|
|
|
- group = next(group for group in groups if group['label']=='< 4 Wochen')
|
|
|
- selection = parse_segment(parse_qs(urlsplit(group['segments'][0]['url']).query)['dashboard_segment'][0])
|
|
|
- self.assertEqual(selection['follow_up'], {'from':'2026-12-29','to':'2027-01-01'})
|
|
|
- self.assertEqual(group['segments'][0]['id'], 'soon')
|
|
|
+ now = datetime.fromisoformat("2026-12-28T23:30:00+00:00")
|
|
|
+ groups = build_dashboard(self.db, "tester", now, stack="wiedervorlage")["views"]["Altersstaffel"]
|
|
|
+ group = next(group for group in groups if group["label"] == "< 4 Wochen")
|
|
|
+ selection = parse_segment(parse_qs(urlsplit(group["segments"][0]["url"]).query)["dashboard_segment"][0])
|
|
|
+ self.assertEqual(selection["follow_up"], {"from": "2026-12-29", "to": "2027-01-01"})
|
|
|
+ self.assertEqual(group["segments"][0]["id"], "soon")
|
|
|
|
|
|
def test_follow_up_stack_roundtrips_for_all_rubrics_and_invoice_scope(self):
|
|
|
for only in [False, True]:
|
|
|
- data = build_dashboard(self.db, 'tester', self.now, only_invoices=only, stack='wiedervorlage')
|
|
|
- for groups in data['views'].values():
|
|
|
- self.assertEqual(sum(group['count'] for group in groups), data['count'])
|
|
|
- self.assertEqual(sum(Decimal(group['amount']) for group in groups), Decimal(data['amount']))
|
|
|
+ data = build_dashboard(self.db, "tester", self.now, only_invoices=only, stack="wiedervorlage")
|
|
|
+ for groups in data["views"].values():
|
|
|
+ self.assertEqual(sum(group["count"] for group in groups), data["count"])
|
|
|
+ self.assertEqual(sum(Decimal(group["amount"]) for group in groups), Decimal(data["amount"]))
|
|
|
for group in groups:
|
|
|
- for segment in group['segments']:
|
|
|
- selection = parse_segment(parse_qs(urlsplit(segment['url']).query)['dashboard_segment'][0])
|
|
|
- self.assertIn('follow_up', selection)
|
|
|
- self.assertNotIn('department', selection)
|
|
|
- response = self.client.get(segment['url'])
|
|
|
+ for segment in group["segments"]:
|
|
|
+ selection = parse_segment(parse_qs(urlsplit(segment["url"]).query)["dashboard_segment"][0])
|
|
|
+ self.assertIn("follow_up", selection)
|
|
|
+ self.assertNotIn("department", selection)
|
|
|
+ response = self.client.get(segment["url"])
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
- self.assertIn('id="list-count">'+str(segment['count']), response.text)
|
|
|
+ self.assertIn('id="list-count">' + str(segment["count"]), response.text)
|
|
|
|
|
|
def test_stack_selection_and_legacy_follow_up_links(self):
|
|
|
- response = self.client.get('/app/forderungen/dashboard?stack=wiedervorlage&metric=anzahl&only_invoices=true')
|
|
|
+ response = self.client.get("/app/forderungen/dashboard?stack=wiedervorlage&metric=anzahl&only_invoices=true")
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
self.assertNotIn('data-dashboard-view="Wiedervorlage"', response.text)
|
|
|
self.assertIn('value="wiedervorlage" selected', response.text)
|
|
|
self.assertIn('name="stack" value="wiedervorlage"', response.text)
|
|
|
- self.assertEqual(self.client.get('/app/forderungen/dashboard?stack=bad').status_code, 422)
|
|
|
- old = self.client.get('/app/forderungen/dashboard?page=Wiedervorlage&metric=anzahl&only_invoices=true', follow_redirects=False)
|
|
|
+ self.assertEqual(self.client.get("/app/forderungen/dashboard?stack=bad").status_code, 422)
|
|
|
+ old = self.client.get(
|
|
|
+ "/app/forderungen/dashboard?page=Wiedervorlage&metric=anzahl&only_invoices=true", follow_redirects=False
|
|
|
+ )
|
|
|
self.assertEqual(old.status_code, 303)
|
|
|
- params = parse_qs(urlsplit(old.headers['location']).query)
|
|
|
- self.assertEqual(params, {'page':['Standort'], 'metric':['anzahl'], 'stack':['wiedervorlage'], 'only_invoices':['true']})
|
|
|
- legacy = pack_segment({'view':'Wiedervorlage','key':{'from':'2026-11-02','to':'2026-11-09'}})
|
|
|
- response = self.client.get('/app/forderungen/liste', params={'dashboard_segment':legacy})
|
|
|
+ params = parse_qs(urlsplit(old.headers["location"]).query)
|
|
|
+ self.assertEqual(
|
|
|
+ params, {"page": ["Standort"], "metric": ["anzahl"], "stack": ["wiedervorlage"], "only_invoices": ["true"]}
|
|
|
+ )
|
|
|
+ legacy = pack_segment({"view": "Wiedervorlage", "key": {"from": "2026-11-02", "to": "2026-11-09"}})
|
|
|
+ response = self.client.get("/app/forderungen/liste", params={"dashboard_segment": legacy})
|
|
|
self.assertIn('id="list-count">2', response.text)
|
|
|
- for bad in [True, 'tomorrow', {'from':'2026-11-02','to':'2026-11-01'}]:
|
|
|
+ for bad in [True, "tomorrow", {"from": "2026-11-02", "to": "2026-11-01"}]:
|
|
|
with self.assertRaises(ValueError):
|
|
|
- parse_segment(pack_segment({'view':'Altersstaffel','key':None,'follow_up':bad}))
|
|
|
+ parse_segment(pack_segment({"view": "Altersstaffel", "key": None, "follow_up": bad}))
|
|
|
|
|
|
def test_same_names_different_user_ids_and_clients_remain_separate(self):
|
|
|
- groups = [group for group in self.data()['views']['Verursacher'] if group['label'].startswith('Alice (')]
|
|
|
+ groups = [group for group in self.data()["views"]["Verursacher"] if group["label"].startswith("Alice (")]
|
|
|
self.assertEqual(len(groups), 2)
|
|
|
- self.assertNotEqual(groups[0]['label'], groups[1]['label'])
|
|
|
- self.assertNotEqual(groups[0]['url'], groups[1]['url'])
|
|
|
+ self.assertNotEqual(groups[0]["label"], groups[1]["label"])
|
|
|
+ self.assertNotEqual(groups[0]["url"], groups[1]["url"])
|
|
|
self.db.connection.execute("INSERT INTO dbo.Benutzer_Rechte VALUES ('tester','2','10','Service')")
|
|
|
- groups = self.data()['views']['Standort']
|
|
|
+ groups = self.data()["views"]["Standort"]
|
|
|
self.assertEqual(len(groups), 2)
|
|
|
for group in groups:
|
|
|
- self.assertEqual(self.client.get(group['url']).status_code, 200)
|
|
|
- self.assertEqual({group['label'] for group in groups}, {'ULM (AHR)', 'ULM (AAM)'})
|
|
|
+ self.assertEqual(self.client.get(group["url"]).status_code, 200)
|
|
|
+ self.assertEqual({group["label"] for group in groups}, {"ULM (AHR)", "ULM (AAM)"})
|
|
|
|
|
|
def test_decimal_sum_and_zero_amount_segments_still_have_links(self):
|
|
|
- self.db.connection.execute("UPDATE dbo.Forderungen_Kopf SET Beleg_offen=0.1 WHERE Client_DB='1' AND Beleg_Nr='R1'")
|
|
|
- self.db.connection.execute("UPDATE dbo.Forderungen_Kopf SET Beleg_offen=0.2 WHERE Client_DB='1' AND Beleg_Nr='X1'")
|
|
|
+ self.db.connection.execute(
|
|
|
+ "UPDATE dbo.Forderungen_Kopf SET Beleg_Offen=0.1 WHERE Client_DB='1' AND Beleg_Nr='R1'"
|
|
|
+ )
|
|
|
+ self.db.connection.execute(
|
|
|
+ "UPDATE dbo.Forderungen_Kopf SET Beleg_Offen=0.2 WHERE Client_DB='1' AND Beleg_Nr='X1'"
|
|
|
+ )
|
|
|
data = self.data()
|
|
|
- self.assertEqual(data['amount'], '10.30')
|
|
|
- zero = build_dashboard(self.db, 'tester', self.now, stack='wiedervorlage')
|
|
|
- self.assertTrue(any(segment['amount']=='0.00' and segment['url'] for group in zero['views']['Verursacher'] for segment in group['segments']))
|
|
|
- self.assertEqual(color('Service'), color('Service'))
|
|
|
+ self.assertEqual(data["amount"], "10.30")
|
|
|
+ zero = build_dashboard(self.db, "tester", self.now, stack="wiedervorlage")
|
|
|
+ self.assertTrue(
|
|
|
+ any(
|
|
|
+ segment["amount"] == "0.00" and segment["url"]
|
|
|
+ for group in zero["views"]["Verursacher"]
|
|
|
+ for segment in group["segments"]
|
|
|
+ )
|
|
|
+ )
|
|
|
+ self.assertEqual(color("Service"), color("Service"))
|
|
|
|
|
|
def test_exact_department_filter_does_not_match_prefixes(self):
|
|
|
self.db.connection.execute("UPDATE dbo.Forderungen_Kopf SET Abteilung='Service2' WHERE Beleg_Nr='CR'")
|
|
|
self.db.connection.execute("INSERT INTO dbo.Benutzer_Rechte VALUES ('tester','1','10','Service2')")
|
|
|
- group = self.data()['views']['Standort'][0]
|
|
|
- segment = next(segment for segment in group['segments'] if segment['department']=='Service')
|
|
|
- exact = self.client.get(segment['url'])
|
|
|
- partial = self.client.get('/app/forderungen/liste?Abteilung=Service')
|
|
|
+ group = self.data()["views"]["Standort"][0]
|
|
|
+ segment = next(segment for segment in group["segments"] if segment["department"] == "Service")
|
|
|
+ exact = self.client.get(segment["url"])
|
|
|
+ partial = self.client.get("/app/forderungen/liste?Abteilung=Service")
|
|
|
self.assertIn('id="list-count">4', exact.text)
|
|
|
self.assertIn('id="list-count">5', partial.text)
|
|
|
|
|
|
def test_forged_segment_cannot_expand_invoice_access(self):
|
|
|
- raw = pack_segment({'view':'Standort','key':{'client':'2','site':'10'}})
|
|
|
- response = self.client.get('/app/forderungen/liste', params={'dashboard_segment':raw})
|
|
|
+ raw = pack_segment({"view": "Standort", "key": {"client": "2", "site": "10"}})
|
|
|
+ response = self.client.get("/app/forderungen/liste", params={"dashboard_segment": raw})
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
self.assertIn('id="list-count">0', response.text)
|
|
|
- self.assertIn('0,00 €', response.text)
|
|
|
+ self.assertIn("0,00 €", response.text)
|
|
|
|
|
|
def test_segment_preserved_through_detail_return_login_and_normal_filters(self):
|
|
|
- url = self.data()['views']['Standort'][0]['url']
|
|
|
- selection = parse_qs(urlsplit(url).query)['dashboard_segment'][0]
|
|
|
- returned = list_return_url(url, '1_R1')
|
|
|
- self.assertEqual(parse_qs(urlsplit(returned).query)['dashboard_segment'], [selection])
|
|
|
- detail = '/app/forderungen/details/1_R1?' + urlencode({'return_to': returned})
|
|
|
- recovered = normalize_page_url('/app/forderungen/details/1_R1/kommentare', current_url='https://proxy.test'+detail)
|
|
|
+ url = self.data()["views"]["Standort"][0]["url"]
|
|
|
+ selection = parse_qs(urlsplit(url).query)["dashboard_segment"][0]
|
|
|
+ returned = list_return_url(url, "1_R1")
|
|
|
+ self.assertEqual(parse_qs(urlsplit(returned).query)["dashboard_segment"], [selection])
|
|
|
+ detail = "/app/forderungen/details/1_R1?" + urlencode({"return_to": returned})
|
|
|
+ recovered = normalize_page_url(
|
|
|
+ "/app/forderungen/details/1_R1/kommentare", current_url="https://proxy.test" + detail
|
|
|
+ )
|
|
|
self.assertEqual(recovered, detail)
|
|
|
- response = self.client.get(url+'&Kunde=Test&page=1&limit=1')
|
|
|
+ response = self.client.get(url + "&Kunde=Test&page=1&limit=1")
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
self.assertIn('hx-include="#liste-filter"', response.text)
|
|
|
sql, values = self.db.calls[-1]
|
|
|
- self.assertIn('F.[Client_DB] = :ds_client', sql)
|
|
|
- self.assertEqual(values['filter_kunde'], '%Test%')
|
|
|
+ self.assertIn("F.[Client_DB] = :ds_client", sql)
|
|
|
+ self.assertEqual(values["filter_kunde"], "%Test%")
|
|
|
self.client.cookies.clear()
|
|
|
- response = self.client.get(url, headers={'HX-Request':'true'})
|
|
|
- target = parse_qs(urlsplit(response.headers['HX-Redirect']).query)['next'][0]
|
|
|
+ response = self.client.get(url, headers={"HX-Request": "true"})
|
|
|
+ target = parse_qs(urlsplit(response.headers["HX-Redirect"]).query)["next"][0]
|
|
|
self.assertEqual(target, url)
|
|
|
|
|
|
def test_bad_segments_are_rejected_before_data_query(self):
|
|
|
- invalid = ['not-json', '{}', '{"view":"Standort","key":{}}', '{"view":"Wiedervorlage","key":{"from":"x","to":null}}',
|
|
|
- '{"view":"Altersstaffel","key":null,"sql":"1=1"}', '{"view":"Verursacher","key":{"id":null}}',
|
|
|
- '{"view":"Standort","key":{"client":true,"site":"10"}}']
|
|
|
+ invalid = [
|
|
|
+ "not-json",
|
|
|
+ "{}",
|
|
|
+ '{"view":"Standort","key":{}}',
|
|
|
+ '{"view":"Wiedervorlage","key":{"from":"x","to":null}}',
|
|
|
+ '{"view":"Altersstaffel","key":null,"sql":"1=1"}',
|
|
|
+ '{"view":"Verursacher","key":{"id":null}}',
|
|
|
+ '{"view":"Standort","key":{"client":true,"site":"10"}}',
|
|
|
+ ]
|
|
|
for value in invalid:
|
|
|
with self.subTest(value=value):
|
|
|
self.db.calls.clear()
|
|
|
- response = self.client.get('/app/forderungen/liste', params={'dashboard_segment':value})
|
|
|
+ response = self.client.get("/app/forderungen/liste", params={"dashboard_segment": value})
|
|
|
self.assertEqual(response.status_code, 400)
|
|
|
self.assertFalse(self.db.calls)
|
|
|
|
|
|
def test_empty_error_and_invalid_dashboard_states(self):
|
|
|
- response = self.client.get('/app/forderungen/dashboard?page=Standort&metric=anzahl')
|
|
|
+ response = self.client.get("/app/forderungen/dashboard?page=Standort&metric=anzahl")
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
- self.assertIn('Forderungen | Dashboard', response.text)
|
|
|
+ self.assertIn("Forderungen | Dashboard", response.text)
|
|
|
self.assertIn('data-dashboard-metric="anzahl"', response.text)
|
|
|
- self.assertEqual(self.client.get('/app/forderungen/dashboard?page=bad').status_code, 422)
|
|
|
+ self.assertEqual(self.client.get("/app/forderungen/dashboard?page=bad").status_code, 422)
|
|
|
self.db.connection.execute("DELETE FROM dbo.Benutzer_Rechte WHERE Benutzer_ID='tester'")
|
|
|
data = self.data()
|
|
|
- self.assertEqual(data['count'], 0); self.assertEqual(data['amount'], '0.00')
|
|
|
- with patch('app.forderung_dashboard.read_invoices', side_effect=RuntimeError('offline')), self.assertLogs('app.routes',level='ERROR'):
|
|
|
- response = self.client.get('/app/forderungen/dashboard')
|
|
|
+ self.assertEqual(data["count"], 0)
|
|
|
+ self.assertEqual(data["amount"], "0.00")
|
|
|
+ with (
|
|
|
+ patch("app.forderung_dashboard.read_invoices", side_effect=RuntimeError("offline")),
|
|
|
+ self.assertLogs("app.routes", level="ERROR"),
|
|
|
+ ):
|
|
|
+ response = self.client.get("/app/forderungen/dashboard")
|
|
|
self.assertEqual(response.status_code, 503)
|
|
|
- self.assertIn('nicht geladen', response.text)
|
|
|
- self.assertNotIn('Anzahl der Belege', response.text)
|
|
|
+ self.assertIn("nicht geladen", response.text)
|
|
|
+ self.assertNotIn("Anzahl der Belege", response.text)
|
|
|
+
|
|
|
|
|
|
-if __name__ == '__main__':
|
|
|
+if __name__ == "__main__":
|
|
|
unittest.main()
|
gc-server3