                                       mock-worker-id)
             (verify-first-call-args-for-indices launch-process
                                                 [0]
-                                                exp-args)))))))
+                                                exp-args))))
+      (testing "testing topology.classpath is added to classpath"
+        (let [topo-cp "/any/path"
+              exp-args (exp-args-fn [] [] (add-to-classpath mock-cp [topo-cp]))
+              mock-supervisor {:conf {STORM-CLUSTER-MODE :distributed}}]
+          (stubbing [read-supervisor-storm-conf {TOPOLOGY-CLASSPATH topo-cp}
+                     supervisor-stormdist-root nil
+                     supervisor/jlp nil
+                     launch-process nil
+                     current-classpath "/base"]
+                    (supervisor/launch-worker mock-supervisor
+                                              mock-storm-id
+                                              mock-port
+                                              mock-worker-id)
+                    (verify-first-call-args-for-indices launch-process
+                                                        [0]
+                                                        exp-args))))
+      (testing "testing topology.environment is added to environment for worker launch"
+        (let [topo-env {"THISVAR" "somevalue" "THATVAR" "someothervalue"}
+              exp-args (exp-args-fn [] [] mock-cp)
+              mock-supervisor {:conf {STORM-CLUSTER-MODE :distributed}}]
+          (stubbing [read-supervisor-storm-conf {TOPOLOGY-ENVIRONMENT topo-env}
+                     supervisor-stormdist-root nil
+                     supervisor/jlp nil
+                     launch-process nil
+                     current-classpath "/base"]
+                    (supervisor/launch-worker mock-supervisor
+                                              mock-storm-id
+                                              mock-port
+                                              mock-worker-id)
+                    (verify-first-call-args-for-indices launch-process
+                                                        [2]
+                                                        (merge topo-env {"LD_LIBRARY_PATH" nil}))))))))
 
 (deftest test-workers-go-bananas
   ;; test that multiple workers are started for a port, and test that

        (let [string-opts "-Dfoo=bar  -Xmx1024m"
              topo-string-opts "-Dkau=aux   -Xmx2048m"
              exp-args (exp-args-fn ["-Dfoo=bar" "-Xmx1024m"]
                                    ["-Dkau=aux" "-Xmx2048m"])
              mock-supervisor {:conf {STORM-CLUSTER-MODE :distributed
                                      WORKER-CHILDOPTS string-opts}}]
          (stubbing [read-supervisor-storm-conf {TOPOLOGY-WORKER-CHILDOPTS

                                (str "-Dstorm.id=" mock-storm-id)
                                (str "-Dworker.id=" mock-worker-id)
                                (str "-Dworker.port=" mock-port)
                                "-cp" classpath
                                "backtype.storm.daemon.worker"
                                mock-storm-id
                                mock-port
