Sfoglia il codice sorgente

Füge Unterstützung für Datumsfilter hinzu und verbessere die Fehlerbehandlung für ungültige Datumsangaben

gc-server3 4 giorni fa
parent
commit
055a857883
3 ha cambiato i file con 59 aggiunte e 3 eliminazioni
  1. 11 1
      app/query_filter.py
  2. 7 1
      app/routes.py
  3. 41 1
      tests/test_list_filters.py

+ 11 - 1
app/query_filter.py

@@ -1,5 +1,6 @@
 import csv
 import csv
 from dataclasses import dataclass, field
 from dataclasses import dataclass, field
+from datetime import date
 from urllib.parse import quote, unquote
 from urllib.parse import quote, unquote
 
 
 EMPTY_FILTER_VALUE = "__empty__"
 EMPTY_FILTER_VALUE = "__empty__"
@@ -15,6 +16,13 @@ def encode_filter_values(values) -> str:
     return ",".join(quote(value, safe="") for value in dict.fromkeys(str(value).strip() for value in values) if value)
     return ",".join(quote(value, safe="") for value in dict.fromkeys(str(value).strip() for value in values) if value)
 
 
 
 
+def parse_filter_date(value: str) -> date:
+    selected = date.fromisoformat(value)
+    if selected.isoformat() != value:
+        raise ValueError("Expected YYYY-MM-DD")
+    return selected
+
+
 def filter_clauses(config, params, exclude=None):
 def filter_clauses(config, params, exclude=None):
     clauses, values = [], {}
     clauses, values = [], {}
     for f in config:
     for f in config:
@@ -25,7 +33,9 @@ def filter_clauses(config, params, exclude=None):
         if f.filter_type == "date":
         if f.filter_type == "date":
             operator = "<=" if "bis" in f.name.lower() else ">="
             operator = "<=" if "bis" in f.name.lower() else ">="
             clauses.append(f"F.[{f.key_column}] {operator} :{parameter}")
             clauses.append(f"F.[{f.key_column}] {operator} :{parameter}")
-            values[parameter] = current
+            # Bind a native date: SQL Server's datetime string conversion depends
+            # on the connection language/DATEFORMAT, even for YYYY-MM-DD.
+            values[parameter] = parse_filter_date(current)
             continue
             continue
         parts = decode_filter_values(current)
         parts = decode_filter_values(current)
         terms = []
         terms = []

+ 7 - 1
app/routes.py

@@ -21,7 +21,7 @@ from sqlalchemy import text
 from sqlalchemy.orm import Session
 from sqlalchemy.orm import Session
 
 
 from app.file_storage import file_primary_key, get_file, get_files
 from app.file_storage import file_primary_key, get_file, get_files
-from app.query_filter import filter_clauses, import_filter_config
+from app.query_filter import filter_clauses, import_filter_config, parse_filter_date
 
 
 from .auth import clear_session_cookie, safe_next_url, set_session_cookie
 from .auth import clear_session_cookie, safe_next_url, set_session_cookie
 from .comment_submission import submit_comment
 from .comment_submission import submit_comment
@@ -181,6 +181,12 @@ def select_get(request: Request):
 
 
 def _list_filter_params(request, config, db, username):
 def _list_filter_params(request, config, db, username):
     params = {f.name: request.query_params[f.name] for f in config if request.query_params.get(f.name, "") != ""}
     params = {f.name: request.query_params[f.name] for f in config if request.query_params.get(f.name, "") != ""}
+    for f in config:
+        if f.filter_type == "date" and f.name in params:
+            try:
+                parse_filter_date(params[f.name])
+            except ValueError as error:
+                raise HTTPException(422, f"Ungültiges Datum für {f.text}. Erwartet wird JJJJ-MM-TT.") from error
     try:
     try:
         legacy = parse_segment(request.query_params.get("dashboard_segment"))
         legacy = parse_segment(request.query_params.get("dashboard_segment"))
     except ValueError as error:
     except ValueError as error:

+ 41 - 1
tests/test_list_filters.py

@@ -68,7 +68,10 @@ class SQLiteListDB:
         if top:
         if top:
             sql = re.sub(r'SELECT TOP \d+', 'SELECT', sql, count=1) + ' LIMIT ' + top[1]
             sql = re.sub(r'SELECT TOP \d+', 'SELECT', sql, count=1) + ' LIMIT ' + top[1]
         sql = sql.replace('nvarchar(max)', 'TEXT')
         sql = sql.replace('nvarchar(max)', 'TEXT')
-        cursor = self.connection.execute(sql, values)
+        # Keep the captured production parameters typed, but adapt dates explicitly
+        # for SQLite rather than relying on its deprecated implicit adapter.
+        cursor = self.connection.execute(sql, {key: value.isoformat() if isinstance(value, date) else value
+                                               for key, value in values.items()})
         columns = [item[0] for item in cursor.description]
         columns = [item[0] for item in cursor.description]
         record = namedtuple('Result', columns)
         record = namedtuple('Result', columns)
         rows = []
         rows = []
@@ -138,6 +141,43 @@ class MultiFilterTests(unittest.TestCase):
         self.assertIn('F.[Beleg_Datum] >= :filter_rechnungsdatumvon', sql)
         self.assertIn('F.[Beleg_Datum] >= :filter_rechnungsdatumvon', sql)
         self.assertIn('F.[Beleg_Datum] <= :filter_rechnungsdatumbis', sql)
         self.assertIn('F.[Beleg_Datum] <= :filter_rechnungsdatumbis', sql)
 
 
+    def test_date_filters_bind_native_dates_and_preserve_day_and_month(self):
+        for field, column in (
+            ('Rechnungsdatum', 'Beleg_Datum'), ('Wiedervorlage', 'Wiedervorlage'),
+        ):
+            for selected in (date(2026, 11, 5), date(2026, 11, 18), date(2028, 2, 29)):
+                with self.subTest(field=field, selected=selected):
+                    self.db.connection.execute('DELETE FROM dbo.Forderungen_Liste')
+                    self.db.add(10, **{column: selected.replace(day=selected.day - 1)})
+                    self.db.add(11, **{column: selected})
+                    self.db.add(12, **{column: selected.replace(day=selected.day - 2)})
+                    for suffix, count in (('Von', 1), ('Bis', 3)):
+                        name = field + suffix
+                        response = self.client.get('/app/forderungen/liste', params={name: selected.isoformat()})
+                        self.assertEqual(response.status_code, 200)
+                        self.assertIn(f'id="list-count">{count}', response.text)
+                        self.assertRegex(response.text, rf'id="{name}"[^>]*value="{selected.isoformat()}"')
+                        bound = self.db.calls[-1][1]['filter_' + name.lower()]
+                        self.assertIs(type(bound), date)
+                        self.assertEqual(bound, selected)
+                    response = self.client.get('/app/forderungen/liste', params={
+                        field + 'Von': selected.isoformat(), field + 'Bis': selected.isoformat(),
+                    })
+                    self.assertIn('id="list-count">1', response.text)
+
+    def test_invalid_dates_are_rejected_before_database_access(self):
+        for field in ('RechnungsdatumVon', 'RechnungsdatumBis', 'WiedervorlageVon', 'WiedervorlageBis'):
+            for value in ('2026-18-11', '18.11.2026', '2026-02-29', '20261118', "2026-11-18' OR 1=1--"):
+                for path in ('/app/forderungen/liste', '/app/forderungen/liste/vorschlaege'):
+                    with self.subTest(field=field, value=value, path=path):
+                        self.db.calls.clear()
+                        params = {field: value}
+                        if path.endswith('vorschlaege'):
+                            params['field'] = 'Kunde'
+                        response = self.client.get(path, params=params)
+                        self.assertEqual(response.status_code, 422)
+                        self.assertEqual(self.db.calls, [])
+
     def test_comma_value_survives_render_and_detail_return(self):
     def test_comma_value_survives_render_and_detail_return(self):
         value = encode_filter_values(['Müller, GmbH', 'Meier'])
         value = encode_filter_values(['Müller, GmbH', 'Meier'])
         response = self.client.get(list_url({'Kunde': value}))
         response = self.client.get(list_url({'Kunde': value}))