integration_test.clj

(defn run-test-job
  "Runs the email job for `timeout` milliseconds before returning whatever
   has been accumulated in `sent-emails`.

   TODO: Consider promoting more generic version of this to samza-config"
  [timeout]
  (let [job (MapConfig. (assoc (into {} (find-job 'notifications-handler.core.email))
                               "job.task.send-fn" (full-name 'mock-send-fn)))]

    (let [job-proc (.run (JobRunner. job) true)]
      (s/stream->seq sent-emails timeout))))

(deftest send-some-emails
  (with-uuids [id tracking-id]
    (let-flow [result (run-test-job 2000)]
      (let [msg {:id id
                 :tracking-id tracking-id
                 :to "[email protected]"
                 :from "[email protected]"
                 :subject "yolo"
                 :body ""
                 :published-at 0
                 :published-by "test"}]

        (println "sending email one")
        (send-email msg)

        (println "sending email two")
        (send-email msg))

      (when (= result :timeout)
        (is false "Timed out waiting for test email job"))

      (is (= 1 (count result)))

      (is (= {:tracking-id tracking-id
              :to "[email protected]"
              :from "[email protected]"
              :subject "yolo"
              :body ""}
             (select-keys (first result)))))))