datapipeline 3.2.0: Benachrichtigungen mit Webhook, Slack und E-Mail

2026-10-09 22:27:00 +02:00
17 changed files with 963 additions and 120 deletions
@@ -2,6 +2,19 @@
 Framework `datapipeline` (Gem). Neueste Version oben.
 ## 3.2.0 – 2026-10-09
 - Benachrichtigungen: `notify notifier, on: :failure | :problems | :always` nach einem Lauf, `DataPipeline::Report`
  (Ereignis, Text, JSON-Payload `schema: 1`), Basisklasse `DataPipeline::Notifier` für eigene Kanäle; verworfene
  Zeilen nur mit `include_rows: true`
 - `DataPipeline::Webhook`: POST mit HMAC-SHA256-Signatur, Wiederholung bei 429/5xx (`Retry-After`), Timeout;
  `Webhook.valid_signature?` für Empfänger
 - `DataPipeline::Email`: Bericht als Text-Mail über SMTP (STARTTLS, TLS, Anmeldung), Gem `net-smtp`
 - `DataPipeline::Slack`: `deliver(report)` mit Block Kit, Maskierung von `&`, `<`, `>`, `mention:` bei Abbruch,
  Incoming Webhooks (`webhook:`), Timeout und Wiederholung; Fehler als `DataPipeline::Slack::Error`
 - Notifier aus der Verbindungsdatei (`notifier(:name)`, `type: webhook | slack | email`); `etl --notify=NAME
  [--notify-on=…]` sendet einen Bericht für alle Pipelines eines Skripts
 ## 3.1.0 – 2026-10-09
 - sqlite: `write_row` schreibt gesammelt je `batch_size` Zeilen (Default 1000) in einer Transaktion;
@@ -7,7 +7,8 @@ gemspec
 # optional plugin gems: needed for the plugin tests, loaded only when a plugin is used
 group :plugins do
  gem 'caxlsx', require: false    # excel write
-  gem 'http', require: false      # rest, slack
+  gem 'http', require: false      # rest, slack, webhook
  gem 'net-smtp', require: false  # email
  gem 'pg', require: false        # postgres
  gem 'roo', require: false       # excel read
  gem 'sqlite3', require: false   # sqlite
@@ -1,7 +1,7 @@
 PATH
  remote: .
  specs:
-    datapipeline (3.1.0)
+    datapipeline (3.2.0)
      csv (~> 3.3)
      logger (~> 1.6)
@@ -26,6 +26,10 @@ GEM
    logger (1.7.0)
    marcel (1.2.1)
    minitest (5.27.0)
    net-protocol (0.4.0)
      timeout
    net-smtp (0.5.2)
      net-protocol
    nokogiri (1.19.4-aarch64-linux-gnu)
      racc (~> 1.4)
    nokogiri (1.19.4-aarch64-linux-musl)
@@ -66,6 +70,7 @@ GEM
    sqlite3 (2.9.6-x86_64-darwin)
    sqlite3 (2.9.6-x86_64-linux-gnu)
    sqlite3 (2.9.6-x86_64-linux-musl)
    timeout (0.6.1)
 PLATFORMS
  aarch64-linux
@@ -84,6 +89,7 @@ DEPENDENCIES
  datapipeline!
  http
  minitest (~> 5.25)
  net-smtp
  pg
  rake
  roo
@@ -1,6 +1,6 @@
 # DataPipeline – ETL-Framework in Ruby
-Stand: 2026-10-09 · Gem `datapipeline` 3.1.0 ([CHANGELOG.md](CHANGELOG.md)) · Ruby 3.4 · Lizenz BSD-3-Clause ([LICENSE](LICENSE))
+Stand: 2026-10-09 · Gem `datapipeline` 3.2.0 ([CHANGELOG.md](CHANGELOG.md)) · Ruby 3.4 · Lizenz BSD-3-Clause ([LICENSE](LICENSE))
 `DataPipeline` liest Daten aus einer Quelle, schickt jede Zeile durch eine Kette von Schritten (prüfen, filtern,
 umbauen, anreichern, abgleichen) und schreibt das Ergebnis in ein oder mehrere Ziele. Eine Pipeline ist ein kurzes
@@ -14,7 +14,7 @@ Festbreiten-Dateien und REST-APIs.
 3. [Pipelines schreiben](#3-pipelines-schreiben) – DSL, Schritte, Prüfen, Fehler, Ergebnis
 4. [Plugins](#4-plugins) – mitgelieferte Quellen und Ziele, eigene Plugins, Abhängigkeiten
 5. [Satzarten-Layouts](#5-satzarten-layouts) – Dateien mit Kopf-, Positions- und Schlusssätzen
-6. [Betrieb](#6-betrieb) – Kommandozeile, Verbindungen, Sperre, Abgleich, Historie, Benachrichtigung, Fehlersuche
+6. [Betrieb](#6-betrieb) – Kommandozeile, Verbindungen, Sperre, Abgleich, Historie, Benachrichtigungen, Fehlersuche
 7. [Umstieg](#7-umstieg) – auf 3.1, von 2.x auf 3.0 (`transform` entfernt), von 2.0 auf 2.1
 8. [Entwicklung](#8-entwicklung) – Aufbau, Tests, Gem bauen
 9. [Ausblick](#9-ausblick)
@@ -32,7 +32,8 @@ Festbreiten-Dateien und REST-APIs.
 | Prüfen | Regeln je Spalte (Pflicht, Länge, Format, Werteliste, Typ), eigene Regeln als Block, Satzarten-Layouts; als Fehler oder nur als Warnung |
 | Fehler | Abbruch, Zeile überspringen oder mit Grund in eine Fehlerdatei schreiben |
 | Datenmengen | Batch im Speicher oder Streaming Zeile für Zeile mit konstantem Speicherbedarf; Postgres-Cursor und Schreiben in Paketen |
-| Betrieb | Kommandozeile `etl` mit Exit-Codes, Dry-Run, Limit, inkrementelle Läufe (Wasserzeichen), Sperre gegen parallele Läufe, Upsert für wiederholbare Läufe, Summenabgleich vor dem Schreiben, Laufhistorie, Verbindungsdatei, Slack |
+| Betrieb | Kommandozeile `etl` mit Exit-Codes, Dry-Run, Limit, inkrementelle Läufe (Wasserzeichen), Sperre gegen parallele Läufe, Upsert für wiederholbare Läufe, Summenabgleich vor dem Schreiben, Laufhistorie, Verbindungsdatei |
 | Benachrichtigung | Berichte an Webhooks (signiert), Slack und E-Mail, je Pipeline oder je Skript; eigene Notifier |
 | Dateiformate | Zeichensatz (z. B. Windows-1252) für CSV und Festbreite, BOM für CSV; Satzarten-Dateien schreiben, prüfen und lesen |
 **Anwendungen:** Das Projekt `etl` nutzt das Gem für die PDS → SAP-Schnittstelle (CUDO) und für das Anlegen von
@@ -106,7 +107,7 @@ der Code hier genau der Datei entspricht.
 | [2.3 Inkrementell und gestreamt](#23-inkrementell-und-gestreamt) | Wasserzeichen, Streaming |
 | [2.4 Eigenes Plugin](#24-eigenes-plugin) | eine Quelle selbst schreiben |
 | [2.5 REST-APIs](#25-rest-apis) | Paging, Wiederholung, Pakete, eine URL je Datensatz |
-| [2.6 Produktiver Lauf](#26-produktiver-lauf) | Verbindungsdatei, Sperre, Kurzschritte, Summenabgleich, Upsert, Zeichensatz, Historie, Slack |
+| [2.6 Produktiver Lauf](#26-produktiver-lauf) | Verbindungsdatei, Sperre, Kurzschritte, Summenabgleich, Upsert, Zeichensatz, Historie, Benachrichtigung |
 ### 2.1 Kundenimport
@@ -624,13 +625,11 @@ FileUtils.mkdir_p(out)
 # Zugangsdaten stehen nicht im Skript, sondern in der Verbindungsdatei (Default: config/connections.yml)
 DataPipeline.connections_file = File.expand_path('config/connections.yml', __dir__)
 # Slack nur, wenn ein Token gesetzt ist; so fällt ein fehlender Token schon vor dem Lauf auf
 slack = DataPipeline::Slack.new(channel: '#etl') if ENV['SLACK_TOKEN']
 DataPipeline.new do
  name 'buchungen_export'
  lock "#{out}/.buchungen.lock"              # läuft schon ein Export, bricht dieser ab (Exit-Code 3)
  history "#{out}/runs.ndjson"               # eine JSON-Zeile je Lauf, auch nach Abbruch
  notify DataPipeline::Slack.new(channel: '#etl') if ENV['SLACK_TOKEN']   # Meldung nur bei Problemen
  read container(:csv, file: "#{DATA}/buchungen.csv", csv: { col_sep: ';' })
@@ -662,10 +661,7 @@ DataPipeline.new do
  write container(:csv, file: "#{out}/buchungen_sap.csv", encoding: 'Windows-1252', csv: { col_sep: ';' },
                        columns: %w[belnr buzei datum konto shkzg dmbtr buchungstext])
-  on_finish do |stats|
+  on_finish { |stats| puts "#{stats.name}: #{stats}" }
    puts "#{stats.name}: #{stats}"
    slack&.send_message("*#{stats.name}*: #{stats}") unless stats.ok?
  end
 end.run
 ```
@@ -679,7 +675,7 @@ end.run
 | `totals`, `reconcile` | Summen werden exakt gerechnet; eine Abweichung bricht vor dem Schreiben ab (6.4) |
 | `upsert: [belnr, buzei]` | ein wiederholter Lauf aktualisiert dieselben Zeilen, statt sie doppelt anzulegen (4.2) |
 | `encoding: 'Windows-1252'` | Zeichensatz der Zieldatei (4.4) |
-| `slack&.send_message` | Meldung nur bei Problemen und nur, wenn `SLACK_TOKEN` gesetzt ist (6.8) |
+| `notify DataPipeline::Slack…` | Meldung nur bei Problemen und nur, wenn `SLACK_TOKEN` gesetzt ist (6.8) |
 ```text
 INFO start
@@ -765,6 +761,7 @@ Tippfehler fallen also sofort auf.
 | `on_error(:abort \| :skip \| :collect, to: storage)` | Umgang mit Fehlern in Schritten |
 | `totals 'spalte', …` | Summen über Quelle und Ergebnis |
 | `reconcile { \|stats\| … }` | Abgleich vor dem Schreiben; eine Meldung bricht ab |
 | `notify(notifier, on: :problems)`, `notifier(:name)` | Bericht an Webhook, Slack oder E-Mail (6.8) |
 | `on_finish { \|stats\| … }` | nach jedem Lauf, auch nach einem Abbruch |
 | `name(str)`, `version(str)` | Name und Version des Laufs für Log, Stats und Historie |
@@ -1137,7 +1134,8 @@ Gemfile der Anwendung, die das Plugin nutzt:
 | `:postgres` | `pg` |
 | `:sqlite` | `sqlite3` |
 | `:excel` | `roo` (lesen), `caxlsx` (schreiben) |
-| `:rest`, `DataPipeline::Slack` | `http` |
+| `:rest`, `DataPipeline::Slack`, `DataPipeline::Webhook` | `http` |
 | `DataPipeline::Email` | `net-smtp` |
 | `:csv`, `:json`, `:fixed` | – |
 Eine CSV- oder JSON-Pipeline lädt keines dieser Gems. Fehlt ein Gem, meldet das Plugin bei der ersten Nutzung, welches
@@ -1299,6 +1297,7 @@ bin/etl run cudo.rb --lock=data/.cudo.lock --history=data/runs.ndjson
 | `--dry-run`, `--limit=N` | für alle Pipelines des Skripts |
 | `--lock=FILE` | Sperre für das ganze Skript (6.3) |
 | `--history=FILE` | Laufhistorie für alle Pipelines des Skripts (6.6) |
 | `--notify=NAME`, `--notify-on=WHEN` | ein Bericht für alle Pipelines des Skripts an den Notifier `NAME` aus der Verbindungsdatei; `WHEN` = `problems` (Default), `failure`, `always` (6.8) |
 | `--verbose`, `--quiet` | Log-Level Debug bzw. Warnung (Default: Info) |
 | jede andere `--name=wert` | `params[:name]` und unverändert an das Skript, z. B. `--out=data`, `--incremental=1` |
@@ -1423,81 +1422,181 @@ Eigene Schritte mit Nebenwirkungen (API-Aufrufe in einem `map`) fragen `dry_run?
 `limit n` verarbeitet nur die ersten n Quellzeilen; im Streaming wird das Lesen dann beendet.
-### 6.8 Benachrichtigung über Slack
+### 6.8 Benachrichtigungen
-`DataPipeline::Slack` schickt Text über die Slack-API `chat.postMessage`. Einrichten:
+`notify` schickt nach einem Lauf einen Bericht an einen Notifier: einen Webhook, Slack oder E-Mail. Ein Fehler beim
-
+Senden wird nur geloggt; Ergebnis und Exit-Code bleiben unverändert.
 1. In Slack eine App anlegen (api.slack.com/apps), unter *OAuth & Permissions* den Bot-Scope `chat:write` vergeben,
   die App im Workspace installieren und den *Bot User OAuth Token* (`xoxb-…`) kopieren.
 2. Die App in den Ziel-Channel einladen (`/invite @appname`), sonst antwortet Slack mit `not_in_channel`.
 3. Das Gem `http` ins Gemfile der Anwendung aufnehmen (4.10).
 4. Den Token als Umgebungsvariable `SLACK_TOKEN` setzen oder mit `token:` übergeben; nicht ins Skript schreiben.
 ```ruby
 slack = DataPipeline::Slack.new                                       # Token aus ENV['SLACK_TOKEN'], Channel '#general'
 slack = DataPipeline::Slack.new(channel: '#etl')                      # anderer Channel (Name oder ID wie 'C0123ABCD')
 slack = DataPipeline::Slack.new(channel: '#etl', token: ENV.fetch('SLACK_ETL_TOKEN'))  # eigener Token je Anwendung
 slack.send_message('Test aus der Pipeline', verbose: true)             # verbose: Antwort von Slack als JSON auf $stdout
 ```
 | Option | Default | Bedeutung |
 | --- | --- | --- |
 | `token:` | `ENV['SLACK_TOKEN']` | Bot-Token; fehlt er, bricht schon `Slack.new` mit `ArgumentError` ab |
 | `channel:` | `'#general'` | Ziel-Channel als Name oder ID |
 `send_message(text, verbose: false)` liefert die Antwort von Slack als Hash. Meldet Slack einen Fehler (z. B.
 `invalid_auth`, `channel_not_found`, `not_in_channel`), wirft die Methode eine Exception mit diesem Code. In
 `on_finish` wird sie nur geloggt (`on_finish: RuntimeError: Slack#send_message failed: …`), der Lauf und sein
 Exit-Code bleiben unverändert. Es gibt keine Wiederholung bei Netzwerkfehlern.
 Eine ausführliche Meldung mit Status, Zählern, Laufzeit und den ersten verworfenen Zeilen. Der Notifier wird vor der
 Pipeline angelegt, damit ein fehlender Token schon beim Start auffällt. Slack setzt `*fett*`, `_kursiv_` und
 `` `code` `` um.
 ```ruby
 slack = DataPipeline::Slack.new(channel: '#etl')
 DataPipeline.new do
  name 'cudo_export'
  notify DataPipeline::Webhook.new(url: ENV['ALERT_URL'], secret: ENV['ALERT_SECRET'])   # bei Problemen
  notify DataPipeline::Slack.new(channel: '#etl', mention: '<!here>'), on: :failure
  notify DataPipeline::Email.new(to: 'buchhaltung@example.com'), on: :always
  notify notifier(:alerts)                                                               # aus der Verbindungsdatei
  # … read, Schritte, write …
  on_finish do |stats|
    next if stats.ok? && stats.warnings.zero?                  # nur melden, wenn etwas zu tun ist
    status = if stats.error then ':red_circle: abgebrochen'
             elsif stats.rejected.positive? then ':large_orange_circle: mit verworfenen Zeilen'
             else ':large_yellow_circle: mit Warnungen'
             end
    lines = ["*Buchungsexport #{stats.version}* #{status}#{' (Dry-Run)' if stats.dry_run}",
             "Zeitraum #{params[:from]} – #{params[:till]}, Start #{stats.started_at.strftime('%d.%m.%Y %H:%M')}, " \
             "Dauer #{stats.duration}s",
             "gelesen #{stats.read} · geschrieben #{stats.written} · verworfen #{stats.rejected} · Warnungen #{stats.warnings}"]
    lines << "Fehler: `#{stats.error.class}: #{stats.error.message}`" if stats.error
    stats.errors.first(5).each do |e|
      lines << "• #{e[:step]}: #{e[:error]} – Beleg #{e[:row]['belnr']}"
    end
    lines << "… und #{stats.rejected - 5} weitere, siehe data/fehler.csv" if stats.rejected > 5
    slack.send_message(lines.join("\n"))
  end
 end.run
 ```
-So sieht die Meldung in Slack aus:
+| Option von `notify` | Wirkung |
 | --- | --- |
 | `on: :problems` (Default) | bei Abbruch, verworfenen Zeilen oder Warnungen |
 | `on: :failure` | nur bei Abbruch |
 | `on: :always` | nach jedem Lauf |
 | `title:` | Titel des Berichts, Default der Name der Pipeline |
 | `dry_run: true` | auch nach einem Dry-Run senden (Default: nicht) |
 | `include_rows: true` | die verworfenen Zeilen mitsenden; Default: nur Schritt und Meldung, weil Zeilen Kundendaten enthalten können |
-```text
+**Bericht** – Jeder Notifier bekommt einen `DataPipeline::Report` mit den Stats eines oder mehrerer Läufe:
-Buchungsexport 1.0 🟠 mit verworfenen Zeilen
+
-Zeitraum 2025-11-01 – 2025-11-30, Start 01.12.2025 06:00, Dauer 4.213s
+| Methode | Inhalt |
-gelesen 1250 · geschrieben 1243 · verworfen 7 · Warnungen 0
+| --- | --- |
-• validate#1: konto: missing – Beleg 5100000123
+| `event` | `run.failed` (Abbruch), `run.rejected` (verworfene Zeilen), `run.warnings`, `run.succeeded`; bei mehreren Läufen der schlimmste |
-• validate#1: dmbtr: not decimal – Beleg 5100000124
+| `title`, `status`, `subject` | z. B. `cudo_export`, `mit verworfenen Zeilen`, `cudo_export: mit verworfenen Zeilen` |
-• validate#2: Betrag 0 – Beleg 5100000130
+| `lines`, `to_s` | Text je Lauf: Zähler, Start, Dauer, die ersten 5 Fehler (ohne Zeilen) |
-• validate#1: shkzg: not in ["S", "H"] – Beleg 5100000131
+| `to_h(include_rows: false)` | JSON-Payload für Webhooks (unten) |
-• validate#1: konto: format /\A\d+\z/ – Beleg 5100000140
+| `id`, `host`, `params`, `runs`, `error` | eindeutige Kennung, Rechner, Parameter der Kommandozeile, Stats, Fehler ausserhalb eines Laufs |
-… und 2 weitere, siehe data/fehler.csv
+
 **Ein Bericht für das ganze Skript** – `notify` meldet jede Pipeline einzeln. Für einen Sammelbericht über alle
 Pipelines eines Skripts gibt es die Kommandozeile; der Notifier kommt aus der Verbindungsdatei:
 ```sh
 bin/etl run cudo.rb --notify=alerts                     # bei Problemen (Default)
 bin/etl run cudo.rb --notify=alerts --notify-on=always
 ```
-`params` gibt es im Pipeline-Block; ausserhalb heisst es `DataPipeline.params`.
+Bricht das Skript selbst ab (z. B. ein Fehler vor der ersten Pipeline), meldet der Bericht `run.failed` mit diesem
 Fehler. Bei belegter Sperre (Exit-Code `3`) und im Dry-Run wird nichts gesendet.
 #### Webhook
 `DataPipeline::Webhook` sendet den Bericht als JSON per `POST`.
 | Option | Default | Bedeutung |
 | --- | --- | --- |
 | `url:` | – | Ziel, Pflicht |
 | `secret:` | – | signiert den Body (unten) |
 | `headers:` | `{}` | weitere Header, z. B. `{ 'Authorization' => "Bearer #{ENV['TOKEN']}" }` |
 | `timeout:` | 10 | Sekunden je Anfrage |
 | `retries:`, `retry_wait:` | 3, 2 | Wiederholung bei 429, 5xx und Netzwerkfehlern, Wartezeit `retry_wait` × Versuch; `Retry-After` wird beachtet; andere 4xx brechen sofort ab |
 ```json
 {"schema":1,"id":"3f6c2a1e-…","event":"run.rejected","title":"cudo_export","host":"etl01",
 "params":{"from":"2026-10-08"},"error":null,
 "runs":[{"name":"cudo_export","version":"1.0","started_at":"2026-10-09T06:00:00.123+02:00","duration":4.213,
          "read":1250,"written":1243,"rejected":7,"warnings":0,"ok":false,"error":null,
          "totals":{"dmbtr":{"read":"98231.4","written":"97810.1"}},
          "errors":[{"step":"validate#1","error":"konto: missing"}]}]}
 ```
 `schema: 1` ist ein Versprechen: Felder können dazukommen, eine Änderung bestehender Felder hiesse `schema: 2`. Jede
 Anfrage trägt die Header `X-DataPipeline-Event` und `X-DataPipeline-Delivery` (die `id`, bei Wiederholungen gleich;
 damit erkennt der Empfänger Doppelte). Mit `secret:` kommen dazu:
 ```text
 X-DataPipeline-Timestamp: 1791525600
 X-DataPipeline-Signature: sha256=<HMAC-SHA256(secret, "<timestamp>.<body>") hexadezimal>
 ```
 Der Empfänger prüft Signatur und Alter (gegen wiederholt eingespielte Anfragen); in Ruby z. B.:
 ```ruby
 ok = DataPipeline::Webhook.valid_signature?(secret: ENV['ALERT_SECRET'], body: request.raw_post,
                                            timestamp: request.headers['X-DataPipeline-Timestamp'],
                                            signature: request.headers['X-DataPipeline-Signature'],
                                            max_age: 300)
 ```
 #### Slack
 `DataPipeline::Slack` meldet einen Bericht als Statuszeile mit Symbol (🔴 abgebrochen, 🟠 verworfene Zeilen,
 🟡 Warnungen, 🟢 ok) und darunter den Text je Lauf. `&`, `<` und `>` aus Daten und Meldungen werden maskiert.
 Zwei Wege zu Slack:
 | Weg | Einrichten | Aufruf |
 | --- | --- | --- |
 | Incoming Webhook (einfacher) | Slack-App anlegen (api.slack.com/apps), *Incoming Webhooks* aktivieren, Webhook für den Channel erzeugen | `Slack.new(webhook: ENV['SLACK_WEBHOOK'])` |
 | Bot-Token | Slack-App mit Bot-Scope `chat:write` installieren, *Bot User OAuth Token* (`xoxb-…`) kopieren, App in den Channel einladen (`/invite @appname`) | `Slack.new(channel: '#etl')`, Token aus `ENV['SLACK_TOKEN']` |
 | Option | Default | Bedeutung |
 | --- | --- | --- |
 | `webhook:` | – | URL eines Incoming Webhook; dann braucht es weder Token noch Channel |
 | `token:` | `ENV['SLACK_TOKEN']` | Bot-Token; ohne Token und ohne Webhook bricht schon `Slack.new` mit `ArgumentError` ab |
 | `channel:` | `'#general'` | Channel als Name oder ID (nur mit Token) |
 | `mention:` | – | z. B. `'<!here>'` oder `'<!subteam^ID>'`, nur bei `run.failed` |
 | `timeout:`, `retries:`, `retry_wait:` | 10, 3, 2 | wie beim Webhook |
 Eigener Text geht weiter mit `send_message`, z. B. in `on_finish`:
 ```ruby
 slack = DataPipeline::Slack.new(channel: '#etl')
 slack.send_message('Monatsabschluss exportiert', verbose: true)    # verbose: Antwort von Slack auf $stdout
 ```
 Meldet Slack einen Fehler (`invalid_auth`, `channel_not_found`, `not_in_channel`), wirft der Aufruf
 `DataPipeline::Slack::Error` mit diesem Code.
 #### E-Mail
 `DataPipeline::Email` sendet den Bericht als Text-Mail (UTF-8) über SMTP. Betreff: `[etl] <Titel>: <Status>`. Die
 Zugangsdaten kommen aus der Umgebung oder als Option:
 | Option | Default | Bedeutung |
 | --- | --- | --- |
 | `to:` | – | eine Adresse oder ein Array, Pflicht |
 | `from:` | `ENV['SMTP_FROM']`, sonst `etl@<rechner>` | Absender |
 | `host:`, `port:` | `ENV['SMTP_HOST']` bzw. `localhost`; `ENV['SMTP_PORT']` bzw. 587 | SMTP-Server |
 | `username:`, `password:` | `ENV['SMTP_USERNAME']`, `ENV['SMTP_PASSWORD']` | Anmeldung (LOGIN); ohne Benutzer keine Anmeldung |
 | `starttls:` | `:auto` | `:auto` (wenn der Server es anbietet), `:always` oder `false` (dann Default-Port 25) |
 | `tls:` | `false` | `true` = TLS von Anfang an (Port 465) |
 | `subject_prefix:` | `'[etl] '` | Anfang des Betreffs |
 | `timeout:` | 15 | Sekunden für Verbindung und Antwort |
 Mit `include_rows: true` stehen die verworfenen Zeilen als JSON am Ende der Mail. Eigener Text:
 `DataPipeline::Email.new(to: …).send_mail('Betreff', "Text")`. Ein Fehler beim Senden wirft
 `DataPipeline::Email::Error`. Das Plugin braucht das Gem `net-smtp`.
 #### Notifier in der Verbindungsdatei
 Wie Storages lassen sich Notifier in `config/connections.yml` ablegen (6.2); `type` ist `webhook`, `slack` oder
 `email`, die anderen Schlüssel sind die Optionen:
 ```yaml
 alerts:
  type: webhook
  url: <%= ENV['ALERT_URL'] %>
  secret: <%= ENV['ALERT_SECRET'] %>
 buchhaltung_mail:
  type: email
  to: [buchhaltung@example.com, etl@example.com]
  host: smtp.example.com
  username: etl@example.com
  password: <%= ENV['SMTP_PASSWORD'] %>
 ```
 ```ruby
 notify notifier(:buchhaltung_mail), on: :always
 ```
 #### Eigene Notifier
 Ein Notifier ist eine Klasse mit Basisklasse `DataPipeline::Notifier` und einer Methode
 `deliver(report, include_rows: false)`, z. B. für Microsoft Teams oder ein Ticketsystem:
 ```ruby
 require 'http'
 class TeamsNotifier < DataPipeline::Notifier
  def initialize(url:) = @url = url
  def deliver report, include_rows: false
    HTTP.post(@url, json: { title: report.subject, text: report.lines.join("\n\n") })
  end
 end
 notify TeamsNotifier.new(url: ENV['TEAMS_WEBHOOK']), on: :failure
 ```
 ### 6.9 Fehlersuche
@@ -1523,11 +1622,19 @@ schnellen Blick auf das Ergebnis.
 ## 7. Umstieg
-### 7.0 Von 3.0 auf 3.1
+### 7.0 Von 3.0 auf 3.1 und 3.2
-3.1 ist abwärtskompatibel. Anders als in 3.0: SQLite schreibt im Streaming (`write_row`) nicht mehr jede Zeile
+3.1 und 3.2 sind abwärtskompatibel. Anders als in 3.0:
-einzeln, sondern je 1000 Zeilen in einer Transaktion (4.3). Wer während eines Laufs mit einer zweiten Verbindung
+
-auf die Zieldatei schaut, sieht die Zeilen also paketweise. Das Gem steht jetzt unter der BSD-3-Clause-Lizenz.
+| Thema | bis 3.0 | jetzt | seit |
 | --- | --- | --- | --- |
 | SQLite, `write_row` im Streaming | jede Zeile einzeln | je 1000 Zeilen in einer Transaktion (4.3); eine zweite Verbindung sieht die Zeilen paketweise | 3.1 |
 | Lizenz | nicht festgelegt | BSD-3-Clause | 3.1 |
 | Slack-Fehler | `RuntimeError` | `DataPipeline::Slack::Error` (weiterhin ein `StandardError`) | 3.2 |
 | `Slack.new` | lädt `http` erst beim Senden | lädt `http` sofort; Timeout 10 s und Wiederholung bei 429/5xx | 3.2 |
 | `send_message(…, verbose: true)` | Antwort als eingerücktes JSON | Antwort wie von Slack geliefert | 3.2 |
 Selbstgebaute Slack-Meldungen in `on_finish` laufen weiter; einfacher ist `notify` (6.8).
 ### 7.1 Von 2.x auf 3.0
@@ -1584,7 +1691,7 @@ ersten Zeilen sollten gleich sein.
 | [lib/datapipeline/lock.rb](lib/datapipeline/lock.rb) | Sperre gegen parallele Läufe |
 | [lib/datapipeline/connections.rb](lib/datapipeline/connections.rb) | Verbindungsdatei |
 | [lib/datapipeline/storage/](lib/datapipeline/storage/) | `storage_plugin.rb` (Basisklasse) und je Plugin eine Datei |
-| [lib/datapipeline/notification/slack.rb](lib/datapipeline/notification/slack.rb) | Slack |
+| [lib/datapipeline/notification/](lib/datapipeline/notification/) | `notifier.rb` (Report, Basisklasse), `webhook.rb`, `slack.rb`, `email.rb` |
 | [lib/datapipeline/version.rb](lib/datapipeline/version.rb) | `DataPipeline::VERSION` |
 | [exe/etl](exe/etl) | Kommando `etl`; [bin/etl](bin/etl) startet es im Repository mit dem Bundle |
 | [examples/](examples/) | Beispiele aus Kapitel 2, Daten in `examples/data/`, Verbindungsdatei in `examples/config/`, Demo-API |
@@ -1592,14 +1699,14 @@ ersten Zeilen sollten gleich sein.
 ### 8.2 Tests
-`bundle exec rake test` startet die Minitest-Suite in [test/](test/), rund 190 Fälle. Einzelne Datei:
+`bundle exec rake test` startet die Minitest-Suite in [test/](test/), rund 210 Fälle. Einzelne Datei:
 `bundle exec rake test TEST=test/core/steps_test.rb`.
 | Ordner | Inhalt |
 | --- | --- |
-| `test/plugins` | je Plugin das gemeinsame Testkit `test/support/storage_contract.rb` (Schreiben und Zurücklesen, zeilenweise, als Quelle und Ziel) plus eigene Fälle; REST gegen einen lokalen Testserver; Laden der Gems erst bei Nutzung |
+| `test/plugins` | je Plugin das gemeinsame Testkit `test/support/storage_contract.rb` (Schreiben und Zurücklesen, zeilenweise, als Quelle und Ziel) plus eigene Fälle; REST, Webhook und Slack gegen einen lokalen HTTP-Testserver, E-Mail gegen einen SMTP-Testserver (`test/support/`); Laden der Gems erst bei Nutzung |
 | `test/core` | DSL, Schritte und Kurzschritte, Batch und Streaming, Fehlerbehandlung, Layouts, Sperre, Verbindungen, Summen, Historie, Wasserzeichen |
-| `test/cli` | `bin/etl` als Prozess: Parameter, Dry-Run, Sperre, Historie, Exit-Codes |
+| `test/cli` | `bin/etl` als Prozess: Parameter, Dry-Run, Sperre, Historie, Sammelbericht `--notify`, Exit-Codes |
 | `test/examples` | die Beispiele aus Kapitel 2: Code in der Doku = Datei, Lauf und Ergebnis; REST gegen die Demo-API |
 Eine `.env` wird in Tests nicht geladen; kein Test erreicht eine Produktivdatenbank. Postgres-Tests legen je Test eine
@@ -1620,10 +1727,10 @@ Ein eigenes Plugin lässt sich mit `MemoryStorage` aus `test/test_helper.rb` als
 - Version in [lib/datapipeline/version.rb](lib/datapipeline/version.rb), Änderungen in [CHANGELOG.md](CHANGELOG.md)
  (neueste oben). Neue Funktionen erhöhen die mittlere Stelle, Breaking Changes die erste (wie 3.0, 7.1).
- Jede Version bekommt ein Tag `vX.Y.Z` (`git tag -a v3.1.0 -m "datapipeline 3.1.0" && git push origin v3.1.0`).
+- Jede Version bekommt ein Tag `vX.Y.Z` (`git tag -a v3.2.0 -m "datapipeline 3.2.0" && git push origin v3.2.0`).
  Anwendungen binden das Gem mit einer Versionsgrenze ein (`'~> 3.1'`, 1.2) oder fest auf ein Tag
-  (`tag: 'v3.1.0'`; dann geht `local.datapipeline` nicht).
+  (`tag: 'v3.2.0'`; dann geht `local.datapipeline` nicht).
- Das `Gemfile` bindet das Gem über `gemspec` ein und ergänzt die Gruppen `plugins` (`pg`, `sqlite3`, `roo`,
+- Das `Gemfile` bindet das Gem über `gemspec` ein und ergänzt die Gruppen `plugins` (`pg`, `sqlite3`, `roo`, `net-smtp`,
  `caxlsx`, `http`) und `test`; `Gemfile.lock` ist versioniert. Ohne Plugin-Gems: `bundle config set without plugins`.
 - Gem bauen: `gem build datapipeline.gemspec`.
 - Lizenz: BSD-3-Clause, siehe [LICENSE](LICENSE); die Datei ist Teil des Gems.
@@ -1640,6 +1747,5 @@ Offene Punkte, Reihenfolge = empfohlene Umsetzung:
   schreiben), `etl list` (Pipelines, Wasserzeichen, letzter Lauf aus der Historie), `--json` für die Stats.
 2. **Qualität:** CI auf git.wicom.li mit `rake test` und Postgres-Service, RuboCop, `DataPipeline::Testing` mit
   `MemoryStorage` als öffentliche Testhilfe für Anwendungen.
-3. **Slack:** fertige Meldung aus den Stats (`notify slack, on: :problems`), Timeout und Wiederholung bei `429`,
+3. **Benachrichtigungen:** Microsoft Teams als mitgelieferter Notifier, HTML-Mail.
   Maskierung von `&`, `<`, `>`, Incoming Webhooks.
 4. **Weitere Plugins:** SFTP-Ablage.
@@ -28,5 +28,5 @@ Gem::Specification.new do |spec|
  spec.add_dependency 'logger', '~> 1.6'
  # optional, loaded only when the plugin is used (add them to the Gemfile of the application):
-  #   pg (postgres), sqlite3 (sqlite), roo + caxlsx (excel), http (rest, slack)
+  #   pg (postgres), sqlite3 (sqlite), roo + caxlsx (excel), http (rest, slack, webhook), net-smtp (email)
 end
@@ -14,13 +14,11 @@ FileUtils.mkdir_p(out)
 # Zugangsdaten stehen nicht im Skript, sondern in der Verbindungsdatei (Default: config/connections.yml)
 DataPipeline.connections_file = File.expand_path('config/connections.yml', __dir__)
 # Slack nur, wenn ein Token gesetzt ist; so fällt ein fehlender Token schon vor dem Lauf auf
 slack = DataPipeline::Slack.new(channel: '#etl') if ENV['SLACK_TOKEN']
 DataPipeline.new do
  name 'buchungen_export'
  lock "#{out}/.buchungen.lock"              # läuft schon ein Export, bricht dieser ab (Exit-Code 3)
  history "#{out}/runs.ndjson"               # eine JSON-Zeile je Lauf, auch nach Abbruch
  notify DataPipeline::Slack.new(channel: '#etl') if ENV['SLACK_TOKEN']   # Meldung nur bei Problemen
  read container(:csv, file: "#{DATA}/buchungen.csv", csv: { col_sep: ';' })
@@ -52,8 +50,5 @@ DataPipeline.new do
  write container(:csv, file: "#{out}/buchungen_sap.csv", encoding: 'Windows-1252', csv: { col_sep: ';' },
                        columns: %w[belnr buzei datum konto shkzg dmbtr buchungstext])
-  on_finish do |stats|
+  on_finish { |stats| puts "#{stats.name}: #{stats}" }
    puts "#{stats.name}: #{stats}"
    slack&.send_message("*#{stats.name}*: #{stats}") unless stats.ok?
  end
 end.run
@@ -11,6 +11,8 @@
 #   --verbose                 log level debug, --quiet only warnings (default info)
 #   --lock=FILE               exclusive lock for the whole script, exit 3 when another run holds it
 #   --history=FILE            one json line per pipeline run (DataPipeline.history)
 #   --notify=NAME             one report for all runs of the script, notifier NAME from the connections file
 #   --notify-on=WHEN          failure, problems (default) or always
 #   --name=value              any other option: DataPipeline.params[:name] and passed on to the script
 #
 # exit code: 0 ok, 1 aborted, 2 finished with rejected rows, 3 locked by another run
@@ -35,6 +37,8 @@ parser = OptionParser.new do |opts|
  opts.on('--quiet') { options[:settings][:log_level] = Logger::WARN }
  opts.on('--lock=FILE', String) { |v| options[:lock] = v }
  opts.on('--history=FILE', String) { |v| options[:settings][:history] = v }
  opts.on('--notify=NAME', String) { |v| options[:notify] = v }
  opts.on('--notify-on=WHEN', String) { |v| options[:notify_on] = v }
 end
 # own options for the parser, everything else (also --name=value of the script) goes to the script
@@ -42,9 +46,9 @@ own = []
 args = []
 argv = ARGV.dup
 while (arg = argv.shift)
-  if arg.match?(/\A--(from|till|param|dry-run|limit|verbose|quiet|lock|history)(=|\z)/)
+  if arg.match?(/\A--(from|till|param|dry-run|limit|verbose|quiet|lock|history|notify|notify-on)(=|\z)/)
    own << arg
-    own << argv.shift if %w[--param --limit --from --till --lock --history].include?(arg)
+    own << argv.shift if %w[--param --limit --from --till --lock --history --notify --notify-on].include?(arg)
  else
    args << arg
  end
@@ -56,7 +60,26 @@ DataPipeline.params.merge!(DataPipeline.parse_params(args)) { |_key, own_value,
 abort parser.help if script.nil?
 abort "etl: #{script} not found\n\n#{parser.help}" unless File.exist?(script)
-DataPipeline.settings(options[:settings].merge(name: File.basename(script, '.rb')))
+name = File.basename(script, '.rb')
 DataPipeline.settings(options[:settings].merge(name:))
 # one report for the whole script; created before the run, so a wrong name fails early
 if options[:notify]
  begin
    notifier = DataPipeline::Notifier.from_connection(options[:notify])
    notify_when = DataPipeline::Notifier.condition(options.fetch(:notify_on, :problems))
  rescue ArgumentError, LoadError => e
    abort "etl: --notify=#{options[:notify]}: #{e.message}"
  end
 end
 summary = lambda do |error = nil|
  return unless notifier && !options[:settings][:dry_run]
  report = DataPipeline::Report.new(DataPipeline.runs, title: name, params: DataPipeline.params, error:)
  notifier.deliver(report) if notify_when.call(report)
 rescue StandardError => e
  warn "#{name}: notify #{options[:notify]}: #{e.class}: #{e.message}"
 end
 begin
  script_lock = DataPipeline::Lock.new(options[:lock]).acquire if options[:lock]   # held until the process ends
@@ -72,9 +95,12 @@ begin
  load File.expand_path(script)
 rescue StandardError => e
  warn "#{File.basename(script)}: #{e.class}: #{e.message}"
  # an aborted pipeline is already in the runs, other errors (script, lock) are reported as such
  summary.call(DataPipeline.runs.any? { |stats| stats.error.equal?(e) } ? nil : e) unless e.is_a?(DataPipeline::LockError)
  exit(e.is_a?(DataPipeline::LockError) ? 3 : 1)
 end
 summary.call
 runs = DataPipeline.runs
 exit 3 if runs.any? { |stats| stats.error.is_a?(DataPipeline::LockError) }
 exit 1 if runs.any?(&:error)
@@ -8,10 +8,11 @@ require_relative 'datapipeline/layout'
 require_relative 'datapipeline/watermark'
 require_relative 'datapipeline/lock'
 require_relative 'datapipeline/connections'
 require_relative 'datapipeline/notification/notifier'
 class DataPipeline
  # plugins are loaded on first use, their gems too: pg (postgres), sqlite3 (sqlite), roo/caxlsx (excel),
-  # http (rest, slack)
+  # http (rest, slack, webhook), net-smtp (email)
  autoload :CsvStorage,      File.expand_path('datapipeline/storage/csv', __dir__)
  autoload :PostgresStorage, File.expand_path('datapipeline/storage/postgres', __dir__)
  autoload :JsonStorage,     File.expand_path('datapipeline/storage/json', __dir__)
@@ -20,6 +21,8 @@ class DataPipeline
  autoload :RestStorage,     File.expand_path('datapipeline/storage/rest', __dir__)
  autoload :FixedStorage,    File.expand_path('datapipeline/storage/fixed', __dir__)
  autoload :Slack,           File.expand_path('datapipeline/notification/slack', __dir__)
  autoload :Webhook,         File.expand_path('datapipeline/notification/webhook', __dir__)
  autoload :Email,           File.expand_path('datapipeline/notification/email', __dir__)
  # passed to header, trailer, map and filter
  #   read:    number of source rows (nil while streaming, until the trailer)
@@ -240,6 +243,26 @@ class DataPipeline
    @reject_target = to
  end
  #
  # notify notifier, on: :problems           -> a report after the run, see DataPipeline::Notifier
  #   on:           :failure (abort), :problems (abort, rejected rows or warnings, default), :always
  #   title:        default the name of the pipeline
  #   dry_run:      true -> also after a dry run (default: not)
  #   include_rows: true -> rejected rows in the report (default: not, they may hold customer data)
  #
  # an error while sending is only logged
  #
  def notify notifier, on: :problems, title: nil, dry_run: false, include_rows: false
    raise ArgumentError, 'notify: notifier needs deliver(report)' unless notifier.respond_to?(:deliver)
    (@notifiers ||= []) << { notifier:, when: Notifier.condition(on), title:, dry_run:, include_rows: }
  end
  # notifier(:alerts) -> Webhook, Slack or Email from the connections file (type webhook, slack, email)
  def notifier name
    Notifier.from_connection(name)
  end
  # on_finish { |stats| ... } -> after every run, also after an abort (stats.error)
  def on_finish(&block)
    @on_finish << block
@@ -484,6 +507,7 @@ class DataPipeline
    print_preview if @stats.dry_run && @stats.error.nil?
    @logger.info "finish #{@stats}"
    append_history
    send_notifications
    @on_finish.each do |hook|
      hook.call(@stats)
    rescue StandardError => e
@@ -504,6 +528,20 @@ class DataPipeline
    @logger.error "history: #{e.class}: #{e.message}"
  end
  def send_notifications
    Array(@notifiers).each do |entry|
      next if @stats.dry_run && !entry[:dry_run]
      report = Report.new([@stats], title: entry[:title], params:)
      next unless entry[:when].call(report)
      entry[:notifier].deliver(report, include_rows: entry[:include_rows])
      @logger.info "notified #{entry[:notifier].class.name.split('::').last}: #{report.event}"
    rescue StandardError => e
      @logger.error "notify #{entry[:notifier].class}: #{e.class}: #{e.message}"
    end
  end
  def add_total side, row
    @stats.totals&.each do |col, sum|
      value = row.key?(col) ? row[col] : row[col.to_sym]
@@ -0,0 +1,93 @@
 require 'securerandom'
 require_relative 'notifier'
 require_relative '../storage/storage_plugin'
 class DataPipeline
  #
  # e-mail over SMTP, text/plain UTF-8
  #
  #   notify DataPipeline::Email.new(to: 'etl@example.com'), on: :failure
  #
  # options (defaults from ENV)
  #   to:        one address or an array (required)
  #   from:      sender, default ENV['SMTP_FROM'] or etl@<hostname>
  #   host:      ENV['SMTP_HOST'], default localhost
  #   port:      ENV['SMTP_PORT'], default 587 (465 with tls: true, 25 without starttls)
  #   username:  ENV['SMTP_USERNAME'], password: ENV['SMTP_PASSWORD']; without username no login
  #   tls:       true -> TLS from the start (port 465)
  #   starttls:  :auto (default, if the server offers it), :always or false
  #   subject_prefix: default '[etl] '
  #   timeout:   seconds for open and read (default 15)
  #
  class Email < Notifier
    class Error < StandardError; end
    def initialize to:, from: nil, host: nil, port: nil, username: nil, password: nil, tls: false,
                   starttls: :auto, subject_prefix: '[etl] ', timeout: 15
      StoragePlugin.require_gem 'net/smtp', 'net-smtp'
      @to = Array(to).map(&:to_s).reject(&:empty?)
      raise ArgumentError, 'Email: to missing' if @to.empty?
      @from = from || ENV['SMTP_FROM'] || "etl@#{Socket.gethostname}"
      @host = host || ENV.fetch('SMTP_HOST', 'localhost')
      @tls = tls
      @starttls = tls || !starttls ? false : starttls.to_sym
      @port = Integer(port || ENV['SMTP_PORT'] || default_port)
      @username = username || ENV['SMTP_USERNAME']
      @password = password || ENV['SMTP_PASSWORD']
      @subject_prefix = subject_prefix
      @timeout = timeout
    end
    def deliver report, include_rows: false
      send_mail("#{@subject_prefix}#{report.subject}", body_of(report, include_rows))
    end
    # any text, e.g. from on_finish
    def send_mail subject, body
      message = build(subject, body)
      smtp = Net::SMTP.new(@host, @port, tls: @tls, starttls: @starttls || false)
      smtp.open_timeout = @timeout
      smtp.read_timeout = @timeout
      login = @username ? { user: @username, secret: @password, authtype: :login } : {}
      smtp.start(helo: Socket.gethostname, **login) { |session| session.send_message(message, @from, *@to) }
      message
    rescue Net::SMTPError, SystemCallError, IOError, Timeout::Error, OpenSSL::SSL::SSLError => e
      raise Error, "#{@host}:#{@port}: #{e.class}: #{e.message}"
    end
    private
    def default_port
      return 465 if @tls
      @starttls ? 587 : 25
    end
    # rows of rejected lines only on request, they may hold customer data
    def body_of report, include_rows
      text = report.to_s
      return text unless include_rows
      rows = report.runs.flat_map { |stats| stats.errors.first(Report::MAX_ERRORS).map { |e| "#{e[:step]}: #{JSON.generate(e[:row])}" } }
      rows.empty? ? text : "#{text}\n\nZeilen:\n#{rows.join("\n")}"
    end
    # RFC 5322 with a base64 body, the subject RFC 2047 encoded when it is not ascii
    def build subject, body
      encoded_subject = subject.ascii_only? ? subject : "=?UTF-8?B?#{[subject].pack('m0')}?="
      [
        "From: #{@from}",
        "To: #{@to.join(', ')}",
        "Subject: #{encoded_subject}",
        "Date: #{Time.now.strftime('%a, %d %b %Y %H:%M:%S %z')}",
        "Message-ID: <#{SecureRandom.uuid}@#{@from.split('@').last}>",
        'MIME-Version: 1.0',
        'Content-Type: text/plain; charset=UTF-8',
        'Content-Transfer-Encoding: base64',
        '',
        [body].pack('m')
      ].join("\r\n")
    end
  end
 end
@@ -0,0 +1,141 @@
 require 'json'
 require 'securerandom'
 require 'socket'
 class DataPipeline
  #
  # what a notifier gets: the stats of one or more runs (one script), with event, title and text lines
  #
  #   event: run.failed (an abort), run.rejected (rejected rows), run.warnings, run.succeeded
  #   to_h:  stable payload for webhooks (schema 1); rows of stats.errors only with include_rows: true,
  #          because they may hold customer data
  #
  class Report
    EVENTS = %w[run.succeeded run.warnings run.rejected run.failed].freeze
    MAX_ERRORS = 20
    attr_reader :id, :title, :runs, :params, :host, :error
    # error: an exception outside of a run (e.g. the script failed before or between pipelines)
    def initialize runs, title: nil, params: {}, error: nil
      @id = SecureRandom.uuid
      @runs = Array(runs)
      @title = title || @runs.filter_map(&:name).uniq.join(', ').then { |t| t.empty? ? 'datapipeline' : t }
      @params = params.to_h
      @host = Socket.gethostname
      @error = error
    end
    # the worst event of all runs
    def event
      return 'run.failed' if @error
      @runs.map { |stats| event_of(stats) }.max_by { |event| EVENTS.index(event) } || 'run.succeeded'
    end
    def failed? = event == 'run.failed'
    def ok? = event == 'run.succeeded'
    def status
      { 'run.failed' => 'abgebrochen', 'run.rejected' => 'mit verworfenen Zeilen',
        'run.warnings' => 'mit Warnungen', 'run.succeeded' => 'ok' }.fetch(event)
    end
    def subject = "#{@title}: #{status}"
    # text for mail, slack and logs: one block per run, the first errors without row data
    def lines errors: 5
      lines = ["#{subject} (#{@host})"]
      lines << "Parameter: #{@params.map { |key, value| "#{key}=#{value}" }.join(' ')}" if @params.any?
      lines << "Fehler: #{@error.class}: #{@error.message}" if @error
      @runs.each do |stats|
        lines << ''
        lines << "#{stats.name || 'Pipeline'}#{" #{stats.version}" if stats.version}: #{stats}"
        lines << "Start #{stats.started_at&.strftime('%d.%m.%Y %H:%M:%S')}, Dauer #{stats.duration}s"
        stats.errors.first(errors).each { |e| lines << "- #{e[:step]}: #{e[:error]}" }
        lines << "- … und #{stats.rejected - errors} weitere" if stats.rejected > errors
      end
      lines
    end
    def to_s = lines.join("\n")
    def to_h include_rows: false
      {
        schema: 1, id: @id, event:, title: @title, host: @host, params: @params,
        error: @error && "#{@error.class}: #{@error.message}",
        runs: @runs.map do |stats|
          stats.to_h.merge(errors: stats.errors.first(MAX_ERRORS).map { |e| include_rows ? e : e.except(:row) })
        end
      }
    end
    private
    def event_of stats
      return 'run.failed' if stats.error
      return 'run.rejected' if stats.rejected.positive?
      return 'run.warnings' if stats.warnings.to_i.positive?
      'run.succeeded'
    end
  end
  #
  # a notifier only needs deliver(report, include_rows: false); see Webhook, Slack, Email
  #
  #   notify DataPipeline::Webhook.new(url: ...), on: :problems
  #
  class Notifier
    # when a notification is sent
    WHEN = {
      always: ->(_report) { true },
      failure: ->(report) { report.failed? },
      problems: ->(report) { !report.ok? }
    }.freeze
    TYPES = { webhook: 'DataPipeline::Webhook', slack: 'DataPipeline::Slack', email: 'DataPipeline::Email' }.freeze
    # notifier from the connections file: type webhook, slack or email, the other keys are its options
    def self.from_connection name
      options = Connections.fetch(name)
      type = options[:type].to_s.to_sym
      klass = TYPES.fetch(type) { raise ArgumentError, "connection #{name.inspect}: type #{type} is no notifier (#{TYPES.keys.join(', ')})" }
      Object.const_get(klass).new(**options.except(:type))
    end
    def self.condition on
      WHEN.fetch(on.to_sym) { raise ArgumentError, "notify on: #{on} unknown, use #{WHEN.keys.join(', ')}" }
    end
    def deliver(_report, include_rows: false)
      raise NotImplementedError, "#{self.class}#deliver is missing"
    end
    private
    # POST with retries on 429/5xx (Retry-After honoured) and network errors; other 4xx raise at once
    def post_with_retries url, body:, headers:, timeout:, retries:, retry_wait:, error:
      attempt = 0
      loop do
        attempt += 1
        begin
          response = HTTP.timeout(timeout).headers(headers).post(url, body:)
        rescue HTTP::Error, SystemCallError, IOError => e
          raise error, "#{url}: #{e.class}: #{e.message}" if attempt > retries
          sleep(retry_wait * attempt)
          next
        end
        return response if response.status.success?
        code = response.code
        unless attempt <= retries && (code == 429 || code >= 500)
          raise error, "#{url}: HTTP #{code} #{response.body.to_s[0, 200]}"
        end
        sleep(response.headers['Retry-After']&.to_f || retry_wait * attempt)
      end
    end
  end
 end
@@ -1,34 +1,88 @@
 require 'http'
 require 'json'
 require_relative 'notifier'
 require_relative '../storage/storage_plugin'
 class DataPipeline
  #
  # Slack.new(channel: '#etl').send_message('posting export done')
  # notify DataPipeline::Slack.new(channel: '#etl'), on: :problems
  # notify DataPipeline::Slack.new(webhook: ENV['SLACK_WEBHOOK'])      # incoming webhook, no token needed
  #
  # options
-  #   token:   bot token, default ENV['SLACK_TOKEN']
+  #   token:      bot token for chat.postMessage, default ENV['SLACK_TOKEN'] (not needed with webhook)
-  #   channel: default '#general'
+  #   channel:    default '#general' (with webhook the channel of the webhook)
  #   webhook:    url of an incoming webhook instead of token and channel
  #   mention:    e.g. '<!here>' or '<!subteam^ID>', only in reports of failed runs
  #   timeout:    seconds per request (default 10)
  #   retries:    repetitions on 429, 5xx and network errors (default 3)
  #   retry_wait: seconds, multiplied by the attempt (default 2)
  #   api_url:    default https://slack.com/api/chat.postMessage
  #
-  class Slack
+  class Slack < Notifier
    API_URL = 'https://slack.com/api/chat.postMessage'.freeze
    MAX_TEXT = 2900 # a section block holds 3000 characters
    class Error < StandardError; end
    DEFAULTS = {
      token: nil,
-      channel: '#general'
+      channel: '#general',
      webhook: nil,
      mention: nil,
      timeout: 10,
      retries: 3,
      retry_wait: 2,
      api_url: API_URL
    }.freeze
    ICONS = { 'run.failed' => ':red_circle:', 'run.rejected' => ':large_orange_circle:',
              'run.warnings' => ':large_yellow_circle:', 'run.succeeded' => ':large_green_circle:' }.freeze
    # & < > have a meaning in slack texts
    def self.escape(text) = text.to_s.gsub('&', '&amp;').gsub('<', '&lt;').gsub('>', '&gt;')
    def initialize options = {}
      StoragePlugin.require_gem 'http'
      @options = DEFAULTS.merge(token: ENV['SLACK_TOKEN']).merge(options)
-      raise ArgumentError, 'Slack: token missing (option token: or ENV SLACK_TOKEN)' unless @options[:token]
+      return if @options[:webhook] || @options[:token]
      raise ArgumentError, 'Slack: token missing (option token: or ENV SLACK_TOKEN, or webhook:)'
    end
-    # returns the parsed slack response, raises when slack reports an error
+    # plain text (slack markup allowed); returns the parsed slack response, raises when slack reports an error
    def send_message(content, verbose: false)
-      response = HTTP.auth("Bearer #{@options[:token]}")
+      post({ text: content }, verbose:)
-                     .post(API_URL, json: { channel: @options[:channel], text: content })
+    end
-      body = JSON.parse(response.body.to_s)
+
-      puts JSON.pretty_generate(body) if verbose
+    # the report as blocks: status line, then the runs as preformatted text
-      raise "Slack#send_message failed: #{body['error']}" unless body['ok']
+    def deliver report, include_rows: false
      icon = ICONS.fetch(report.event)
      mention = report.failed? && @options[:mention] ? " #{@options[:mention]}" : ''
      head = "#{icon} *#{self.class.escape(report.title)}*: #{report.status}#{mention}"
      details = self.class.escape(report.lines.drop(1).join("\n").strip)
      details = "#{details[0, MAX_TEXT]}\n…" if details.length > MAX_TEXT
      blocks = [{ type: 'section', text: { type: 'mrkdwn', text: head } }]
      blocks << { type: 'section', text: { type: 'mrkdwn', text: "```#{details}```" } } unless details.empty?
      post({ text: "#{report.subject}#{mention}", blocks: })
    end
    private
    def post payload, verbose: false
      url = @options[:webhook] || @options[:api_url]
      headers = { 'Content-Type' => 'application/json; charset=utf-8' }
      unless @options[:webhook]
        headers['Authorization'] = "Bearer #{@options[:token]}"
        payload = payload.merge(channel: @options[:channel])
      end
      response = post_with_retries(url, body: JSON.generate(payload), headers:, timeout: @options[:timeout],
                                        retries: @options[:retries], retry_wait: @options[:retry_wait], error: Error)
      text = response.body.to_s
      puts text if verbose
      return text if @options[:webhook] # incoming webhooks answer "ok"
      body = JSON.parse(text)
      raise Error, "Slack#send_message failed: #{body['error']}" unless body['ok']
      body
    end
@@ -0,0 +1,63 @@
 require 'openssl'
 require_relative 'notifier'
 require_relative '../storage/storage_plugin'
 class DataPipeline
  #
  # POST of the report as json (Report#to_h, schema 1)
  #
  #   notify DataPipeline::Webhook.new(url: ENV['ALERT_URL'], secret: ENV['ALERT_SECRET'])
  #
  # options
  #   url:        target (required)
  #   secret:     signs the body: X-DataPipeline-Signature: sha256=HMAC_SHA256(secret, "<timestamp>.<body>")
  #               with X-DataPipeline-Timestamp (unix seconds)
  #   headers:    more headers, e.g. { 'Authorization' => 'Bearer ...' }
  #   timeout:    seconds per request (default 10)
  #   retries:    repetitions on 429, 5xx and network errors (default 3), Retry-After is honoured
  #   retry_wait: seconds, multiplied by the attempt (default 2)
  #
  # every request carries X-DataPipeline-Event and X-DataPipeline-Delivery (the report id, the same on retries)
  #
  class Webhook < Notifier
    class Error < StandardError; end
    def initialize url:, secret: nil, headers: {}, timeout: 10, retries: 3, retry_wait: 2
      StoragePlugin.require_gem 'http'
      @url = url.to_s
      raise ArgumentError, 'Webhook: url missing' if @url.empty?
      @secret = secret
      @headers = headers.to_h { |key, value| [key.to_s, value.to_s] }
      @timeout = timeout
      @retries = retries
      @retry_wait = retry_wait
    end
    def deliver report, include_rows: false
      body = JSON.generate(report.to_h(include_rows:))
      post_with_retries(@url, body:, headers: headers_for(report, body), timeout: @timeout, retries: @retries,
                              retry_wait: @retry_wait, error: Error)
    end
    # for receivers: true when signature and timestamp (max_age seconds) of a request are valid
    def self.valid_signature? secret:, body:, timestamp:, signature:, max_age: 300
      return false if timestamp.to_s.empty? || (Time.now.to_i - timestamp.to_i).abs > max_age
      expected = "sha256=#{OpenSSL::HMAC.hexdigest('SHA256', secret, "#{timestamp}.#{body}")}"
      expected.bytesize == signature.to_s.bytesize && OpenSSL.fixed_length_secure_compare(expected, signature.to_s)
    end
    private
    def headers_for report, body
      headers = { 'Content-Type' => 'application/json', 'User-Agent' => "datapipeline/#{VERSION}",
                  'X-DataPipeline-Event' => report.event, 'X-DataPipeline-Delivery' => report.id }.merge(@headers)
      return headers unless @secret
      timestamp = Time.now.to_i.to_s
      signature = OpenSSL::HMAC.hexdigest('SHA256', @secret, "#{timestamp}.#{body}")
      headers.merge('X-DataPipeline-Timestamp' => timestamp, 'X-DataPipeline-Signature' => "sha256=#{signature}")
    end
  end
 end
@@ -1,5 +1,5 @@
 class DataPipeline
-  VERSION = '3.1.0'.freeze
+  VERSION = '3.2.0'.freeze
  # kept for scripts that print the version (rake info:info); the history is in CHANGELOG.md
  class Version
@@ -1,5 +1,6 @@
 require_relative '../test_helper'
 require 'open3'
 require_relative '../support/test_http_server'
 # bin/etl as a subprocess: arguments, params, dry run, exit codes
 class EtlCliTest < PipelineTest
@@ -127,4 +128,44 @@ class EtlCliTest < PipelineTest
      assert_equal ['export_kunden', 3, true], entry.values_at('name', 'read', 'ok')
    end
  end
  def test_notify_one_report_for_all_pipelines_of_the_script
    server = TestHttpServer.new { [200, 'ok'] }
    with_tmpdir do |dir|
      Dir.mkdir("#{dir}/config")
      File.write("#{dir}/config/connections.yml", "alerts:\n  type: webhook\n  url: #{server.url}/hook\n")
      path = script(dir, 'zwei.rb', <<~RUBY)
        2.times do |i|
          DataPipeline.new do
            name "teil\#{i}"
            on_error :skip
            read container(:csv, file: 'in.csv')
            validate 'a' => { format: /\\A\\d+\\z/ } if i == 1
            write container(:csv, file: "out\#{i}.csv")
          end.run
        end
      RUBY
      _out, err, status = etl(dir, 'run', path, '--notify=alerts')
      assert_equal 2, status.exitstatus, err
      assert_equal 1, server.requests.size
      payload = JSON.parse(server.requests.last[:body])
      assert_equal ['run.rejected', 'zwei', %w[teil0 teil1]], [payload['event'], payload['title'], payload['runs'].map { |r| r['name'] }]
      _out, err, status = etl(dir, 'run', path, '--notify=alerts', '--notify-on=failure')
      assert_equal 2, status.exitstatus, err
      assert_equal 1, server.requests.size # nothing failed
      broken = script(dir, 'kaputt.rb', "raise 'Skriptfehler'\n")
      _out, _err, status = etl(dir, 'run', broken, '--notify=alerts')
      assert_equal 1, status.exitstatus
      payload = JSON.parse(server.requests.last[:body])
      assert_equal ['run.failed', 'RuntimeError: Skriptfehler', []], payload.values_at('event', 'error', 'runs')
      _out, err, status = etl(dir, 'run', path, '--notify=gibtsnicht')
      assert_equal 1, status.exitstatus
      assert_includes err, 'etl: --notify=gibtsnicht: connection "gibtsnicht" not in config/connections.yml (alerts)'
    end
  ensure
    server&.stop
  end
 end
@@ -0,0 +1,206 @@
 require_relative '../test_helper'
 require_relative '../support/test_http_server'
 require_relative '../support/test_smtp_server'
 # Report, Webhook, Slack, Email and notify against local test servers
 class NotificationTest < PipelineTest
  ROWS = [{ 'id' => '1', 'konto' => '1000' }, { 'id' => '2', 'konto' => '' }].freeze
  # collects reports, for notify without network
  class MemoryNotifier < DataPipeline::Notifier
    attr_reader :reports
    def initialize(fail: false) = (@reports = []; @fail = fail)
    def deliver report, include_rows: false
      raise 'kaputt' if @fail
      @reports << [report, include_rows]
    end
  end
  def teardown
    @server&.stop
    DataPipeline.connections_file = nil
    super
  end
  def stats_with_rejected
    run_pipeline(ROWS.map(&:dup)) { name 'export'; on_error :skip; validate 'konto' => { required: true } }.last
  end
  def report(stats = stats_with_rejected) = DataPipeline::Report.new([stats], params: { from: '2026-10-01' })
  def test_report_event_lines_and_payload_without_rows
    report = report()
    assert_equal ['run.rejected', 'export: mit verworfenen Zeilen'], [report.event, report.subject]
    assert_includes report.to_s, '- validate#1: konto: missing'
    assert_includes report.to_s, 'Parameter: from=2026-10-01'
    payload = report.to_h
    assert_equal [1, 'run.rejected', 'export'], payload.values_at(:schema, :event, :title)
    assert_equal [{ step: 'validate#1', error: 'konto: missing' }], payload[:runs].first[:errors]
    assert_equal({ 'id' => '2', 'konto' => '' }, report.to_h(include_rows: true)[:runs].first[:errors].first[:row])
  end
  def test_report_worst_event_of_several_runs_and_script_error
    ok = run_pipeline(ROWS.map(&:dup)).last
    assert_equal 'run.succeeded', DataPipeline::Report.new([ok]).event
    assert_equal 'run.rejected', DataPipeline::Report.new([ok, stats_with_rejected]).event
    failed = DataPipeline::Report.new([ok], title: 'skript', error: RuntimeError.new('kaputt'))
    assert_equal %w[run.failed skript], [failed.event, failed.title]
    assert_includes failed.to_s, 'Fehler: RuntimeError: kaputt'
  end
  def test_webhook_signed_post
    @server = TestHttpServer.new { [200, 'ok'] }
    report = report()
    DataPipeline::Webhook.new(url: "#{@server.url}/hook", secret: 's3cret', headers: { 'X-Team' => 'etl' }).deliver(report)
    request = @server.requests.last
    headers = request[:headers]
    assert_equal ['/hook', 'run.rejected', report.id, 'etl'],
                 [request[:path], headers['x-datapipeline-event'], headers['x-datapipeline-delivery'], headers['x-team']]
    assert_equal 1, JSON.parse(request[:body])['schema']
    refute_includes request[:body], '"row"'
    assert DataPipeline::Webhook.valid_signature?(secret: 's3cret', body: request[:body],
                                                  timestamp: headers['x-datapipeline-timestamp'],
                                                  signature: headers['x-datapipeline-signature'])
    refute DataPipeline::Webhook.valid_signature?(secret: 'falsch', body: request[:body],
                                                  timestamp: headers['x-datapipeline-timestamp'],
                                                  signature: headers['x-datapipeline-signature'])
    refute DataPipeline::Webhook.valid_signature?(secret: 's3cret', body: request[:body], timestamp: (Time.now.to_i - 600).to_s,
                                                  signature: headers['x-datapipeline-signature'])
  end
  def test_webhook_retries_429_and_5xx_but_not_4xx
    answers = [[503, 'busy'], [429, 'slow', { 'Retry-After' => '0' }], [200, 'ok']]
    @server = TestHttpServer.new { answers.shift }
    DataPipeline::Webhook.new(url: @server.url, retry_wait: 0).deliver(report)
    assert_equal 3, @server.requests.size
    assert_equal 1, @server.requests.map { |r| r[:headers]['x-datapipeline-delivery'] }.uniq.size
    @server.stop
    @server = TestHttpServer.new { [400, 'bad'] }
    error = assert_raises(DataPipeline::Webhook::Error) { DataPipeline::Webhook.new(url: @server.url, retry_wait: 0).deliver(report) }
    assert_includes error.message, 'HTTP 400'
    assert_equal 1, @server.requests.size
  end
  def test_webhook_network_error_after_retries
    port = TCPServer.new('127.0.0.1', 0).then { |s| s.addr[1].tap { s.close } }
    error = assert_raises(DataPipeline::Webhook::Error) do
      DataPipeline::Webhook.new(url: "http://127.0.0.1:#{port}", retries: 1, retry_wait: 0, timeout: 2).deliver(report)
    end
    assert_includes error.message, '127.0.0.1'
  end
  def test_slack_api_blocks_escape_and_mention
    @server = TestHttpServer.new { [200, { 'ok' => true }] }
    slack = DataPipeline::Slack.new(token: 'xoxb-1', channel: '#etl', api_url: @server.url, mention: '<!here>')
    failed = DataPipeline::Report.new([stats_with_rejected], title: 'A & <B>', error: RuntimeError.new('x < y'))
    slack.deliver(failed)
    request = @server.requests.last
    body = JSON.parse(request[:body])
    assert_equal ['Bearer xoxb-1', '#etl'], [request[:headers]['authorization'], body['channel']]
    assert_equal ':red_circle: *A &amp; &lt;B&gt;*: abgebrochen <!here>', body['blocks'][0]['text']['text']
    assert_includes body['blocks'][1]['text']['text'], 'RuntimeError: x &lt; y'
    slack.deliver(report)
    refute_includes JSON.parse(@server.requests.last[:body])['blocks'][0]['text']['text'], '<!here>' # mention only on failure
  end
  def test_slack_error_and_incoming_webhook
    @server = TestHttpServer.new { |req| req[:path] == '/api' ? [200, { 'ok' => false, 'error' => 'not_in_channel' }] : [200, 'ok'] }
    error = assert_raises(DataPipeline::Slack::Error) { DataPipeline::Slack.new(token: 't', api_url: "#{@server.url}/api").send_message('x') }
    assert_includes error.message, 'not_in_channel'
    token = ENV.delete('SLACK_TOKEN')
    hook = DataPipeline::Slack.new(webhook: "#{@server.url}/hook")
    assert_equal 'ok', hook.deliver(report)
    request = @server.requests.last
    assert_nil request[:headers]['authorization']
    refute JSON.parse(request[:body]).key?('channel')
  ensure
    ENV['SLACK_TOKEN'] = token if token
  end
  def test_email_over_smtp_with_login
    @server = TestSmtpServer.new
    mail = DataPipeline::Email.new(to: %w[a@example.com b@example.com], from: 'etl@example.com', host: '127.0.0.1',
                                   port: @server.port, starttls: false, username: 'user', password: 'geheim')
    mail.deliver(DataPipeline::Report.new([stats_with_rejected], title: 'Grösse Ä'))
    sent = @server.mails.last
    assert_equal ['etl@example.com', %w[a@example.com b@example.com], %w[user geheim]], sent.values_at(:from, :to, :auth)
    head, body = sent[:data].split("\r\n\r\n", 2)
    assert_includes head, "Subject: =?UTF-8?B?#{['[etl] Grösse Ä: mit verworfenen Zeilen'].pack('m0')}?="
    assert_includes head, 'Content-Type: text/plain; charset=UTF-8'
    text = body.unpack1('m').force_encoding('UTF-8')
    assert_includes text, '- validate#1: konto: missing'
    refute_includes text, 'Zeilen:'
  end
  def test_email_rows_on_request_and_errors
    @server = TestSmtpServer.new
    mail = DataPipeline::Email.new(to: 'a@example.com', host: '127.0.0.1', port: @server.port, starttls: false)
    mail.deliver(report, include_rows: true)
    assert_includes @server.mails.last[:data].split("\r\n\r\n", 2).last.unpack1('m').force_encoding('UTF-8'), '{"id":"2","konto":""}'
    assert_nil @server.mails.last[:auth]
    assert_raises(ArgumentError) { DataPipeline::Email.new(to: []) }
    port = TCPServer.new('127.0.0.1', 0).then { |s| s.addr[1].tap { s.close } }
    assert_raises(DataPipeline::Email::Error) { DataPipeline::Email.new(to: 'a@b', host: '127.0.0.1', port:, starttls: false).send_mail('x', 'y') }
  end
  def test_notify_conditions_dry_run_and_rows
    always = MemoryNotifier.new
    failure = MemoryNotifier.new
    problems = MemoryNotifier.new
    rows = MemoryNotifier.new
    run_pipeline(ROWS.map(&:dup)) do
      name 'export'
      on_error :skip
      validate 'konto' => { required: true }
      notify always, on: :always
      notify failure, on: :failure
      notify problems
      notify rows, include_rows: true, title: 'Export Konten'
    end
    assert_equal [1, 0, 1], [always, failure, problems].map { |n| n.reports.size }
    assert_equal ['Export Konten', true], [rows.reports.first[0].title, rows.reports.first[1]]
    dry = MemoryNotifier.new
    dry_too = MemoryNotifier.new
    capture_stdout { run_pipeline(ROWS.map(&:dup), dry_run: true) { notify dry, on: :always; notify dry_too, on: :always, dry_run: true } }
    assert_equal [0, 1], [dry.reports.size, dry_too.reports.size]
    assert_raises(ArgumentError) { run_pipeline { notify Object.new } }
    assert_raises(ArgumentError) { run_pipeline { notify MemoryNotifier.new, on: :sometimes } }
  end
  def test_notify_error_is_only_logged
    log = StringIO.new
    stats = DataPipeline.new do
      logger Logger.new(log)
      read MemoryStorage.new(rows: ROWS.map(&:dup))
      write MemoryStorage.new
      notify MemoryNotifier.new(fail: true), on: :always
    end.run
    assert stats.ok?
    assert_includes log.string, 'notify NotificationTest::MemoryNotifier: RuntimeError: kaputt'
  end
  def test_notifier_from_connections_file
    with_tmpdir do |dir|
      File.write("#{dir}/c.yml", <<~YAML)
        alerts: { type: webhook, url: "http://127.0.0.1:9/hook", secret: "<%= 'abc' %>" }
        mail: { type: email, to: [etl@example.com], host: smtp.example.com }
        db: { type: sqlite, database: x.sqlite3 }
      YAML
      DataPipeline.connections_file = "#{dir}/c.yml"
      pipeline = DataPipeline.new { logger PipelineHelper::QUIET }
      assert_kind_of DataPipeline::Webhook, pipeline.notifier(:alerts)
      assert_equal 'abc', pipeline.notifier(:alerts).instance_variable_get(:@secret)
      assert_kind_of DataPipeline::Email, pipeline.notifier(:mail)
      assert_includes assert_raises(ArgumentError) { pipeline.notifier(:db) }.message, 'is no notifier'
    end
  end
 end
@@ -5,7 +5,7 @@ require 'json'
 #
 # minimal http server for DataPipeline::RestStorage tests, one request per connection
 #   server = TestHttpServer.new { |req| [200, { 'items' => [] }] }
-#   req: { method:, path:, query:, headers:, body: }  -> [status, hash/array or string]
+#   req: { method:, path:, query:, headers:, body: }  -> [status, hash/array or string, optional headers]
 #
 class TestHttpServer
  attr_reader :requests
@@ -40,9 +40,10 @@ class TestHttpServer
    request = { method:, path:, query: URI.decode_www_form(query.to_s).to_h, headers:, body: }
    @requests << request
-    status, payload = @handler.call(request)
+    status, payload, extra = @handler.call(request)
    text = payload.is_a?(String) ? payload : JSON.generate(payload)
-    client.write "HTTP/1.1 #{status} X\r\nContent-Type: application/json\r\n" \
+    extra = Hash(extra).map { |key, value| "#{key}: #{value}\r\n" }.join
    client.write "HTTP/1.1 #{status} X\r\nContent-Type: application/json\r\n#{extra}" \
                 "Content-Length: #{text.bytesize}\r\nConnection: close\r\n\r\n#{text}"
  rescue StandardError => e
    warn "TestHttpServer: #{e.class}: #{e.message}"
@@ -0,0 +1,59 @@
 require 'socket'
 #
 # minimal smtp server for DataPipeline::Email tests: EHLO, AUTH LOGIN, MAIL, RCPT, DATA, QUIT
 #   server = TestSmtpServer.new
 #   server.mails -> [{ from:, to: [...], data:, auth: [user, password] }]
 #
 class TestSmtpServer
  attr_reader :mails
  def initialize
    @mails = []
    @server = TCPServer.new('127.0.0.1', 0)
    @thread = Thread.new { loop { serve(@server.accept) } }
  end
  def port = @server.addr[1]
  def stop
    @thread.kill
    @server.close
  end
  private
  def serve client
    mail = { to: [] }
    client.write "220 test ESMTP\r\n"
    while (line = client.gets)
      case line
      when /\AEHLO/i then client.write "250-test\r\n250 AUTH LOGIN PLAIN\r\n"
      when /\AAUTH LOGIN/i
        client.write "334 VXNlcm5hbWU6\r\n"
        user = client.gets.unpack1('m')
        client.write "334 UGFzc3dvcmQ6\r\n"
        mail[:auth] = [user, client.gets.unpack1('m')]
        client.write "235 ok\r\n"
      when /\AMAIL FROM:<(.*)>/i then mail[:from] = $1; client.write "250 ok\r\n"
      when /\ARCPT TO:<(.*)>/i then mail[:to] << $1; client.write "250 ok\r\n"
      when /\ADATA/i
        client.write "354 go\r\n"
        data = +''
        while (part = client.gets) && part != ".\r\n"
          data << part
        end
        mail[:data] = data
        @mails << mail
        mail = { to: [] }
        client.write "250 queued\r\n"
      when /\AQUIT/i then client.write "221 bye\r\n"; break
      else client.write "250 ok\r\n"
      end
    end
  rescue StandardError => e
    warn "TestSmtpServer: #{e.class}: #{e.message}"
  ensure
    client.close
  end
 end