From bfbf29ce65147191a35e8b021c7ad3fc3fc304ec Mon Sep 17 00:00:00 2001
From: eltrac <hi@eltr.ac>
Date: Thu, 15 Oct 2026 17:21:09 +0800
Subject: test: test concurrent proceesor; assert config option types

---
 src/weepinbell/processor/facade.clj        |  5 ++-
 test/weepinbell/processor_starter_test.clj | 62 ++++++++++++++++++++++--------
 2 files changed, 50 insertions(+), 17 deletions(-)

diff --git a/src/weepinbell/processor/facade.clj b/src/weepinbell/processor/facade.clj
index 29febc2..1360b30 100644
--- a/src/weepinbell/processor/facade.clj
+++ b/src/weepinbell/processor/facade.clj
@@ -1,6 +1,6 @@
 (ns weepinbell.processor.facade
   (:require [clojure.spec.alpha :as s]
-            [clojure.core.async :refer [go-loop <! put! close! chan <!!]]
+            [clojure.core.async :refer [go-loop <! put! chan <!!]]
             [taoensso.timbre :as timbre]
             [weepinbell.utils.digest :refer [digest]]
             [weepinbell.storage.spec]
@@ -15,6 +15,9 @@
 (s/def ::callbacks
   (s/keys :req-un [::on-receive ::on-create ::on-delete ::on-update]))
 (s/def ::storage (partial s/valid? :weepinbell.storage.spec/vtable))
+(s/def ::base-url string?)
+(s/def ::allowed-target-domains sequential?)
+(s/def ::pipeline-concurrency int?)
 (s/def ::config
   (s/keys :req-un [::base-url ::allowed-target-domains ::pipeline-concurrency]))
 (s/def ::processor-meta
diff --git a/test/weepinbell/processor_starter_test.clj b/test/weepinbell/processor_starter_test.clj
index 8b604a5..500bd9f 100644
--- a/test/weepinbell/processor_starter_test.clj
+++ b/test/weepinbell/processor_starter_test.clj
@@ -1,12 +1,17 @@
 (ns weepinbell.processor-starter-test
   (:require [clojure.test :refer [deftest testing is]]
-            [clojure.core.async :refer [chan put!]]
+            [clojure.core.async :refer [chan put! poll!]]
             [weepinbell.processor.facade :refer [start-processor]]))
 
-(defn- test-processor [test-wmc received?]
-  {:pre [(instance? clojure.lang.IPending received?)]}
+(defn- test-processor 
+  ([test-wmc received?]
+   (test-processor test-wmc received? 1 nil))
+  ([test-wmc received? concurrency]
+   (test-processor test-wmc received? concurrency nil))
+  ([test-wmc received? concurrency on-receive]
   (let [pmeta {:wmc test-wmc
-               :callbacks {:on-receive (fn [_] (deliver received? true))
+               :callbacks {:on-receive (or on-receive 
+                                           (fn [_] (put! received? true)))
                            :on-create (fn [_])
                            :on-delete (fn [_])
                            :on-update (fn [_])}
@@ -17,22 +22,47 @@
                          :get-by-domain (fn [& _])}
                :config {:base-url "https://test.com"
                         :allowed-target-domains ["test.com"]
-                        :pipeline-concurrency 1}}]
-    (start-processor pmeta)))
+                        :pipeline-concurrency concurrency}}]
+    (start-processor pmeta))))
 
 (deftest start-processor-test
-  (testing "graceful shutdown"
-    (let [test-wmc (chan)
-          received? (promise)
+  (testing "processor starting and taking webmention"
+    (let [test-wmc (chan), received? (chan)
           shutdown (test-processor test-wmc received?)]
-      (shutdown)
       (put! test-wmc {:s "s" :t "t"})
-      (is (not (deref received? 50 false)))))
-  (testing "processor starting and taking webmention"
-    (let [test-wmc (chan)
-          received? (promise)
+      (Thread/sleep 50)
+      (is (poll! received?))
+      (shutdown)))
+  (testing "graceful shutdown, should not take webmention after shutdown"
+    (let [test-wmc (chan), received? (chan)
           shutdown (test-processor test-wmc received?)]
+      (shutdown)
       (put! test-wmc {:s "s" :t "t"})
-      (is (deref received? 50 false))
-      (shutdown))))
+      (Thread/sleep 50)
+      (is (nil? (poll! received?))))))
+
+(deftest concurrent-processor-test 
+  (testing "processor processing all webmentions before shutting down"
+    (let [test-wmc (chan), received? (chan)
+          concurrency 3
+          shutdown (test-processor test-wmc received? 3)]
+      (is (fn? shutdown))
+      (dotimes [_ concurrency] (put! test-wmc {:s "s" :t "t"}))
+      (shutdown)
+      (Thread/sleep 50)
+      (dotimes [_ concurrency] (is (poll! received?)))))
+  (testing "processor still works if Exceptions were thrown"
+    (let [test-wmc (chan), received? (chan), fail-once (chan)]
+      (put! fail-once true)
+      (let [on-receive (fn [_]
+                         ;; throw Exception at the first call
+                         (if (poll! fail-once)
+                           (throw (Exception.))
+                           (put! received? true)))
+            concurrency 5
+            shutdown (test-processor test-wmc received? concurrency on-receive)]
+        (dotimes [_ concurrency] (put! test-wmc {:s "s" :t "t"}))
+        (Thread/sleep 50)
+        (dotimes [_ (- concurrency 1)] (is (poll! received?)))
+        (shutdown)))))
 
-- 
cgit v1.2.3