|
|
@@ -9,10 +9,12 @@ from decimal import Decimal
|
|
|
from unittest.mock import patch
|
|
|
from urllib.parse import parse_qs, urlencode, urlsplit
|
|
|
|
|
|
-from app.dashboard_segment import pack_segment, parse_segment, segment_sql
|
|
|
+from app.dashboard_segment import pack_segment, parse_segment
|
|
|
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
|
|
|
+from tests.test_forderung_list import invoice
|
|
|
+from app.query_filter import decode_filter_values
|
|
|
|
|
|
|
|
|
class DashboardTests(unittest.TestCase):
|
|
|
@@ -24,7 +26,7 @@ class DashboardTests(unittest.TestCase):
|
|
|
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))
|
|
|
+ needed = set(invoice()._fields) | 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 = (
|
|
|
@@ -38,7 +40,7 @@ class DashboardTests(unittest.TestCase):
|
|
|
ALTER TABLE dbo.Forderungen_Details ADD Reklamation TEXT;
|
|
|
ALTER TABLE dbo.Forderungen_Details ADD Mahnen_aussetzen TEXT;
|
|
|
CREATE TABLE dbo.Forderungen_Kommentar_letzte (Client_DB TEXT, Beleg_Nr TEXT, Kommentar_Fibu TEXT, Kommentar_Abteilung TEXT, Begründung TEXT);
|
|
|
- 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);
|
|
|
+ CREATE TABLE dbo.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;
|
|
|
@@ -64,6 +66,23 @@ class DashboardTests(unittest.TestCase):
|
|
|
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'"
|
|
|
)
|
|
|
+ # The production query now reads a DWH view. Model that view in SQLite;
|
|
|
+ # keep the detail overlays live so update/authorization tests still run real SQL.
|
|
|
+ overlays = {
|
|
|
+ 'Verursacher_Benutzer_ID': 'COALESCE(D.Verursacher_Benutzer_ID, F.Verursacher_Benutzer_ID)',
|
|
|
+ 'Verursacher_Text': 'COALESCE(B.Name, D.Verursacher_Benutzer_ID, F.Verursacher_Text)',
|
|
|
+ 'Wiedervorlage': 'COALESCE(D.Wiedervorlage, F.Beleg_Faellig_Datum)',
|
|
|
+ 'Bearbeitet': "COALESCE(D.Bearbeitet, 'N')",
|
|
|
+ 'Abwarten': "CASE WHEN D.Reklamation='J' THEN 'Rekl.' WHEN D.Versicherung='J' THEN 'Vers.' "
|
|
|
+ "WHEN D.Rechtsanwalt='J' THEN 'RA' WHEN D.Mahnen_aussetzen='J' THEN 'N' ELSE '' END",
|
|
|
+ }
|
|
|
+ projection = ', '.join(f'{overlays.get(name, "F.[" + name + "]")} AS [{name}]' for name in columns)
|
|
|
+ self.db.connection.execute(
|
|
|
+ 'CREATE VIEW dbo.Forderungen_Liste AS SELECT ' + projection +
|
|
|
+ ' FROM Forderungen_Kopf F LEFT JOIN Forderungen_Details D ON F.Client_DB=D.Client_DB AND F.Beleg_Nr=D.Beleg_Nr'
|
|
|
+ ' LEFT JOIN Benutzer_Rolle B ON D.Verursacher_Benutzer_ID=B.Benutzer_ID'
|
|
|
+ ' LEFT JOIN Garantie2 G ON F.Client_DB=G.Client_DB AND F.Beleg_Nr=G.Beleg_Nr'
|
|
|
+ )
|
|
|
self.db.connection.commit()
|
|
|
execute = self.db.execute
|
|
|
|
|
|
@@ -75,6 +94,7 @@ class DashboardTests(unittest.TestCase):
|
|
|
)
|
|
|
query = (
|
|
|
query.replace("isnull(", "ifnull(")
|
|
|
+ .replace("nvarchar(max)", "TEXT")
|
|
|
.replace("dateadd(day, 4, getdate())", "'2026-11-06'")
|
|
|
.replace("getdate()", "'2026-11-02'")
|
|
|
)
|
|
|
@@ -99,7 +119,10 @@ class DashboardTests(unittest.TestCase):
|
|
|
return rows
|
|
|
|
|
|
return SQLRows()
|
|
|
- return execute(query, values)
|
|
|
+ result = execute(query, values)
|
|
|
+ if ' AS value, MIN(' in query:
|
|
|
+ result.rows = [tuple(row.values()) for row in result.rows]
|
|
|
+ return result
|
|
|
|
|
|
adapter = patch.object(self.db, "execute", side_effect=dialect)
|
|
|
adapter.start()
|
|
|
@@ -131,11 +154,12 @@ class DashboardTests(unittest.TestCase):
|
|
|
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"])
|
|
|
+ params = parse_qs(urlsplit(item["url"]).query)
|
|
|
+ self.assertEqual(params['RechnungGutschrift'], ['Rechnung'])
|
|
|
+ self.assertNotIn('dashboard_segment', params)
|
|
|
response = self.client.get(item["url"])
|
|
|
self.assertIn('id="list-count">' + str(item["count"]), response.text)
|
|
|
- self.assertIn("Nur Rechnungen", response.text)
|
|
|
+ self.assertIn('name="RechnungGutschrift" value="Rechnung"', response.text)
|
|
|
self.db.connection.execute(
|
|
|
"UPDATE dbo.Forderungen_Kopf SET Rechnung_Gutschrift='Rechnungskorrektur' WHERE Beleg_Nr='CR'"
|
|
|
)
|
|
|
@@ -170,9 +194,9 @@ class DashboardTests(unittest.TestCase):
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
count = int(re.search(r'id="list-count">(\d+)', response.text)[1])
|
|
|
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.assertNotIn('name="dashboard_segment"', response.text)
|
|
|
+ self.assertNotIn("Dashboardfilter entfernen", response.text)
|
|
|
+ self.assertIn("Aktive Filter (", response.text)
|
|
|
from app.jinja_templates import number_format
|
|
|
|
|
|
self.assertIn(number_format(Decimal(segment["amount"])) + " €", response.text)
|
|
|
@@ -215,8 +239,11 @@ class DashboardTests(unittest.TestCase):
|
|
|
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]['url'], group['url'])
|
|
|
+ params = parse_qs(urlsplit(group['url']).query)
|
|
|
+ self.assertEqual(decode_filter_values(params['Staffel'][0]), ['< 4 Wochen'])
|
|
|
+ self.assertNotIn('WiedervorlageVon', params)
|
|
|
+ self.assertNotIn('WiedervorlageBis', params)
|
|
|
self.assertEqual(group["segments"][0]["id"], "soon")
|
|
|
|
|
|
def test_follow_up_stack_roundtrips_for_all_rubrics_and_invoice_scope(self):
|
|
|
@@ -227,12 +254,15 @@ class DashboardTests(unittest.TestCase):
|
|
|
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)
|
|
|
+ params = parse_qs(urlsplit(segment['url']).query)
|
|
|
+ self.assertNotIn('dashboard_segment', params)
|
|
|
+ self.assertNotIn('WiedervorlageVon', params)
|
|
|
+ self.assertNotIn('WiedervorlageBis', params)
|
|
|
+ self.assertNotIn('Abteilung', params)
|
|
|
+ self.assertEqual(segment['url'], group['url'])
|
|
|
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(group["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")
|
|
|
@@ -251,16 +281,17 @@ class DashboardTests(unittest.TestCase):
|
|
|
)
|
|
|
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)
|
|
|
+ self.assertIn('id="list-count">5', response.text)
|
|
|
+ self.assertEqual(urlsplit(str(response.url)).query, '')
|
|
|
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}))
|
|
|
|
|
|
- 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 (")]
|
|
|
- self.assertEqual(len(groups), 2)
|
|
|
- self.assertNotEqual(groups[0]["label"], groups[1]["label"])
|
|
|
- self.assertNotEqual(groups[0]["url"], groups[1]["url"])
|
|
|
+ def test_same_names_share_normal_filter_but_clients_remain_separate(self):
|
|
|
+ groups = [group for group in self.data()["views"]["Verursacher"] if group["label"] == 'Alice']
|
|
|
+ self.assertEqual(len(groups), 1)
|
|
|
+ self.assertEqual(groups[0]['count'], 2)
|
|
|
+ self.assertEqual(parse_qs(urlsplit(groups[0]['url']).query), {'Verursacher': ['Alice']})
|
|
|
self.db.connection.execute("INSERT INTO dbo.Benutzer_Rechte VALUES ('tester','2','10','Service')")
|
|
|
groups = self.data()["views"]["Standort"]
|
|
|
self.assertEqual(len(groups), 2)
|
|
|
@@ -295,7 +326,62 @@ class DashboardTests(unittest.TestCase):
|
|
|
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)
|
|
|
+ self.assertIn('id="list-count">4', partial.text)
|
|
|
+
|
|
|
+ def test_dashboard_filters_can_be_changed_without_hidden_constraints(self):
|
|
|
+ from app.navigation import list_url
|
|
|
+ group = self.data()['views']['Standort'][0]
|
|
|
+ segment = next(item for item in group['segments'] if item['department'] == 'Service')
|
|
|
+ params = {name: values[0] for name, values in parse_qs(urlsplit(segment['url']).query).items()}
|
|
|
+ self.assertEqual(params, {'Mandant': '1', 'Filiale': '10', 'Abteilung': 'Service'})
|
|
|
+ response = self.client.get(segment['url'])
|
|
|
+ self.assertIn('id="list-count">4', response.text)
|
|
|
+ self.assertIn('data-remove="Service"', response.text)
|
|
|
+ params.pop('Abteilung')
|
|
|
+ response = self.client.get(list_url(params))
|
|
|
+ self.assertIn('id="list-count">5', response.text)
|
|
|
+ self.assertNotIn('name="dashboard_segment"', response.text)
|
|
|
+ params['Staffel'] = '%3C%204%20Wochen'
|
|
|
+ response = self.client.get(list_url(params))
|
|
|
+ self.assertIn('id="list-count">1', response.text)
|
|
|
+
|
|
|
+ def test_old_dashboard_link_redirects_to_visible_filters(self):
|
|
|
+ legacy = pack_segment({'view': 'Standort', 'key': {'client': '1', 'site': '10'}, 'department': 'Service'})
|
|
|
+ response = self.client.get('/app/forderungen/liste', params={'dashboard_segment': legacy}, follow_redirects=False)
|
|
|
+ self.assertEqual(response.status_code, 307)
|
|
|
+ self.assertEqual(response.headers['location'], '/app/forderungen/liste?Mandant=1&Filiale=10&Abteilung=Service')
|
|
|
+ response = self.client.get(response.headers['location'])
|
|
|
+ self.assertIn('id="list-count">4', response.text)
|
|
|
+ self.assertIn('data-remove="Service"', response.text)
|
|
|
+ self.assertNotIn('dashboard_segment', response.text)
|
|
|
+ self.assertNotIn('Dashboardfilter', response.text)
|
|
|
+
|
|
|
+ def test_old_person_id_is_translated_to_normal_name_filter(self):
|
|
|
+ legacy = pack_segment({'view': 'Verursacher', 'key': {'id': 'alice'}, 'only_invoices': True})
|
|
|
+ response = self.client.get('/app/forderungen/liste', params={'dashboard_segment': legacy})
|
|
|
+ params = parse_qs(urlsplit(str(response.url)).query)
|
|
|
+ self.assertEqual(params, {'Verursacher': ['Alice'], 'RechnungGutschrift': ['Rechnung']})
|
|
|
+ self.assertIn('id="list-count">2', response.text)
|
|
|
+ self.assertIn('data-remove="Alice"', response.text)
|
|
|
+ self.assertNotIn('dashboard_segment', response.text)
|
|
|
+
|
|
|
+ def test_explicit_filters_override_old_link_and_login_recovers_visible_filters(self):
|
|
|
+ legacy = pack_segment({'view': 'Standort', 'key': {'client': '1', 'site': '10'}, 'department': 'Service'})
|
|
|
+ response = self.client.get('/app/forderungen/liste', params={'dashboard_segment': legacy, 'Filiale': '20'})
|
|
|
+ self.assertEqual(parse_qs(urlsplit(str(response.url)).query)['Filiale'], ['20'])
|
|
|
+ self.assertIn('id="list-count">0', response.text)
|
|
|
+ self.client.cookies.clear()
|
|
|
+ response = self.client.get('/app/forderungen/liste', params={'dashboard_segment': legacy}, headers={'HX-Request': 'true'})
|
|
|
+ target = parse_qs(urlsplit(response.headers['HX-Redirect']).query)['next'][0]
|
|
|
+ self.assertEqual(target, '/app/forderungen/liste?Mandant=1&Filiale=10&Abteilung=Service')
|
|
|
+
|
|
|
+ def test_new_links_preserve_names_with_commas_and_spaces(self):
|
|
|
+ self.db.connection.execute("UPDATE dbo.Benutzer_Rolle SET Name='Link, Florian' WHERE Benutzer_ID='alice'")
|
|
|
+ group = next(item for item in self.data()['views']['Verursacher'] if item['label'] == 'Link, Florian')
|
|
|
+ self.assertIn('Verursacher=Link%252C%2520Florian', group['url'])
|
|
|
+ response = self.client.get(group['url'])
|
|
|
+ self.assertIn('id="list-count">1', response.text)
|
|
|
+ self.assertIn('data-remove="Link, Florian"', response.text)
|
|
|
|
|
|
def test_forged_segment_cannot_expand_invoice_access(self):
|
|
|
raw = pack_segment({"view": "Standort", "key": {"client": "2", "site": "10"}})
|
|
|
@@ -306,9 +392,9 @@ class DashboardTests(unittest.TestCase):
|
|
|
|
|
|
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]
|
|
|
+ filters = parse_qs(urlsplit(url).query)
|
|
|
returned = list_return_url(url, "1_R1")
|
|
|
- self.assertEqual(parse_qs(urlsplit(returned).query)["dashboard_segment"], [selection])
|
|
|
+ self.assertEqual(parse_qs(urlsplit(returned).query), {**filters, 'focus': ['1_R1']})
|
|
|
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
|
|
|
@@ -318,7 +404,7 @@ class DashboardTests(unittest.TestCase):
|
|
|
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.assertIn("F.[Client_DB] = :filter_mandant", sql)
|
|
|
self.assertEqual(values["filter_kunde"], "%Test%")
|
|
|
self.client.cookies.clear()
|
|
|
response = self.client.get(url, headers={"HX-Request": "true"})
|
gc-server3