|
|
@@ -0,0 +1,411 @@
|
|
|
+"""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
|
|
|
+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.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_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()
|
gc-server3