|
|
@@ -0,0 +1,170 @@
|
|
|
+"""Follow-up date and audit comment are one protected transaction."""
|
|
|
+from datetime import date, datetime
|
|
|
+import unittest
|
|
|
+from unittest.mock import patch
|
|
|
+from urllib.parse import parse_qs, urlencode, urlsplit
|
|
|
+
|
|
|
+from app.navigation import normalize_page_url
|
|
|
+from app.wiedervorlage import BERLIN, _as_date
|
|
|
+from tests import test_verursacher as fixture
|
|
|
+
|
|
|
+
|
|
|
+class WiedervorlageTests(unittest.TestCase):
|
|
|
+ def setUp(self):
|
|
|
+ self.fixture = fixture.VerursacherTests()
|
|
|
+ self.fixture.setUp()
|
|
|
+ self.addCleanup(self.fixture.doCleanups)
|
|
|
+ self.client, self.db = self.fixture.client, self.fixture.db
|
|
|
+ self.db.connection.executescript("""
|
|
|
+ ALTER TABLE dbo.Forderungen_Kopf ADD Beleg_Faellig_Datum TEXT;
|
|
|
+ UPDATE dbo.Forderungen_Kopf SET Beleg_Faellig_Datum = '2026-10-15';
|
|
|
+ """)
|
|
|
+ self.db.connection.commit()
|
|
|
+ self.origin = '/app/forderungen/liste?Filiale=10&WiedervorlageBis=2026-10-31'
|
|
|
+ execute = self.db.execute
|
|
|
+ def adapt_dates(query, values=None):
|
|
|
+ values = {key: value.isoformat() if isinstance(value, (date, datetime)) else value
|
|
|
+ for key, value in (values or {}).items()}
|
|
|
+ return execute(query, values)
|
|
|
+ adapter = patch.object(self.db, 'execute', side_effect=adapt_dates)
|
|
|
+ adapter.start(); self.addCleanup(adapter.stop)
|
|
|
+ self.now = datetime(2026, 10, 30, 0, 5, 12, tzinfo=BERLIN)
|
|
|
+ clock = patch('app.wiedervorlage._now_berlin', return_value=self.now)
|
|
|
+ clock.start(); self.addCleanup(clock.stop)
|
|
|
+
|
|
|
+ def url(self, key='1_R1'):
|
|
|
+ return f'/app/forderungen/liste/{key}/wiedervorlage?' + urlencode({'return_to': self.origin})
|
|
|
+
|
|
|
+ def get(self, key='1_R1'):
|
|
|
+ return self.client.get(self.url(key), headers={'HX-Request': 'true'})
|
|
|
+
|
|
|
+ def post(self, value='2026-11-06', key='1_R1', **extra):
|
|
|
+ return self.client.post(self.url(key), data={'wiedervorlage': value, **extra},
|
|
|
+ headers={'HX-Request': 'true', 'X-CSRF-Token': 'csrf'})
|
|
|
+
|
|
|
+ def row(self):
|
|
|
+ return self.fixture.row()
|
|
|
+
|
|
|
+ def comments(self):
|
|
|
+ return self.db.connection.execute("SELECT Client_DB, Beleg_Nr, Benutzer_ID, Kommentar, Timestamp FROM dbo.Forderungen_Kommentar WHERE Kommentar LIKE 'WV auf %'").fetchall()
|
|
|
+
|
|
|
+ def test_modal_prefills_list_fallback_and_berlin_shortcuts_without_writing(self):
|
|
|
+ response = self.get()
|
|
|
+ self.assertEqual(response.status_code, 200)
|
|
|
+ self.assertIn('value="2026-10-15"', response.text)
|
|
|
+ self.assertIn('data-gc-date="2026-10-30"', response.text)
|
|
|
+ self.assertIn('data-gc-date="2026-11-06"', response.text)
|
|
|
+ self.assertNotIn('<html', response.text)
|
|
|
+ self.assertIsNone(self.row()); self.assertEqual(self.comments(), [])
|
|
|
+ self.assertEqual(self.db.commits, 0)
|
|
|
+
|
|
|
+ def test_shortcuts_cross_year_and_leap_day(self):
|
|
|
+ for now, expected in [(datetime(2026, 12, 28, tzinfo=BERLIN), '2027-01-04'),
|
|
|
+ (datetime(2028, 2, 22, tzinfo=BERLIN), '2028-02-29')]:
|
|
|
+ with patch('app.wiedervorlage._now_berlin', return_value=now):
|
|
|
+ self.assertIn('data-gc-date="'+expected+'"', self.get().text)
|
|
|
+
|
|
|
+ def test_insert_date_comment_actor_timestamp_and_filtered_focus(self):
|
|
|
+ response = self.post(Benutzer_ID='other', Client_DB='2', Beleg_Nr='R2')
|
|
|
+ self.assertEqual(response.status_code, 200)
|
|
|
+ self.assertEqual(self.row()[2:], (None, 'N', 'N', '2026-11-06'))
|
|
|
+ comments = self.comments()
|
|
|
+ self.assertEqual(len(comments), 1)
|
|
|
+ self.assertEqual(comments[0][:4], ('1', 'R1', 'tester', 'WV auf 06.11.2026 angepasst'))
|
|
|
+ self.assertEqual(datetime.fromisoformat(comments[0][4]), self.now.replace(tzinfo=None))
|
|
|
+ self.assertEqual(self.db.commits, 1)
|
|
|
+ destination = response.headers['HX-Redirect']
|
|
|
+ self.assertEqual(parse_qs(urlsplit(destination).query),
|
|
|
+ {'Filiale': ['10'], 'WiedervorlageBis': ['2026-10-31'], 'focus': ['1_R1']})
|
|
|
+
|
|
|
+ def test_update_preserves_other_fields_and_preselects_saved_date(self):
|
|
|
+ self.db.connection.execute("INSERT INTO dbo.Forderungen_Details VALUES ('1','R1','alice','N','J','2026-10-20')")
|
|
|
+ self.db.connection.commit()
|
|
|
+ self.assertIn('value="2026-10-20"', self.get().text)
|
|
|
+ self.post()
|
|
|
+ self.assertEqual(self.row()[2:], ('alice', 'N', 'J', '2026-11-06'))
|
|
|
+ self.assertIn('value="2026-11-06"', self.get().text)
|
|
|
+
|
|
|
+ def test_missing_date_can_be_assigned(self):
|
|
|
+ self.db.connection.execute("UPDATE dbo.Forderungen_Kopf SET Beleg_Faellig_Datum = NULL WHERE Client_DB='1' AND Beleg_Nr='R1'")
|
|
|
+ self.db.connection.commit()
|
|
|
+ self.assertIn('value="" required', self.get().text)
|
|
|
+ self.assertEqual(self.post().status_code, 200)
|
|
|
+ self.assertEqual(self.row()[-1], '2026-11-06')
|
|
|
+ self.assertEqual(len(self.comments()), 1)
|
|
|
+
|
|
|
+ def test_same_date_and_repeated_submission_do_not_duplicate_comments(self):
|
|
|
+ self.post('2026-10-15')
|
|
|
+ self.assertIsNone(self.row()); self.assertEqual(self.comments(), [])
|
|
|
+ self.post(); self.post()
|
|
|
+ self.assertEqual(len(self.comments()), 1)
|
|
|
+
|
|
|
+ def test_invalid_or_empty_dates_never_write(self):
|
|
|
+ for value in ['', '2026-02-30', '2026-13-01', '30.10.2026', '20261030', "x'; DROP TABLE dbo.Forderungen_Details"]:
|
|
|
+ with self.subTest(value=value):
|
|
|
+ response = self.post(value)
|
|
|
+ self.assertEqual(response.status_code, 422)
|
|
|
+ self.assertIn('gültiges Wiedervorlagedatum', response.text)
|
|
|
+ self.assertIsNone(self.row()); self.assertEqual(self.comments(), [])
|
|
|
+ self.assertEqual(self.db.commits, 0)
|
|
|
+
|
|
|
+ def test_comment_failure_rolls_back_existing_date_and_allows_retry(self):
|
|
|
+ self.db.connection.execute("INSERT INTO dbo.Forderungen_Details VALUES ('1','R1','alice','N','J','2026-10-20')")
|
|
|
+ self.db.connection.commit()
|
|
|
+ execute = self.db.execute
|
|
|
+ def failing(query, values=None):
|
|
|
+ if 'INSERT INTO dbo.Forderungen_Kommentar' in str(query):
|
|
|
+ raise RuntimeError('comment unavailable')
|
|
|
+ return execute(query, values)
|
|
|
+ with patch.object(self.db, 'execute', side_effect=failing), self.assertLogs('app.wiedervorlage', level='ERROR'):
|
|
|
+ response = self.post()
|
|
|
+ self.assertEqual(response.status_code, 503)
|
|
|
+ self.assertNotIn('HX-Redirect', response.headers)
|
|
|
+ self.assertIn('value="2026-11-06"', response.text)
|
|
|
+ self.assertEqual(self.row()[2:], ('alice', 'N', 'J', '2026-10-20'))
|
|
|
+ self.assertEqual(self.comments(), [])
|
|
|
+ self.post()
|
|
|
+ self.assertEqual(len(self.comments()), 1)
|
|
|
+
|
|
|
+ def test_commit_failure_rolls_back_new_record_and_comment(self):
|
|
|
+ with patch.object(self.db, 'commit', side_effect=RuntimeError('offline')), self.assertLogs('app.wiedervorlage', level='ERROR'):
|
|
|
+ response = self.post()
|
|
|
+ self.assertEqual(response.status_code, 503)
|
|
|
+ self.assertIsNone(self.row()); self.assertEqual(self.comments(), [])
|
|
|
+
|
|
|
+ def test_rights_csrf_and_native_redirect(self):
|
|
|
+ self.assertEqual(self.get('1_R2').status_code, 404)
|
|
|
+ self.assertEqual(self.post(key='1_R2').status_code, 404)
|
|
|
+ self.assertEqual(self.client.post(self.url(), data={'wiedervorlage':'2026-11-06'}).status_code, 403)
|
|
|
+ native = self.client.get(self.url(), follow_redirects=False)
|
|
|
+ self.assertEqual(native.status_code, 303)
|
|
|
+ self.assertNotIn('/wiedervorlage', native.headers['location'])
|
|
|
+ self.assertIsNone(self.row())
|
|
|
+ native_save = self.client.post(self.url(), data={'wiedervorlage':'2026-11-06'},
|
|
|
+ headers={'X-CSRF-Token':'csrf'}, follow_redirects=False)
|
|
|
+ self.assertEqual(native_save.status_code, 303)
|
|
|
+ self.assertNotIn('/wiedervorlage', native_save.headers['location'])
|
|
|
+ self.assertEqual(len(self.comments()), 1)
|
|
|
+
|
|
|
+ def test_session_expiry_normalizes_full_list_without_replaying_post(self):
|
|
|
+ self.client.cookies.clear()
|
|
|
+ response = self.post()
|
|
|
+ target = parse_qs(urlsplit(response.headers['HX-Redirect']).query)['next'][0]
|
|
|
+ self.assertNotIn('/wiedervorlage', target)
|
|
|
+ self.assertEqual(target, normalize_page_url(self.url()))
|
|
|
+ self.assertIsNone(self.row()); self.assertEqual(self.comments(), [])
|
|
|
+ self.assertEqual(normalize_page_url('/app/forderungen/liste/1_R1/wiedervorlage',
|
|
|
+ current_url='https://proxy.test'+self.origin), target)
|
|
|
+ self.assertEqual(normalize_page_url('/app/forderungen/liste/1_R1/wiedervorlage?return_to=https://evil.test'),
|
|
|
+ '/app/forderungen/liste?focus=1_R1')
|
|
|
+
|
|
|
+ def test_database_date_types(self):
|
|
|
+ value = datetime(2026, 10, 30, 12, 30)
|
|
|
+ self.assertEqual(_as_date(value), value.date())
|
|
|
+ self.assertEqual(_as_date(value.date()), value.date())
|
|
|
+ self.assertEqual(_as_date('2026-10-30 00:00:00'), value.date())
|
|
|
+ self.assertIsNone(_as_date(None))
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == '__main__':
|
|
|
+ unittest.main()
|