93 lines
3.5 KiB
Python
93 lines
3.5 KiB
Python
import datetime
|
|
import os
|
|
from unittest import mock
|
|
|
|
import yaml
|
|
|
|
import config as config_module
|
|
from config import Config, verify_password
|
|
from models import get_session, Job
|
|
from scheduler import StreamScheduler
|
|
|
|
|
|
def _add_job(**kw):
|
|
session = get_session()
|
|
try:
|
|
job = Job(name=kw.pop("name", "j"), stream_url="https://e.example/s", schedule_enabled=True, **kw)
|
|
session.add(job)
|
|
session.commit()
|
|
return job.id
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def test_bad_legacy_job_does_not_break_tick(client):
|
|
_add_job(name="kaputt1", schedule_start="00:00", schedule_stop="xx")
|
|
_add_job(name="kaputt2", schedule_start="25:99", schedule_days="foo-bar")
|
|
_add_job(name="kaputt3", schedule_start="00:00", schedule_once=True, schedule_date="kein-datum")
|
|
good = _add_job(name="gut", schedule_start="00:00", schedule_stop="", schedule_days="*")
|
|
|
|
rec = mock.MagicMock()
|
|
sched = StreamScheduler(rec)
|
|
with mock.patch("scheduler.logger") as log:
|
|
sched._tick()
|
|
sched._tick()
|
|
|
|
started = [c.kwargs["job_id"] for c in rec.start_recording.call_args_list]
|
|
assert started == [good]
|
|
# Fehler nur einmal pro Job geloggt, nicht bei jedem Tick
|
|
assert log.error.call_count == 1
|
|
|
|
|
|
def test_started_keys_pruned(client):
|
|
sched = StreamScheduler(mock.MagicMock())
|
|
old = (datetime.date.today() - datetime.timedelta(days=5)).isoformat()
|
|
today = datetime.date.today().isoformat()
|
|
sched._started_keys = {f"1|False|{old}|10:00", f"2|False|{today}|10:00", "kaputt"}
|
|
sched._prune_started_keys(datetime.datetime.now())
|
|
assert sched._started_keys == {f"2|False|{today}|10:00"}
|
|
|
|
|
|
def test_first_start_generates_password(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(config_module, "PBKDF2_ITERATIONS", 1000)
|
|
path = tmp_path / "config.yml"
|
|
cfg = Config(str(path))
|
|
assert cfg.generated_password
|
|
data = yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
assert data["auth"]["enabled"] is True
|
|
assert "password" not in data["auth"]
|
|
assert verify_password(cfg.generated_password, data["auth"]["password_hash"])
|
|
assert cfg.generated_password not in path.read_text(encoding="utf-8")
|
|
if os.name == "posix":
|
|
assert (os.stat(path).st_mode & 0o777) == 0o600
|
|
|
|
# Zweiter Start: kein neues Passwort
|
|
assert Config(str(path)).generated_password is None
|
|
|
|
|
|
def test_plaintext_password_migration(tmp_path):
|
|
path = tmp_path / "config.yml"
|
|
path.write_text("auth:\n enabled: false\n username: admin\n password: altes-passwort\n", encoding="utf-8")
|
|
cfg = Config(str(path))
|
|
assert cfg.generated_password is None
|
|
data = yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
assert "password" not in data["auth"]
|
|
assert data["auth"]["enabled"] is False
|
|
assert verify_password("altes-passwort", data["auth"]["password_hash"])
|
|
|
|
|
|
def test_legacy_default_password_counts_as_unset(tmp_path):
|
|
path = tmp_path / "config.yml"
|
|
path.write_text("auth:\n enabled: false\n username: admin\n password: stream-recorder\n", encoding="utf-8")
|
|
Config(str(path))
|
|
data = yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
assert data["auth"]["password_hash"] == ""
|
|
|
|
|
|
def test_legacy_default_password_kept_when_auth_enabled(tmp_path):
|
|
path = tmp_path / "config.yml"
|
|
path.write_text("auth:\n enabled: true\n username: admin\n password: stream-recorder\n", encoding="utf-8")
|
|
cfg = Config(str(path))
|
|
assert cfg.default_password_active is True
|
|
assert verify_password("stream-recorder", cfg.get("auth", "password_hash"))
|