         (is (= (.value record) "all streams lead to kafka"))))))
 
 (deftest test-word-count-topology
-  (testing "It works"
+  (testing "Complete table written to output topic"
     (let [builder (KStreamBuilder.)
           _ (build-count-words builder)
           driver (ProcessorTopologyTestDriver. count-words-props builder

  (:require [clojure.test :refer :all]
            [clstreams.core :refer :all])
  (:import [org.apache.kafka.test ProcessorTopologyTestDriver]
           [org.apache.kafka.common.serialization StringSerializer StringDeserializer]
           [org.apache.kafka.streams.kstream KStreamBuilder]))

(def string-serializer (StringSerializer.))
(def string-deserializer (StringDeserializer.))

(deftest test-copy-topology
  (testing "It works"
    (let [builder (KStreamBuilder.)
          _ (build-copy builder)
          driver (ProcessorTopologyTestDriver. count-words-props builder

                string-serializer string-serializer)
      (.process driver "streams-file-input" "" "join kafka summit"
                string-serializer string-serializer)
      (let [output (collect-output #(.readOutput driver "streams-wordcount-output"
                                                 string-deserializer long-deserializer))]
        (let [expected {"all" 1, "streams" 2, "lead" 1, "to" 1,
                        "kafka" 3, "hello" 1, "join" 1, "summit" 1}]
          (is (= output expected) "Output topic contents are correct")
          (is (= (-> driver (.getKeyValueStore "Counts") .all iterator-seq key-value-map)
                 expected) "Store contents are correct"))))))
