<commit_before>(ns riemann.test.core
  (:require riemann.transport.tcp
            riemann.streams
            [riemann.logging :as logging])
  (:use riemann.client
        riemann.common
        riemann.core
        riemann.index
        clojure.test
        riemann.time
        riemann.time.controlled
        [clojure.algo.generic.functor :only [fmap]]
        [riemann.time :only [unix-time]]))

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

(defmacro tim
  "Evaluates expr and returns the time it took in seconds"
  [expr]
  `(let [start# (. System (nanoTime))
         ret# ~expr]
     (/ (- (. System (nanoTime)) start#) 1000000000.0)))

(deftest serialization
         (let [core (core)
               out (ref [])
               server (logging/suppress "riemann.transport.tcp"
                                        (riemann.transport.tcp/tcp-server core))
               stream (riemann.streams/append out)
               client (riemann.client/tcp-client)
               events [{:host "shiiiiire!"}
                       {:service "baaaaaginnnns!"}
                       {:state "middling"}
                       {:description "quite dire, really"}
                       {:tags ["oh" "sam"]}
                       {:metric -1000.0}
                       {:metric Double/MAX_VALUE}
                       {:metric Long/MIN_VALUE}
                       {:time 1234}
                       {:ttl 12.0}]]
           
           (try
             (swap! (core :servers) conj server)
             (swap! (core :streams) conj stream)

             ; Send events
             (doseq [e events] (send-event client e))

             (doseq [[in out] (map (fn [a b] [a b]) events (deref out))]
               (is (every? (fn [k] (= (k in) (k out))) (keys in))))

             (finally
               (close-client client)
               (logging/suppress ["riemann.core" "riemann.transport.tcp"] 
                                 (stop core))))))

(deftest query-test
         (let [core (core)
               index (index)
               server (logging/suppress "riemann.transport.tcp"
                                        (riemann.transport.tcp/tcp-server core))
               client (riemann.client/tcp-client)]

           (try
             (swap! (core :servers) conj server)
             (reset! (core :index) index)

             ; Send events
             (update-index core {:metric 1 :time 1})
             (update-index core {:metric 2 :time 3})
             (update-index core {:host "kitten" 
                                 :tags ["whiskers" "paws"] :time 2})
             (update-index core {:service "miao" :host "cat" :time 3})

             (let [r (set (query client "metric = 2 or service = \"miao\" or tagged \"whiskers\""))]
               (is (= r
                      #{(event {:metric 2, :time 3})
                        (event {:host "kitten" :tags ["whiskers" "paws"] :time 2})
                        (event {:host "cat", :service "miao", :time 3})} r)))

             (finally
               (close-client client)
               (logging/suppress ["riemann.core" "riemann.transport.tcp"]
                 (stop core))))))

(deftest expires
         (let [core (core)
               index (index)
               res (ref nil)
               expired-stream (riemann.streams/expired 
                                (fn [e] (dosync (ref-set res e))))
               reaper (periodically-expire core 0.001)]

                (reset! (:index core) index)
                (swap! (:streams core) conj expired-stream)

           ; Insert events
           (update-index core {:service 1 :ttl 0.01 :time (unix-time)})
           (update-index core {:service 2 :ttl 1 :time (unix-time)})

           (advance! 0.011)

           ; Wait for reaper to eat them
           (Thread/sleep 30)

           ; Kill reaper
           (future-cancel reaper)
           
           ; Check that index does not contain these states
           (is (= [2] (map (fn [e] (:service e)) index)))

           ; Check that expired-stream received them.
           (is (= @res
                  {:service 1
                   :host nil
                   :time 0.011
                   :state "expired"}))))

(comment
  (deftest sum<commit_msg>Up the time tolerance for reaper expiry tests.<commit_after>(ns riemann.test.core
  (:require riemann.transport.tcp
            riemann.streams
            [riemann.logging :as logging])
  (:use riemann.client
        riemann.common
        riemann.core
        riemann.index
        clojure.test
        riemann.time
        riemann.time.controlled
        [clojure.algo.generic.functor :only [fmap]]
        [riemann.time :only [unix-time]]))

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

(defmacro tim
  "Evaluates expr and returns the time it took in seconds"
  [expr]
  `(let [start# (. System (nanoTime))
         ret# ~expr]
     (/ (- (. System (nanoTime)) start#) 1000000000.0)))

(deftest serialization
         (let [core (core)
               out (ref [])
               server (logging/suppress "riemann.transport.tcp"
                                        (riemann.transport.tcp/tcp-server core))
               stream (riemann.streams/append out)
               client (riemann.client/tcp-client)
               events [{:host "shiiiiire!"}
                       {:service "baaaaaginnnns!"}
                       {:state "middling"}
                       {:description "quite dire, really"}
                       {:tags ["oh" "sam"]}
                       {:metric -1000.0}
                       {:metric Double/MAX_VALUE}
                       {:metric Long/MIN_VALUE}
                       {:time 1234}
                       {:ttl 12.0}]]
           
           (try
             (swap! (core :servers) conj server)
             (swap! (core :streams) conj stream)

             ; Send events
             (doseq [e events] (send-event client e))

             (doseq [[in out] (map (fn [a b] [a b]) events (deref out))]
               (is (every? (fn [k] (= (k in) (k out))) (keys in))))

             (finally
               (close-client client)
               (logging/suppress ["riemann.core" "riemann.transport.tcp"] 
                                 (stop core))))))

(deftest query-test
         (let [core (core)
               index (index)
               server (logging/suppress "riemann.transport.tcp"
                                        (riemann.transport.tcp/tcp-server core))
               client (riemann.client/tcp-client)]

           (try
             (swap! (core :servers) conj server)
             (reset! (core :index) index)

             ; Send events
             (update-index core {:metric 1 :time 1})
             (update-index core {:metric 2 :time 3})
             (update-index core {:host "kitten" 
                                 :tags ["whiskers" "paws"] :time 2})
             (update-index core {:service "miao" :host "cat" :time 3})

             (let [r (set (query client "metric = 2 or service = \"miao\" or tagged \"whiskers\""))]
               (is (= r
                      #{(event {:metric 2, :time 3})
                        (event {:host "kitten" :tags ["whiskers" "paws"] :time 2})
                        (event {:host "cat", :service "miao", :time 3})} r)))

             (finally
               (close-client client)
               (logging/suppress ["riemann.core" "riemann.transport.tcp"]
                 (stop core))))))

(deftest expires
         (let [core (core)
               index (index)
               res (atom nil)
               expired-stream (riemann.streams/expired 
                                (fn [e] (reset! res e)))
               reaper (periodically-expire core 0.001)]

           (reset! (:index core) index)
           (swap! (:streams core) conj expired-stream)

           ; Insert events
           (update-index core {:service 1 :ttl 0.01 :time (unix-time)})
           (update-index core {:service 2 :ttl 1 :time (unix-time)})

           (advance! 0.011)

           ; Wait for reaper to eat them
           (Thread/sleep 100)

           ; Kill reaper
           (future-cancel reaper)
           
           ; Check that index does not contain these states
           (is (= [2] (map (fn [e] (:service e)) index)))

           ; Check that expired-stream received them.
           (is (= @res
                  {:service 1
                   :host nil
                   :time 0.011
                   :state "expired"}))))

(comment
  (deftest sum
{'fraction_non_alphanumeric': 0.12455020313406849, 'fraction_numerical': 0.013116656993615786, 'mean_word_length': 1.3508867667121418, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2103674', 'n_tokens_mistral': 2625, 'n_tokens_neox': 2351, 'n_words': 681}