Reisacher-Forderung/tests/test_logging.py

"""Logging is tested against isolated engines, ASGI apps and temporary files only."""
import tests
import ast
import asyncio
from datetime import datetime, timedelta, timezone
import io
import json
import logging
import multiprocessing
import pyodbc
from pathlib import Path
import sqlite3
import tempfile
import unittest
from types import SimpleNamespace
from unittest.mock import MagicMock, patch

import httpx
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import HTMLResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
from fastapi.testclient import TestClient
from sqlalchemy import Column, Integer, create_engine, text
from sqlalchemy.exc import DBAPIError
from sqlalchemy.orm import declarative_base
from sqlalchemy.pool import StaticPool

from app.auth import AuthenticationMiddleware
from app.config import settings, Settings
from app.db import LoggingSession, install_engine_logging
from app.http_logging import install_request_logging
from app.logging_setup import (ApplicationFileHandler, configure_logging, request_context,
                               run_cli, sanitize_sql)
from app.oauth_token import create_access_token
from app.permissions import Entitlement
import app.logging_setup as setup

SECRET = 'DO_NOT_LOG_SECRET_93'


def config(directory, **kwargs):
    values = dict(LOG_DIR=str(directory), LOG_LEVEL='INFO', LOG_TO_FILE=True, LOG_TO_CONSOLE=True,
                  LOG_RETENTION_DAYS=30, LOG_SQL_STATEMENT=True)
    values.update(kwargs)
    return SimpleNamespace(**values)


def process_writer(directory, worker):
    configure_logging(config(directory, LOG_TO_CONSOLE=False))
    for index in range(35):
        logging.getLogger('app.worker').info('Worker event', extra={'event':'worker', 'count':worker*1000+index})
    for handler in list(logging.getLogger().handlers):
        if getattr(handler, '_gc_handler', False): handler.close()


class LoggingTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory(prefix='gcepic-logging-')
        self.root = Path(self.temp.name)
        root = logging.getLogger()
        self.old_handlers, self.old_level = root.handlers[:], root.level
        root.handlers.clear()
        self.old_config = setup._configuration
        self.logger_states = {name:(logger.level, logger.disabled, logger.propagate, logger.handlers[:])
                              for name,logger in root.manager.loggerDict.items() if isinstance(logger, logging.Logger)}
        self.console = io.StringIO()
        self.configuration = config(self.root)
        configure_logging(self.configuration, console_stream=self.console)
        self.patch = patch.object(settings, 'LOG_SQL_STATEMENT', True)
        self.patch.start()

    def tearDown(self):
        self.patch.stop()
        root = logging.getLogger()
        for handler in list(root.handlers):
            root.removeHandler(handler)
            if getattr(handler, '_gc_handler', False): handler.close()
        root.handlers[:] = self.old_handlers; root.setLevel(self.old_level)
        for name, state in self.logger_states.items():
            logger = logging.getLogger(name)
            logger.setLevel(state[0]); logger.disabled=state[1]; logger.propagate=state[2]; logger.handlers[:]=state[3]
        setup._configuration = self.old_config
        self.temp.cleanup()

    def records(self):
        file = self.root/'application.log'
        return [json.loads(line) for line in file.read_text(encoding='utf-8').splitlines()] if file.exists() else []

    def engine(self, **kwargs):
        engine = create_engine('sqlite://', poolclass=StaticPool, connect_args={'check_same_thread':False}, hide_parameters=True, **kwargs)
        install_engine_logging(engine)
        self.addCleanup(engine.dispose)
        return engine

    def test_literals_comments_and_identifiers_are_sanitized(self):
        sql = "SELECT [col1], :bind2, N'O''Brien', 123, -1.23e+4, 0xAFF, .5, \"private\" /* private /* nested */ end */ -- private\nFROM [table2]"
        safe = sanitize_sql(sql)
        self.assertNotIn('private', safe); self.assertNotIn('Brien', safe)
        for value in ['123','1.23','0xAFF','.5']: self.assertNotIn(value, safe)
        for name in ['[col1]', ':bind2', '[table2]']: self.assertIn(name, safe)

    def test_idempotent_initialization_and_configured_outputs(self):
        root = logging.getLogger()
        count = len(root.handlers)
        configure_logging(self.configuration, console_stream=self.console)
        self.assertEqual(len(root.handlers), count)
        logging.getLogger('app.test').info('Started', extra={'event':'started'})
        self.assertEqual(len(self.records()), 1)
        self.assertEqual(self.console.getvalue().count('Started'), 1)
        self.assertRegex(self.records()[0]['timestamp'], r'\.\d{3}[+-]\d{2}:\d{2}$')
        self.assertEqual(Settings(_env_file=None, LOG_LEVEL='debug').LOG_LEVEL, 'DEBUG')

    def test_sql_error_logged_once_even_when_caught_and_logged_again(self):
        engine = self.engine()
        with LoggingSession(engine) as db:
            for value in ['first-secret', 'second-secret']:
                try:
                    db.execute(text("SELECT :value FROM missing_table WHERE 91827=1 AND 'private-value'='private-value'").execution_options(query_name='test.missing'), {'value':value})
                except Exception:
                    logging.getLogger('app.service').exception('Operation failed')
        records = self.records()
        self.assertEqual(len(records), 2)
        self.assertTrue(all(row['event']=='database_error' for row in records))
        self.assertEqual(records[0]['query_fingerprint'], records[1]['query_fingerprint'])
        self.assertEqual(records[0]['query_name'], 'test.missing')
        self.assertIn('missing_table', records[0]['sql'])
        self.assertTrue(records[0]['exceptions'][0]['frames'])
        combined=json.dumps(records)+self.console.getvalue()
        for value in ['first-secret','second-secret','private-value','91827','[parameters:']:
            self.assertNotIn(value, combined)

    def test_connect_and_fetch_errors_are_logged_without_driver_messages(self):
        def unavailable(): raise sqlite3.OperationalError(SECRET)
        engine = self.engine(creator=unavailable)
        with LoggingSession(engine) as db, self.assertRaises(Exception):
            db.execute(text('SELECT 1').execution_options(query_name='connect.test'))
        self.assertEqual(len(self.records()), 1)
        self.assertEqual(self.records()[0]['query_name'], 'connect.test')
        class Cursor(sqlite3.Cursor):
            def execute(self, sql, params=()):
                self.bad_fetch = 'fetch_marker' in sql
                return super().execute(sql, params)
            def fetchall(self):
                if getattr(self,'bad_fetch',False): raise sqlite3.OperationalError(SECRET)
                return super().fetchall()
        class Connection(sqlite3.Connection):
            def cursor(self,*args,**kwargs): return super().cursor(factory=Cursor)
        engine = self.engine(creator=lambda: sqlite3.connect(':memory:',factory=Connection))
        with LoggingSession(engine) as db, self.assertRaises(Exception):
            db.execute(text('SELECT 1 AS fetch_marker')).fetchall()
        self.assertEqual(len(self.records()), 2)
        self.assertEqual(self.records()[-1]['operation'], 'fetch')
        self.assertNotIn(SECRET, json.dumps(self.records())+self.console.getvalue())

    def test_sql_server_diagnostics_include_codes_and_redacted_descriptions(self):
        driver=pyodbc.ProgrammingError('42S22', "[42S22] [Microsoft][ODBC Driver 18 for SQL Server][SQL Server]Invalid column name 'PrivateColumn'. (207) (SQLExecDirectW); [42000] [Microsoft][ODBC Driver 18 for SQL Server][SQL Server]Statement(s) could not be prepared. (8180)")
        error=DBAPIError("SELECT :value, 'inline-private' FROM sample", {'value':SECRET}, driver)
        setup.log_query_error(error,error.statement,'test.sql_server')
        logging.getLogger('app.service').error('Handled',exc_info=(type(error),error,None))
        rows=self.records()
        self.assertEqual(len(rows),1)
        self.assertEqual(rows[0]['db_code'],207)
        self.assertEqual(rows[0]['sqlstate'],'42S22')
        self.assertIn('Invalid column name',rows[0]['db_message'])
        self.assertIn("'PrivateColumn'",rows[0]['db_message'])
        self.assertIn("'PrivateColumn'",self.console.getvalue())
        self.assertEqual([item['code'] for item in rows[0]['db_diagnostics']],[207,8180])
        self.assertEqual(rows[0]['db_diagnostics'][1]['sqlstate'],'42000')
        self.assertIn('could not be prepared',self.console.getvalue())
        for secret in [SECRET,'inline-private','[parameters:']:
            self.assertNotIn(secret,json.dumps(rows)+self.console.getvalue())

    def test_invalid_column_names_are_visible_only_in_canonical_error_207(self):
        with patch.object(settings,'LOG_SQL_STATEMENT',False):
            for description in ["Ungültiger Spaltenname 'Kunde_Größe_2'.", "Invalid column name 'Customer''s (2026)'.", 'Ungültiger Spaltenname "Kunde_Gesamt_Offen".', 'Ungültiger Spaltenname „Kunde_Gesamt_Offen“.']:
                driver=pyodbc.ProgrammingError('42S22','[Microsoft][ODBC Driver 18 for SQL Server][SQL Server]'+description+' (207) (SQLExecDirectW)')
                setup.log_query_error(driver,parameters={'value':SECRET})
                row=self.records()[-1]
                self.assertEqual(row['db_message'],description)
                self.assertEqual(row['db_diagnostics'][0]['message'],description)
                self.assertNotIn('sql',row)
        for code in [207,245]:
            description=f"Invalid column name 'PrivateColumn'. Extra value '{SECRET}'. ({code}) (SQLExecDirectW)"
            setup.log_query_error(pyodbc.ProgrammingError('42S22',description))
            self.assertNotIn('PrivateColumn',self.records()[-1]['db_message'])
        self.assertNotIn(SECRET,json.dumps(self.records())+self.console.getvalue())

    def test_german_binding_errors_keep_each_invalid_column_name(self):
        raw=('\r\n[42S22] [Microsoft][ODBC Driver 18 for SQL Server][SQL Server]Ungültiger Spaltenname "Spalte_A". (207) (SQLExecDirectW);\r\n'
             '[42S22] [Microsoft][ODBC Driver 18 for SQL Server][SQL Server]Ungültiger Spaltenname "Spalte_B". (207); '
             '[42S22] [Microsoft][ODBC Driver 18 for SQL Server][SQL Server]Die Sicht oder Funktion "dbo.PrivateView" konnte aufgrund von Bindungsfehlern nicht verwendet werden. (4413); '
             '[42S22] [Microsoft][ODBC Driver 18 for SQL Server][SQL Server]Anweisung(en) konnte(n) nicht vorbereitet werden. (8180)')
        setup.log_query_error(pyodbc.ProgrammingError('42S22',raw),parameters={'token':SECRET})
        row=self.records()[-1]
        self.assertIn('"Spalte_A"',row['db_message'])
        self.assertIn('"Spalte_B"',row['db_diagnostics'][1]['message'])
        self.assertEqual([item['code'] for item in row['db_diagnostics']],[207,207,4413,8180])
        self.assertIn('Spalte_A',self.console.getvalue())
        self.assertIn('Spalte_B',self.console.getvalue())
        self.assertNotIn('PrivateView',json.dumps(row)+self.console.getvalue())
        self.assertNotIn(SECRET,json.dumps(row)+self.console.getvalue())

    def test_sql_server_diagnostics_redact_echoed_values_and_work_without_sql_logging(self):
        descriptions=[
            ('23000',f"Violation of UNIQUE KEY constraint 'key'. The duplicate key value is ({SECRET}, 12345). (2627) (SQLExecDirectW)",2627),
            ('22018',f"Conversion failed when converting the nvarchar value '{SECRET}' to data type int. (245) (SQLExecDirectW)",245),
            ('42000',f"Custom validation failed: {SECRET}; inline-private; password=credential-private (50001) (SQLExecDirectW)",50001),
            ('HYT00',"Login timeout expired (0) (SQLDriverConnect)",0),
        ]
        with patch.object(settings,'LOG_SQL_STATEMENT',False):
            for state,description,code in descriptions:
                driver=pyodbc.Error(state,'[Microsoft][ODBC Driver 18 for SQL Server][SQL Server]'+description)
                error=DBAPIError("SELECT 'inline-private'",{'value':SECRET},driver)
                setup.log_query_error(error,error.statement,'test.redaction')
                self.assertEqual(self.records()[-1]['db_code'],code)
                self.assertNotIn('sql',self.records()[-1])
        combined=json.dumps(self.records())+self.console.getvalue()
        for secret in [SECRET,'12345','inline-private','credential-private']:
            self.assertNotIn(secret,combined)
        self.assertIn('UNIQUE KEY',combined)
        self.assertIn('Conversion failed',combined)
        self.assertIn('Login timeout expired',combined)

    def test_commit_rollback_and_flush_errors_are_captured(self):
        class Connection(sqlite3.Connection):
            fail_commit=False; fail_rollback=False
            def commit(self):
                if self.fail_commit:
                    self.fail_commit=False; raise sqlite3.OperationalError(SECRET)
                return super().commit()
            def rollback(self):
                if self.fail_rollback:
                    self.fail_rollback=False; raise sqlite3.OperationalError(SECRET)
                return super().rollback()
        engine=self.engine(creator=lambda:sqlite3.connect(':memory:',factory=Connection))
        with LoggingSession(engine) as db:
            db.execute(text('CREATE TABLE sample (id INTEGER PRIMARY KEY)')); db.commit()
            db.execute(text('INSERT INTO sample VALUES (1)'))
            db.connection().connection.driver_connection.fail_commit=True
            with self.assertRaises(Exception): db.commit()
            db.rollback()
            db.execute(text('SELECT 1'))
            db.connection().connection.driver_connection.fail_rollback=True
            with self.assertRaises(Exception): db.rollback()
        Base=declarative_base()
        class Model(Base):
            __tablename__='unique_sample'
            id=Column(Integer, primary_key=True)
        Base.metadata.create_all(engine)
        with LoggingSession(engine) as db:
            db.add_all([Model(id=1),Model(id=1)])
            with self.assertRaises(Exception): db.flush()
            db.rollback()
        operations=[row['operation'] for row in self.records() if row.get('event')=='database_error']
        self.assertIn('commit',operations); self.assertIn('rollback',operations); self.assertIn('flush',operations)
        self.assertEqual(len(operations),3)
        self.assertNotIn(SECRET, json.dumps(self.records())+self.console.getvalue())

    def app(self):
        app=FastAPI()
        @app.get('/status/{code}')
        async def response(code:int): return HTMLResponse(SECRET,status_code=code)
        @app.get('/http-error')
        async def http_error(): raise HTTPException(400, SECRET)
        @app.get('/number')
        async def number(value:int): return value
        @app.get('/ok')
        async def ok(request:Request):
            request.state.principal=SimpleNamespace(username=request.headers.get('x-test-user','tester'))
            logging.getLogger('app.service').info('Service event')
            from starlette.concurrency import run_in_threadpool
            await run_in_threadpool(lambda:logging.getLogger('app.service').info('Thread event'))
            await asyncio.sleep(.01)
            return {'ok':True}
        @app.get('/health')
        async def health(): return {'ok':True}
        failure=RuntimeError(SECRET)
        @app.get('/crash')
        async def crash(): raise failure
        @app.get('/stream')
        async def stream():
            async def chunks():
                yield b'first'
                raise RuntimeError(SECRET)
            return StreamingResponse(chunks())
        app.mount('/static',StaticFiles(directory=self.root),name='static')
        install_request_logging(app)
        return app, failure

    def test_http_codes_validation_static_and_exception_deduplication(self):
        app,failure=self.app()
        with TestClient(app,raise_server_exceptions=False) as client:
            for code in [400,401,403,404,405,422,500,503]:
                response=client.get('/status/'+str(code)+'?token='+SECRET,headers={'Authorization':SECRET,'Cookie':'token='+SECRET})
                self.assertEqual(response.status_code,code)
                row=self.records()[-1]
                self.assertEqual(row['level'],'ERROR' if code>=500 else 'WARNING')
                self.assertEqual(row['request_id'],response.headers['X-Request-ID'])
                self.assertNotIn('?',row['path'])
            self.assertEqual(client.get('/number?value='+SECRET).status_code,422)
            self.assertEqual(client.post('/ok',json={'password':SECRET}).status_code,405)
            self.assertEqual(client.get('/static/missing.txt').status_code,404)
            self.assertEqual(client.get('/http-error').status_code,400)
            crash=client.get('/crash')
            self.assertEqual(crash.status_code,500)
            self.assertTrue(crash.headers['X-Request-ID'])
            repeated=client.get('/crash')
            self.assertNotEqual(crash.headers['X-Request-ID'],repeated.headers['X-Request-ID'])
        logging.getLogger('uvicorn.error').error('Exception in ASGI application',exc_info=(RuntimeError,failure,failure.__traceback__))
        technical=[row for row in self.records() if row.get('event')=='unhandled_exception']
        self.assertEqual(len(technical),2)
        self.assertNotIn(SECRET,json.dumps(self.records())+self.console.getvalue())

    def test_caught_query_error_in_http_200_keeps_request_context(self):
        app=FastAPI()
        engine=self.engine()
        @app.get('/caught-query')
        def caught(request:Request):
            request.state.principal=SimpleNamespace(username='tester')
            with LoggingSession(engine) as db:
                try:
                    db.execute(text('SELECT :secret FROM missing_http_table').execution_options(query_name='http.caught'), {'secret':SECRET})
                except Exception:
                    logging.getLogger('app.service').exception('Query handled')
            return HTMLResponse('Formularfehler',status_code=200)
        install_request_logging(app)
        with TestClient(app) as client:
            response=client.get('/caught-query')
        rows=self.records()
        self.assertEqual([row['event'] for row in rows], ['database_error','http_completed'])
        self.assertTrue(all(row['request_id']==response.headers['x-request-id'] for row in rows))
        self.assertTrue(all(row['user']=='tester' for row in rows))
        self.assertEqual(rows[-1]['status'],200)
        self.assertNotIn(SECRET,json.dumps(rows)+self.console.getvalue())

    def test_stream_error_keeps_sent_status_and_health_success_is_debug(self):
        app,_=self.app()
        with TestClient(app,raise_server_exceptions=False) as client:
            client.get('/health'); self.assertEqual(self.records(),[])
            response=client.get('/stream')
        self.assertEqual(response.status_code,200)
        summary=next(row for row in self.records() if row.get('event')=='http_completed')
        self.assertEqual(summary['status'],200); self.assertEqual(summary['outcome'],'stream_failed')
        self.assertEqual(summary['level'],'ERROR')
        self.assertNotIn(SECRET,json.dumps(self.records()))

    def test_parallel_request_and_thread_contexts_are_isolated(self):
        app,_=self.app()
        async def requests():
            async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app),base_url='http://test') as client:
                return await asyncio.gather(*[client.get('/ok?secret='+SECRET,headers={'x-test-user':user,'X-Request-ID':SECRET}) for user in ['alpha','beta']])
        responses=asyncio.run(requests())
        ids={response.headers['x-request-id'] for response in responses}
        self.assertEqual(len(ids),2); self.assertNotIn(SECRET,ids)
        users={row['request_id']:row['user'] for row in self.records() if row.get('event')=='http_completed'}
        self.assertEqual(set(users.values()),{'alpha','beta'})
        for row in self.records(): self.assertEqual(row['user'],users[row['request_id']])
        self.assertIsNone(request_context.get())

    def test_early_auth_errors_and_htmx_login_are_observed(self):
        app=FastAPI()
        @app.post('/write')
        def write(): return {'ok':True}
        app.add_middleware(AuthenticationMiddleware)
        install_request_logging(app)
        entitlement=(Entitlement('1','10','SC','Mitarbeiter',100),)
        with patch.multiple(settings,TRUSTED_PROXY_IPS=['testclient'],DEBUG_AUTH=False,COOKIE_SECURE=False), \
             patch('app.auth.SessionLocal',return_value=MagicMock()), \
             patch('app.auth.load_entitlements',return_value=entitlement), \
             patch('app.auth.lookup_directory_user',return_value=None), TestClient(app) as client:
            denied=client.post('/write',headers={'X-Remote-User':'invalid'})
            self.assertEqual(denied.status_code,401)
            login=client.post('/write',headers={'HX-Request':'true'})
            self.assertEqual(login.status_code,200); self.assertIn('HX-Redirect',login.headers)
            token=create_access_token(username='tester',name='Tester',email='',role='Mitarbeiter',auth_source='ldap',csrf='csrf')
            client.cookies.set(settings.SESSION_COOKIE_NAME,token)
            self.assertEqual(client.post('/write').status_code,403)
        statuses=[row['status'] for row in self.records() if row.get('event')=='http_completed']
        self.assertEqual(statuses,[401,200,403])
        self.assertTrue(any(row.get('event')=='auth_login_required' for row in self.records()))

    def test_retention_rotation_parallel_processes_and_fallback(self):
        old=self.root/'application.log.2001-01-01'; old.write_text('expired',encoding='utf-8')
        preserved=self.root/'webservice.log'; preserved.write_text('keep',encoding='utf-8')
        handler=next(handler for handler in logging.getLogger().handlers if isinstance(handler,ApplicationFileHandler))
        handler.cleanup_archives(); self.assertFalse(old.exists()); self.assertTrue(preserved.exists())
        logging.getLogger('app.test').info('before rotation')
        handler.clh._do_lock()
        try: handler.doRollover()
        finally: handler.clh._do_unlock()
        logging.getLogger('app.test').info('after rotation')
        self.assertEqual(self.records()[-1]['message'],'after rotation')
        with patch.object(handler.clh,'do_write',side_effect=PermissionError(SECRET)), patch('sys.stderr',new=io.StringIO()) as fallback:
            logging.getLogger('app.test').error('safe failure')
        self.assertIn('Konsolenfallback',fallback.getvalue()); self.assertNotIn(SECRET,fallback.getvalue())
        directory=self.root/'workers'; directory.mkdir()
        seed=ApplicationFileHandler(str(directory/'application.log'))
        seed.setFormatter(setup.JsonFormatter()); seed.addFilter(setup.ContextFilter())
        seed.handle(logging.LogRecord('app.worker',logging.INFO,__file__,0,'Before shared rotation',(),None))
        seed.clh._do_lock()
        try:
            seed.rolloverAt=int(datetime.now(timezone.utc).timestamp())-10
            seed.write_rollover_time()
        finally:
            seed.clh._do_unlock()
            seed.close()
        processes=[multiprocessing.get_context('spawn').Process(target=process_writer,args=(str(directory),n)) for n in range(3)]
        for process in processes: process.start()
        for process in processes:
            process.join(20)
            if process.is_alive(): process.terminate(); self.fail('Logging worker timed out')
            self.assertEqual(process.exitcode,0)
        rows=[json.loads(line) for line in (directory/'application.log').read_text(encoding='utf-8').splitlines()]
        self.assertEqual(len(rows),105); self.assertEqual(len({row['count'] for row in rows}),105)
        archived=[json.loads(line) for archive in directory.glob('application.log.*') for line in archive.read_text(encoding='utf-8').splitlines()]
        self.assertEqual([row['message'] for row in archived],['Before shared rotation'])

    def test_cli_failure_and_disabled_sql_text(self):
        with patch('app.logging_setup.configure_logging'), self.assertRaises(SystemExit):
            run_cli(lambda: (_ for _ in ()).throw(RuntimeError(SECRET)))
        self.assertEqual(self.records()[-1]['event'],'cli_failed')
        with patch.object(settings,'LOG_SQL_STATEMENT',False), LoggingSession(self.engine()) as db:
            with self.assertRaises(Exception): db.execute(text('SELECT * FROM missing'))
        self.assertNotIn('sql',self.records()[-1]); self.assertNotIn(SECRET,self.console.getvalue())

    def test_project_python_has_no_print_calls(self):
        root=Path(__file__).resolve().parent.parent
        paths=list((root/'app').rglob('*.py'))+list(root.glob('*.py'))+list((root/'tests').glob('*.py'))
        for path in paths:
            tree=ast.parse(path.read_text(encoding='utf-8-sig'))
            for node in ast.walk(tree):
                self.assertFalse(isinstance(node,ast.Call) and isinstance(node.func,ast.Name) and node.func.id=='print',str(path))

    def test_file_configuration_failure_falls_back_without_exposing_exception(self):
        output=io.StringIO()
        with patch('app.logging_setup.ApplicationFileHandler',side_effect=PermissionError(SECRET)):
            configure_logging(config(self.root/'denied',LOG_TO_CONSOLE=False),console_stream=output)
        logging.getLogger('app.service').error('Still available')
        self.assertIn('Konsolenfallback',output.getvalue())
        self.assertIn('Still available',output.getvalue())
        self.assertNotIn(SECRET,output.getvalue())

    def test_cleanup_queries_are_parameterized_and_cli_imports_work(self):
        import app.cleanup_comments as cleanup
        import filter_export
        rows=[SimpleNamespace(Client_DB='1',Beleg_Nr='R1',Timestamp=datetime(2026,1,1),
                              Benutzer_ID='tester',Kommentar='AZ: 123/45 Schaden '+SECRET+' bei '+SECRET+' eingereicht')]
        db=MagicMock()
        db.execute.return_value.fetchall.return_value=rows
        with patch('app.cleanup_comments.SessionLocal',return_value=db):
            db.__enter__.return_value=db
            cleanup.main()
            cleanup.get_details_from_comments()
        writes=[call for call in db.execute.call_args_list if 'INSERT' in str(call.args[0]) or 'UPDATE' in str(call.args[0])]
        self.assertEqual(len(writes),2)
        for call in writes:
            self.assertNotIn(SECRET,str(call.args[0]))
            self.assertIn(SECRET,str(call.args[1]))
        filter_export.import_filter_config2()
        self.assertNotIn(SECRET,self.console.getvalue())

    def test_mail_and_docuware_metadata_never_dump_payloads(self):
        import app.imap as imap
        import app.docuware_search as docuware_search
        payload=SECRET.encode()
        message=SimpleNamespace(subject='WRG123456 '+SECRET,
                                attachments=[SimpleNamespace(filename=SECRET+'.pdf',payload=payload,size=len(payload))])
        db=MagicMock(); db.__enter__.return_value=db
        db.execute.return_value.fetchall.return_value=[('1','WRG123456')]
        mailbox=MagicMock(); mailbox.login.return_value.__enter__.return_value=mailbox
        mailbox.fetch.return_value=[message]
        with patch.dict('os.environ',{'LOCAL_STORAGE':str(self.root/'files')}), \
             patch('app.imap.SessionLocal',return_value=db),patch('app.imap.MailBox',return_value=mailbox):
            imap.main()
        connection=SimpleNamespace(organizations=[SimpleNamespace(file_cabinets=[SECRET],baskets=[SECRET])])
        with patch('app.docuware_search.docuware.connect',return_value=connection):
            docuware_search.main()
        self.assertTrue(any(row.get('event')=='mail_attachment_saved' and row['size_bytes']==len(payload) for row in self.records()))
        self.assertNotIn(SECRET,json.dumps(self.records())+self.console.getvalue())


if __name__=='__main__': unittest.main()