Formalizing the test-machine command API

.gitignore

@@ -9,6 +9,7 @@
 /jackdaw-[a-z]*/checkouts
 /jackdaw-[a-z]*/target
 /target
+/test-results
 /logs
 pom.xml
 pom.xml.asc

CHANGELOG.md

@@ -1,6 +1,14 @@
 # Changelog
 
-## Fixed
+## Unreleased
+
+### Added
+
+* Start formalizing test-machine commands with fspec'd functions
+
+## [0.7.2] - [2020-02-07]
+
+### Fixed
 
 * Fixed bug in Avro deserialisation, when handling a union of enum types
 

src/jackdaw/test/commands.clj

@@ -1,9 +1,11 @@
 (ns jackdaw.test.commands
   ""
   (:require
+   [clojure.spec.alpha :as s]
    [jackdaw.test.commands.base :as base]
    [jackdaw.test.commands.write :as write]
-   [jackdaw.test.commands.watch :as watch]))
+   [jackdaw.test.commands.watch :as watch])
+  (:refer-clojure :exclude [do]))
 
 (def base-commands base/command-map)
 (def write-command write/command-map)
@@ -37,3 +39,80 @@
   [machine handler]
   (assoc machine
          :command-handler handler))
+
+;; Test Command API
+
+(s/def ::topic-id (s/or :keyword keyword?
+                        :string string?))
+(s/def ::test-message any?)
+(s/def ::write-options map?)
+(s/def ::watch-options map?)
+
+(defn do
+  "Invoke the provided function, passing a snapshot of the test journal
+
+   Use this to perform side-effects without representing their result in the journal"
+  [do-fn]
+  `[:do ~do-fn])
+
+(s/fdef do
+  :args ifn?
+  :ret vector?)
+
+(defn do!
+  "Invoke the provided function, passing the journal `ref`
+
+   Use this to perform side-effects when you want to represent the result in the journal
+   (e.g. insert test-data into an external database AND into the journal with the expectation
+   that it will eventually appear in kafka via some external system like kafka-connect)"
+  [do-fn]
+  `[:do! ~do-fn])
+
+(s/fdef do!
+  :args ifn?
+  :ret vector?)
+
+(defn write!
+  "Write a message to the topic identified in the topic-metadata by `topic-id`
+
+   `:message` is typically a map to be serialized by the Serde configured in the topic-metadata
+              but it can be whatever the configued Serde is capable of serializing
+   `:options` is an optional map containing additional information describing how the test-message
+              should be created. The following properties are supported.
+
+      `:key`             An explicit key to associate with the test message
+      `:key-fn`          A function to derive the key from the test message
+      `:partition`       The partition to which the test message should be written
+      `:partition-fn`    A function to derive the partition to which the test message should be written"
+  ([topic-id message]
+   `[:write! ~topic-id ~message])
+
+  ([topic-id message options]
+   `[:write! ~topic-id ~message ~options]))
+
+(s/fdef write!
+  :args (s/cat :topic-id ::topic-id
+               :message ::test-message
+               :options (s/? ::write-options))
+  :ret vector?)
+
+(defn watch
+  "Watch the test-journal until the `watch-fn` predicate returns true
+
+   `:watch-fn` is a function that takes the journal and returns true once the journal
+               contains evidence of the test being complete
+   `:options` is an optional map containing additional information describing how the watch
+              function will be run. The following properties are supported.
+
+      `:info` Diagnostic information to be included in the response when a watch fails
+              to observe the expected data in the journal
+      `:timeout` The number of milliseconds to wait before giving up"
+  ([watch-fn]
+   `[:watch ~watch-fn])
+  ([watch-fn options]
+   `[:watch ~watch-fn ~options]))
+
+(s/fdef watch
+  :args (s/cat :watch-fn ifn?
+               :options (s/? ::watch-options))
+  :ret vector?)

src/jackdaw/test/commands/base.clj

@@ -1,6 +1,6 @@
 (ns jackdaw.test.commands.base
   (:require
-    [clojure.pprint :as pprint]))
+   [clojure.pprint :as pprint]))
 
 (def command-map
   {:stop (constantly true)

test/jackdaw/test_test.clj

@@ -3,6 +3,7 @@
    [clojure.test :refer :all]
    [jackdaw.streams :as k]
    [jackdaw.test :as jd.test]
+   [jackdaw.test.commands :as cmd]
    [jackdaw.test.fixtures :as fix]
    [jackdaw.test.serde :as serde]
    [jackdaw.test.transports :as trns]
@@ -115,10 +116,9 @@
   (testing "write then watch"
     (fix/with-fixtures [(fix/topic-fixture kafka-config {"foo" foo-topic})]
       (with-open [t (jd.test/test-machine (kafka-transport))]
-        (let [write [:write! "foo" {:id "msg1" :payload "yolo"}]
-              watch [:watch (by-id "foo" "msg1")
-                     {:info "failed to find foo with id=msg1"}]
-
+        (let [write (cmd/write! "foo" {:id "msg1" :payload "yolo"})
+              watch (cmd/watch (by-id "foo" "msg1")
+                               {:info "failed to find foo with id=msg1"})
               {:keys [results journal]} (jd.test/run-test t [write watch])
               [write-result watch-result] results]
 
@@ -139,13 +139,13 @@
 (deftest test-reuse-machine
   (fix/with-fixtures [(fix/topic-fixture kafka-config {"foo" foo-topic})]
     (with-open [t (jd.test/test-machine (kafka-transport))]
-      (let [prog1 [[:write! "foo" {:id "msg2" :payload "yolo"}]
-                   [:watch (by-id "foo" "msg2")
-                    {:info "failed to find foo with id=msg2"}]]
+      (let [prog1 [(cmd/write! "foo" {:id "msg2" :payload "yolo"})
+                   (cmd/watch (by-id "foo" "msg2")
+                              {:info "failed to find foo with id=msg2"})]
 
-            prog2 [[:write! "foo" {:id "msg3" :payload "you only live twice"}]
-                   [:watch (by-id "foo" "msg3")
-                    {:info "failed to find foo with id=msg3"}]]]
+            prog2 [(cmd/write! "foo" {:id "msg3" :payload "you only live twice"})
+                   (cmd/watch (by-id "foo" "msg3")
+                              {:info "failed to find foo with id=msg3"})]]
 
         (testing "run test sequence and inspect results"
           (let [{:keys [results journal]} (jd.test/run-test t prog1)]
@@ -207,14 +207,15 @@
                                                            :out test-out}})
         (fn [machine]
           (jd.test/run-test machine
-                            [[:write! :in {:id "1" :payload "foo"} {:key-fn :id}]
-                             [:write! :in {:id "2" :payload "bar"} {:key-fn :id}]
-                             [:watch (fn [journal]
-                                       (when (->> (get-in journal [:topics :out])
-                                                  (filter (fn [r]
-                                                            (= (get-in r [:value :id]) "2")))
-                                                  (not-empty))
-                                         true)) {:timeout 2000}]])))
+                            [(cmd/write! :in {:id "1" :payload "foo"} {:key-fn :id})
+                             (cmd/write! :in {:id "2" :payload "bar"} {:key-fn :id})
+                             (cmd/watch (fn [journal]
+                                          (when (->> (get-in journal [:topics :out])
+                                                     (filter (fn [r]
+                                                               (= (get-in r [:value :id]) "2")))
+                                                     (not-empty))
+                                            true))
+                                        {:timeout 2000})])))
       (catch Exception e
         (reset! error-raised e)))
     (is (not @error-raised))))
@@ -230,13 +231,13 @@
                                                            :out test-out}})
         (fn [machine]
           (jd.test/run-test machine
-                            [[:write! :in {:id "1" :payload "foo"} {:key-fn :id}]
-                             [:write! :in {:id "2" :payload "bar"} {:key-fn :id}]
-                             [:watch (fn [journal]
-                                       (->> (get-in journal [:topics :out])
-                                            (filter (fn [r]
-                                                      (= (:id r) "2")))
-                                            (not-empty)))]])))
+                            [(cmd/write! :in {:id "1" :payload "foo"} {:key-fn :id})
+                             (cmd/write! :in {:id "2" :payload "bar"} {:key-fn :id})
+                             (cmd/watch (fn [journal]
+                                          (->> (get-in journal [:topics :out])
+                                               (filter (fn [r]
+                                                         (= (:id r) "2")))
+                                               (not-empty))))])))
       (catch Exception e
         (reset! error-raised e)))
     (is @error-raised)))
@@ -252,13 +253,13 @@
                                                            :out test-out}})
         (fn [machine]
           (jd.test/run-test machine
-                            [[:write! :in {:id "1" :payload "foo"} {:key-fn bad-key-fn}]
-                             [:write! :in {:id "2" :payload "bar"} {:key-fn :id}]
-                             [:watch (fn [journal]
-                                       (->> (get-in journal [:topics :out])
-                                            (filter (fn [r]
-                                                      (= (:id r) "2")))
-                                            (not-empty)))]])))
+                            [(cmd/write! :in {:id "1" :payload "foo"} {:key-fn bad-key-fn})
+                             (cmd/write! :in {:id "2" :payload "bar"} {:key-fn :id})
+                             (cmd/watch (fn [journal]
+                                          (->> (get-in journal [:topics :out])
+                                               (filter (fn [r]
+                                                         (= (:id r) "2")))
+                                               (not-empty))))])))
       (catch Exception e
         (reset! error-raised e)))
     (is @error-raised)))
@@ -274,10 +275,10 @@
                                                            :out test-out}})
         (fn [machine]
           (jd.test/run-test machine
-                            [[:write! :in {:id "1" :payload "foo"} {:key-fn :id}]
-                             [:write! :in {:id "2" :payload "bar"} {:key-fn :id}]
-                             [:watch (fn [journal]
-                                       (bad-watch-fn journal))]])))
+                            [(cmd/write! :in {:id "1" :payload "foo"} {:key-fn :id})
+                             (cmd/write! :in {:id "2" :payload "bar"} {:key-fn :id})
+                             (cmd/watch (fn [journal]
+                                          (bad-watch-fn journal)))])))
       (catch Exception e
         (reset! error-raised e)))
     (is @error-raised)))