MajoorWaldi
c7f93f0eab
- Improved URL parameter handling in `appendAssetFilterQueryParams` for better readability and maintainability. - Added error handling for performance marks in `startTimer` to prevent potential issues. - Enhanced `setRuntimeStatePatch` to avoid modifying blocked keys in the runtime state. - Introduced hard caps on recent file and execution start tracking to prevent memory growth. - Reduced workflow cache size to limit memory usage. - Implemented safe escaping for asset IDs in `findAssetElement` to mitigate XSS risks. - Ensured proper grid reference invalidation during asset manager panel cleanup. - Enhanced debounce utility with cancel and flush methods for better control. - Updated SQLite facade to improve error handling and schema population. - Simplified custom roots store error messages for clarity. - Added validation for upload filenames to prevent reserved names and control characters. - Improved path resolution logic to avoid race conditions with symlinks. - Enhanced health check routes with raw socket peer IP validation for security. - Added collision handling in staging file resolution to prevent infinite loops. - Updated tests to reflect changes in query building and validation logic.
40 lines
1.2 KiB
Python
40 lines
1.2 KiB
Python
import pytest
|
|
|
|
from mjr_am_backend.adapters.db import sqlite_facade as s
|
|
|
|
|
|
def test_validate_in_base_query_more_cases():
|
|
ok, msg = s._validate_in_base_query("")
|
|
assert not ok and "empty" in msg.lower()
|
|
|
|
ok2, msg2 = s._validate_in_base_query("SELECT * FROM t")
|
|
assert not ok2 and "{in_clause}" in msg2.lower()
|
|
|
|
ok3, msg3 = s._validate_in_base_query("SELECT * FROM t WHERE {IN_CLAUSE}; DROP TABLE x")
|
|
assert not ok3 and "forbidden" in msg3.lower()
|
|
|
|
ok4, msg4 = s._validate_in_base_query("WITH q AS (SELECT 1) SELECT * FROM q WHERE {IN_CLAUSE}")
|
|
assert ok4 and not msg4
|
|
|
|
|
|
def test_build_in_query_invalid_count_and_column():
|
|
b1, q1, p1 = s._build_in_query("DELETE FROM t WHERE {IN_CLAUSE}", "id", 0)
|
|
assert b1 and q1 == "" and p1 == 0
|
|
|
|
b2, q2, p2 = s._build_in_query("SELECT * FROM t WHERE {IN_CLAUSE}", "id", 0)
|
|
assert b2 and q2 == "" and p2 == 0
|
|
|
|
|
|
def test_async_loop_thread_run_closes_coro_on_deadlock():
|
|
t = s._AsyncLoopThread(run_timeout_s=2.0)
|
|
_ = t.start()
|
|
t._thread_ident = __import__("threading").get_ident()
|
|
c = __import__("asyncio").sleep(0)
|
|
with pytest.raises(RuntimeError):
|
|
t.run(c)
|
|
try:
|
|
c.close()
|
|
except Exception:
|
|
pass
|
|
t.stop()
|