@Test
public void testProperties() throws Exception
{

  Configuration conf = new Configuration(false);
  conf.set(StreamingApplication.DT_PREFIX + "operator.o1.prop.subject", subject);
  conf.set(StreamingApplication.DT_PREFIX + "operator.o1.prop.content", content);
  conf.set(StreamingApplication.DT_PREFIX + "operator.o1.prop.from", from);
  conf.set(StreamingApplication.DT_PREFIX + "operator.o1.prop.smtpHost", "127.0.0.1");
  conf.set(StreamingApplication.DT_PREFIX + "operator.o1.prop.smtpUserName", from);
  conf.set(StreamingApplication.DT_PREFIX + "operator.o1.prop.smtpPassword", "<password>");
  conf.set(StreamingApplication.DT_PREFIX + "operator.o1.prop.recipients.TO", to + "," + cc);
  conf.set(StreamingApplication.DT_PREFIX + "operator.o1.prop.recipients.CC", cc);
  final AtomicReference<SmtpOutputOperator> o1 = new AtomicReference<SmtpOutputOperator>();
  StreamingApplication app = new StreamingApplication()
  {
    @Override
    public void populateDAG(DAG dag, Configuration conf)
    {
      o1.set(dag.addOperator("o1", new SmtpOutputOperator()));
    }
  };
  LocalMode lma = LocalMode.newInstance();
  lma.prepareDAG(app, conf);
  Assert.assertEquals("checking TO list", to + "," + cc, o1.get().getRecipients().get("TO"));
  Assert.assertEquals("checking CC list", cc, o1.get().getRecipients().get("CC"));
}