(ns riemann.streams-test
  (:require [riemann.common :refer :all :exclude [match]]
            [riemann.folds :as folds]
            [riemann.index :as index]
            [riemann.logging :as logging]
            [riemann.streams :refer :all]
            [riemann.test :refer [run-stream run-stream-intervals test-stream
                                  with-test-stream test-stream-intervals]]
            [riemann.time :refer :all]
            [riemann.time.controlled :refer :all]
            [clojure.test :refer :all])
  (:import (java.util.concurrent Executor
                                 CountDownLatch)))

(logging/init)

(use-fixtures :once control-time!)
(use-fixtures :each reset-time!)

(defn evs
  "Generate events based on the given event, with given metrics"
  [base-event & metrics]
  (vec (map #(assoc base-event :metric %)
            metrics)))

(defn em
  "Generate events with given metrics"
  [& metrics]
  (vec (map (fn [m] {:metric m}) metrics)))

(deftest smap*-test
         (testing "passes nil values"
                  (test-stream (smap* identity)
                               [1 nil false 3]
                               [1 nil false 3])))

(deftest smap-test
         (testing "increment"
                  (test-stream (smap inc) [6 3 -1] [7 4 0]))

         (testing "ignores nil values"
                  (test-stream (smap identity)
                               [1 nil false 3]
                               [1 false 3])))

(deftest smapcat-test
  (testing "doubles"
    (test-stream (smapcat #(vector % %)) [0 1 2 3]
                 [0 0 1 1 2 2 3 3])))

(deftest sflatten-test
  (test-stream (sflatten) [[0 1 2 3] [4 5 6 7]]
               [0 1 2 3 4 5 6 7]))

(deftest sdo-test
  (let [vals1   (atom [])
        vals2   (atom [])
        add1    #(swap! vals1 conj %)
        add2    #(swap! vals2 conj %)]
    (run-stream (sdo add1 add2) [1 2 3])
    (is (= @vals1 [1 2 3]))
    (is (= @vals2 [1 2 3]))
    (run-stream (sdo add1) [4 5 6])
    (is (= @vals1 [1 2 3 4 5 6]))
    (run-stream (sdo) [6 7 8])
    (is (= @vals1 [1 2 3 4 5 6]))
    (is (= @vals2 [1 2 3]))))

(deftest exception-stream-test
         (testing "direct fn"
                  (logging/suppress
                    ["riemann.streams" "riemann.streams-test"]

                    (let [out        (atom [])
                          exceptions (atom [])
                          stream (exception-stream
                                   #(swap! exceptions conj %)
                                   (fn [e]
                                     (swap! out conj e)
                                     (throw
                                       (RuntimeException. "foo"))))]
                      (stream :hi)

                      (is (= @out [:hi]))
                      (is (= 1 (count @exceptions)))
                      (let [e (first @exceptions)]
                        (is (= (:service e) "riemann exception"))
                        (is (= (:state e)   "error"))
                        (is (= (:time e)    0))
                        (is (re-find #".*RuntimeException.*" (:description e)))
                        (is (= #{"exception" "java.lang.RuntimeException"}
                               (set (:tags e))))))))

         (testing "scheduled fn"
                  (logging/suppress
                    "riemann.streams"
                    (reset-time!)
                    (let [e     (atom nil)
                          s (exception-stream #(reset! e %)
                                              (rate 1
                                                    (fn [e] (/ 1 0))))]
                      (s {:metric 2})
                      (is (nil? @e))
                      (advance! 1)
                      (is (= "error" (:state @e))))))

         (testing "exception handling takes care of Exception"
                  (logging/suppress
                   "riemann.streams"
                   (let [exceptions (atom [])]
                     (binding [riemann.streams/*exception-stream* #(swap! exceptions conj %)]
                       (run-stream (sdo #(throw (Exception. %)))
                                   ["boo!"]))
                     (is (= 1 (count @exceptions)))))))

(deftest execute-on-test
         (let [output (atom [])
               n      100
               events (range n)
               latch  (CountDownLatch. n)
               stream (execute-on
                        (reify Executor
                          (execute [this task]
                                   (future
                                     (task)
                                     (.countDown latch))))
                        (fn [event]
                          (swap! output conj event)))]

           ; Run streams against events.
           (dorun (map stream events))

           ; Check that our event was applied to the conj stream.
           (.await latch)
           (is (= (set @output) (set events)))))

(deftest sreduce-test
         (testing "explicit value"
                  (test-stream (sreduce + 1) [1 2 3] [2 4 7]))

         (testing "implicit value"
                  (test-stream (sreduce +) [1 2 3 4] [3 6 10])))

(deftest counter-test
         (testing "passes through events without metrics"
                  (test-stream (counter)
                               [{} {:state "expired"} {:service "foo"}]
                               [{} {:state "expired"} {:service "foo"}]))

         (testing "counts"
                  (test-stream (counter)
                               [{:metric 2} {} {:metric 3}]
                               [{:metric 2} {} {:metric 5}])

                  (test-stream (counter 100)
                               [{:metric 2} {} {:metric 3}]
                               [{:metric 102} {} {:metric 105}]))

         (testing "resets"
                  (test-stream (counter 100)
                               [{:metric 1}
                                {:metric 200 :tags ["reset"]}
                                {:metric 5}]
                               [{:metric 101}
                                {:metric 200 :tags ["reset"]}
                                {:metric 205}])))

(deftest match-test
         ; Regular strings.
         (test-stream (match :service "foo")
                      [{}
                       {:service "bar"}
                       {:service "foo"}]
                      [{:service "foo"}])

         ; Sets
         (test-stream (match :metric #{0 2})
                      [{}
                       {:metric 1}
                       {:metric 2}]
                      [{:metric 2}])

  Regexen
         (test-stream (match :state #"^mi")
                      [{}
                       {:state "migas"}
                       {:state "other breakfast foods"}]
                      [{:state "migas"}])

         ; Functions
         (test-stream (match identity 2)
                      [1 2 3]
                      [2])

         ; Maps
         (test-stream (match identity {:state #"^mi", :host "other"})
                      [{}
                       {:state "migas"}
                       {:state "migas", :host "other"}]
                      [{:state "migas", :host "other"}])

  ; Nils
  (test-stream (match :host nil)
               [{}
                {:host nil}
                {:host :foo}]
               [{}
                {:host nil}]))

(deftest untag-test
         (testing "single tag"
                  (test-stream (untag "foo")
                               [{}
                                {:service :a :tags []}
                                {:service :a :tags ["foo"]}
                                {:service :b :tags ["foo" "bar" "baz"]}]
                               [{:tags []}
                                {:service :a :tags []}
                                {:service :a :tags []}
                                {:service :b :tags ["bar" "baz"]}]))

         (testing "multiple tags"
                  (test-stream (untag ["foo" "bar"])
                               [{}
                                {:service :a :tags []}
                                {:service :a :tags ["foo"]}
                                {:service :b :tags ["foo" "bar" "baz"]}]
                               [{:tags []}
                                {:service :a :tags []}
                                {:service :a :tags []}
                                {:service :b :tags ["baz"]}])))

(deftest tag-test
         ; Single tag
         (test-stream (tag "foo")
                      [{}
                       {:service :a :tags ["foo"]}
                       {:service :b :tags ["bar" "baz"]}]
                      [{:tags ["foo"]}
                       {:service :a :tags ["foo"]}
                       {:service :b :tags ["foo" "bar" "baz"]}])

         ; Multiple tags
         (test-stream (tag ["foo" "bar"])
                      [{}
                       {:service :a :tags ["foo"]}
                       {:service :b :tags ["foo" "baz"]}]
                      [{:tags ["foo" "bar"]}
                       {:service :a :tags ["foo" "bar"]}
                       {:service :b :tags ["foo" "bar" "baz"]}]))

(deftest tagged-all-test
         (test-stream (tagged-all ["kitten" "cat"])
                      [{:tags ["kitten" "cat"]}
                       {:tags ["kitten", "cat", "meow"]}
                       {:tags ["dog" "cat"]}
                       {:tags ["cat"]}
                       {:tags []}
                       {}]
                      [{:tags ["kitten" "cat"]}
                       {:tags ["kitten", "cat", "meow"]}])

         (test-stream (tagged-all "meow")
                      [{:tags ["meow" "bark"]}
                       {:tags ["meow"]}
                       {:tags ["bark"]}
                       {}]
                      [{:tags ["meow" "bark"]}
                       {:tags ["meow"]}])

         (testing "return values"
                  (is (true? ((tagged-all ["meow" "bark"])
                                {:tags ["meow" "bark" "grr"]})))
                  (is (nil? ((tagged-all ["meow" "bar"])
                               {:tags ["meow"]})))))

(deftest tagged-any-test
         (test-stream (tagged-any ["kitten" "cat"])
                      [{:tags ["kitten" "cat"]}
                       {:tags ["cat", "dog"]}
                       {:tags ["kitten"]}
                       {:tags ["dog"]}
                       {:tags []}
                       {}]
                      [{:tags ["kitten" "cat"]}
                       {:tags ["cat", "dog"]}
                       {:tags ["kitten"]}])

         (test-stream (tagged-any "meow")
                      [{:tags ["meow" "bark"]}
                       {:tags ["meow"]}
                       {:tags ["bark"]}
                       {}]
                      [{:tags ["meow" "bark"]}
                       {:tags ["meow"]}])

         (testing "return values"
                  (is (true? ((tagged-any ["meow" "bark"])
                                {:tags ["meow" "moo"]})))
                  (is (nil? ((tagged-any ["meow" "bar"])
                               {:tags ["moo"]})))))

(deftest split*-test
         (testing "with one branch"
                  (test-stream (split* identity)
                               [true false nil 2]
                               [true 2]))

         (testing "with a default stream"
                  ;; dispatch with default value
                  (let [sup    (fn [threshold] #(< threshold (:metric %)))
                        res    (atom [])
                        events [{:metric 15} {:metric 8} {:metric 2}]
                        expect [{:metric 15 :state :critical}
                                {:metric 8 :state :warn}
                                {:metric 2 :state :ok}]
                        stream (split* (sup 10) (with :state :critical
                                                      (partial swap! res conj))
                                       (sup 5) (with :state :warn
                                                     (partial swap! res conj))
                                       (with :state :ok
                                             (partial swap! res conj)))]
                    (dorun (map stream events))
                    (is (= expect @res))))

         (testing "with no default stream"
                  (let [sup    (fn [threshold] #(< threshold (:metric %)))
                        res    (atom [])
                        events [{:metric 15} {:metric 8} {:metric 2}]
                        expect [{:metric 15 :state :critical}
                                {:metric 8 :state :warn}]
                        stream (split*
                                 (sup 10) (with :state :critical
                                                (partial swap! res conj))
                                 (sup 5) (with :state :warn
                                               (partial swap! res conj)))]
                    (dorun (map stream events))
                    (is (= expect @res)))))

(deftest split-test
         ;; same test as above, using implicit rewrites
         (testing "with a default stream"
                  (let [sup    (fn [threshold] #(< threshold (:metric %)))
                        res    (atom [])
                        events [{:metric 15} {:metric 8} {:metric 2}]
                        expect [{:metric 15 :state :critical}
                                {:metric 8 :state :warn}
                                {:metric 2 :state :ok}]
                        stream (split
                                 (> metric 10) (with :state :critical
                                                     (partial swap! res conj))
                                 (> metric 5) (with :state :warn
                                                    (partial swap! res conj))
                                 (with :state :ok
                                       (partial swap! res conj)))]
                    (dorun (map stream events))
                    (is (= expect @res))))

         (testing "evaluates streams once"
                  (let [res (atom [])
                        stream (split
                                 true
                                 (let [state (atom 0)]
                                   (fn [_]
                                     (swap! res conj
                                            (swap! state inc)))))]
                    (stream :x)
                    (stream :x)
                    (stream :x)
                    (is (= @res [1 2 3])))))

(deftest splitp-test
         ;; same test as above, using splitp
         (testing "basics"
                  (let [res    (atom [])
                        events [{:metric 15} {:metric 8} {:metric 2}]
                        expect [{:metric 15 :state :crit}
                                {:metric 8 :state :warn}
                                {:metric 2 :state :ok}]
                        stream (splitp <= metric
                                       10 (with :state :crit
                                                (partial swap! res conj))
                                       5  (with :state :warn
                                                (partial swap! res conj))
                                       (with :state :ok
                                             (partial swap! res conj)))]
                    (dorun (map stream events))
                    (is (= expect @res))))

         (testing "Without a default"
                  (is (thrown? IllegalArgumentException
                               ((splitp = true
                                        false :doesnt-happen)
                                  :foo))))

         (testing "Evaluates child streams once at creation time"
                  (let [a (atom 0)
                        b (atom 0)
                        s (splitp = state
                                  :foo (do (swap! a inc) identity)
                                       (do (swap! b inc) identity))]
                    (is (= 1 @a))
                    (is (= 1 @b))
                    (s {:state :foo})
                    (s {:state :foo})
                    (s {:state :bar})
                    (s {:state :baz})
                    (is (= 1 @a))
                    (is (= 1 @b)))))

(deftest where*-test
         (test-stream (where* identity)
                      [true false nil 2]
                      [true 2])

         (test-stream (where* expired?)
                      [{:time -1 :ttl 0.5}
                       {:time 0 :ttl 1}]
                      [{:time -1 :ttl 0.5}])

         ; Complex closure with else clause
         (let [good (atom [])
               bad  (atom [])
               s (where* (fn [event]
                           (or (= "good" (:service event))
                               (< 2 (:metric event))))
                         (partial swap! good conj)
                         (else (partial swap! bad conj)))
               events [{:service "good" :metric 0}
                       {:service "bad" :metric 0}
                       {:metric 1}
                       {:service "bad" :metric 1}
                       {:service "bad" :metric 3}]]

           ; Run stream
           (doseq [e events] (s e))

           (is (= @good
                  [{:service "good" :metric 0}
                   {:service "bad" :metric 3}]))
           (is (= @bad
                  [{:service "bad" :metric 0}
                   {:metric 1}
                   {:service "bad" :metric 1}])))

         (testing "evaluates children once"
                  ; Where should evaluate its children exactly once.
                  (let [x (atom 0)
                        s (where* (fn [e] true) (do (swap! x inc) identity))]
                    (is (= @x 1))
                    (s {:service "test"})
                    (is (= @x 1))
                    (s {:service "test"})
                    (is (= @x 1)))))

(deftest where*-return-value
         ; Where*'s return value should be whether the predicate matched.
         (is (= true ((where* expired? (fn [e] "hi"))
                        (expire {}))))
         (is (= 2 ((where* (constantly 2)) :zoom))))

(deftest where-test
         (testing "field"
                  (let [r (ref [])
                        s (where (or (state "ok" "good")
                                     (= "weird" state))
                                 (fn [e] (dosync (alter r conj e))))
                        events [{:state "ok"}
                                {:state "good"}
                                {:state "weird"}
                                {:state "error"}]
                        expect [{:state "ok"}
                                {:state "good"}
                                {:state "weird"}]]
                    (doseq [e events] (s e))
                    (is (= expect (deref r)))))

         (testing "event binding"
           (test-stream (where (= event 2))
                        [1 2 3 2]
                        [2 2]))

         (testing "regex"
                  (test-stream (where (service #"^foo"))
                               [{}
                                {:service "foo"}
                                {:service "food"}]
                               [{:service "foo"}
                                {:service "food"}]))

         (testing "functions"
                  (test-stream (where (and metric (even? metric)))
                               [{}
                                {:metric 1}
                                {:metric 2}]
                               [{:metric 2}]))

         (testing "functions as values"
  May not support this later . I do n't think it 's all that
                  ; useful given you can just use (even? metric).
                  (test-stream (where (and metric (metric even?)))
                               [{}
                                {:metric 1}
                                {:metric 2}]
                               [{:metric 2}]))

         (testing "using sets as predicates"
                  (test-stream (where (service #{"service1" "service2"}))
                               [{:service "service1"}
                                {:service "service2"}
                                {:service "service-doesnt-match"}]
                               [{:service "service1"}
                                {:service "service2"}]))


         (testing "variable"
                  ; Verify that the macro allows variables to be used in
                  ; predicates.
                  (let [regex #"cat"]
                    (test-stream (where (service regex))
                                 [{:service "kitten"}
                                  {:service "cats"}]
                                 [{:service "cats"}])))


         (testing "tagged (one tag)"
                  (let [r (atom [])
                        s (where (tagged "foo") (append r))
                        events [{}
                                {:tags []}
                                {:tags ["blah"]}
                                {:tags ["foo"]}
                                {:tags ["foo" "bar"]}]]
                    (doseq [e events] (s e))
                    (is (= (deref r)
                           [{:tags ["foo"]} {:tags ["foo" "bar"]}]))))

         (testing "tagged-all (one tag)"
                  (let [r (atom [])
                        s (where (tagged-all "foo") (append r))
                        events [{}
                                {:tags []}
                                {:tags ["blah"]}
                                {:tags ["foo"]}
                                {:tags ["foo" "bar"]}]]
                    (doseq [e events] (s e))
                    (is (= (deref r)
                           [{:tags ["foo"]} {:tags ["foo" "bar"]}]))))

         (testing "tagged-any (one tag)"
                  (let [r (atom [])
                        s (where (tagged-any "foo") (append r))
                        events [{}
                                {:tags []}
                                {:tags ["blah"]}
                                {:tags ["foo"]}
                                {:tags ["foo" "bar"]}]]
                    (doseq [e events] (s e))
                    (is (= (deref r)
                           [{:tags ["foo"]} {:tags ["foo" "bar"]}]))))

         (testing "tagged (multiple tags)"
                  (let [r (atom [])
                        s (where (tagged ["foo" "bar"]) (append r))
                        events [{}
                                {:tags []}
                                {:tags ["blah"]}
                                {:tags ["foo"]}
                                {:tags ["foo" "bar"]}
                                {:tags ["foo" "bar" "baz"]}]]
                    (doseq [e events] (s e))
                    (is (= (deref r)
                           [{:tags ["foo" "bar"]} {:tags ["foo" "bar" "baz"]}]))))

         (testing "tagged-all (multiple tags)"
                  (let [r (atom [])
                        s (where (tagged-all ["foo" "bar"]) (append r))
                        events [{}
                                {:tags []}
                                {:tags ["blah"]}
                                {:tags ["foo"]}
                                {:tags ["foo" "bar"]}
                                {:tags ["foo" "bar" "baz"]}]]
                    (doseq [e events] (s e))
                    (is (= (deref r)
                           [{:tags ["foo" "bar"]} {:tags ["foo" "bar" "baz"]}]))))

         (testing "tagged-any (multiple tags)"
                  (let [r (atom [])
                        s (where (tagged-any ["foo" "bar"]) (append r))
                        events [{}
                                {:tags []}
                                {:tags ["blah"]}
                                {:tags ["foo"]}
                                {:tags ["foo" "bar"]}
                                {:tags ["baz" "bar"]}]]
                    (doseq [e events] (s e))
                    (is (= (deref r)
                           [{:tags ["foo"]} {:tags ["foo" "bar"]} {:tags ["baz" "bar"]}]))))

         (testing "else"
                  ; Where should take an else clause.
                  (let [a (atom [])
                        b (atom [])]
                    (run-stream
                      (where (service #"a")
                             #(swap! a conj (:service %))
                             (else #(swap! b conj (:service %))))
                      [{:service "cat"}
                       {:service "dog"}
                       {:service nil}
                       {:service "badger"}])
                    (is (= @a ["cat" "badger"]))
                    (is (= @b ["dog" nil]))))

         (testing "evaluates children once"
                  ; Where should evaluate its children exactly once.
                  (let [x (atom 0)
                        s (where true (do (swap! x inc) identity))]
                    (is (= @x 1))
                    (s {:service "test"})
                    (is (= @x 1))
                    (s {:service "test"})
                    (is (= @x 1))))

         (testing "return value"
                  ; Where's return value should be whether the predicate
                  ; matched.
                  (is (= true  ((where (service "foo")) {:service "foo"})))
                  (is (= false ((where (service "foo")) {:service "bar"})))

                  (is (= true  ((where (tagged "foo")) {:tags ["foo"]})))
                  (is (= false ((where (tagged "foo")) {:tags ["bar"]})))
                  (is (= true  ((where (tagged-all "foo")) {:tags ["foo"]})))
                  (is (= false ((where (tagged-all "foo")) {:tags ["bar"]})))
                  (is (= true  ((where (tagged-any "foo")) {:tags ["foo"]})))
                  (is (= false ((where (tagged-any "foo")) {:tags ["bar"]})))

                  (is (= true  ((where (tagged ["foo" "bar"])) {:tags ["foo" "bar"]})))
                  (is (= false ((where (tagged ["foo" "bar"])) {:tags ["bar"]})))
                  (is (= true  ((where (tagged-all ["foo" "bar"])) {:tags ["foo" "bar"]})))
                  (is (= false ((where (tagged-all ["foo" "bar"])) {:tags ["bar"]})))
                  (is (= true ((where (tagged-any ["foo" "bar"])) {:tags ["foo" "bar"]})))
                  (is (= false ((where (tagged-any ["foo" "bar"])) {:tags ["blah"]})))

                  (is (= true ((where (service "foo")
                                      (fn [event] 2))
                                 {:service "foo"})))
                  (is (= false ((where (service "foo")
                                       (else (fn [event] 2)))
                                  {:service "bar"})))

                  (is (= 2 ((where 2) :wheeee!)))
                  (is (= nil ((where nil) :zoooom!)))))

(deftest default-kv
         (let [r (atom nil)
               s (default :service "foo" (register r))]
           (s {:service nil})
           (is (= {:service "foo"} (deref r)))

           (s {:service "foo"})
           (is (= {:service "foo"} (deref r)))

           (s {:service "bar" :test "baz"})
           (is (= {:service "bar" :test "baz"} (deref r)))))

(deftest default-map
         (let [r (atom nil)
               s (default {:service "foo" :state nil} (register r))]
           (s (event {:service nil}))
           (is (= "foo" (:service (deref r))))
           (is (= nil (:state (deref r))))

           (s (event {:service "foo"}))
           (is (= "foo" (:service (deref r))))
           (is (= nil (:state (deref r))))

           (s (event {:service "bar" :host "baz" :state "evil"}))
           (is (= "bar" (:service (deref r))))
           (is (= "baz" (:host (deref r))))
           (is (= "evil" (:state (deref r))))))

(deftest with-kv
         (let [r (ref nil)
               s (with :service "foo" (fn [e] (dosync (ref-set r e))))]
           (s {:service nil})
           (is (= {:service "foo"} (deref r)))

           (s {:service "foo"})
           (is (= {:service "foo"} (deref r)))

           (s {:service "bar" :test "baz"})
           (is (= {:service "foo" :test "baz"} (deref r)))

           (s [{:service "bar" :test "baz"}])
           (is (= [{:service "foo" :test "baz"}] (deref r)))))

(deftest with-map
         (let [r (ref nil)
               s (with {:service "foo" :state nil} (fn [e] (dosync (ref-set r e))))
               empty-s (with {} (fn [e] (dosync (ref-set r e))))]
           (s (event {:service nil}))
           (is (= "foo" (:service (deref r))))
           (is (= nil (:state (deref r))))

           (s (event {:service "foo"}))
           (is (= "foo" (:service (deref r))))
           (is (= nil (:state (deref r))))

           (s (event {:service "bar" :test "baz" :state "evil"}))
           (is (= "foo" (:service (deref r))))
           (is (= nil (:state (deref r))))

           (empty-s (event {:service "bar" :test "baz"}))
           (is (= "bar" (:service (deref r))))))

(deftest by-single
  Each test stream keeps track of the first host it sees , and confirms
         ; that each subsequent event matches that host.
         (let [i (ref 0)
               s (by :host
                     (let [host (ref nil)]
                       (fn [event]
                         (dosync
                           (alter i inc)
                           (when (nil? (deref host))
                             (ref-set host (event :host)))
                           (is (= (deref host) (event :host)))))))
               events (map (fn [h] {:host h}) [:a :a :b :a :c :b])]
           (doseq [event events]
             (s event))
           (is (= (count events) (deref i)))))

(deftest by-multiple
  Each test stream keeps track of the first host / service it sees , and
         ; confirms that each subsequent event matches that host.
         (let [i (ref 0)
               s (by [:host :service]
                     (let [host (ref nil)
                           service (ref nil)]
                       (fn [event]
                         (dosync
                           (alter i inc)

                           (when (nil? (deref host))
                             (ref-set host (event :host)))
                           (when (nil? (deref service))
                             (ref-set service (event :service)))

                           (is (= (deref host) (event :host)))
                           (is (= (deref service) (event :service)))))))

               events (map (fn [h] {:host (first h)
                                    :service (last h)})
                           [[1 :a]
                            [1 :b]
                            [1 :a]
                            [2 :a]
                            [2 :a]
                            [1 :a]
                            [2 :a]
                            [1 :b]])]
           (doseq [event events]
             (s event))
           (is (= (count events) (deref i)))))

(deftest by-builder-test
  (testing "one child"
    (let [i (ref 0)
          s (by-builder [host :host]
                        (let [host-ref (ref nil)]
                          (fn [event]
                            (dosync
                             (alter i inc)
                             (when (nil? (deref host-ref))
                               (ref-set host-ref (event :host)))
                             (is (= (deref host-ref) (event :host) host))))))
          events (map (fn [h] {:host h}) [:a :a :b :a :c :b])]
      (doseq [event events]
        (s event))
      (is (= (count events) (deref i)))))

  (testing "two children"
    (let [i (ref 0)
          s (by-builder [host :host]
                        (let [host-ref (ref nil)]
                          (fn [event]
                            (dosync
                             (alter i inc)
                             (when (nil? (deref host-ref))
                               (ref-set host-ref (event :host)))
                             (is (= (deref host-ref) (event :host) host)))))
                        (let [host-ref (ref nil)]
                          (fn [event]
                            (dosync
                             (alter i inc)
                             (when (nil? (deref host-ref))
                               (ref-set host-ref (event :host)))
                             (is (= (deref host-ref) (event :host) host))))))
          events (map (fn [h] {:host h}) [:a :a :b :a :c :b])]
      (doseq [event events]
        (s event))
      (is (= (* 2 (count events)) (deref i)))))

  (testing "multiple fields"
    (let [i (ref 0)
          s (by-builder [[host service] [:host :service]]
                (let [host-ref (ref nil)
                      service-ref (ref nil)]
                  (fn [event]
                    (dosync
                     (alter i inc)
                     (when (nil? (deref host-ref))
                       (ref-set host-ref (event :host)))
                     (when (nil? (deref service-ref))
                       (ref-set service-ref (event :service)))

                     (is (= (deref host-ref) (event :host) host))
                     (is (= (deref service-ref) (event :service)) service)))))

          events (map (fn [h] {:host (first h)
                               :service (last h)})
                      [[1 :a]
                       [1 :b]
                       [1 :a]
                       [2 :a]
                       [2 :a]
                       [1 :a]
                       [2 :a]
                       [1 :b]])]
      (doseq [event events]
        (s event))
      (is (= (count events) (deref i))))))

(deftest by-evaluates-children-once-per-branch
         (let [i (atom 0)
               s (by :metric (do (swap! i inc) identity))]
           (is (= @i 0))
           (s {:metric 1})
           (is (= @i 1))
           (s {:metric 2})
           (is (= @i 2))
           (s {:metric 1})
           (is (= @i 2))))

(deftest pipe-test
  (testing "One stage"
    (test-stream
      (pipe !)
      [1 2 3]
      [1 2 3]))

  (testing "Chained stages"
    (test-stream
      (pipe ! (smap inc !) (smap (partial * 2) !))
      [1 2 3]
      [4 6 8]))

  (testing "Multiple markers"
    (test-stream
      (pipe - (splitp < metric
                      0.9 (with :state :critical -)
                      0.5 (with :state :warning  -)
                          (with :state :ok       -))
            (smap identity -))
      [{:metric 0.1}
       {:metric 0.7}
       {:metric 1.0}]
      [{:metric 0.1 :state :ok}
       {:metric 0.7 :state :warning}
       {:metric 1.0 :state :critical}]))

  (testing "Rebinding a marker"
    (test-stream
      (pipe - (let [! -
                    - clojure.core/-]
                (smap - !)))
      [-1 0 1]
      [1 0 -1])))

(deftest fill-in-test
         (test-stream-intervals
           (fill-in 0.01 {:metric 0})
           []
           [])

         ; Quick succession
         (is (= (map :metric (run-stream-intervals
                               (fill-in 0.01 {:metric 0})
                               (interpose nil (em 1 2 3))))
                [1 2 3]))

         ; With a gap and expiry
         (is (= (map :metric (run-stream-intervals
                               (fill-in 0.05 {:metric 0})
                               [{:metric 1} 0.06
                                {:metric 2} nil
                                {:metric 3} 0.08
                                {:metric 4 :state "expired"} 0.06
                                {:metric 5}]))
                [1 0 2 3 0 4 5]))
         )

(deftest fill-in-last-test
  (test-stream-intervals
    (fill-in-last 0.01 {:metric 0})
    []
    [])

  ; Quick succession
  (let [output (run-stream-intervals
                 (fill-in-last 0.01 {:metric 0})
                 (interpose nil (evs {:host "foo" :service "bar"}
                                     1 2 3)))]
    (is (= (map :metric output)
           [1 2 3]))
    (is (= (map :host output)
           ["foo" "foo" "foo"]))
    (is (= (map :service output)
           ["bar" "bar" "bar"])))

  ; With a gap and expiry
  (let [output (run-stream-intervals
                 (fill-in-last 0.05 {:metric 0})
                 [{:host "a" :metric 1} 0.06
                  {:host "b" :metric 2} nil
                  {:host "c" :metric 3} 0.08
                  {:host "d" :metric 4 :state "expired"} 0.06
                  {:host "e" :metric 5}])]
    (is (= (map :metric output)
           [1 0 2 3 0 4 5]))
    (is (= (map :host output)
           ["a" "a" "b" "c" "c" "d" "e"]))))

(deftest interpolate-constant-test
         (test-stream-intervals
           (interpolate-constant 0.01)
           []
           [])

         ; Should forward a single state
         (is (= (map :metric (run-stream-intervals
                               (interpolate-constant 0.1)
                               [{:metric 1} 0.05]))
                [1]))

  Should forward first state and ignore immediate successors
         (is (= (map :metric (run-stream-intervals
                               (interpolate-constant 0.1)
                               [{:metric 1} 0.05
                                {:metric 2} nil
                                {:metric 3}]))
                [1]))

         ; Should fill in missing states regularly
         (is (= (map :metric (run-stream-intervals
                               (interpolate-constant 0.1)
                               (interpose 0.22 (em 1 2 3 4))))
                [1 1 1 2 2 3 3]))

         ; Should forward final "expired" state.
         (is (= (map :metric (run-stream-intervals
                               (interpolate-constant 0.1)
                               (interpose 0.22
                                          [{:metric 1}
                                           {:metric 2}
                                           {:metric 3}
                                           {:metric 4 :state "expired"}])))
                [1 1 1 2 2 3 3 4]))

         ; Should not fill during expired times.
         (is (= (map :metric (run-stream-intervals
                               (interpolate-constant 0.05)
                               [{:metric 1 :state "expired"}
                                0.12
                                {:metric 2}
                                0.12
                                {:metric 3 :state "expired"}
                                0.12]))
                [1 2 2 2 3]))
         )

(deftest ddt-immediate-test
         ; Empty -> empty
         (test-stream (ddt) [] [])
         ; Ignore stream without metrics
         (test-stream (ddt) [{} {} {} {}] [])
  Do nothing the first time
         (test-stream (ddt) [{:metric 1 :time 0}] [])
  Differentiate
         (test-stream (ddt)
                      [{:metric 0 :time 0}
                       {:metric 0 :time 1}
                       {:metric 2 :time 2}
                       {:metric -4 :time 4}]
                      [{:metric 0 :time 1}
                       {:metric 2 :time 2}
                       {:metric -3 :time 4}]))

(deftest ddt-interval-test
         (testing "Quick burst without crossing interval"
                  (is (= (map :metric (run-stream-intervals
                                        (ddt 0.1)
                                        [{:metric 1} nil
                                         {:metric 2} nil
                                         {:metric 3}]))
                         [])))

         (testing "1 event per interval"
                  (test-stream-intervals
                    (ddt 1)
  Note that the swap occurs just prior to events at time 1 .
                    [{:time 0 :metric -1} 99/100
                     {:time 1 :metric 0} 1
                     {:time 2 :metric -5} 1]
                    [{:time 1 :metric 1}
                     {:time 2 :metric -5}]))

         (testing "n events per interval"
                  (test-stream-intervals
                    (ddt 1)
                    [{:time 0 :metric -1} 1/100
                     {:time 1/2 :metric 100} 1/2 ; Ignored
                     {:time 2/2 :metric 1} 1/2
                     {:time 3/2 :metric nil} 1/2 ; Ignored
                     {:time 4/2 :metric -3} 1/2]
                    [{:time 2/2 :metric 2}
                     {:time 4/2 :metric -4}]))

         (testing "emits zeroes when no events arrive in an interval"
                  (test-stream-intervals (ddt 2)
                                         [{:time 0 :metric 0} 1
                                          {:time 1 :metric 1} 2
                                          {:time 3 :metric 2} 3]
                                         [{:time 2 :metric 1}
                                          {:time 4 :metric 1/2}
                                          {:time 6 :metric 0}])))

(deftest rate-slow-even
         (let [output (ref [])
               interval 1
               intervals 5
               gen-rate 10
               total (* gen-rate intervals)
               gen-period (/ interval gen-rate)
               r (rate interval
                        (fn [event] (dosync (alter output conj event))))]

           ; Generate events
           (dotimes [_ intervals]
             (dotimes [_ gen-rate]
               (advance! (+ (unix-time) gen-period))
               (r {:metric 1 :time (unix-time)})))

           ; Give all futures time to complete
           (advance! (+ (unix-time) gen-period))

           ; Verify output states
           (let [o (deref output)]

             ; All events recorded
             (is (approx-equal total (reduce + (map :metric o))))

             ; Middle events should have the correct rate.
             (is (every? (fn [measured-rate]
                           (approx-equal gen-rate measured-rate))
                         (map :metric (drop 1 (drop-last o)))))

  First and last events should be complementary
             (let [first-last (+ (:metric (first o))
                                 (:metric (last o)))]
               (is (or (approx-equal (* 2 gen-rate) first-last)
                       (approx-equal gen-rate first-last))))

             )))

(deftest rate-threaded
         (let [output (atom nil)
               interval 5/2
               total 10000
               threads 4
               r (rate interval
                       (fn [event] (dosync (reset! output event))))

               ; Generate events
               workers (map (fn [t] (future
                                      (dotimes [i (/ total threads)]
                                        (r {:metric 1 :time (unix-time)}))))
                            (range threads))]

           ; Wait for workers
           (dorun (map deref workers))
           (advance! interval)

           ; All events recorded
           (is (= (/ total interval) (:metric @output)))))

(deftest rate-without-input
         (test-stream-intervals
           (rate 1)
           [{:metric 1 :service "foo"} 0.5
            {:metric 1 :service "bar"} 0.5
            {:metric 1 :service "baz" :ttl 3} 3
            {:state "expired"}]
           [{:time 1 :metric 2 :service "bar"}
            {:time 2 :metric 1 :service "baz" :ttl 3}
            {:time 3 :metric 0 :service "baz" :ttl 2}
            {:time 4 :metric 0 :service "baz" :ttl 1}]))

(deftest rate-expiration
         (test-stream-intervals
           (rate 1)
           [{:time 0 :metric 1 :ttl 3} 5]
           [{:time 1 :metric 1 :ttl 3}
            {:time 2 :metric 0 :ttl 2}
            {:time 3 :metric 0 :ttl 1}]))

(deftest fold-interval-test
         (test-stream-intervals
           (riemann.streams/fold-interval 1 :metric (partial apply +))
           [{:metric 2} 0.1
            {:metric 4} 0.2
            {:metric 2} 0.3
            {:metric 4} 1.0
            {:metric 100} 0.1
            {:metric 100} 1.0]
           (em 12 200)))

(deftest fold-interval-metric-test
         (test-stream-intervals
           (riemann.streams/fold-interval-metric 1 (partial apply +))
           [{:metric 2} 0.1
            {:metric 4} 0.2
            {:metric 2} 0.3
            {:metric 4} 1.0
            {:metric 100} 0.1
            {:metric 100} 1.0]
           (em 12 200)))

(deftest changed-test
  (let [states (partial map (partial hash-map :state))]
    (testing ":state"
      (test-stream (changed :state)
                   (states [:ok :bad :bad :ok :ok :ok :evil :bad])
                   (states [:ok :bad :ok :evil :bad])))

    (testing ":state with :init"
      (test-stream (changed :state {:init :ok})
                   (states [:ok :bad :bad :ok :ok :evil :bad])
                   (states [:bad :ok :evil :bad])))

    (testing "with previous event and arbitrary fn"
      (test-stream (changed #(:state %) {:init :ok, :pairs? true})
                   (states [:ok :bad :bad :ok :ok :evil :bad])
                   (partition 2 1 (states [:ok :bad :ok :evil :bad]))))))

(deftest changed-with-exception-test
        (logging/suppress
          "riemann.streams"
           (let [exceptions (atom [])]
             (binding [riemann.streams/*exception-stream* #(swap! exceptions conj %)]
               (run-stream (changed :state (fn [x]
                                             (throw (Exception. "boo!"))))
                           [(riemann.common/event {:state :critical})]))
             (is (= 1 (count @exceptions))))))

(deftest changed-state-test
  Each test stream keeps track of the first host / service it sees , and
         ; confirms that each subsequent event matches that host, and that
         ; each event is different from the previous state.
         (let [i (ref 0)
               s (changed-state
                   (let [host (ref nil)
                         service (ref nil)
                         state (ref nil)]
                     (fn [event]
                       (dosync
                         (alter i inc)

                         (is (not= (deref state) (:state event)))
                         (ref-set state (:state event))

                         (when (nil? (deref host))
                           (ref-set host (event :host)))
                         (when (nil? (deref service))
                           (ref-set service (event :service)))

                         (is (= (deref host) (event :host)))
                         (is (= (deref service) (event :service)))))))

               events [{:host 1 :service 1 :state 1}
                       {:host 2 :service 1 :state 1}
                       {:host 1 :service 1 :state 1}
                       {:host 1 :service 1 :state 2}
                       {:host 2 :service 1 :state 2}
                       {:host 2 :service 2 :state 1}
                       {:host 2 :service 1 :state 1}]]

           (doseq [event events]
             (s event))
           (is (= 6 (deref i)))))

(deftest over-test
(logging/suppress ["riemann.streams"]
         (let [output (ref [])
               r (over 1.5
                         (fn [e] (dosync (alter output conj e))))
               metrics [0.5 1 1.5 2 2.5]
               expect [2 2.5]]

           (doseq [m metrics] (r {:metric m}))
           (is (= expect (vec (map (fn [s] (:metric s)) (deref output))))))))

(deftest under-test
(logging/suppress ["riemann.streams"]
         (let [output (ref [])
               r (under 1.5
                         (fn [e] (dosync (alter output conj e))))
               metrics [0.5 1 1.5 2 2.5]
               expect [0.5 1]]

           (doseq [m metrics] (r {:metric m}))
           (is (= expect (vec (map (fn [s] (:metric s)) (deref output))))))))

(deftest ewma-timeless-test
         (test-stream (ewma-timeless 0)
                      (em 1 10 20 -100 4)
                      (em 0 0  0   0   0))
         (test-stream (ewma-timeless 1)
                      (em 5 13 1 -10 3)
                      (em 5 13 1 -10 3))
         (test-stream (ewma-timeless 1/2)
                      (em 1   1   1   1     1    )
                      (em 1/2 3/4 7/8 15/16 31/32)))

(deftest ewma-test
  (let [output1 (atom [])
        output2 (atom [])
        output5 (atom [])
        seconds 50
        s1 (ewma 1
          (fn [event] (swap! output1 conj event)))
        s2 (ewma 2
          (fn [event] (swap! output2 conj event)))
        s5 (ewma 5
          (fn [event] (swap! output5 conj event)))]

    ; Generate enough events to allow the means to converge
    (dotimes [n seconds]
      (s1 {:metric 1 :time (inc n)})
      (s2 {:metric 1 :time (inc n)})
      (s5 {:metric 1 :time (inc n)}))

    ; Verify that the means have converged
    (is (every? (fn [measured-rate]
                  (approx-equal 1 measured-rate))
                  (map :metric (drop 10 @output1))))
    (is (every? (fn [measured-rate]
                  (approx-equal 1 measured-rate))
                  (map :metric (drop 15 @output2))))
    (is (every? (fn [measured-rate]
                  (approx-equal 1 measured-rate))
                  (map :metric (drop 40 @output5))))

  Verify halflives with spaced out : metric 0 events
    (s1 {:metric 0 :time (+ 1 seconds)})
    (s1 {:metric 0 :time (+ 2 seconds)})
    (is (approx-equal 1/2 (:metric (last (drop-last @output1)))))
    (is (approx-equal 1/4 (:metric (last @output1))))

    (s2 {:metric 0 :time (+ 2 seconds)})
    (s2 {:metric 0 :time (+ 4 seconds)})
    (is (approx-equal 1/2 (:metric (last (drop-last @output2)))))
    (is (approx-equal 1/4 (:metric (last @output2))))

    (s5 {:metric 0 :time (+ 5 seconds)})
    (s5 {:metric 0 :time (+ 10 seconds)})
    (is (approx-equal 1/2 (:metric (last (drop-last @output5)))))
    (is (approx-equal 1/4 (:metric (last @output5)))))

  ; Verify that metrics are weighted by :time, regardless of order of arrival
  (test-stream-intervals (ewma 1)
    [{:metric 1 :time 0} 1
     {:metric 1 :time 2} 1
     {:metric 1 :time 1} 1]
    [{:metric 0.5 :time 0}
     {:metric 0.625 :time 2}
     {:metric 0.875 :time 2}]))

(deftest top-test
  (let [e (fn [s m & tags] {:service s :metric m :tags tags})
        a (fn [m & tags] (apply e :a m tags))
        b (fn [m & tags] (apply e :b m tags))
        c (fn [m & tags] (apply e :c m tags))
        d (fn [m & tags] (apply e :d m tags))]

           ; A single event
           (test-stream (top 1 :metric)
                        [(a 1)]
                        [(a 1)])

           ; Repeating the same service
           (test-stream (top 1 :metric)
                        [(a 1) (a 2) (a 1) (a 3)]
                        [(a 1) (a 2) (a 1) (a 3)])

           ; Displacing a smaller event
           (test-stream (top 2 :metric)
                        [(a 1) (b 2) (c 3) (a 1)          (b 2)]
                        [(a 1) (b 2) (c 3) (expire (a 1)) (b 2)])

           ; Allowing in a smaller event when there's room
           (test-stream (top 2 :metric)
                        [(a 2) (b 2) (c 1)          (a 5) (c 1)          (a 0) (c 1)]
                        [(a 2) (b 2) (expire (c 1)) (a 5) (expire (c 1)) (a 0) (c 1)])

           ; Ignoring smaller events
           (test-stream (top 2 :metric)
                        [(a 1) (b 2) (c 3)         (d 1)          (a 2)          (b nil) (d 2)]
                        [(a 1) (b 2) (c 3) (expire (d 1)) (expire (a 2)) (expire (b nil)) (d 2)])

           ; Events without metrics
           (test-stream (top 1 :metric)
                        [(a nil)          (b 1) (a nil)]
                        [(expire (a nil)) (b 1) (expire (a nil))])

           ; Events without metrics displace previous events
           (test-stream (top 1 :metric)
                        [(b 2) (b nil)          (a 1)]
                        [(b 2) (expire (b nil)) (a 1)])

           ; Expiring a nonexistent event
           (test-stream (top 2 :metric)
                        [(expire (a 2))]
                        [(expire (a 2))])

           ; Expiring an existing event
           (test-stream (top 1 :metric)
                        [(a 2) (expire (a 1)) (b 1)]
                        [(a 2) (expire (a 1)) (b 1)])

           ; Ring
           (test-stream (top 2 :metric)
                        [(a 1) (b 2) (c 3) (d 4)         (a 2)          (b 3)  (c 4) (d 5)]
                        [(a 1) (b 2) (c 3) (d 4) (expire (a 2)) (expire (b 3)) (c 4) (d 5)])

           ; Demotion
           (with-test-stream index
             (top 1 :metric index bit-bucket true)
             [(a 1) (b 2)]
             [(a 1) (expire (a 1)) (b 2)])))

(deftest throttle-test
         (test-stream-intervals (throttle 3 2)
                                [{:state "1"} 0
                                 {:state "2"} 0
  t = 0
  t = 1
  t = 2
  t = 3
  t = 3
  t = 3
  t = 5
                                [{:state "1"}
                                 {:state "2"}
                                 {:state "3"}

                                 {:state "5"}
                                 {:state "expired"}
                                 {:state "expired"}

                                 {:state "expired"}]))

(deftest rollup-test
         (testing "single output event per interval"
           (test-stream-intervals
             (rollup 1 2)
             [:a 0 :b 0 :c 2 :d 0 :e 0 :f 2]
             [[:a] [:b :c] [:d :e :f]]))

         (testing "basic rollups"
                  (test-stream-intervals
                    (rollup 2 1)
                    [ 1 0 2 0 3 1   4 0 5 0 6 0 7 1       :foo 10]
                    [[1] [2]   [3] [4]           [5 6 7] [:foo]  ]))
           [ [ 1 ] [ 2 ]    [ 3 ] [ 4 ] [ 5 ]        [ 6 7 ] [: foo ] ]

         (testing "expired events"
                  (test-stream-intervals
                    (rollup 2 3)
                    [ 1 0 {:state "expired"} 0 :a 1 :b 1 :c 1 ]
                    [[1] [{:state "expired"}]              [:a :b :c]])

                  (let [e {:state "expired"}]
                    (test-stream-intervals
                      (rollup 2 2)
                      [ e 0 e 0 e 1 e 1]
                      [[e] [e]       [e e]]))))

(deftest batch-test
  (testing "nothing"
    (test-stream-intervals (batch 2 3) [] []))

  (testing "incomplete batches"
    (test-stream-intervals
      (batch 2 3)
      [:a 3 :b 1 :c 2 :d 3]
      [[:a] [:b :c] [:d]]))

  (testing "overflowing batches"
    (test-stream-intervals
      (batch 2 3)
      [:a 1 :b 1 :c 1 :d 1 :e 1 :f 1]
      [[:a :b] [:c] [:d :e] [:f]])))

(deftest coalesce-test
         (let [out (atom [])
               s (coalesce #(reset! out %))
               a1 {:service :a :state "one" :time 0}
               b1 {:service :b :state "one" :time 0}
               a2 {:service :a :state "two" :time 3 :ttl 2}
               c1 {:service :c :state "one" :time 0}
               b2 {:service :b :state "two" :time 0}]

           (s a1)
           (advance! 1.1)
           (is (= (set @out) #{a1}))

           (s b1)
           (advance! 2.1)
           (is (= (set @out) #{a1 b1}))

           (s a2)
           (advance! 3.1)
           (is (= (set @out) #{a2 b1}))

  Wait for  expiry of a2
           ; Should receive expired a2 once

  a2 arrived at 3 , expired at 5 ,
  and the final expiry flush happens at 6
           ; (due to the batching in coalesce)
           (s c1)
           (advance! 6.1)
           (is (= (set @out) #{a2 b1 c1}))

           (s b2)
           (advance! 7.1)
           (is (= (set @out) #{b2 c1}))))

(deftest stable-test
  Does n't emit until dt seconds have passed .
         (test-stream (stable 3 :x)
                      [{:x 1 :time 0} {:x 1 :time 1} {:x 1 :time 2}]
                      [])

         ; Constant values are emitted after dt seconds
         (test-stream (stable 3 :x)
                      [{:x 1 :time 0} {:x 1 :time 1} {:x 1 :time 3}]
                      [{:x 1 :time 0} {:x 1 :time 1} {:x 1 :time 3}])

         ; Ignores spikes
         (test-stream (stable 3 :x)
                      [{:x 0 :time 0}
                       {:x 0 :time 3}
                       {:x 1 :time 4}
                       {:x 1 :time 5}
                       {:x 0 :time 6}
                       {:x 0 :time 9}]
                      [{:x 0 :time 0}
                       {:x 0 :time 3}
                       ; spike elided
                       {:x 0 :time 6}
                       {:x 0 :time 9}])

         ; Ignores flapping
         (test-stream (stable 3 :x)
                      [{:x 0 :time 0}
                       {:x 0 :time 10}
                       {:x 1 :time 11}
                       {:x 0 :time 11}
                       {:x 1 :time 12}
                       {:x 5 :time 13}
                       {:x 2 :time 14}
                       {:x 2 :time 17}]
                      [{:x 0 :time 0}
                       {:x 0 :time 10}
                       {:x 2 :time 14}
                       {:x 2 :time 17}])

  Triggers after dt seconds of stability , even without new events .
         (test-stream-intervals (stable 10 :x)
                                [{:x 0 :time 0} 1
                                 {:x 1 :time 1} 10
                                 {:x 2 :time 11} 1]
                                [{:x 1 :time 1}])

         ; Triggers after dt seconds with new events.
         (test-stream-intervals (stable 10 :x)
                                [{:x 0 :time 0} 1
                                 {:x 0 :time 1} 4
                                 {:x 0 :time 5} 6
                                 {:x 1 :time 11} 1]
                                [{:x 0 :time 0}
                                 {:x 0 :time 1}
                                 {:x 0 :time 5}]))

(deftest project-test
         ; Empty -> empty
         (test-stream (project [(service :foo) (service :bar)])
                      []
                      [])

         ; Without anything to project to, does nothing
         (test-stream (project [])
                      [1 2 3]
                      [])

         ; Basic test: ignores non-matching events, updates state properly
         (test-stream (project [(service "foo") (service "bar")])
                      [{:service "cat"}
                       {:service "foo" :a 1}
                       {:service "foo" :a 2}
                       {:service "meow"}
                       {:service "bar" :b 3}
                       {:service "foo" :b 4}]
                      [[{:service "foo" :a 1} nil]
                       [{:service "foo" :a 2} nil]
                       [{:service "foo" :a 2} {:service "bar" :b 3}]
                       [{:service "foo" :b 4} {:service "bar" :b 3}]])

         ; Passes on initially expired events correctly
         (test-stream
           (project [(service "foo") (service "bar")])
           [{:service "foo" :state "expired"}
            {:service "foo" :state "expired"}
            {:service "cat"}]
           [[{:service "foo" :time 0 :state "expired"} nil]
            [{:service "foo" :time 0 :state "expired"} nil]])

         ; Expires existing events
         (test-stream
           (project [(service "foo") (service "bar")])
           [{:service "foo" :state "ok"}
            {:service "bar" :state "ok"}
            {:service "bar" :state "expired"}
            {:service "foo" :state "expired"}
            {:service "bar" :state "expired"}]
           [[{:service "foo" :state "ok"}
             nil]
            [{:service "foo" :state "ok"}
             {:service "bar" :state "ok"}]
            [{:service "foo" :state "ok"}
             {:service "bar" :state "expired" :time 0}]
            [{:service "foo" :state "expired" :time 0}
             nil]
            [nil
             {:service "bar" :state "expired" :time 0}]])

         ; Expiration test: expires own events when the time comes.
         (test-stream-intervals
           (project [(service "foo") (service "bar")])
           [{:service "foo" :state "ok" :time 0 :ttl 1}
            2
            {:service "bar" :state "ok"}
            1
            {:service "bar" :state "ok2"}]
           [[{:service "foo" :state "ok" :time 0 :ttl 1} nil]
            [{:service "foo" :state "expired" :time 2}
             {:service "bar" :state "ok"}]
            [nil {:service "bar" :state "ok2"}]]))

(deftest adjust-test
  (let [out (atom nil)
        s (adjust [:state str " 2"] (register out))]

    (s {})
    (is (= (deref out) {:state " 2"}))

    (s {:state "hey" :service "bar"})
    (is (= (deref out) {:state "hey 2" :service "bar"})))

  (let [out (atom nil)
        s (adjust #(assoc % :metric (count (:tags %)))
                  (register out))]

    (s {:service "a" :tags []})
    (is (= (deref out) {:service "a" :tags [] :metric 0}))

    (s {:service "a" :tags ["foo" "bar"]})
    (is (= (deref out) {:service "a" :tags ["foo" "bar"] :metric 2}))))

(deftest moving-event-window-test
  Zero - width windows .
         (test-stream (moving-event-window 0) [] [])
         (test-stream (moving-event-window 0) [1 2] [[] []])

         ; n-width windows
         (test-stream (moving-event-window 2) [1 2 3] [[1] [1 2] [2 3]]))

(deftest fixed-event-window-test
  Zero - width windows .
         (test-stream (fixed-event-window 0) [] [])
         (test-stream (fixed-event-window 0) [1 2] [])

         ; n-width windows
         (test-stream (fixed-event-window 2) [1] [])
         (test-stream (fixed-event-window 2) [1 2] [[1 2]])
         (test-stream (fixed-event-window 2) [1 2 3 4 5] [[1 2] [3 4]]))

(deftest runs-test
  Zero - width runs
         (test-stream (runs 0 :state)
                      []
                      [])

         (test-stream (runs 0 :state)
                      [{:state 2} {:state 3} {:state 4}]
                      [])

         ;; n-width runs
         (test-stream (runs 3 :state)
                      [{:state 1} {:state 2} {:state 2}]
                      [])

         (test-stream (runs 3 :state)
                      [{:state 1} {:state 1} {:metric "0.5" :state 2} {:metric "0.6" :state 2} {:state 3} {:state 4 :metric "5.0"} {:state 4 :metric "5.1"} {:state 4 :metric "5.2"}]
                      [{:state 4 :metric "5.2"}]
                      )
  )

(deftest moving-time-window-test
  Zero - second windows .
         (test-stream (moving-time-window 0) [] [])
         (test-stream (moving-time-window 0) [{:time 1} {:time 2}] [])

         ; n-width windows
         (test-stream (moving-time-window 2) [] [])
         (test-stream (moving-time-window 2) [{:time 1}] [[{:time 1}]])
         (test-stream (moving-time-window 2)
                      [{:time 1} {:time 2} {:time 3} {:time 4}]
                      [[{:time 1}]
                       [{:time 1} {:time 2}]
                       [{:time 2} {:time 3}]
                       [{:time 3} {:time 4}]])

         ; With out-of-order events
         (test-stream (moving-time-window 2)
                      [{:time 5}
                       {:time 1}
                       {:time 2}
                       {:time 6}
                       {:time 3}
                       {:time 8}
                       {:time 4}
                       {:time 8}
                       {:time 5}
                       {:time 9}]
                      [[{:time 5}]
                       [{:time 5} {:time 6}]
                       [{:time 8}]
                       [{:time 8} {:time 8}]
                       [{:time 8} {:time 8} {:time 9}]]))

(deftest fixed-time-window-test
  Zero - time windows .
         (is (thrown? IllegalArgumentException (fixed-time-window 0)))

         ; n-width windows
         (test-stream (fixed-time-window 2) [] [])
         (test-stream (fixed-time-window 2) [{:time 1}] [])
         (test-stream (fixed-time-window 2)
                      [{:time 1} {:time 2} {:time 3} {:time 4} {:time 5}]
                      [[{:time 1} {:time 2}]
                       [{:time 3} {:time 4}]])

         ; With a gap
         (test-stream (fixed-time-window 2) [{:time 1} {:time 7}]
                      [[{:time 1}] [] []])

         ; With out-of-order events
         (test-stream (fixed-time-window 2)
                      [{:time 5}
                       {:time 1}
                       {:time 2}
                       {:time 6}
                       {:time 3}
                       {:time 8}
                       {:time 4}
                       {:time 8}
                       {:time 5}
                       {:time 9}
                       {:time 11}]
                      [[{:time 5} {:time 6}]
                       [{:time 8} {:time 8}]
                       [{:time 9}]]))

(deftest fixed-offset-time-window-test
  Zero - time windows .
         (is (thrown? IllegalArgumentException (fixed-offset-time-window 0)))

         ; n-width windows
         (test-stream (fixed-offset-time-window 2) [] [])
         (test-stream (fixed-offset-time-window 2) [{:time 1}] [])
         (test-stream (fixed-offset-time-window 2)
                      [{:time 1} {:time 2} {:time 3} {:time 4} {:time 5} {:time 6}]
                      [[{:time 1}]
                       [{:time 2} {:time 3}]
                       [{:time 4} {:time 5}]])

         ; With a gap
         (test-stream (fixed-offset-time-window 2) [{:time 1} {:time 7}]
                      [[{:time 1}] [] []])

         ; With out-of-order events
         (test-stream (fixed-offset-time-window 2)
                      [{:time 5}
                       {:time 1}
                       {:time 2}
                       {:time 6}
                       {:time 3}
                       {:time 8}
                       {:time 4}
                       {:time 8}
                       {:time 5}
                       {:time 9}
                       {:time 11}]
                      [[{:time 5}]
                       [{:time 6}]
                       [{:time 8} {:time 8} {:time 9}]]))

(deftest part-time-simple-test
         ; Record windows of [start-time e1 e2 e3 end-time]
         (advance! 1)
         (let [effects (atom [])
               p (part-time-simple
                   5
                   (fn [_] [])
                   conj
                   (fn [window start end] (swap! effects conj
                                                 (concat [start] window [end]))))]
           (advance! 1) (p :t1)
           (advance! 2) (p :t2)
           (advance! 4) (p :t4)
           (advance! 5) (p :t5)
           (advance! 6) (p :t6)
           (advance! 8) (p :t8)
           (advance! 9) (p :t9)
           (advance! 99) (p :t99)
           (advance! 100) (p :t100)
           (advance! 101) (p :t101)
           (advance! 102) (p :t102)
           (advance! 200)
           (is (= @effects
                  [[1 :t1 :t2 :t4 :t5 6]
                  [6 :t6 :t8 :t9 11]
                  [96 :t99 :t100 101]
                  [101 :t101 :t102 106]]))))

(deftest apdex-test
         (test-stream-intervals
           (apdex 1 (state "ok") (state "warning"))
           (concat
             (interpose 0 (shuffle (concat (repeat 5 {:state "ok"})
                                           (repeat 7 {:state "warning"})
                                           (repeat 3 {:state "expired"})
                                           (repeat 1 {:state "fail"}))))
             [0 {:state "final"} 3])
           [{:state "final" :metric (/ (+ 5 (/ 7 2)) (+ 5 7 2))}]))

(deftest ^:focus clock-skew-test
         (test-stream-intervals (clock-skew)
                                [; foo bar baz
                                 {:time 1 :host :foo} 1
  2
                                 {:time 3 :host :bar} 1
  3 4
                                 {:time 9 :host :baz} 1
                                 ; 4 5 10
                                 {:time 1 :host :foo} 1
                                 ; 2 6 11
                                 {:time 100 :host :foo}]
                                [{:time 1 :host :foo :metric 0}
                                 {:time 3 :host :bar :metric 0}
                                 {:time 9 :host :baz :metric 5}
                                 {:time 1 :host :foo :metric -4}
                                 {:time 100 :host :foo :metric 89}]))

(deftest not-expired-test
  (testing "state expired"
    (test-stream (not-expired)
                 [{:state "expired"}
                  {:state "ok"}
                  {:state "warning"}
                  {:state "expired"}]
                 [{:state "ok"}
                  {:state "warning"}]))
  (testing "expired time"
    (let [out (atom [])
          child #(swap! out conj %)
          s (not-expired child)]
      (s {:time 0 :ttl 10})
      (advance! 10)
      (s {:time 1 :ttl 5})
      (s {:time 6 :ttl 2})
      (s {:time 10 :ttl 2})
      (is (= @out [{:time 0 :ttl 10}
                   {:time 10 :ttl 2}])))))

(deftest expired-test
  (testing "state expired"
    (test-stream (expired)
                 [{:state "expired"}
                  {:state "ok"}
                  {:state "warning"}
                  {:state "expired"}]
                 [{:state "expired"}
                  {:state "expired"}]))
  (testing "expired time"
    (let [out (atom [])
          child #(swap! out conj %)
          s (expired child)]
      (s {:time 0 :ttl 10})
      (advance! 10)
      (s {:time 1 :ttl 5})
      (s {:time 6 :ttl 2})
      (s {:time 10 :ttl 2})
      (is (= @out [{:time 1 :ttl 5}
                   {:time 6 :ttl 2}])))))

(deftest predict-linear-test
  (test-stream (predict-linear 10 600)
               [{:metric 1 :time 1}]
               [])
  (test-stream (predict-linear 5 60)
               [{:metric 1 :time 1}
                {:metric 2 :time 2}
                {:metric 3 :time 3}
                {:metric 4 :time 4}
                {:metric 5 :time 5}]
               [{:metric 62 :time 2}
                {:metric 63 :time 3}
                {:metric 64 :time 4}
                {:metric 65 :time 5}])
  (test-stream (predict-linear 2 86400)
               [{:metric 100 :time 0}
                {:metric 50 :time 43200}]
               [{:metric -50 :time 43200}])
  (test-stream (predict-linear 2 60)
               [{:metric 100 :time 0}
                {:metric 100 :time 1}]
               [{:metric 100 :time 1}])
  (test-stream (predict-linear 5 60 60)
               [{:metric 1 :time 1}
                {:metric 2 :time 2}
                {:metric 3 :time 3}
                {:metric 4 :time 4}
                {:metric 5 :time 5}]
               [{:metric 62 :time 2}
                {:metric 63 :time 3}
                {:metric 64 :time 4}
                {:metric 65 :time 5}])
  (test-stream (predict-linear 2 60 60)
               [{:metric 100 :time 0}
                {:metric 100 :time 1}
                {:metric 500 :time 2}]
               [{:metric 100 :time 1}
                {:metric 100 :time 2}]))

 Run streams against events.
 Check that our event was applied to the conj stream.
 Regular strings.
 Sets
 Functions
 Maps
 Nils
 Single tag
 Multiple tags
 dispatch with default value
 same test as above, using implicit rewrites
 same test as above, using splitp
 Complex closure with else clause
 Run stream
 Where should evaluate its children exactly once.
 Where*'s return value should be whether the predicate matched.
 useful given you can just use (even? metric).
 Verify that the macro allows variables to be used in
 predicates.
 Where should take an else clause.
 Where should evaluate its children exactly once.
 Where's return value should be whether the predicate
 matched.
 that each subsequent event matches that host.
 confirms that each subsequent event matches that host.
 Quick succession
 With a gap and expiry
 Quick succession
 With a gap and expiry
 Should forward a single state
 Should fill in missing states regularly
 Should forward final "expired" state.
 Should not fill during expired times.
 Empty -> empty
 Ignore stream without metrics
 Ignored
 Ignored
 Generate events
 Give all futures time to complete
 Verify output states
 All events recorded
 Middle events should have the correct rate.
 Generate events
 Wait for workers
 All events recorded
 confirms that each subsequent event matches that host, and that
 each event is different from the previous state.
 Generate enough events to allow the means to converge
 Verify that the means have converged
 Verify that metrics are weighted by :time, regardless of order of arrival
 A single event
 Repeating the same service
 Displacing a smaller event
 Allowing in a smaller event when there's room
 Ignoring smaller events
 Events without metrics
 Events without metrics displace previous events
 Expiring a nonexistent event
 Expiring an existing event
 Ring
 Demotion
 Should receive expired a2 once
 (due to the batching in coalesce)
 Constant values are emitted after dt seconds
 Ignores spikes
 spike elided
 Ignores flapping
 Triggers after dt seconds with new events.
 Empty -> empty
 Without anything to project to, does nothing
 Basic test: ignores non-matching events, updates state properly
 Passes on initially expired events correctly
 Expires existing events
 Expiration test: expires own events when the time comes.
 n-width windows
 n-width windows
 n-width runs
 n-width windows
 With out-of-order events
 n-width windows
 With a gap
 With out-of-order events
 n-width windows
 With a gap
 With out-of-order events
 Record windows of [start-time e1 e2 e3 end-time]
 foo bar baz
 4 5 10
 2 6 11
(ns riemann.streams-test
  (:require [riemann.common :refer :all :exclude [match]]
            [riemann.folds :as folds]
            [riemann.index :as index]
            [riemann.logging :as logging]
            [riemann.streams :refer :all]
            [riemann.test :refer [run-stream run-stream-intervals test-stream
                                  with-test-stream test-stream-intervals]]
            [riemann.time :refer :all]
            [riemann.time.controlled :refer :all]
            [clojure.test :refer :all])
  (:import (java.util.concurrent Executor
                                 CountDownLatch)))

(logging/init)

(use-fixtures :once control-time!)
(use-fixtures :each reset-time!)

(defn evs
  "Generate events based on the given event, with given metrics"
  [base-event & metrics]
  (vec (map #(assoc base-event :metric %)
            metrics)))

(defn em
  "Generate events with given metrics"
  [& metrics]
  (vec (map (fn [m] {:metric m}) metrics)))

(deftest smap*-test
         (testing "passes nil values"
                  (test-stream (smap* identity)
                               [1 nil false 3]
                               [1 nil false 3])))

(deftest smap-test
         (testing "increment"
                  (test-stream (smap inc) [6 3 -1] [7 4 0]))

         (testing "ignores nil values"
                  (test-stream (smap identity)
                               [1 nil false 3]
                               [1 false 3])))

(deftest smapcat-test
  (testing "doubles"
    (test-stream (smapcat #(vector % %)) [0 1 2 3]
                 [0 0 1 1 2 2 3 3])))

(deftest sflatten-test
  (test-stream (sflatten) [[0 1 2 3] [4 5 6 7]]
               [0 1 2 3 4 5 6 7]))

(deftest sdo-test
  (let [vals1   (atom [])
        vals2   (atom [])
        add1    #(swap! vals1 conj %)
        add2    #(swap! vals2 conj %)]
    (run-stream (sdo add1 add2) [1 2 3])
    (is (= @vals1 [1 2 3]))
    (is (= @vals2 [1 2 3]))
    (run-stream (sdo add1) [4 5 6])
    (is (= @vals1 [1 2 3 4 5 6]))
    (run-stream (sdo) [6 7 8])
    (is (= @vals1 [1 2 3 4 5 6]))
    (is (= @vals2 [1 2 3]))))

(deftest exception-stream-test
         (testing "direct fn"
                  (logging/suppress
                    ["riemann.streams" "riemann.streams-test"]

                    (let [out        (atom [])
                          exceptions (atom [])
                          stream (exception-stream
                                   #(swap! exceptions conj %)
                                   (fn [e]
                                     (swap! out conj e)
                                     (throw
                                       (RuntimeException. "foo"))))]
                      (stream :hi)

                      (is (= @out [:hi]))
                      (is (= 1 (count @exceptions)))
                      (let [e (first @exceptions)]
                        (is (= (:service e) "riemann exception"))
                        (is (= (:state e)   "error"))
                        (is (= (:time e)    0))
                        (is (re-find #".*RuntimeException.*" (:description e)))
                        (is (= #{"exception" "java.lang.RuntimeException"}
                               (set (:tags e))))))))

         (testing "scheduled fn"
                  (logging/suppress
                    "riemann.streams"
                    (reset-time!)
                    (let [e     (atom nil)
                          s (exception-stream #(reset! e %)
                                              (rate 1
                                                    (fn [e] (/ 1 0))))]
                      (s {:metric 2})
                      (is (nil? @e))
                      (advance! 1)
                      (is (= "error" (:state @e))))))

         (testing "exception handling takes care of Exception"
                  (logging/suppress
                   "riemann.streams"
                   (let [exceptions (atom [])]
                     (binding [riemann.streams/*exception-stream* #(swap! exceptions conj %)]
                       (run-stream (sdo #(throw (Exception. %)))
                                   ["boo!"]))
                     (is (= 1 (count @exceptions)))))))

(deftest execute-on-test
         (let [output (atom [])
               n      100
               events (range n)
               latch  (CountDownLatch. n)
               stream (execute-on
                        (reify Executor
                          (execute [this task]
                                   (future
                                     (task)
                                     (.countDown latch))))
                        (fn [event]
                          (swap! output conj event)))]

           (dorun (map stream events))

           (.await latch)
           (is (= (set @output) (set events)))))

(deftest sreduce-test
         (testing "explicit value"
                  (test-stream (sreduce + 1) [1 2 3] [2 4 7]))

         (testing "implicit value"
                  (test-stream (sreduce +) [1 2 3 4] [3 6 10])))

(deftest counter-test
         (testing "passes through events without metrics"
                  (test-stream (counter)
                               [{} {:state "expired"} {:service "foo"}]
                               [{} {:state "expired"} {:service "foo"}]))

         (testing "counts"
                  (test-stream (counter)
                               [{:metric 2} {} {:metric 3}]
                               [{:metric 2} {} {:metric 5}])

                  (test-stream (counter 100)
                               [{:metric 2} {} {:metric 3}]
                               [{:metric 102} {} {:metric 105}]))

         (testing "resets"
                  (test-stream (counter 100)
                               [{:metric 1}
                                {:metric 200 :tags ["reset"]}
                                {:metric 5}]
                               [{:metric 101}
                                {:metric 200 :tags ["reset"]}
                                {:metric 205}])))

(deftest match-test
         (test-stream (match :service "foo")
                      [{}
                       {:service "bar"}
                       {:service "foo"}]
                      [{:service "foo"}])

         (test-stream (match :metric #{0 2})
                      [{}
                       {:metric 1}
                       {:metric 2}]
                      [{:metric 2}])

  Regexen
         (test-stream (match :state #"^mi")
                      [{}
                       {:state "migas"}
                       {:state "other breakfast foods"}]
                      [{:state "migas"}])

         (test-stream (match identity 2)
                      [1 2 3]
                      [2])

         (test-stream (match identity {:state #"^mi", :host "other"})
                      [{}
                       {:state "migas"}
                       {:state "migas", :host "other"}]
                      [{:state "migas", :host "other"}])

  (test-stream (match :host nil)
               [{}
                {:host nil}
                {:host :foo}]
               [{}
                {:host nil}]))

(deftest untag-test
         (testing "single tag"
                  (test-stream (untag "foo")
                               [{}
                                {:service :a :tags []}
                                {:service :a :tags ["foo"]}
                                {:service :b :tags ["foo" "bar" "baz"]}]
                               [{:tags []}
                                {:service :a :tags []}
                                {:service :a :tags []}
                                {:service :b :tags ["bar" "baz"]}]))

         (testing "multiple tags"
                  (test-stream (untag ["foo" "bar"])
                               [{}
                                {:service :a :tags []}
                                {:service :a :tags ["foo"]}
                                {:service :b :tags ["foo" "bar" "baz"]}]
                               [{:tags []}
                                {:service :a :tags []}
                                {:service :a :tags []}
                                {:service :b :tags ["baz"]}])))

(deftest tag-test
         (test-stream (tag "foo")
                      [{}
                       {:service :a :tags ["foo"]}
                       {:service :b :tags ["bar" "baz"]}]
                      [{:tags ["foo"]}
                       {:service :a :tags ["foo"]}
                       {:service :b :tags ["foo" "bar" "baz"]}])

         (test-stream (tag ["foo" "bar"])
                      [{}
                       {:service :a :tags ["foo"]}
                       {:service :b :tags ["foo" "baz"]}]
                      [{:tags ["foo" "bar"]}
                       {:service :a :tags ["foo" "bar"]}
                       {:service :b :tags ["foo" "bar" "baz"]}]))

(deftest tagged-all-test
         (test-stream (tagged-all ["kitten" "cat"])
                      [{:tags ["kitten" "cat"]}
                       {:tags ["kitten", "cat", "meow"]}
                       {:tags ["dog" "cat"]}
                       {:tags ["cat"]}
                       {:tags []}
                       {}]
                      [{:tags ["kitten" "cat"]}
                       {:tags ["kitten", "cat", "meow"]}])

         (test-stream (tagged-all "meow")
                      [{:tags ["meow" "bark"]}
                       {:tags ["meow"]}
                       {:tags ["bark"]}
                       {}]
                      [{:tags ["meow" "bark"]}
                       {:tags ["meow"]}])

         (testing "return values"
                  (is (true? ((tagged-all ["meow" "bark"])
                                {:tags ["meow" "bark" "grr"]})))
                  (is (nil? ((tagged-all ["meow" "bar"])
                               {:tags ["meow"]})))))

(deftest tagged-any-test
         (test-stream (tagged-any ["kitten" "cat"])
                      [{:tags ["kitten" "cat"]}
                       {:tags ["cat", "dog"]}
                       {:tags ["kitten"]}
                       {:tags ["dog"]}
                       {:tags []}
                       {}]
                      [{:tags ["kitten" "cat"]}
                       {:tags ["cat", "dog"]}
                       {:tags ["kitten"]}])

         (test-stream (tagged-any "meow")
                      [{:tags ["meow" "bark"]}
                       {:tags ["meow"]}
                       {:tags ["bark"]}
                       {}]
                      [{:tags ["meow" "bark"]}
                       {:tags ["meow"]}])

         (testing "return values"
                  (is (true? ((tagged-any ["meow" "bark"])
                                {:tags ["meow" "moo"]})))
                  (is (nil? ((tagged-any ["meow" "bar"])
                               {:tags ["moo"]})))))

(deftest split*-test
         (testing "with one branch"
                  (test-stream (split* identity)
                               [true false nil 2]
                               [true 2]))

         (testing "with a default stream"
                  (let [sup    (fn [threshold] #(< threshold (:metric %)))
                        res    (atom [])
                        events [{:metric 15} {:metric 8} {:metric 2}]
                        expect [{:metric 15 :state :critical}
                                {:metric 8 :state :warn}
                                {:metric 2 :state :ok}]
                        stream (split* (sup 10) (with :state :critical
                                                      (partial swap! res conj))
                                       (sup 5) (with :state :warn
                                                     (partial swap! res conj))
                                       (with :state :ok
                                             (partial swap! res conj)))]
                    (dorun (map stream events))
                    (is (= expect @res))))

         (testing "with no default stream"
                  (let [sup    (fn [threshold] #(< threshold (:metric %)))
                        res    (atom [])
                        events [{:metric 15} {:metric 8} {:metric 2}]
                        expect [{:metric 15 :state :critical}
                                {:metric 8 :state :warn}]
                        stream (split*
                                 (sup 10) (with :state :critical
                                                (partial swap! res conj))
                                 (sup 5) (with :state :warn
                                               (partial swap! res conj)))]
                    (dorun (map stream events))
                    (is (= expect @res)))))

(deftest split-test
         (testing "with a default stream"
                  (let [sup    (fn [threshold] #(< threshold (:metric %)))
                        res    (atom [])
                        events [{:metric 15} {:metric 8} {:metric 2}]
                        expect [{:metric 15 :state :critical}
                                {:metric 8 :state :warn}
                                {:metric 2 :state :ok}]
                        stream (split
                                 (> metric 10) (with :state :critical
                                                     (partial swap! res conj))
                                 (> metric 5) (with :state :warn
                                                    (partial swap! res conj))
                                 (with :state :ok
                                       (partial swap! res conj)))]
                    (dorun (map stream events))
                    (is (= expect @res))))

         (testing "evaluates streams once"
                  (let [res (atom [])
                        stream (split
                                 true
                                 (let [state (atom 0)]
                                   (fn [_]
                                     (swap! res conj
                                            (swap! state inc)))))]
                    (stream :x)
                    (stream :x)
                    (stream :x)
                    (is (= @res [1 2 3])))))

(deftest splitp-test
         (testing "basics"
                  (let [res    (atom [])
                        events [{:metric 15} {:metric 8} {:metric 2}]
                        expect [{:metric 15 :state :crit}
                                {:metric 8 :state :warn}
                                {:metric 2 :state :ok}]
                        stream (splitp <= metric
                                       10 (with :state :crit
                                                (partial swap! res conj))
                                       5  (with :state :warn
                                                (partial swap! res conj))
                                       (with :state :ok
                                             (partial swap! res conj)))]
                    (dorun (map stream events))
                    (is (= expect @res))))

         (testing "Without a default"
                  (is (thrown? IllegalArgumentException
                               ((splitp = true
                                        false :doesnt-happen)
                                  :foo))))

         (testing "Evaluates child streams once at creation time"
                  (let [a (atom 0)
                        b (atom 0)
                        s (splitp = state
                                  :foo (do (swap! a inc) identity)
                                       (do (swap! b inc) identity))]
                    (is (= 1 @a))
                    (is (= 1 @b))
                    (s {:state :foo})
                    (s {:state :foo})
                    (s {:state :bar})
                    (s {:state :baz})
                    (is (= 1 @a))
                    (is (= 1 @b)))))

(deftest where*-test
         (test-stream (where* identity)
                      [true false nil 2]
                      [true 2])

         (test-stream (where* expired?)
                      [{:time -1 :ttl 0.5}
                       {:time 0 :ttl 1}]
                      [{:time -1 :ttl 0.5}])

         (let [good (atom [])
               bad  (atom [])
               s (where* (fn [event]
                           (or (= "good" (:service event))
                               (< 2 (:metric event))))
                         (partial swap! good conj)
                         (else (partial swap! bad conj)))
               events [{:service "good" :metric 0}
                       {:service "bad" :metric 0}
                       {:metric 1}
                       {:service "bad" :metric 1}
                       {:service "bad" :metric 3}]]

           (doseq [e events] (s e))

           (is (= @good
                  [{:service "good" :metric 0}
                   {:service "bad" :metric 3}]))
           (is (= @bad
                  [{:service "bad" :metric 0}
                   {:metric 1}
                   {:service "bad" :metric 1}])))

         (testing "evaluates children once"
                  (let [x (atom 0)
                        s (where* (fn [e] true) (do (swap! x inc) identity))]
                    (is (= @x 1))
                    (s {:service "test"})
                    (is (= @x 1))
                    (s {:service "test"})
                    (is (= @x 1)))))

(deftest where*-return-value
         (is (= true ((where* expired? (fn [e] "hi"))
                        (expire {}))))
         (is (= 2 ((where* (constantly 2)) :zoom))))

(deftest where-test
         (testing "field"
                  (let [r (ref [])
                        s (where (or (state "ok" "good")
                                     (= "weird" state))
                                 (fn [e] (dosync (alter r conj e))))
                        events [{:state "ok"}
                                {:state "good"}
                                {:state "weird"}
                                {:state "error"}]
                        expect [{:state "ok"}
                                {:state "good"}
                                {:state "weird"}]]
                    (doseq [e events] (s e))
                    (is (= expect (deref r)))))

         (testing "event binding"
           (test-stream (where (= event 2))
                        [1 2 3 2]
                        [2 2]))

         (testing "regex"
                  (test-stream (where (service #"^foo"))
                               [{}
                                {:service "foo"}
                                {:service "food"}]
                               [{:service "foo"}
                                {:service "food"}]))

         (testing "functions"
                  (test-stream (where (and metric (even? metric)))
                               [{}
                                {:metric 1}
                                {:metric 2}]
                               [{:metric 2}]))

         (testing "functions as values"
  May not support this later . I do n't think it 's all that
                  (test-stream (where (and metric (metric even?)))
                               [{}
                                {:metric 1}
                                {:metric 2}]
                               [{:metric 2}]))

         (testing "using sets as predicates"
                  (test-stream (where (service #{"service1" "service2"}))
                               [{:service "service1"}
                                {:service "service2"}
                                {:service "service-doesnt-match"}]
                               [{:service "service1"}
                                {:service "service2"}]))


         (testing "variable"
                  (let [regex #"cat"]
                    (test-stream (where (service regex))
                                 [{:service "kitten"}
                                  {:service "cats"}]
                                 [{:service "cats"}])))


         (testing "tagged (one tag)"
                  (let [r (atom [])
                        s (where (tagged "foo") (append r))
                        events [{}
                                {:tags []}
                                {:tags ["blah"]}
                                {:tags ["foo"]}
                                {:tags ["foo" "bar"]}]]
                    (doseq [e events] (s e))
                    (is (= (deref r)
                           [{:tags ["foo"]} {:tags ["foo" "bar"]}]))))

         (testing "tagged-all (one tag)"
                  (let [r (atom [])
                        s (where (tagged-all "foo") (append r))
                        events [{}
                                {:tags []}
                                {:tags ["blah"]}
                                {:tags ["foo"]}
                                {:tags ["foo" "bar"]}]]
                    (doseq [e events] (s e))
                    (is (= (deref r)
                           [{:tags ["foo"]} {:tags ["foo" "bar"]}]))))

         (testing "tagged-any (one tag)"
                  (let [r (atom [])
                        s (where (tagged-any "foo") (append r))
                        events [{}
                                {:tags []}
                                {:tags ["blah"]}
                                {:tags ["foo"]}
                                {:tags ["foo" "bar"]}]]
                    (doseq [e events] (s e))
                    (is (= (deref r)
                           [{:tags ["foo"]} {:tags ["foo" "bar"]}]))))

         (testing "tagged (multiple tags)"
                  (let [r (atom [])
                        s (where (tagged ["foo" "bar"]) (append r))
                        events [{}
                                {:tags []}
                                {:tags ["blah"]}
                                {:tags ["foo"]}
                                {:tags ["foo" "bar"]}
                                {:tags ["foo" "bar" "baz"]}]]
                    (doseq [e events] (s e))
                    (is (= (deref r)
                           [{:tags ["foo" "bar"]} {:tags ["foo" "bar" "baz"]}]))))

         (testing "tagged-all (multiple tags)"
                  (let [r (atom [])
                        s (where (tagged-all ["foo" "bar"]) (append r))
                        events [{}
                                {:tags []}
                                {:tags ["blah"]}
                                {:tags ["foo"]}
                                {:tags ["foo" "bar"]}
                                {:tags ["foo" "bar" "baz"]}]]
                    (doseq [e events] (s e))
                    (is (= (deref r)
                           [{:tags ["foo" "bar"]} {:tags ["foo" "bar" "baz"]}]))))

         (testing "tagged-any (multiple tags)"
                  (let [r (atom [])
                        s (where (tagged-any ["foo" "bar"]) (append r))
                        events [{}
                                {:tags []}
                                {:tags ["blah"]}
                                {:tags ["foo"]}
                                {:tags ["foo" "bar"]}
                                {:tags ["baz" "bar"]}]]
                    (doseq [e events] (s e))
                    (is (= (deref r)
                           [{:tags ["foo"]} {:tags ["foo" "bar"]} {:tags ["baz" "bar"]}]))))

         (testing "else"
                  (let [a (atom [])
                        b (atom [])]
                    (run-stream
                      (where (service #"a")
                             #(swap! a conj (:service %))
                             (else #(swap! b conj (:service %))))
                      [{:service "cat"}
                       {:service "dog"}
                       {:service nil}
                       {:service "badger"}])
                    (is (= @a ["cat" "badger"]))
                    (is (= @b ["dog" nil]))))

         (testing "evaluates children once"
                  (let [x (atom 0)
                        s (where true (do (swap! x inc) identity))]
                    (is (= @x 1))
                    (s {:service "test"})
                    (is (= @x 1))
                    (s {:service "test"})
                    (is (= @x 1))))

         (testing "return value"
                  (is (= true  ((where (service "foo")) {:service "foo"})))
                  (is (= false ((where (service "foo")) {:service "bar"})))

                  (is (= true  ((where (tagged "foo")) {:tags ["foo"]})))
                  (is (= false ((where (tagged "foo")) {:tags ["bar"]})))
                  (is (= true  ((where (tagged-all "foo")) {:tags ["foo"]})))
                  (is (= false ((where (tagged-all "foo")) {:tags ["bar"]})))
                  (is (= true  ((where (tagged-any "foo")) {:tags ["foo"]})))
                  (is (= false ((where (tagged-any "foo")) {:tags ["bar"]})))

                  (is (= true  ((where (tagged ["foo" "bar"])) {:tags ["foo" "bar"]})))
                  (is (= false ((where (tagged ["foo" "bar"])) {:tags ["bar"]})))
                  (is (= true  ((where (tagged-all ["foo" "bar"])) {:tags ["foo" "bar"]})))
                  (is (= false ((where (tagged-all ["foo" "bar"])) {:tags ["bar"]})))
                  (is (= true ((where (tagged-any ["foo" "bar"])) {:tags ["foo" "bar"]})))
                  (is (= false ((where (tagged-any ["foo" "bar"])) {:tags ["blah"]})))

                  (is (= true ((where (service "foo")
                                      (fn [event] 2))
                                 {:service "foo"})))
                  (is (= false ((where (service "foo")
                                       (else (fn [event] 2)))
                                  {:service "bar"})))

                  (is (= 2 ((where 2) :wheeee!)))
                  (is (= nil ((where nil) :zoooom!)))))

(deftest default-kv
         (let [r (atom nil)
               s (default :service "foo" (register r))]
           (s {:service nil})
           (is (= {:service "foo"} (deref r)))

           (s {:service "foo"})
           (is (= {:service "foo"} (deref r)))

           (s {:service "bar" :test "baz"})
           (is (= {:service "bar" :test "baz"} (deref r)))))

(deftest default-map
         (let [r (atom nil)
               s (default {:service "foo" :state nil} (register r))]
           (s (event {:service nil}))
           (is (= "foo" (:service (deref r))))
           (is (= nil (:state (deref r))))

           (s (event {:service "foo"}))
           (is (= "foo" (:service (deref r))))
           (is (= nil (:state (deref r))))

           (s (event {:service "bar" :host "baz" :state "evil"}))
           (is (= "bar" (:service (deref r))))
           (is (= "baz" (:host (deref r))))
           (is (= "evil" (:state (deref r))))))

(deftest with-kv
         (let [r (ref nil)
               s (with :service "foo" (fn [e] (dosync (ref-set r e))))]
           (s {:service nil})
           (is (= {:service "foo"} (deref r)))

           (s {:service "foo"})
           (is (= {:service "foo"} (deref r)))

           (s {:service "bar" :test "baz"})
           (is (= {:service "foo" :test "baz"} (deref r)))

           (s [{:service "bar" :test "baz"}])
           (is (= [{:service "foo" :test "baz"}] (deref r)))))

(deftest with-map
         (let [r (ref nil)
               s (with {:service "foo" :state nil} (fn [e] (dosync (ref-set r e))))
               empty-s (with {} (fn [e] (dosync (ref-set r e))))]
           (s (event {:service nil}))
           (is (= "foo" (:service (deref r))))
           (is (= nil (:state (deref r))))

           (s (event {:service "foo"}))
           (is (= "foo" (:service (deref r))))
           (is (= nil (:state (deref r))))

           (s (event {:service "bar" :test "baz" :state "evil"}))
           (is (= "foo" (:service (deref r))))
           (is (= nil (:state (deref r))))

           (empty-s (event {:service "bar" :test "baz"}))
           (is (= "bar" (:service (deref r))))))

(deftest by-single
  Each test stream keeps track of the first host it sees , and confirms
         (let [i (ref 0)
               s (by :host
                     (let [host (ref nil)]
                       (fn [event]
                         (dosync
                           (alter i inc)
                           (when (nil? (deref host))
                             (ref-set host (event :host)))
                           (is (= (deref host) (event :host)))))))
               events (map (fn [h] {:host h}) [:a :a :b :a :c :b])]
           (doseq [event events]
             (s event))
           (is (= (count events) (deref i)))))

(deftest by-multiple
  Each test stream keeps track of the first host / service it sees , and
         (let [i (ref 0)
               s (by [:host :service]
                     (let [host (ref nil)
                           service (ref nil)]
                       (fn [event]
                         (dosync
                           (alter i inc)

                           (when (nil? (deref host))
                             (ref-set host (event :host)))
                           (when (nil? (deref service))
                             (ref-set service (event :service)))

                           (is (= (deref host) (event :host)))
                           (is (= (deref service) (event :service)))))))

               events (map (fn [h] {:host (first h)
                                    :service (last h)})
                           [[1 :a]
                            [1 :b]
                            [1 :a]
                            [2 :a]
                            [2 :a]
                            [1 :a]
                            [2 :a]
                            [1 :b]])]
           (doseq [event events]
             (s event))
           (is (= (count events) (deref i)))))

(deftest by-builder-test
  (testing "one child"
    (let [i (ref 0)
          s (by-builder [host :host]
                        (let [host-ref (ref nil)]
                          (fn [event]
                            (dosync
                             (alter i inc)
                             (when (nil? (deref host-ref))
                               (ref-set host-ref (event :host)))
                             (is (= (deref host-ref) (event :host) host))))))
          events (map (fn [h] {:host h}) [:a :a :b :a :c :b])]
      (doseq [event events]
        (s event))
      (is (= (count events) (deref i)))))

  (testing "two children"
    (let [i (ref 0)
          s (by-builder [host :host]
                        (let [host-ref (ref nil)]
                          (fn [event]
                            (dosync
                             (alter i inc)
                             (when (nil? (deref host-ref))
                               (ref-set host-ref (event :host)))
                             (is (= (deref host-ref) (event :host) host)))))
                        (let [host-ref (ref nil)]
                          (fn [event]
                            (dosync
                             (alter i inc)
                             (when (nil? (deref host-ref))
                               (ref-set host-ref (event :host)))
                             (is (= (deref host-ref) (event :host) host))))))
          events (map (fn [h] {:host h}) [:a :a :b :a :c :b])]
      (doseq [event events]
        (s event))
      (is (= (* 2 (count events)) (deref i)))))

  (testing "multiple fields"
    (let [i (ref 0)
          s (by-builder [[host service] [:host :service]]
                (let [host-ref (ref nil)
                      service-ref (ref nil)]
                  (fn [event]
                    (dosync
                     (alter i inc)
                     (when (nil? (deref host-ref))
                       (ref-set host-ref (event :host)))
                     (when (nil? (deref service-ref))
                       (ref-set service-ref (event :service)))

                     (is (= (deref host-ref) (event :host) host))
                     (is (= (deref service-ref) (event :service)) service)))))

          events (map (fn [h] {:host (first h)
                               :service (last h)})
                      [[1 :a]
                       [1 :b]
                       [1 :a]
                       [2 :a]
                       [2 :a]
                       [1 :a]
                       [2 :a]
                       [1 :b]])]
      (doseq [event events]
        (s event))
      (is (= (count events) (deref i))))))

(deftest by-evaluates-children-once-per-branch
         (let [i (atom 0)
               s (by :metric (do (swap! i inc) identity))]
           (is (= @i 0))
           (s {:metric 1})
           (is (= @i 1))
           (s {:metric 2})
           (is (= @i 2))
           (s {:metric 1})
           (is (= @i 2))))

(deftest pipe-test
  (testing "One stage"
    (test-stream
      (pipe !)
      [1 2 3]
      [1 2 3]))

  (testing "Chained stages"
    (test-stream
      (pipe ! (smap inc !) (smap (partial * 2) !))
      [1 2 3]
      [4 6 8]))

  (testing "Multiple markers"
    (test-stream
      (pipe - (splitp < metric
                      0.9 (with :state :critical -)
                      0.5 (with :state :warning  -)
                          (with :state :ok       -))
            (smap identity -))
      [{:metric 0.1}
       {:metric 0.7}
       {:metric 1.0}]
      [{:metric 0.1 :state :ok}
       {:metric 0.7 :state :warning}
       {:metric 1.0 :state :critical}]))

  (testing "Rebinding a marker"
    (test-stream
      (pipe - (let [! -
                    - clojure.core/-]
                (smap - !)))
      [-1 0 1]
      [1 0 -1])))

(deftest fill-in-test
         (test-stream-intervals
           (fill-in 0.01 {:metric 0})
           []
           [])

         (is (= (map :metric (run-stream-intervals
                               (fill-in 0.01 {:metric 0})
                               (interpose nil (em 1 2 3))))
                [1 2 3]))

         (is (= (map :metric (run-stream-intervals
                               (fill-in 0.05 {:metric 0})
                               [{:metric 1} 0.06
                                {:metric 2} nil
                                {:metric 3} 0.08
                                {:metric 4 :state "expired"} 0.06
                                {:metric 5}]))
                [1 0 2 3 0 4 5]))
         )

(deftest fill-in-last-test
  (test-stream-intervals
    (fill-in-last 0.01 {:metric 0})
    []
    [])

  (let [output (run-stream-intervals
                 (fill-in-last 0.01 {:metric 0})
                 (interpose nil (evs {:host "foo" :service "bar"}
                                     1 2 3)))]
    (is (= (map :metric output)
           [1 2 3]))
    (is (= (map :host output)
           ["foo" "foo" "foo"]))
    (is (= (map :service output)
           ["bar" "bar" "bar"])))

  (let [output (run-stream-intervals
                 (fill-in-last 0.05 {:metric 0})
                 [{:host "a" :metric 1} 0.06
                  {:host "b" :metric 2} nil
                  {:host "c" :metric 3} 0.08
                  {:host "d" :metric 4 :state "expired"} 0.06
                  {:host "e" :metric 5}])]
    (is (= (map :metric output)
           [1 0 2 3 0 4 5]))
    (is (= (map :host output)
           ["a" "a" "b" "c" "c" "d" "e"]))))

(deftest interpolate-constant-test
         (test-stream-intervals
           (interpolate-constant 0.01)
           []
           [])

         (is (= (map :metric (run-stream-intervals
                               (interpolate-constant 0.1)
                               [{:metric 1} 0.05]))
                [1]))

  Should forward first state and ignore immediate successors
         (is (= (map :metric (run-stream-intervals
                               (interpolate-constant 0.1)
                               [{:metric 1} 0.05
                                {:metric 2} nil
                                {:metric 3}]))
                [1]))

         (is (= (map :metric (run-stream-intervals
                               (interpolate-constant 0.1)
                               (interpose 0.22 (em 1 2 3 4))))
                [1 1 1 2 2 3 3]))

         (is (= (map :metric (run-stream-intervals
                               (interpolate-constant 0.1)
                               (interpose 0.22
                                          [{:metric 1}
                                           {:metric 2}
                                           {:metric 3}
                                           {:metric 4 :state "expired"}])))
                [1 1 1 2 2 3 3 4]))

         (is (= (map :metric (run-stream-intervals
                               (interpolate-constant 0.05)
                               [{:metric 1 :state "expired"}
                                0.12
                                {:metric 2}
                                0.12
                                {:metric 3 :state "expired"}
                                0.12]))
                [1 2 2 2 3]))
         )

(deftest ddt-immediate-test
         (test-stream (ddt) [] [])
         (test-stream (ddt) [{} {} {} {}] [])
  Do nothing the first time
         (test-stream (ddt) [{:metric 1 :time 0}] [])
  Differentiate
         (test-stream (ddt)
                      [{:metric 0 :time 0}
                       {:metric 0 :time 1}
                       {:metric 2 :time 2}
                       {:metric -4 :time 4}]
                      [{:metric 0 :time 1}
                       {:metric 2 :time 2}
                       {:metric -3 :time 4}]))

(deftest ddt-interval-test
         (testing "Quick burst without crossing interval"
                  (is (= (map :metric (run-stream-intervals
                                        (ddt 0.1)
                                        [{:metric 1} nil
                                         {:metric 2} nil
                                         {:metric 3}]))
                         [])))

         (testing "1 event per interval"
                  (test-stream-intervals
                    (ddt 1)
  Note that the swap occurs just prior to events at time 1 .
                    [{:time 0 :metric -1} 99/100
                     {:time 1 :metric 0} 1
                     {:time 2 :metric -5} 1]
                    [{:time 1 :metric 1}
                     {:time 2 :metric -5}]))

         (testing "n events per interval"
                  (test-stream-intervals
                    (ddt 1)
                    [{:time 0 :metric -1} 1/100
                     {:time 2/2 :metric 1} 1/2
                     {:time 4/2 :metric -3} 1/2]
                    [{:time 2/2 :metric 2}
                     {:time 4/2 :metric -4}]))

         (testing "emits zeroes when no events arrive in an interval"
                  (test-stream-intervals (ddt 2)
                                         [{:time 0 :metric 0} 1
                                          {:time 1 :metric 1} 2
                                          {:time 3 :metric 2} 3]
                                         [{:time 2 :metric 1}
                                          {:time 4 :metric 1/2}
                                          {:time 6 :metric 0}])))

(deftest rate-slow-even
         (let [output (ref [])
               interval 1
               intervals 5
               gen-rate 10
               total (* gen-rate intervals)
               gen-period (/ interval gen-rate)
               r (rate interval
                        (fn [event] (dosync (alter output conj event))))]

           (dotimes [_ intervals]
             (dotimes [_ gen-rate]
               (advance! (+ (unix-time) gen-period))
               (r {:metric 1 :time (unix-time)})))

           (advance! (+ (unix-time) gen-period))

           (let [o (deref output)]

             (is (approx-equal total (reduce + (map :metric o))))

             (is (every? (fn [measured-rate]
                           (approx-equal gen-rate measured-rate))
                         (map :metric (drop 1 (drop-last o)))))

  First and last events should be complementary
             (let [first-last (+ (:metric (first o))
                                 (:metric (last o)))]
               (is (or (approx-equal (* 2 gen-rate) first-last)
                       (approx-equal gen-rate first-last))))

             )))

(deftest rate-threaded
         (let [output (atom nil)
               interval 5/2
               total 10000
               threads 4
               r (rate interval
                       (fn [event] (dosync (reset! output event))))

               workers (map (fn [t] (future
                                      (dotimes [i (/ total threads)]
                                        (r {:metric 1 :time (unix-time)}))))
                            (range threads))]

           (dorun (map deref workers))
           (advance! interval)

           (is (= (/ total interval) (:metric @output)))))

(deftest rate-without-input
         (test-stream-intervals
           (rate 1)
           [{:metric 1 :service "foo"} 0.5
            {:metric 1 :service "bar"} 0.5
            {:metric 1 :service "baz" :ttl 3} 3
            {:state "expired"}]
           [{:time 1 :metric 2 :service "bar"}
            {:time 2 :metric 1 :service "baz" :ttl 3}
            {:time 3 :metric 0 :service "baz" :ttl 2}
            {:time 4 :metric 0 :service "baz" :ttl 1}]))

(deftest rate-expiration
         (test-stream-intervals
           (rate 1)
           [{:time 0 :metric 1 :ttl 3} 5]
           [{:time 1 :metric 1 :ttl 3}
            {:time 2 :metric 0 :ttl 2}
            {:time 3 :metric 0 :ttl 1}]))

(deftest fold-interval-test
         (test-stream-intervals
           (riemann.streams/fold-interval 1 :metric (partial apply +))
           [{:metric 2} 0.1
            {:metric 4} 0.2
            {:metric 2} 0.3
            {:metric 4} 1.0
            {:metric 100} 0.1
            {:metric 100} 1.0]
           (em 12 200)))

(deftest fold-interval-metric-test
         (test-stream-intervals
           (riemann.streams/fold-interval-metric 1 (partial apply +))
           [{:metric 2} 0.1
            {:metric 4} 0.2
            {:metric 2} 0.3
            {:metric 4} 1.0
            {:metric 100} 0.1
            {:metric 100} 1.0]
           (em 12 200)))

(deftest changed-test
  (let [states (partial map (partial hash-map :state))]
    (testing ":state"
      (test-stream (changed :state)
                   (states [:ok :bad :bad :ok :ok :ok :evil :bad])
                   (states [:ok :bad :ok :evil :bad])))

    (testing ":state with :init"
      (test-stream (changed :state {:init :ok})
                   (states [:ok :bad :bad :ok :ok :evil :bad])
                   (states [:bad :ok :evil :bad])))

    (testing "with previous event and arbitrary fn"
      (test-stream (changed #(:state %) {:init :ok, :pairs? true})
                   (states [:ok :bad :bad :ok :ok :evil :bad])
                   (partition 2 1 (states [:ok :bad :ok :evil :bad]))))))

(deftest changed-with-exception-test
        (logging/suppress
          "riemann.streams"
           (let [exceptions (atom [])]
             (binding [riemann.streams/*exception-stream* #(swap! exceptions conj %)]
               (run-stream (changed :state (fn [x]
                                             (throw (Exception. "boo!"))))
                           [(riemann.common/event {:state :critical})]))
             (is (= 1 (count @exceptions))))))

(deftest changed-state-test
  Each test stream keeps track of the first host / service it sees , and
         (let [i (ref 0)
               s (changed-state
                   (let [host (ref nil)
                         service (ref nil)
                         state (ref nil)]
                     (fn [event]
                       (dosync
                         (alter i inc)

                         (is (not= (deref state) (:state event)))
                         (ref-set state (:state event))

                         (when (nil? (deref host))
                           (ref-set host (event :host)))
                         (when (nil? (deref service))
                           (ref-set service (event :service)))

                         (is (= (deref host) (event :host)))
                         (is (= (deref service) (event :service)))))))

               events [{:host 1 :service 1 :state 1}
                       {:host 2 :service 1 :state 1}
                       {:host 1 :service 1 :state 1}
                       {:host 1 :service 1 :state 2}
                       {:host 2 :service 1 :state 2}
                       {:host 2 :service 2 :state 1}
                       {:host 2 :service 1 :state 1}]]

           (doseq [event events]
             (s event))
           (is (= 6 (deref i)))))

(deftest over-test
(logging/suppress ["riemann.streams"]
         (let [output (ref [])
               r (over 1.5
                         (fn [e] (dosync (alter output conj e))))
               metrics [0.5 1 1.5 2 2.5]
               expect [2 2.5]]

           (doseq [m metrics] (r {:metric m}))
           (is (= expect (vec (map (fn [s] (:metric s)) (deref output))))))))

(deftest under-test
(logging/suppress ["riemann.streams"]
         (let [output (ref [])
               r (under 1.5
                         (fn [e] (dosync (alter output conj e))))
               metrics [0.5 1 1.5 2 2.5]
               expect [0.5 1]]

           (doseq [m metrics] (r {:metric m}))
           (is (= expect (vec (map (fn [s] (:metric s)) (deref output))))))))

(deftest ewma-timeless-test
         (test-stream (ewma-timeless 0)
                      (em 1 10 20 -100 4)
                      (em 0 0  0   0   0))
         (test-stream (ewma-timeless 1)
                      (em 5 13 1 -10 3)
                      (em 5 13 1 -10 3))
         (test-stream (ewma-timeless 1/2)
                      (em 1   1   1   1     1    )
                      (em 1/2 3/4 7/8 15/16 31/32)))

(deftest ewma-test
  (let [output1 (atom [])
        output2 (atom [])
        output5 (atom [])
        seconds 50
        s1 (ewma 1
          (fn [event] (swap! output1 conj event)))
        s2 (ewma 2
          (fn [event] (swap! output2 conj event)))
        s5 (ewma 5
          (fn [event] (swap! output5 conj event)))]

    (dotimes [n seconds]
      (s1 {:metric 1 :time (inc n)})
      (s2 {:metric 1 :time (inc n)})
      (s5 {:metric 1 :time (inc n)}))

    (is (every? (fn [measured-rate]
                  (approx-equal 1 measured-rate))
                  (map :metric (drop 10 @output1))))
    (is (every? (fn [measured-rate]
                  (approx-equal 1 measured-rate))
                  (map :metric (drop 15 @output2))))
    (is (every? (fn [measured-rate]
                  (approx-equal 1 measured-rate))
                  (map :metric (drop 40 @output5))))

  Verify halflives with spaced out : metric 0 events
    (s1 {:metric 0 :time (+ 1 seconds)})
    (s1 {:metric 0 :time (+ 2 seconds)})
    (is (approx-equal 1/2 (:metric (last (drop-last @output1)))))
    (is (approx-equal 1/4 (:metric (last @output1))))

    (s2 {:metric 0 :time (+ 2 seconds)})
    (s2 {:metric 0 :time (+ 4 seconds)})
    (is (approx-equal 1/2 (:metric (last (drop-last @output2)))))
    (is (approx-equal 1/4 (:metric (last @output2))))

    (s5 {:metric 0 :time (+ 5 seconds)})
    (s5 {:metric 0 :time (+ 10 seconds)})
    (is (approx-equal 1/2 (:metric (last (drop-last @output5)))))
    (is (approx-equal 1/4 (:metric (last @output5)))))

  (test-stream-intervals (ewma 1)
    [{:metric 1 :time 0} 1
     {:metric 1 :time 2} 1
     {:metric 1 :time 1} 1]
    [{:metric 0.5 :time 0}
     {:metric 0.625 :time 2}
     {:metric 0.875 :time 2}]))

(deftest top-test
  (let [e (fn [s m & tags] {:service s :metric m :tags tags})
        a (fn [m & tags] (apply e :a m tags))
        b (fn [m & tags] (apply e :b m tags))
        c (fn [m & tags] (apply e :c m tags))
        d (fn [m & tags] (apply e :d m tags))]

           (test-stream (top 1 :metric)
                        [(a 1)]
                        [(a 1)])

           (test-stream (top 1 :metric)
                        [(a 1) (a 2) (a 1) (a 3)]
                        [(a 1) (a 2) (a 1) (a 3)])

           (test-stream (top 2 :metric)
                        [(a 1) (b 2) (c 3) (a 1)          (b 2)]
                        [(a 1) (b 2) (c 3) (expire (a 1)) (b 2)])

           (test-stream (top 2 :metric)
                        [(a 2) (b 2) (c 1)          (a 5) (c 1)          (a 0) (c 1)]
                        [(a 2) (b 2) (expire (c 1)) (a 5) (expire (c 1)) (a 0) (c 1)])

           (test-stream (top 2 :metric)
                        [(a 1) (b 2) (c 3)         (d 1)          (a 2)          (b nil) (d 2)]
                        [(a 1) (b 2) (c 3) (expire (d 1)) (expire (a 2)) (expire (b nil)) (d 2)])

           (test-stream (top 1 :metric)
                        [(a nil)          (b 1) (a nil)]
                        [(expire (a nil)) (b 1) (expire (a nil))])

           (test-stream (top 1 :metric)
                        [(b 2) (b nil)          (a 1)]
                        [(b 2) (expire (b nil)) (a 1)])

           (test-stream (top 2 :metric)
                        [(expire (a 2))]
                        [(expire (a 2))])

           (test-stream (top 1 :metric)
                        [(a 2) (expire (a 1)) (b 1)]
                        [(a 2) (expire (a 1)) (b 1)])

           (test-stream (top 2 :metric)
                        [(a 1) (b 2) (c 3) (d 4)         (a 2)          (b 3)  (c 4) (d 5)]
                        [(a 1) (b 2) (c 3) (d 4) (expire (a 2)) (expire (b 3)) (c 4) (d 5)])

           (with-test-stream index
             (top 1 :metric index bit-bucket true)
             [(a 1) (b 2)]
             [(a 1) (expire (a 1)) (b 2)])))

(deftest throttle-test
         (test-stream-intervals (throttle 3 2)
                                [{:state "1"} 0
                                 {:state "2"} 0
  t = 0
  t = 1
  t = 2
  t = 3
  t = 3
  t = 3
  t = 5
                                [{:state "1"}
                                 {:state "2"}
                                 {:state "3"}

                                 {:state "5"}
                                 {:state "expired"}
                                 {:state "expired"}

                                 {:state "expired"}]))

(deftest rollup-test
         (testing "single output event per interval"
           (test-stream-intervals
             (rollup 1 2)
             [:a 0 :b 0 :c 2 :d 0 :e 0 :f 2]
             [[:a] [:b :c] [:d :e :f]]))

         (testing "basic rollups"
                  (test-stream-intervals
                    (rollup 2 1)
                    [ 1 0 2 0 3 1   4 0 5 0 6 0 7 1       :foo 10]
                    [[1] [2]   [3] [4]           [5 6 7] [:foo]  ]))
           [ [ 1 ] [ 2 ]    [ 3 ] [ 4 ] [ 5 ]        [ 6 7 ] [: foo ] ]

         (testing "expired events"
                  (test-stream-intervals
                    (rollup 2 3)
                    [ 1 0 {:state "expired"} 0 :a 1 :b 1 :c 1 ]
                    [[1] [{:state "expired"}]              [:a :b :c]])

                  (let [e {:state "expired"}]
                    (test-stream-intervals
                      (rollup 2 2)
                      [ e 0 e 0 e 1 e 1]
                      [[e] [e]       [e e]]))))

(deftest batch-test
  (testing "nothing"
    (test-stream-intervals (batch 2 3) [] []))

  (testing "incomplete batches"
    (test-stream-intervals
      (batch 2 3)
      [:a 3 :b 1 :c 2 :d 3]
      [[:a] [:b :c] [:d]]))

  (testing "overflowing batches"
    (test-stream-intervals
      (batch 2 3)
      [:a 1 :b 1 :c 1 :d 1 :e 1 :f 1]
      [[:a :b] [:c] [:d :e] [:f]])))

(deftest coalesce-test
         (let [out (atom [])
               s (coalesce #(reset! out %))
               a1 {:service :a :state "one" :time 0}
               b1 {:service :b :state "one" :time 0}
               a2 {:service :a :state "two" :time 3 :ttl 2}
               c1 {:service :c :state "one" :time 0}
               b2 {:service :b :state "two" :time 0}]

           (s a1)
           (advance! 1.1)
           (is (= (set @out) #{a1}))

           (s b1)
           (advance! 2.1)
           (is (= (set @out) #{a1 b1}))

           (s a2)
           (advance! 3.1)
           (is (= (set @out) #{a2 b1}))

  Wait for  expiry of a2

  a2 arrived at 3 , expired at 5 ,
  and the final expiry flush happens at 6
           (s c1)
           (advance! 6.1)
           (is (= (set @out) #{a2 b1 c1}))

           (s b2)
           (advance! 7.1)
           (is (= (set @out) #{b2 c1}))))

(deftest stable-test
  Does n't emit until dt seconds have passed .
         (test-stream (stable 3 :x)
                      [{:x 1 :time 0} {:x 1 :time 1} {:x 1 :time 2}]
                      [])

         (test-stream (stable 3 :x)
                      [{:x 1 :time 0} {:x 1 :time 1} {:x 1 :time 3}]
                      [{:x 1 :time 0} {:x 1 :time 1} {:x 1 :time 3}])

         (test-stream (stable 3 :x)
                      [{:x 0 :time 0}
                       {:x 0 :time 3}
                       {:x 1 :time 4}
                       {:x 1 :time 5}
                       {:x 0 :time 6}
                       {:x 0 :time 9}]
                      [{:x 0 :time 0}
                       {:x 0 :time 3}
                       {:x 0 :time 6}
                       {:x 0 :time 9}])

         (test-stream (stable 3 :x)
                      [{:x 0 :time 0}
                       {:x 0 :time 10}
                       {:x 1 :time 11}
                       {:x 0 :time 11}
                       {:x 1 :time 12}
                       {:x 5 :time 13}
                       {:x 2 :time 14}
                       {:x 2 :time 17}]
                      [{:x 0 :time 0}
                       {:x 0 :time 10}
                       {:x 2 :time 14}
                       {:x 2 :time 17}])

  Triggers after dt seconds of stability , even without new events .
         (test-stream-intervals (stable 10 :x)
                                [{:x 0 :time 0} 1
                                 {:x 1 :time 1} 10
                                 {:x 2 :time 11} 1]
                                [{:x 1 :time 1}])

         (test-stream-intervals (stable 10 :x)
                                [{:x 0 :time 0} 1
                                 {:x 0 :time 1} 4
                                 {:x 0 :time 5} 6
                                 {:x 1 :time 11} 1]
                                [{:x 0 :time 0}
                                 {:x 0 :time 1}
                                 {:x 0 :time 5}]))

(deftest project-test
         (test-stream (project [(service :foo) (service :bar)])
                      []
                      [])

         (test-stream (project [])
                      [1 2 3]
                      [])

         (test-stream (project [(service "foo") (service "bar")])
                      [{:service "cat"}
                       {:service "foo" :a 1}
                       {:service "foo" :a 2}
                       {:service "meow"}
                       {:service "bar" :b 3}
                       {:service "foo" :b 4}]
                      [[{:service "foo" :a 1} nil]
                       [{:service "foo" :a 2} nil]
                       [{:service "foo" :a 2} {:service "bar" :b 3}]
                       [{:service "foo" :b 4} {:service "bar" :b 3}]])

         (test-stream
           (project [(service "foo") (service "bar")])
           [{:service "foo" :state "expired"}
            {:service "foo" :state "expired"}
            {:service "cat"}]
           [[{:service "foo" :time 0 :state "expired"} nil]
            [{:service "foo" :time 0 :state "expired"} nil]])

         (test-stream
           (project [(service "foo") (service "bar")])
           [{:service "foo" :state "ok"}
            {:service "bar" :state "ok"}
            {:service "bar" :state "expired"}
            {:service "foo" :state "expired"}
            {:service "bar" :state "expired"}]
           [[{:service "foo" :state "ok"}
             nil]
            [{:service "foo" :state "ok"}
             {:service "bar" :state "ok"}]
            [{:service "foo" :state "ok"}
             {:service "bar" :state "expired" :time 0}]
            [{:service "foo" :state "expired" :time 0}
             nil]
            [nil
             {:service "bar" :state "expired" :time 0}]])

         (test-stream-intervals
           (project [(service "foo") (service "bar")])
           [{:service "foo" :state "ok" :time 0 :ttl 1}
            2
            {:service "bar" :state "ok"}
            1
            {:service "bar" :state "ok2"}]
           [[{:service "foo" :state "ok" :time 0 :ttl 1} nil]
            [{:service "foo" :state "expired" :time 2}
             {:service "bar" :state "ok"}]
            [nil {:service "bar" :state "ok2"}]]))

(deftest adjust-test
  (let [out (atom nil)
        s (adjust [:state str " 2"] (register out))]

    (s {})
    (is (= (deref out) {:state " 2"}))

    (s {:state "hey" :service "bar"})
    (is (= (deref out) {:state "hey 2" :service "bar"})))

  (let [out (atom nil)
        s (adjust #(assoc % :metric (count (:tags %)))
                  (register out))]

    (s {:service "a" :tags []})
    (is (= (deref out) {:service "a" :tags [] :metric 0}))

    (s {:service "a" :tags ["foo" "bar"]})
    (is (= (deref out) {:service "a" :tags ["foo" "bar"] :metric 2}))))

(deftest moving-event-window-test
  Zero - width windows .
         (test-stream (moving-event-window 0) [] [])
         (test-stream (moving-event-window 0) [1 2] [[] []])

         (test-stream (moving-event-window 2) [1 2 3] [[1] [1 2] [2 3]]))

(deftest fixed-event-window-test
  Zero - width windows .
         (test-stream (fixed-event-window 0) [] [])
         (test-stream (fixed-event-window 0) [1 2] [])

         (test-stream (fixed-event-window 2) [1] [])
         (test-stream (fixed-event-window 2) [1 2] [[1 2]])
         (test-stream (fixed-event-window 2) [1 2 3 4 5] [[1 2] [3 4]]))

(deftest runs-test
  Zero - width runs
         (test-stream (runs 0 :state)
                      []
                      [])

         (test-stream (runs 0 :state)
                      [{:state 2} {:state 3} {:state 4}]
                      [])

         (test-stream (runs 3 :state)
                      [{:state 1} {:state 2} {:state 2}]
                      [])

         (test-stream (runs 3 :state)
                      [{:state 1} {:state 1} {:metric "0.5" :state 2} {:metric "0.6" :state 2} {:state 3} {:state 4 :metric "5.0"} {:state 4 :metric "5.1"} {:state 4 :metric "5.2"}]
                      [{:state 4 :metric "5.2"}]
                      )
  )

(deftest moving-time-window-test
  Zero - second windows .
         (test-stream (moving-time-window 0) [] [])
         (test-stream (moving-time-window 0) [{:time 1} {:time 2}] [])

         (test-stream (moving-time-window 2) [] [])
         (test-stream (moving-time-window 2) [{:time 1}] [[{:time 1}]])
         (test-stream (moving-time-window 2)
                      [{:time 1} {:time 2} {:time 3} {:time 4}]
                      [[{:time 1}]
                       [{:time 1} {:time 2}]
                       [{:time 2} {:time 3}]
                       [{:time 3} {:time 4}]])

         (test-stream (moving-time-window 2)
                      [{:time 5}
                       {:time 1}
                       {:time 2}
                       {:time 6}
                       {:time 3}
                       {:time 8}
                       {:time 4}
                       {:time 8}
                       {:time 5}
                       {:time 9}]
                      [[{:time 5}]
                       [{:time 5} {:time 6}]
                       [{:time 8}]
                       [{:time 8} {:time 8}]
                       [{:time 8} {:time 8} {:time 9}]]))

(deftest fixed-time-window-test
  Zero - time windows .
         (is (thrown? IllegalArgumentException (fixed-time-window 0)))

         (test-stream (fixed-time-window 2) [] [])
         (test-stream (fixed-time-window 2) [{:time 1}] [])
         (test-stream (fixed-time-window 2)
                      [{:time 1} {:time 2} {:time 3} {:time 4} {:time 5}]
                      [[{:time 1} {:time 2}]
                       [{:time 3} {:time 4}]])

         (test-stream (fixed-time-window 2) [{:time 1} {:time 7}]
                      [[{:time 1}] [] []])

         (test-stream (fixed-time-window 2)
                      [{:time 5}
                       {:time 1}
                       {:time 2}
                       {:time 6}
                       {:time 3}
                       {:time 8}
                       {:time 4}
                       {:time 8}
                       {:time 5}
                       {:time 9}
                       {:time 11}]
                      [[{:time 5} {:time 6}]
                       [{:time 8} {:time 8}]
                       [{:time 9}]]))

(deftest fixed-offset-time-window-test
  Zero - time windows .
         (is (thrown? IllegalArgumentException (fixed-offset-time-window 0)))

         (test-stream (fixed-offset-time-window 2) [] [])
         (test-stream (fixed-offset-time-window 2) [{:time 1}] [])
         (test-stream (fixed-offset-time-window 2)
                      [{:time 1} {:time 2} {:time 3} {:time 4} {:time 5} {:time 6}]
                      [[{:time 1}]
                       [{:time 2} {:time 3}]
                       [{:time 4} {:time 5}]])

         (test-stream (fixed-offset-time-window 2) [{:time 1} {:time 7}]
                      [[{:time 1}] [] []])

         (test-stream (fixed-offset-time-window 2)
                      [{:time 5}
                       {:time 1}
                       {:time 2}
                       {:time 6}
                       {:time 3}
                       {:time 8}
                       {:time 4}
                       {:time 8}
                       {:time 5}
                       {:time 9}
                       {:time 11}]
                      [[{:time 5}]
                       [{:time 6}]
                       [{:time 8} {:time 8} {:time 9}]]))

(deftest part-time-simple-test
         (advance! 1)
         (let [effects (atom [])
               p (part-time-simple
                   5
                   (fn [_] [])
                   conj
                   (fn [window start end] (swap! effects conj
                                                 (concat [start] window [end]))))]
           (advance! 1) (p :t1)
           (advance! 2) (p :t2)
           (advance! 4) (p :t4)
           (advance! 5) (p :t5)
           (advance! 6) (p :t6)
           (advance! 8) (p :t8)
           (advance! 9) (p :t9)
           (advance! 99) (p :t99)
           (advance! 100) (p :t100)
           (advance! 101) (p :t101)
           (advance! 102) (p :t102)
           (advance! 200)
           (is (= @effects
                  [[1 :t1 :t2 :t4 :t5 6]
                  [6 :t6 :t8 :t9 11]
                  [96 :t99 :t100 101]
                  [101 :t101 :t102 106]]))))

(deftest apdex-test
         (test-stream-intervals
           (apdex 1 (state "ok") (state "warning"))
           (concat
             (interpose 0 (shuffle (concat (repeat 5 {:state "ok"})
                                           (repeat 7 {:state "warning"})
                                           (repeat 3 {:state "expired"})
                                           (repeat 1 {:state "fail"}))))
             [0 {:state "final"} 3])
           [{:state "final" :metric (/ (+ 5 (/ 7 2)) (+ 5 7 2))}]))

(deftest ^:focus clock-skew-test
         (test-stream-intervals (clock-skew)
                                 {:time 1 :host :foo} 1
  2
                                 {:time 3 :host :bar} 1
  3 4
                                 {:time 9 :host :baz} 1
                                 {:time 1 :host :foo} 1
                                 {:time 100 :host :foo}]
                                [{:time 1 :host :foo :metric 0}
                                 {:time 3 :host :bar :metric 0}
                                 {:time 9 :host :baz :metric 5}
                                 {:time 1 :host :foo :metric -4}
                                 {:time 100 :host :foo :metric 89}]))

(deftest not-expired-test
  (testing "state expired"
    (test-stream (not-expired)
                 [{:state "expired"}
                  {:state "ok"}
                  {:state "warning"}
                  {:state "expired"}]
                 [{:state "ok"}
                  {:state "warning"}]))
  (testing "expired time"
    (let [out (atom [])
          child #(swap! out conj %)
          s (not-expired child)]
      (s {:time 0 :ttl 10})
      (advance! 10)
      (s {:time 1 :ttl 5})
      (s {:time 6 :ttl 2})
      (s {:time 10 :ttl 2})
      (is (= @out [{:time 0 :ttl 10}
                   {:time 10 :ttl 2}])))))

(deftest expired-test
  (testing "state expired"
    (test-stream (expired)
                 [{:state "expired"}
                  {:state "ok"}
                  {:state "warning"}
                  {:state "expired"}]
                 [{:state "expired"}
                  {:state "expired"}]))
  (testing "expired time"
    (let [out (atom [])
          child #(swap! out conj %)
          s (expired child)]
      (s {:time 0 :ttl 10})
      (advance! 10)
      (s {:time 1 :ttl 5})
      (s {:time 6 :ttl 2})
      (s {:time 10 :ttl 2})
      (is (= @out [{:time 1 :ttl 5}
                   {:time 6 :ttl 2}])))))

(deftest predict-linear-test
  (test-stream (predict-linear 10 600)
               [{:metric 1 :time 1}]
               [])
  (test-stream (predict-linear 5 60)
               [{:metric 1 :time 1}
                {:metric 2 :time 2}
                {:metric 3 :time 3}
                {:metric 4 :time 4}
                {:metric 5 :time 5}]
               [{:metric 62 :time 2}
                {:metric 63 :time 3}
                {:metric 64 :time 4}
                {:metric 65 :time 5}])
  (test-stream (predict-linear 2 86400)
               [{:metric 100 :time 0}
                {:metric 50 :time 43200}]
               [{:metric -50 :time 43200}])
  (test-stream (predict-linear 2 60)
               [{:metric 100 :time 0}
                {:metric 100 :time 1}]
               [{:metric 100 :time 1}])
  (test-stream (predict-linear 5 60 60)
               [{:metric 1 :time 1}
                {:metric 2 :time 2}
                {:metric 3 :time 3}
                {:metric 4 :time 4}
                {:metric 5 :time 5}]
               [{:metric 62 :time 2}
                {:metric 63 :time 3}
                {:metric 64 :time 4}
                {:metric 65 :time 5}])
  (test-stream (predict-linear 2 60 60)
               [{:metric 100 :time 0}
                {:metric 100 :time 1}
                {:metric 500 :time 2}]
               [{:metric 100 :time 1}
                {:metric 100 :time 2}]))
