2 Commits
Author SHA1 Message Date
Florian WildeandClaude Opus 5.5 b7577c4552 feat: Regeln kopieren – vorausgefüllt, zunächst deaktiviert
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-15 09:19:59 +02:00
Florian WildeandClaude Opus 5.5 43b634a135 feat: Anbindungen kopieren – inklusive Zugangsdaten
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-12 00:50:00 +02:00
14 changed files with 571 additions and 27 deletions
+21 -1
View File
@@ -8,6 +8,24 @@ Die Version steht an genau einer Stelle: `src/netmover/__init__.py`.
## [Unreleased]
## [0.9.0] – 2026-10-15
### Hinzugefügt
- **Regeln kopieren** (Knopf „Kopieren“): alle Einstellungen inklusive Archiv
übernommen, Name „… (Kopie)“. Die Kopie ist zunächst **deaktiviert**, damit sie nicht
mit dem Original um dieselben Dateien konkurriert; Protokoll und Buchführung
beginnen neu.
## [0.8.0] – 2026-10-12
### Hinzugefügt
- **Anbindungen kopieren** (Knopf „Kopieren“): Formular vorausgefüllt, Name
„… (Kopie)“. Eigene Zugangsdaten der Vorlage werden übernommen (eigenes Exemplar),
solange kein neues Passwort eingegeben wird; ein eigener SSH-Schlüssel wird
weiterverwendet; gespeicherte Zugangsdaten bleiben ausgewählt. „Testen“ im
Kopie-Formular nutzt die übernommenen Zugangsdaten. Der SFTP-Hostschlüssel wird nur
bei gleichem Server und Port übernommen.
## [0.7.0] – 2026-10-12
### Hinzugefügt
@@ -259,7 +277,9 @@ Erste Version.
- Löschen von Zugangsdaten, die noch von einem Mount benutzt werden, trennt den
Mount nicht mehr stillschweigend ab, sondern wird von der Datenbank abgelehnt.
[Unreleased]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.7.0...HEAD
[Unreleased]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.9.0...HEAD
[0.9.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.8.0...v0.9.0
[0.8.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.7.0...v0.8.0
[0.7.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.6.0...v0.7.0
[0.6.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.5.0...v0.6.0
[0.5.0]: https://git.wilde-server.de/florian.wilde/netmover/compare/v0.4.0...v0.5.0
+28 -1
View File
@@ -21,10 +21,37 @@ Zeiten in Ortszeit (Europe/Berlin).
| 2026-10-11 | 00:54 | 01:16 | 0:22 | – | SFTP-Quelle, Zugangsdaten, Anmeldung (0.5.0) |
| 2026-10-11 | 08:43 | 08:55 | 0:12 | – | Anmeldung je Netz (0.6.0) |
| 2026-10-12 | 00:14 | 00:34 | 0:20 | – | Archiv je Regel (0.7.0) |
| **Summe** | | | **3:49** | | |
| 2026-10-12 | 00:46 | 00:55 | 0:09 | – | Anbindungen kopieren (0.8.0) |
| 2026-10-15 | 09:17 | 09:26 | 0:09 | – | Regeln kopieren (0.9.0) |
| **Summe** | | | **4:07** | | |
---
## 2026-10-15 · 09:17–09:26 · Regeln kopieren (0.9.0)
**Erledigt** – Knopf „Kopieren“, Route `/rules/{id}/copy` (GET Formular, POST
speichern), `rule_copy_name`. 5 Tests (362 gesamt); Gegenprobe „Kopie aktiv“ schlägt
an (2 Tests).
**Entscheidung** – Kopie startet deaktiviert: Mit identischer Quelle würden Original
und Kopie sonst sofort parallel dieselben Dateien bearbeiten (doppelte Lieferung bzw.
Kollisionen). Protokoll und Buchführung (`copied_files`, `archived_files`) werden
bewusst nicht mitkopiert – die Kopie ist eine neue Regel.
## 2026-10-12 · 00:46–00:55 · Anbindungen kopieren (0.8.0)
**Erledigt** – Knopf „Kopieren“, Route `/mounts/{id}/copy` (GET Formular, POST
speichern), `save_mount(copy_from=…)`, Formular-Test mit Vorlage. 7 Tests (357 gesamt);
Gegenprobe ohne Übernahme der Zugangsdaten: 3 Tests schlagen an.
**Entscheidungen**
- Eigene Zugangsdaten werden als eigenes Exemplar kopiert (nicht geteilt) – eigene
Zugangsdaten verschwinden mit ihrer Anbindung, eine geteilte Nutzung wäre
dafür der Fall „Zugangsdaten“.
- Eigenes Schlüsselpaar bleibt gleich (schon in `authorized_keys`).
- Hostschlüssel nur bei gleichem Server/Port übernehmen, sonst würde der Test an
einem anderen Server mit „Hostschlüssel geändert“ scheitern.
## 2026-10-12 · 00:14–00:34 · Archiv je Regel (0.7.0)
**Erledigt**
+2
View File
@@ -28,6 +28,8 @@ Legende: ✅ erledigt · 🔨 in Arbeit · 📋 geplant · 💡 Idee (noch nicht
| Version | Thema |
|---|---|
| 0.9.0 | Regeln kopieren (Kopie zunächst deaktiviert) |
| 0.8.0 | Anbindungen kopieren inklusive Zugangsdaten |
| 0.7.0 | Archiv je Regel: Kopie ins Ziel und ins Archiv, Quelle erst danach löschen |
| 0.6.0 | Anmeldung je Netz: ohne Passwort / mit Passwort / abgewiesen, Schutz vor Aussperren |
| 0.5.0 | SFTP als Quelle; gespeicherte Zugangsdaten für SMB/SFTP-Passwörter (Seite „Zugangsdaten“); Anmeldung abschaltbar (vertrauenswürdige Netze / ganz) |
+1 -1
View File
@@ -4,7 +4,7 @@
services:
netmover:
build: .
image: netmover:0.7.0
image: netmover:0.9.0
container_name: netmover
restart: unless-stopped
cap_add:
+30 -1
View File
@@ -72,6 +72,22 @@ Stabilitätsfenster dort mindestens 2 Sekunden betragen.
| **Beim Start automatisch einhängen** | Nach einem Neustart des Servers wieder einhängen. Fällt das NAS aus, versucht NetMover es außerdem jede Minute erneut. |
| **Aktiv** | Deaktivierte Anbindungen werden von Regeln nicht benutzt. |
### Kopieren {#anbindung-kopieren}
Für eine ähnliche Anbindung – etwa eine zweite Freigabe auf demselben NAS – bei der
Vorlage auf **Kopieren** klicken. Das Formular ist mit allen Werten vorausgefüllt
(Name „… (Kopie)“); meist muss nur die Freigabe geändert werden.
- **Zugangsdaten werden übernommen**, solange das Passwortfeld leer bleibt. Die Kopie
bekommt ein eigenes Exemplar – das Löschen der Vorlage betrifft sie nicht.
Gespeicherte Zugangsdaten sind einfach vorausgewählt.
- Ein **eigener SSH-Schlüssel** der Vorlage wird weiter benutzt; ein neuer Eintrag in
`authorized_keys` ist nicht nötig.
- **Testen** funktioniert im Kopie-Formular bereits mit den übernommenen Zugangsdaten.
Tipp: Nutzen mehrere Anbindungen dieselben Zugangsdaten, lohnt sich ein Eintrag unter
[Zugangsdaten](#zugangsdaten) – dann gilt ein neues Passwort sofort überall.
### Testen {#anbindung-testen}
**Testen** prüft drei Dinge einzeln und sagt im Klartext, was nicht klappt:
@@ -210,6 +226,17 @@ Synology-Verwaltungsordner (`@eaDir`, `#recycle`).
| **Modus setzen** | z. B. `0644`; leer = wie die Quelldatei |
| **Quarantäneordner** | Dateien, die mehrmals hintereinander scheitern, werden dorthin verschoben (relativ zur Quell-Anbindung), statt es endlos zu versuchen. Leer = keine Quarantäne. |
### Regel kopieren {#regel-kopieren}
Für eine ähnliche Regel – etwa für einen zweiten Quell-Unterordner – bei der Vorlage
auf **Kopieren** klicken. Alle Einstellungen (Filter, Ziel, Archiv, Sicherheit,
Rechte) sind übernommen, der Name lautet „… (Kopie)“.
Die Kopie ist zunächst **deaktiviert**: Mit gleicher Quelle würde sie sonst sofort mit
dem Original um dieselben Dateien konkurrieren. Also Quelle bzw. Ziel anpassen,
speichern, **Probelauf**, dann **Aktivieren**. Protokoll und Buchführung beginnen bei
der Kopie neu.
### Probelauf {#probelauf}
**Probelauf** zeigt in einer Tabelle, welche Datei wohin ginge – **ohne etwas zu
@@ -412,7 +439,9 @@ deaktiviert, Zugangsdaten müssen neu eingegeben werden.
### Jede Person in einen eigenen Ordner {#beispiel-unterordner}
Scannt jede Person in einen eigenen Unterordner (`Scans/max`, `Scans/erika`): pro
Person eine Regel mit eigenem **Unterordner** in der Quelle und eigenem Ziel.
Person eine Regel mit eigenem **Unterordner** in der Quelle und eigenem Ziel. Am
schnellsten: die erste Regel einrichten, dann **Kopieren** und nur Unterordner und
Ziel ändern.
### Kopie als zusätzliche Sicherung {#beispiel-kopie}
+1 -1
View File
@@ -1,3 +1,3 @@
"""NetMover – NFS/SMB-Mounts verwalten und Dateien zeitgesteuert verschieben."""
__version__ = "0.7.0"
__version__ = "0.9.0"
+48 -5
View File
@@ -294,7 +294,11 @@ def save_mount(
mounts: MountManager,
data: MountInput,
mount_id: int | None = None,
copy_from: Mount | None = None,
) -> tuple[Mount | None, Errors]:
"""``copy_from``: Vorlage beim Kopieren – eigene Zugangsdaten (Passwort bzw.
Schlüsselpaar) werden übernommen, solange kein neues Passwort eingegeben ist;
die Kopie bekommt ein eigenes Exemplar."""
errors: Errors = {}
m = s.get(Mount, mount_id) if mount_id is not None else None
if mount_id is not None and m is None:
@@ -349,6 +353,14 @@ def save_mount(
setattr(m, f, getattr(data, f))
if not data.type.is_kernel_mount:
m.auto_mount = False
if (
mount_id is None
and copy_from is not None
and data.type is MountType.SFTP
and copy_from.type is MountType.SFTP
and (copy_from.server, copy_from.port) == (data.server, data.port)
):
m.host_key = copy_from.host_key # gleicher Server – Hostschlüssel bleibt gültig
# Zugangsdaten: entweder ein gespeicherter Schlüssel aus der Schlüsselverwaltung
# (geteilt, wird hier nie verändert) oder eigene Zugangsdaten nur dieser Anbindung.
@@ -386,21 +398,32 @@ def save_mount(
else:
m.username = None
cred = own
template = (
copy_from.credential
if copy_from is not None
and copy_from.credential is not None
and not copy_from.credential.managed
and copy_from.credential.kind is data.auth
else None
)
if cred is None or cred.kind is not data.auth:
if data.auth is CredentialKind.SSH_KEY:
private, public = generate_ssh_keypair(f"netmover@{m.slug}")
secret, pub = private, public
if template is not None and (data.auth is CredentialKind.SSH_KEY or not data.password):
# Kopie: dasselbe Passwort bzw. Schlüsselpaar (steht schon in authorized_keys)
secret_enc, pub = template.secret_enc, template.public_key
elif data.auth is CredentialKind.SSH_KEY:
private, pub = generate_ssh_keypair(f"netmover@{m.slug}")
secret_enc = secrets.encrypt(private)
else:
if not data.password:
errors["password"] = "Passwort fehlt." # noqa: S105
s.rollback()
return None, errors
secret, pub = data.password, None
secret_enc, pub = secrets.encrypt(data.password), None
if cred is None:
cred = Credential(name=f"mount:{m.slug}", managed=False)
s.add(cred)
cred.kind = data.auth
cred.secret_enc = secrets.encrypt(secret)
cred.secret_enc = secret_enc
cred.public_key = pub
elif data.password and data.auth is CredentialKind.PASSWORD:
cred.secret_enc = secrets.encrypt(data.password)
@@ -421,6 +444,16 @@ def save_mount(
return m, errors
def copy_name(s: Session, name: str) -> str:
"""„NAS (Kopie)“, „NAS (Kopie 2)“, … – frei in der Datenbank."""
base = f"{name} (Kopie)"[:100]
candidate, n = base, 1
while s.scalar(select(Mount.id).where(Mount.name == candidate)) is not None:
n += 1
candidate = f"{name} (Kopie {n})"[:100]
return candidate
def delete_mount(s: Session, mounts: MountManager, mount_id: int) -> None:
m = s.get(Mount, mount_id)
if m is None:
@@ -635,6 +668,16 @@ def parse_rule_form(s: Session, form: Mapping[str, str]) -> tuple[RuleInput, Err
return d, errors
def rule_copy_name(s: Session, name: str) -> str:
"""„Scans (Kopie)“, „Scans (Kopie 2)“, … – frei in der Datenbank."""
base = f"{name} (Kopie)"[:100]
candidate, n = base, 1
while s.scalar(select(Rule.id).where(Rule.name == candidate)) is not None:
n += 1
candidate = f"{name} (Kopie {n})"[:100]
return candidate
def save_rule(s: Session, d: RuleInput, rule_id: int | None = None) -> tuple[Rule | None, Errors]:
dup = s.scalar(select(Rule.id).where(Rule.name == d.name))
if dup is not None and dup != rule_id:
+67 -13
View File
@@ -38,11 +38,13 @@ from netmover.admin import (
AdminError,
MountInput,
RuleInput,
copy_name,
credential_for_test,
delete_mount,
mount_input_from_model,
parse_mount_form,
parse_rule_form,
rule_copy_name,
rule_input_from_model,
save_mount,
save_rule,
@@ -591,10 +593,12 @@ def _mount_form(
errors: dict[str, str],
mount: Mount | None = None,
status_code: int = 200,
copy_from: Mount | None = None,
) -> Response:
owner = mount or copy_from
pub = (
mount.credential.public_key
if mount is not None and mount.credential and not mount.credential.managed
owner.credential.public_key
if owner is not None and owner.credential and not owner.credential.managed
else None
)
return render(
@@ -607,8 +611,9 @@ def _mount_form(
mount=mount,
public_key=pub,
has_secret=bool(
mount is not None and mount.credential is not None and not mount.credential.managed
owner is not None and owner.credential is not None and not owner.credential.managed
),
copy_from=copy_from,
key_options=_key_options(request),
password_options=_password_options(request),
)
@@ -628,16 +633,33 @@ async def mount_edit(request: Request, mount_id: int) -> Response:
return _mount_form(request, mount_input_from_model(m), {}, m)
@protected.get("/mounts/{source_id}/copy", response_class=HTMLResponse)
async def mount_copy(request: Request, source_id: int) -> Response:
"""Formular für eine neue Anbindung, vorausgefüllt aus einer bestehenden."""
with nm(request).session_factory() as s:
src = s.get(Mount, source_id)
if src is None:
return redirect("/mounts")
data = mount_input_from_model(src)
data.name = copy_name(s, src.name)
return _mount_form(request, data, {}, copy_from=src)
@protected.post("/mounts/new", response_class=HTMLResponse)
@protected.post("/mounts/{mount_id}/edit", response_class=HTMLResponse)
async def mount_save(request: Request, mount_id: int | None = None) -> Response:
@protected.post("/mounts/{source_id}/copy", response_class=HTMLResponse)
async def mount_save(
request: Request, mount_id: int | None = None, source_id: int | None = None
) -> Response:
state = nm(request)
data, errors = parse_mount_form(_form_dict(await request.form()))
with session_scope(state.session_factory) as s:
existing = s.get(Mount, mount_id) if mount_id is not None else None
copy_from = s.get(Mount, source_id) if source_id is not None else None
if not errors:
try:
m, errors = save_mount(s, state.secrets, state.mounts, data, mount_id)
m, errors = save_mount(
s, state.secrets, state.mounts, data, mount_id, copy_from=copy_from
)
except AdminError as exc:
errors = {"_": str(exc)}
m = None
@@ -646,7 +668,10 @@ async def mount_save(request: Request, mount_id: int | None = None) -> Response:
if errors or m is None:
s.rollback()
existing = s.get(Mount, mount_id) if mount_id is not None else None
return _mount_form(request, data, errors, existing, status_code=400)
copy_from = s.get(Mount, source_id) if source_id is not None else None
return _mount_form(
request, data, errors, existing, status_code=400, copy_from=copy_from
)
saved_id, name = m.id, m.name
is_sftp_key = m.type is MountType.SFTP and data.auth is CredentialKind.SSH_KEY
flash(request, f"Anbindung „{name}“ gespeichert.")
@@ -668,12 +693,17 @@ async def mount_test_form(request: Request) -> Response:
data, errors = parse_mount_form(form)
if errors:
return render(request, "partials/check_result.html", errors=errors, report=None)
existing_id = int(form["mount_id"]) if form.get("mount_id", "").isdigit() else None
ref = form.get("mount_id") or form.get("copy_from") or ""
existing_id = int(ref) if ref.isdigit() else None
with state.session_factory() as s:
existing = s.get(Mount, existing_id) if existing_id else None
cred = credential_for_test(data, existing, state.secrets, s)
cfg = data.config()
if data.type is MountType.SFTP and existing is not None:
if (
data.type is MountType.SFTP
and existing is not None
and (existing.server, existing.port) == (data.server, data.port)
):
cfg = dataclasses.replace(cfg, host_key=existing.host_key)
report = await run_in_threadpool(state.mounts.check, cfg, cred, None)
return render(request, "partials/check_result.html", report=report, errors={})
@@ -873,6 +903,7 @@ def _rule_form(
errors: dict[str, str],
rule_id: int | None,
status_code: int = 200,
copy_of: str | None = None,
) -> Response:
with nm(request).session_factory() as s:
mounts = s.scalars(select(Mount).order_by(Mount.name)).all()
@@ -886,6 +917,7 @@ def _rule_form(
d=d,
errors=errors,
rule_id=rule_id,
copy_of=copy_of,
src_mounts=mounts,
src_options=[("", "– wählen –")] + [(m.id, f"{m.name} ({m.type.value})") for m in mounts],
dst_options=[("", "– wählen –")] + [(m.id, f"{m.name} ({m.type.value})") for m in mounts],
@@ -913,25 +945,47 @@ async def rule_edit(request: Request, rule_id: int) -> Response:
return _rule_form(request, d, {}, rule_id)
@protected.get("/rules/{source_id}/copy", response_class=HTMLResponse)
async def rule_copy(request: Request, source_id: int) -> Response:
"""Neue Regel, vorausgefüllt aus einer bestehenden – zunächst deaktiviert, damit sie
nicht sofort mit dem Original um dieselben Dateien konkurriert."""
with nm(request).session_factory() as s:
src = s.get(Rule, source_id)
if src is None:
return redirect("/rules")
d = rule_input_from_model(src)
d.name = rule_copy_name(s, src.name)
d.enabled = False
source_name = src.name
return _rule_form(request, d, {}, None, copy_of=source_name)
@protected.post("/rules/new", response_class=HTMLResponse)
@protected.post("/rules/{rule_id}/edit", response_class=HTMLResponse)
async def rule_save(request: Request, rule_id: int | None = None) -> Response:
@protected.post("/rules/{source_id}/copy", response_class=HTMLResponse)
async def rule_save(
request: Request, rule_id: int | None = None, source_id: int | None = None
) -> Response:
state = nm(request)
form = _form_dict(await request.form())
with session_scope(state.session_factory) as s:
source = s.get(Rule, source_id) if source_id is not None else None
copy_of = source.name if source is not None else None
d, errors = parse_rule_form(s, form)
r = None
if not errors:
try:
r, errors = save_rule(s, d, rule_id)
r, errors = save_rule(s, d, rule_id) # Kopie: rule_id ist None → neue Regel
except AdminError as exc:
errors = {"_": str(exc)}
if errors or r is None:
s.rollback()
return _rule_form(request, d, errors, rule_id, status_code=400)
name = r.name
return _rule_form(request, d, errors, rule_id, status_code=400, copy_of=copy_of)
name, enabled = r.name, r.enabled
state.scheduler.sync()
flash(request, f"Regel „{name}“ gespeichert.")
if copy_of and not enabled:
flash(request, "Die Kopie ist deaktiviert. Probelauf machen, dann aktivieren.", "info")
return redirect("/rules")
+4 -2
View File
@@ -1,12 +1,14 @@
{% extends "base.html" %}
{% from "macros.html" import field, select, check %}
{% block title %}{{ 'Anbindung bearbeiten' if mount else 'Neue Anbindung' }}{% endblock %}
{% block title %}{{ 'Anbindung bearbeiten' if mount else ('Anbindung kopieren' if copy_from else 'Neue Anbindung') }}{% endblock %}
{% block content %}
<h1>{{ 'Anbindung „' ~ mount.name ~ '“ bearbeiten' if mount else 'Neue Anbindung' }}</h1>
<h1>{% if mount %}Anbindung „{{ mount.name }}“ bearbeiten{% elif copy_from %}Kopie von „{{ copy_from.name }}“{% else %}Neue Anbindung{% endif %}</h1>
{% if copy_from %}<p class="msg info small">Alle Werte stammen aus „{{ copy_from.name }}“. Zugangsdaten werden übernommen, solange du das Passwort leer lässt – die Kopie bekommt ihr eigenes Exemplar.{% if copy_from.credential and copy_from.credential.public_key and not copy_from.credential.managed %} Der SSH-Schlüssel bleibt derselbe, ein neuer Eintrag in <code>authorized_keys</code> ist nicht nötig.{% endif %}</p>{% endif %}
{% if errors.get('_') %}<p class="msg error">{{ errors['_'] }}</p>{% endif %}
<form class="form" method="post" data-dynamic>
<input type="hidden" name="csrf_token" value="{{ csrf }}">
{% if mount %}<input type="hidden" name="mount_id" value="{{ mount.id }}">{% endif %}
{% if copy_from %}<input type="hidden" name="copy_from" value="{{ copy_from.id }}">{% endif %}
<fieldset><legend>Verbindung <a class="legend-help" href="/help#anbindung-felder" title="Hilfe">?</a></legend><div class="fields">
{{ field('name', 'Name', d.name, errors=errors, attrs='required maxlength="100"', hint=('Verzeichnis: ' ~ mount.mountpoint if mount and mount.mountpoint else '')) }}
{{ select('type', 'Typ', [('cifs', 'SMB/CIFS'), ('nfs4', 'NFS 4'), ('nfs3', 'NFS 3'), ('sftp', 'SFTP')], d.type.value, errors=errors) }}
+1
View File
@@ -22,6 +22,7 @@
{% endif %}
{% endif %}
<a class="btn small" href="/mounts/{{ m.id }}/edit">Bearbeiten</a>
<a class="btn small" href="/mounts/{{ m.id }}/copy" title="Neue Anbindung mit denselben Werten und Zugangsdaten">Kopieren</a>
<form class="inline" method="post" action="/mounts/{{ m.id }}/delete" data-confirm="„{{ m.name }}“ wirklich löschen?">{{ csrf_field(csrf) }}<button class="small danger">Löschen</button></form>
</div>
</div>
+5 -2
View File
@@ -1,8 +1,11 @@
{% extends "base.html" %}
{% from "macros.html" import field, select, check %}
{% block title %}{{ 'Regel bearbeiten' if rule_id else 'Neue Regel' }}{% endblock %}
{% block title %}{{ 'Regel bearbeiten' if rule_id else ('Regel kopieren' if copy_of else 'Neue Regel') }}{% endblock %}
{% block content %}
<h1>{{ 'Regel bearbeiten' if rule_id else 'Neue Regel' }}</h1>
<h1>{% if rule_id %}Regel bearbeiten{% elif copy_of %}Kopie von „{{ copy_of }}“{% else %}Neue Regel{% endif %}</h1>
{% if copy_of %}<p class="msg info small">Alle Einstellungen stammen aus „{{ copy_of }}“. Die Kopie ist zunächst
<strong>deaktiviert</strong>, damit sie nicht mit dem Original um dieselben Dateien konkurriert – Quelle bzw. Ziel
anpassen, speichern, Probelauf machen, dann aktivieren. Protokoll und Buchführung beginnen bei der Kopie neu.</p>{% endif %}
{% if errors.get('_') %}<p class="msg error">{{ errors['_'] }}</p>{% endif %}
{% if not src_mounts %}<p class="msg info">Zuerst eine Anbindung als Quelle <a href="/mounts/new">anlegen</a>.</p>{% endif %}
<form class="form" method="post" data-dynamic>
+1
View File
@@ -18,6 +18,7 @@
<button class="small" hx-post="/rules/{{ r.id }}/dryrun" hx-target="#dry-{{ r.id }}" hx-swap="innerHTML">Probelauf <span class="htmx-indicator">…</span></button>
<form class="inline" method="post" action="/rules/{{ r.id }}/toggle">{{ csrf_field(csrf) }}<button class="small">{{ 'Deaktivieren' if r.enabled else 'Aktivieren' }}</button></form>
<a class="btn small" href="/rules/{{ r.id }}/edit">Bearbeiten</a>
<a class="btn small" href="/rules/{{ r.id }}/copy" title="Neue Regel mit denselben Einstellungen">Kopieren</a>
<form class="inline" method="post" action="/rules/{{ r.id }}/delete" data-confirm="Regel „{{ r.name }}“ und ihr Protokoll löschen?">{{ csrf_field(csrf) }}<button class="small danger">Löschen</button></form>
</div>
</div>
+194
View File
@@ -0,0 +1,194 @@
"""0.8.0: Anbindungen kopieren – inklusive Zugangsdaten."""
from __future__ import annotations
from typing import Any
from fastapi.testclient import TestClient
from sqlalchemy import select
from sqlalchemy.orm import Session
from netmover.admin import MountInput, copy_name, delete_mount, save_mount
from netmover.crypto import SecretBox
from netmover.models import Credential, CredentialKind, Mount, MountType
from netmover.mounting.fake import FakeMountBackend
from netmover.mounting.manager import MountManager
from tests.webutil import post
def _cifs(**kw: object) -> MountInput:
data: dict[str, object] = {
"name": "NAS",
"type": MountType.CIFS,
"server": "10.0.10.6",
"share": "Scans",
"username": "scanner",
"password": "Pa$$wort",
}
data.update(kw)
return MountInput(**data) # type: ignore[arg-type]
def test_copy_reuses_own_password_as_independent_copy(
session: Session, secretbox: SecretBox, manager: MountManager
) -> None:
src, _ = save_mount(session, secretbox, manager, _cifs())
assert src is not None and src.credential is not None
copy, err = save_mount(
session,
secretbox,
manager,
_cifs(name="Archiv", share="Archiv", password=""),
copy_from=src,
)
assert not err and copy is not None and copy.credential is not None
assert copy.credential.id != src.credential.id and not copy.credential.managed
assert secretbox.decrypt(copy.credential.secret_enc) == "Pa$$wort"
assert copy.credential.username == "scanner"
# Vorlage löschen – die Kopie behält ihre Zugangsdaten
delete_mount(session, manager, src.id)
session.flush()
assert secretbox.decrypt(copy.credential.secret_enc) == "Pa$$wort"
def test_copy_with_new_password_uses_new_one(
session: Session, secretbox: SecretBox, manager: MountManager
) -> None:
src, _ = save_mount(session, secretbox, manager, _cifs())
copy, _ = save_mount(
session, secretbox, manager, _cifs(name="B", share="B", password="anders"), copy_from=src
)
assert copy is not None and copy.credential is not None
assert secretbox.decrypt(copy.credential.secret_enc) == "anders"
def test_copy_with_stored_credentials_shares_them(
session: Session, secretbox: SecretBox, manager: MountManager
) -> None:
from netmover.credentials import create_password
cred = create_password(session, secretbox, "NAS-Scanner", "scanner", "x")
src, _ = save_mount(
session, secretbox, manager, _cifs(username="", password="", cred_id=str(cred.id))
)
copy, _ = save_mount(
session,
secretbox,
manager,
_cifs(name="B", share="B", username="", password="", cred_id=str(cred.id)),
copy_from=src,
)
assert (
copy is not None and copy.credential is cred and src is not None and src.credential is cred
)
def test_copy_sftp_keeps_keypair_and_host_key_only_for_same_server(
session: Session, secretbox: SecretBox, manager: MountManager
) -> None:
sftp = {
"type": MountType.SFTP,
"server": "10.0.10.17",
"share": "/consume",
"port": 22,
"username": "paperless",
"auth": CredentialKind.SSH_KEY,
"password": "",
}
src, _ = save_mount(session, secretbox, manager, _cifs(name="Paperless", **sftp))
assert src is not None and src.credential is not None
src.host_key = "ssh-ed25519 AAAAhost"
same, _ = save_mount(
session,
secretbox,
manager,
_cifs(name="Paperless 2", **{**sftp, "share": "/consume2"}),
copy_from=src,
)
other, _ = save_mount(
session,
secretbox,
manager,
_cifs(name="Anderer", **{**sftp, "server": "10.0.10.99"}),
copy_from=src,
)
assert same is not None and other is not None
for m in (same, other):
assert m.credential is not None and m.credential.id != src.credential.id
assert m.credential.public_key == src.credential.public_key # gleiches Schlüsselpaar
assert m.credential.secret_enc == src.credential.secret_enc
assert same.host_key == "ssh-ed25519 AAAAhost" and other.host_key is None
def test_copy_name(session: Session, secretbox: SecretBox, manager: MountManager) -> None:
save_mount(session, secretbox, manager, _cifs())
assert copy_name(session, "NAS") == "NAS (Kopie)"
save_mount(session, secretbox, manager, _cifs(name="NAS (Kopie)", share="x"))
assert copy_name(session, "NAS") == "NAS (Kopie 2)"
# -- Oberfläche --------------------------------------------------------------------------------
MOUNT = {
"name": "NAS",
"type": "cifs",
"server": "nas",
"share": "scans",
"username": "scanner",
"password": "Geheim,123",
"auto_mount": "on",
"enabled": "on",
}
def _mid(state: Any, name: str) -> int:
with state.session_factory() as s:
mid = s.scalar(select(Mount.id).where(Mount.name == name))
assert mid is not None
return mid
def test_copy_button_form_test_and_save(
client: TestClient, state: Any, backend: FakeMountBackend
) -> None:
backend.export("nas", "scans")
backend.export("nas", "archiv")
post(client, "/mounts/new", MOUNT)
src_id = _mid(state, "NAS")
assert f'href="/mounts/{src_id}/copy"' in client.get("/mounts").text
form = client.get(f"/mounts/{src_id}/copy").text
assert "Kopie von „NAS“" in form and 'value="NAS (Kopie)"' in form
assert 'value="scanner"' in form and "Geheim,123" not in form
assert f'name="copy_from" value="{src_id}"' in form
# Testen mit übernommenem Passwort (Feld leer)
copy_form = {
**MOUNT,
"name": "NAS (Kopie)",
"share": "archiv",
"password": "",
"copy_from": str(src_id),
}
r = post(client, "/mounts/test", copy_form)
assert "Einhängen erfolgreich" in r.text
spec = [c for c in backend.calls if c.action == "mount"][-1].spec
assert spec is not None and spec.credentials is not None
assert spec.credentials.password == "Geheim,123"
r = post(client, f"/mounts/{src_id}/copy", copy_form)
assert r.status_code == 303, r.text
with state.session_factory() as s:
copy = s.get(Mount, _mid(state, "NAS (Kopie)"))
assert copy is not None and copy.share == "archiv" and copy.slug == "nas-kopie"
assert copy.credential is not None
assert state.secrets.decrypt(copy.credential.secret_enc) == "Geheim,123"
assert len(s.scalars(select(Credential)).all()) == 2
def test_copy_validation_error_keeps_copy_context(client: TestClient, state: Any) -> None:
post(client, "/mounts/new", MOUNT)
src_id = _mid(state, "NAS")
r = post(client, f"/mounts/{src_id}/copy", {**MOUNT, "name": "NAS", "password": ""})
assert r.status_code == 400 and "bereits vergeben" in r.text and "Kopie von „NAS“" in r.text
+168
View File
@@ -0,0 +1,168 @@
"""0.9.0: Regeln kopieren – vorausgefüllt, zunächst deaktiviert."""
from __future__ import annotations
from pathlib import Path
from typing import Any
from fastapi.testclient import TestClient
from sqlalchemy import select
from netmover.db import session_scope
from netmover.models import (
ArchiveKind,
ArchiveLayout,
CopiedFile,
Mount,
MountType,
Rule,
RuleAction,
Run,
)
from tests.webutil import post
def _setup(state: Any, tmp_path: Path) -> tuple[int, int]:
consume = tmp_path / "consume"
consume.mkdir()
with session_scope(state.session_factory) as s:
m = Mount(
name="NAS",
slug="nas",
type=MountType.CIFS,
server="nas",
share="scans",
mountpoint=str(tmp_path / "mnt" / "nas"),
)
a = Mount(
name="Archiv",
slug="archiv",
type=MountType.CIFS,
server="nas",
share="archiv",
mountpoint=str(tmp_path / "mnt" / "archiv"),
)
s.add_all([m, a])
s.flush()
r = Rule(
name="Scans → Paperless",
src_mount_id=m.id,
src_subpath="max",
dst_path=str(consume),
pattern="*.pdf;*.jpg",
recursive=True,
interval_seconds=45,
stability_seconds=7,
action=RuleAction.COPY,
set_mode="0640",
quarantine_path="_q",
archive_kind=ArchiveKind.MOUNT,
archive_mount_id=a.id,
archive_path="scans",
archive_layout=ArchiveLayout.DATE,
)
s.add(r)
s.flush()
s.add(Run(rule_id=r.id))
s.add(CopiedFile(rule_id=r.id, src_relpath="a.pdf", size=1, mtime_ns=1, dst_path="x"))
return r.id, m.id
def _rule(state: Any, name: str) -> Rule:
with state.session_factory() as s:
r = s.scalars(select(Rule).where(Rule.name == name)).one()
s.expunge(r)
return r
def _form_from_page(html: str) -> dict[str, str]:
"""Wie ein Browser: alle Felder des Formulars mit ihren vorausgefüllten Werten."""
import re
form: dict[str, str] = {}
for m in re.finditer(r'<input[^>]*name="([^"]+)"[^>]*>', html):
tag, name = m.group(0), m.group(1)
if 'type="checkbox"' in tag:
if " checked" in tag:
form[name] = "on"
else:
v = re.search(r'value="([^"]*)"', tag)
form[name] = v.group(1) if v else ""
for m in re.finditer(r'<select[^>]*name="([^"]+)"[^>]*>(.*?)</select>', html, re.S):
sel = re.search(r'<option value="([^"]*)" selected>', m.group(2))
form[m.group(1)] = sel.group(1) if sel else ""
return form
def test_copy_form_is_prefilled_and_disabled(
client: TestClient, state: Any, tmp_path: Path
) -> None:
rid, _ = _setup(state, tmp_path)
assert f'href="/rules/{rid}/copy"' in client.get("/rules").text
html = client.get(f"/rules/{rid}/copy").text
assert "Kopie von „Scans → Paperless“" in html and "deaktiviert" in html
form = _form_from_page(html)
assert form["name"] == "Scans → Paperless (Kopie)"
assert "enabled" not in form # Häkchen „Aktiv“ nicht gesetzt
assert form["pattern"] == "*.pdf;*.jpg" and form["src_subpath"] == "max"
assert form["archive_kind"] == "mount" and form["archive_layout"] == "date"
def test_copy_saves_new_rule_with_all_settings(
client: TestClient, state: Any, tmp_path: Path
) -> None:
rid, _ = _setup(state, tmp_path)
form = _form_from_page(client.get(f"/rules/{rid}/copy").text)
form["src_subpath"] = "erika" # typische Anpassung
r = post(client, f"/rules/{rid}/copy", form)
assert r.status_code == 303, r.text
assert "Die Kopie ist deaktiviert" in client.get("/rules").text
orig, copy = _rule(state, "Scans → Paperless"), _rule(state, "Scans → Paperless (Kopie)")
assert copy.id != orig.id and copy.enabled is False and orig.enabled is True
assert copy.src_subpath == "erika" and orig.src_subpath == "max"
for f in (
"src_mount_id",
"dst_path",
"pattern",
"recursive",
"interval_seconds",
"stability_seconds",
"action",
"set_mode",
"quarantine_path",
"archive_kind",
"archive_mount_id",
"archive_path",
"archive_layout",
):
assert getattr(copy, f) == getattr(orig, f), f
# Protokoll und Buchführung beginnen neu
with state.session_factory() as s:
assert s.scalars(select(Run).where(Run.rule_id == copy.id)).all() == []
assert s.scalars(select(CopiedFile).where(CopiedFile.rule_id == copy.id)).all() == []
assert state.scheduler.next_run(copy.id) is None # deaktiviert → kein Termin
def test_copy_names_count_up(client: TestClient, state: Any, tmp_path: Path) -> None:
rid, _ = _setup(state, tmp_path)
for expected in ("Scans → Paperless (Kopie)", "Scans → Paperless (Kopie 2)"):
form = _form_from_page(client.get(f"/rules/{rid}/copy").text)
assert form["name"] == expected
post(client, f"/rules/{rid}/copy", form)
def test_copy_validation_error_keeps_context(
client: TestClient, state: Any, tmp_path: Path
) -> None:
rid, _ = _setup(state, tmp_path)
form = _form_from_page(client.get(f"/rules/{rid}/copy").text)
form["name"] = "Scans → Paperless" # schon vergeben
r = post(client, f"/rules/{rid}/copy", form)
assert r.status_code == 400 and "bereits vergeben" in r.text and "Kopie von" in r.text
with state.session_factory() as s:
assert len(s.scalars(select(Rule)).all()) == 1
def test_copy_of_missing_rule_redirects(client: TestClient) -> None:
assert client.get("/rules/4711/copy").headers["location"] == "/rules"