[email] fixed bugs and added unit tests

Original commit: elastic/x-pack-elasticsearch@3b5406d4c8

pom.xml

@@ -44,6 +44,13 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>com.google.apis</groupId>
+            <artifactId>google-api-services-gmail</artifactId>
+            <version>v1-rev23-1.19.1</version>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>org.codehaus.groovy</groupId>
             <artifactId>groovy-all</artifactId>

src/main/java/org/elasticsearch/alerts/AlertsPlugin.java

@@ -5,6 +5,7 @@
  */
 package org.elasticsearch.alerts;
 
+import org.elasticsearch.alerts.actions.email.service.InternalEmailService;
 import org.elasticsearch.alerts.support.init.InitializingService;
 import org.elasticsearch.common.collect.ImmutableList;
 import org.elasticsearch.common.component.LifecycleComponent;
@@ -47,7 +48,8 @@ public class AlertsPlugin extends AbstractPlugin {
                 // the initialization service must be first in the list
                 // as other services may depend on one of the initialized
                 // constructs
-                InitializingService.class);
+                InitializingService.class,
+                InternalEmailService.class);
     }
 
     @Override

src/main/java/org/elasticsearch/alerts/actions/email/EmailAction.java

@@ -22,6 +22,7 @@ import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
 
 import java.io.IOException;
+import java.util.Objects;
 
 /**
  */
@@ -29,23 +30,23 @@ public class EmailAction extends Action<EmailAction.Result> {
 
     public static final String TYPE = "email";
 
-    private final Email.Builder email;
+    final Email emailPrototype;
-    private final Authentication auth;
+    final Authentication auth;
-    private final Profile profile;
+    final Profile profile;
-    private final String account;
+    final String account;
-    private final Template subject;
+    final Template subject;
-    private final Template textBody;
+    final Template textBody;
-    private final Template htmlBody;
+    final Template htmlBody;
-    private final boolean attachPayload;
+    final boolean attachPayload;
 
-    private final EmailService emailService;
+    final EmailService emailService;
 
-    public EmailAction(ESLogger logger, EmailService emailService, Email.Builder email, Authentication auth, Profile profile,
+    public EmailAction(ESLogger logger, EmailService emailService, Email emailPrototype, Authentication auth, Profile profile,
                        String account, Template subject, Template textBody, Template htmlBody, boolean attachPayload) {
 
         super(logger);
         this.emailService = emailService;
-        this.email = email;
+        this.emailPrototype = emailPrototype;
         this.auth = auth;
         this.profile = profile;
         this.account = account;
@@ -64,6 +65,10 @@ public class EmailAction extends Action<EmailAction.Result> {
     public Result execute(ExecutionContext ctx, Payload payload) throws IOException {
         ImmutableMap<String, Object> model = templateModel(ctx, payload);
 
+        Email.Builder email = Email.builder()
+                .id(ctx.id())
+                .copyFrom(emailPrototype);
+
         email.id(ctx.id());
         email.subject(subject.render(model));
         email.textBody(textBody.render(model));
@@ -73,7 +78,7 @@ public class EmailAction extends Action<EmailAction.Result> {
         }
 
         if (attachPayload) {
-            Attachment.Bytes attachment = new Attachment.XContent.Yaml("payload", "payload.yml", "alert execution output", payload);
+            Attachment.Bytes attachment = new Attachment.XContent.Yaml("payload", "payload.yml", payload);
             email.attach(attachment);
         }
 
@@ -86,22 +91,71 @@ public class EmailAction extends Action<EmailAction.Result> {
         }
     }
 
+    @Override
+    public int hashCode() {
+        return Objects.hash(emailPrototype, auth, profile, account, subject, textBody, htmlBody, attachPayload);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+        if (obj == null || getClass() != obj.getClass()) {
+            return false;
+        }
+        final EmailAction other = (EmailAction) obj;
+        return Objects.equals(this.emailPrototype, other.emailPrototype)
+                && Objects.equals(this.auth, other.auth)
+                && Objects.equals(this.profile, other.profile)
+                && Objects.equals(this.account, other.account)
+                && Objects.equals(this.subject, other.subject)
+                && Objects.equals(this.textBody, other.textBody)
+                && Objects.equals(this.htmlBody, other.htmlBody)
+                && Objects.equals(this.attachPayload, other.attachPayload);
+    }
+
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
         if (account != null) {
-            builder.field(EmailAction.Parser.ACCOUNT_FIELD.getPreferredName(), account);
+            builder.field(Parser.ACCOUNT_FIELD.getPreferredName(), account);
         }
         if (profile != null) {
-            builder.field(EmailAction.Parser.PROFILE_FIELD.getPreferredName(), profile);
+            builder.field(Parser.PROFILE_FIELD.getPreferredName(), profile);
+        }
+        if (auth != null) {
+            builder.field(Parser.USER_FIELD.getPreferredName(), auth.user());
+            builder.field(Parser.PASSWORD_FIELD.getPreferredName(), auth.password());
+        }
+        builder.field(Parser.ATTACH_PAYLOAD_FIELD.getPreferredName(), attachPayload);
+        if (emailPrototype.from() != null) {
+            builder.field(Email.FROM_FIELD.getPreferredName(), emailPrototype.from());
+        }
+        if (emailPrototype.to() != null && !emailPrototype.to().isEmpty()) {
+            builder.field(Email.TO_FIELD.getPreferredName(), (ToXContent) emailPrototype.to());
+        }
+        if (emailPrototype.cc() != null && !emailPrototype.cc().isEmpty()) {
+            builder.field(Email.CC_FIELD.getPreferredName(), (ToXContent) emailPrototype.cc());
+        }
+        if (emailPrototype.bcc() != null && !emailPrototype.bcc().isEmpty()) {
+            builder.field(Email.BCC_FIELD.getPreferredName(), (ToXContent) emailPrototype.bcc());
+        }
+        if (emailPrototype.replyTo() != null && !emailPrototype.replyTo().isEmpty()) {
+            builder.field(Email.REPLY_TO_FIELD.getPreferredName(), (ToXContent) emailPrototype.replyTo());
         }
-        builder.field(Email.TO_FIELD.getPreferredName(), (ToXContent) email.to());
         if (subject != null) {
             builder.field(Email.SUBJECT_FIELD.getPreferredName(), subject);
         }
         if (textBody != null) {
             builder.field(Email.TEXT_BODY_FIELD.getPreferredName(), textBody);
         }
+        if (htmlBody != null) {
+            builder.field(Email.HTML_BODY_FIELD.getPreferredName(), htmlBody);
+        }
+        if (emailPrototype.priority() != null) {
+            builder.field(Email.PRIORITY_FIELD.getPreferredName(), emailPrototype.priority());
+        }
         return builder.endObject();
     }
 
@@ -136,7 +190,7 @@ public class EmailAction extends Action<EmailAction.Result> {
             String password = null;
             String account = null;
             Profile profile = null;
-            Email.Builder email = Email.builder();
+            Email.Builder email = Email.builder().id("prototype");
             Template subject = null;
             Template textBody = null;
             Template htmlBody = null;
@@ -202,12 +256,9 @@ public class EmailAction extends Action<EmailAction.Result> {
                 }
             }
 
-            if (email.to() == null || email.to().isEmpty()) {
+            Authentication auth = user != null ? new Authentication(user, password) : null;
-                throw new ActionSettingsException("could not parse email action. [to] was not found or was empty");
-            }
 
-            Authentication auth = new Authentication(user, password);
+            return new EmailAction(logger, emailService, email.build(), auth, profile, account, subject, textBody, htmlBody, attachPayload);
-            return new EmailAction(logger, emailService, email, auth, profile, account, subject, textBody, htmlBody, attachPayload);
         }
 
         @Override
@@ -263,7 +314,6 @@ public class EmailAction extends Action<EmailAction.Result> {
             super(type, success);
         }
 
-
         public static class Success extends Result {
 
             private final EmailService.EmailSent sent;
@@ -297,6 +347,10 @@ public class EmailAction extends Action<EmailAction.Result> {
                 this.reason = reason;
             }
 
+            public String reason() {
+                return reason;
+            }
+
             @Override
             protected XContentBuilder xContentBody(XContentBuilder builder, Params params) throws IOException {
                 return builder.field("reason", reason);

src/main/java/org/elasticsearch/alerts/actions/email/service/Account.java

@@ -43,8 +43,12 @@ public class Account {
         // applying the defaults on missing emails fields
         email = config.defaults.apply(email);
 
+        if (email.to == null) {
+            throw new EmailException("email must have [to] recipient");
+        }
+
         Transport transport = session.getTransport(SMTP_PROTOCOL);
-        String user = auth != null ? auth.username() : null;
+        String user = auth != null ? auth.user() : null;
         if (user == null) {
             user = config.smtp.user;
             if (user == null) {
@@ -85,10 +89,10 @@ public class Account {
 
         static final String SMTP_SETTINGS_PREFIX = "mail.smtp.";
 
-        private final String name;
+        final String name;
-        private final Profile profile;
+        final Profile profile;
-        private final Smtp smtp;
+        final Smtp smtp;
-        private final EmailDefaults defaults;
+        final EmailDefaults defaults;
 
         public Config(String name, Settings settings) {
             this.name = name;
@@ -106,16 +110,16 @@ public class Account {
 
         static class Smtp {
 
-            private final String host;
+            final String host;
-            private final int port;
+            final int port;
-            private final String user;
+            final String user;
-            private final String password;
+            final String password;
-            private final Properties properties;
+            final Properties properties;
 
             public Smtp(Settings settings) {
-                host = settings.get("host");
+                host = settings.get("host", settings.get("localaddress", settings.get("local_address")));
-                port = settings.getAsInt("port", settings.getAsInt("localport", 25));
+                port = settings.getAsInt("port", settings.getAsInt("localport", settings.getAsInt("local_port", 25)));
-                user = settings.get("user", settings.get("from", settings.get("local_address", null)));
+                user = settings.get("user", settings.get("from", null));
                 password = settings.get("password", null);
                 properties = loadSmtpProperties(settings);
             }
@@ -165,7 +169,7 @@ public class Account {
          * needed on each alert (e.g. if all the emails are always sent to the same recipients
          * one could set those here and leave them out on the alert definition).
          */
-        class EmailDefaults {
+        static class EmailDefaults {
 
             final Email.Address from;
             final Email.AddressList replyTo;
@@ -186,16 +190,59 @@ public class Account {
             }
 
             Email apply(Email email) {
-                return Email.builder()
+                Email.Builder builder = Email.builder().copyFrom(email);
-                        .from(from)
+                if (email.from == null) {
-                        .replyTo(replyTo)
+                    builder.from(from);
-                        .priority(priority)
+                }
-                        .to(to)
+                if (email.replyTo == null) {
-                        .cc(cc)
+                    builder.replyTo(replyTo);
-                        .bcc(bcc)
+                }
-                        .subject(subject)
+                if (email.priority == null) {
-                        .copyFrom(email)
+                    builder.priority(priority);
-                        .build();
+                }
+                if (email.to == null) {
+                    builder.to(to);
+                }
+                if (email.cc == null) {
+                    builder.cc(cc);
+                }
+                if (email.bcc == null) {
+                    builder.bcc(bcc);
+                }
+                if (email.subject == null) {
+                    builder.subject(subject);
+                }
+                return builder.build();
+            }
+
+            @Override
+            public boolean equals(Object o) {
+                if (this == o) return true;
+                if (o == null || getClass() != o.getClass()) return false;
+
+                EmailDefaults that = (EmailDefaults) o;
+
+                if (bcc != null ? !bcc.equals(that.bcc) : that.bcc != null) return false;
+                if (cc != null ? !cc.equals(that.cc) : that.cc != null) return false;
+                if (from != null ? !from.equals(that.from) : that.from != null) return false;
+                if (priority != that.priority) return false;
+                if (replyTo != null ? !replyTo.equals(that.replyTo) : that.replyTo != null) return false;
+                if (subject != null ? !subject.equals(that.subject) : that.subject != null) return false;
+                if (to != null ? !to.equals(that.to) : that.to != null) return false;
+
+                return true;
+            }
+
+            @Override
+            public int hashCode() {
+                int result = from != null ? from.hashCode() : 0;
+                result = 31 * result + (replyTo != null ? replyTo.hashCode() : 0);
+                result = 31 * result + (priority != null ? priority.hashCode() : 0);
+                result = 31 * result + (to != null ? to.hashCode() : 0);
+                result = 31 * result + (cc != null ? cc.hashCode() : 0);
+                result = 31 * result + (bcc != null ? bcc.hashCode() : 0);
+                result = 31 * result + (subject != null ? subject.hashCode() : 0);
+                return result;
             }
         }
     }

src/main/java/org/elasticsearch/alerts/actions/email/service/Accounts.java

@@ -8,7 +8,6 @@ package org.elasticsearch.alerts.actions.email.service;
 import org.elasticsearch.common.logging.ESLogger;
 import org.elasticsearch.common.settings.Settings;
 
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -21,33 +20,29 @@ public class Accounts {
     private final Map<String, Account> accounts;
 
     public Accounts(Settings settings, ESLogger logger) {
-        settings = settings.getAsSettings("account");
+        Settings accountsSettings = settings.getAsSettings("account");
-        Map<String, Account> accounts = new HashMap<>();
+        accounts = new HashMap<>();
-        for (String name : settings.names()) {
+        for (String name : accountsSettings.names()) {
-            Account.Config config = new Account.Config(name, settings.getAsSettings(name));
+            Account.Config config = new Account.Config(name, accountsSettings.getAsSettings(name));
             Account account = new Account(config, logger);
             accounts.put(name, account);
         }
 
-        if (accounts.isEmpty()) {
-            this.accounts = Collections.emptyMap();
-            this.defaultAccountName = null;
-        } else {
-            this.accounts = accounts;
         String defaultAccountName = settings.get("default_account");
         if (defaultAccountName == null) {
+            if (accounts.isEmpty()) {
+                this.defaultAccountName = null;
+            } else {
                 Account account = accounts.values().iterator().next();
-                logger.error("default account set to [{}]", account.name());
+                logger.info("default account set to [{}]", account.name());
                 this.defaultAccountName = account.name();
+            }
         } else if (!accounts.containsKey(defaultAccountName)) {
-                Account account = accounts.values().iterator().next();
+            throw new EmailSettingsException("could not fine default account [" + defaultAccountName + "]");
-                this.defaultAccountName = account.name();
-                logger.error("could not find configured default account [{}]. falling back on account [{}]", defaultAccountName, account.name());
         } else {
             this.defaultAccountName = defaultAccountName;
         }
     }
-    }
 
     /**
      * Returns the account associated with the given name. If there is not such account, {@code null} is returned.

src/main/java/org/elasticsearch/alerts/actions/email/service/Attachment.java

@@ -6,7 +6,6 @@
 package org.elasticsearch.alerts.actions.email.service;
 
 import org.elasticsearch.alerts.actions.email.service.support.BodyPartSource;
-import org.elasticsearch.common.base.Charsets;
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentType;
@@ -26,16 +25,8 @@ import java.nio.file.Path;
 */
 public abstract class Attachment extends BodyPartSource {
 
-    public Attachment(String id) {
+    protected Attachment(String id, String name, String contentType) {
-        super(id);
+        super(id, name, contentType);
-    }
-
-    public Attachment(String id, String name) {
-        super(id, name);
-    }
-
-    public Attachment(String id, String name, String description) {
-        super(id, name, description);
     }
 
     @Override
@@ -43,68 +34,12 @@ public abstract class Attachment extends BodyPartSource {
         MimeBodyPart part = new MimeBodyPart();
         part.setContentID(id);
         part.setFileName(name);
-        part.setDescription(description, Charsets.UTF_8.name());
         part.setDisposition(Part.ATTACHMENT);
         writeTo(part);
         return part;
     }
 
-    protected abstract void writeTo(MimeBodyPart part) throws MessagingException;
+    public abstract String type();
-
-    public static class File extends Attachment {
-
-        static final String TYPE = "file";
-
-        private final Path path;
-        private final DataSource dataSource;
-        private final String contentType;
-
-        public File(String id, Path path) {
-            this(id, path.getFileName().toString(), path);
-        }
-
-        public File(String id, Path path, String contentType) {
-            this(id, path.getFileName().toString(), path, contentType);
-        }
-
-        public File(String id, String name, Path path) {
-            this(id, name, name, path);
-        }
-        public File(String id, String name, Path path, String contentType) {
-            this(id, name, name, path, contentType);
-        }
-
-        public File(String id, String name, String description, Path path) {
-            this(id, name, description, path, null);
-        }
-
-        public File(String id, String name, String description, Path path, String contentType) {
-            super(id, name, description);
-            this.path = path;
-            this.dataSource = new FileDataSource(path.toFile());
-            this.contentType = contentType;
-        }
-
-        public Path path() {
-            return path;
-        }
-
-        public String type() {
-            return TYPE;
-        }
-
-        String contentType() {
-            return contentType != null ? contentType : dataSource.getContentType();
-        }
-
-        @Override
-        public void writeTo(MimeBodyPart part) throws MessagingException {
-            DataSource dataSource = new FileDataSource(path.toFile());
-            DataHandler handler = contentType != null ?
-                    new DataHandler(dataSource, contentType) :
-                    new DataHandler(dataSource);
-            part.setDataHandler(handler);
-        }
 
     /**
      * intentionally not emitting path as it may come as an information leak
@@ -115,10 +50,49 @@ public abstract class Attachment extends BodyPartSource {
                 .field("type", type())
                 .field("id", id)
                 .field("name", name)
-                    .field("description", description)
+                .field("content_type", contentType)
-                    .field("content_type", contentType())
                 .endObject();
     }
+
+    protected abstract void writeTo(MimeBodyPart part) throws MessagingException;
+
+    public static class File extends Attachment {
+
+        static final String TYPE = "file";
+
+        private final Path path;
+        private final DataSource dataSource;
+
+        public File(String id, Path path) {
+            this(id, path.getFileName().toString(), path);
+        }
+
+        public File(String id, Path path, String contentType) {
+            this(id, path.getFileName().toString(), path, contentType);
+        }
+
+        public File(String id, String name, Path path) {
+            this(id, name, path, fileTypeMap.getContentType(path.toFile()));
+        }
+
+        public File(String id, String name, Path path, String contentType) {
+            super(id, name, contentType);
+            this.path = path;
+            this.dataSource = new FileDataSource(path.toFile());
+        }
+
+        public Path path() {
+            return path;
+        }
+
+        public String type() {
+            return TYPE;
+        }
+
+        @Override
+        public void writeTo(MimeBodyPart part) throws MessagingException {
+            part.setDataHandler(new DataHandler(dataSource));
+        }
     }
 
     public static class Bytes extends Attachment {
@@ -126,20 +100,18 @@ public abstract class Attachment extends BodyPartSource {
         static final String TYPE = "bytes";
 
         private final byte[] bytes;
-        private final String contentType;
 
         public Bytes(String id, byte[] bytes, String contentType) {
             this(id, id, bytes, contentType);
         }
 
-        public Bytes(String id, String name, byte[] bytes, String contentType) {
+        public Bytes(String id, String name, byte[] bytes) {
-            this(id, name, name, bytes, contentType);
+            this(id, name, bytes, fileTypeMap.getContentType(name));
         }
 
-        public Bytes(String id, String name, String description, byte[] bytes, String contentType) {
+        public Bytes(String id, String name, byte[] bytes, String contentType) {
-            super(id, name, description);
+            super(id, name, contentType);
             this.bytes = bytes;
-            this.contentType = contentType;
         }
 
         public String type() {
@@ -150,27 +122,12 @@ public abstract class Attachment extends BodyPartSource {
             return bytes;
         }
 
-        public String contentType() {
-            return contentType;
-        }
-
         @Override
         public void writeTo(MimeBodyPart part) throws MessagingException {
             DataSource dataSource = new ByteArrayDataSource(bytes, contentType);
             DataHandler handler = new DataHandler(dataSource);
             part.setDataHandler(handler);
         }
-
-        @Override
-        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-            return builder.startObject()
-                    .field("type", type())
-                    .field("id", id)
-                    .field("name", name)
-                    .field("description", description)
-                    .field("content_type", contentType)
-                    .endObject();
-        }
     }
 
     public static class XContent extends Bytes {
@@ -180,11 +137,7 @@ public abstract class Attachment extends BodyPartSource {
         }
 
         protected XContent(String id, String name, ToXContent content, XContentType type) {
-            this(id, name, name, content, type);
+            super(id, name, bytes(name, content, type), mimeType(type));
-        }
-
-        protected XContent(String id, String name, String description, ToXContent content, XContentType type) {
-            super(id, name, description, bytes(name, content, type), mimeType(type));
         }
 
         static String mimeType(XContentType type) {
@@ -202,7 +155,7 @@ public abstract class Attachment extends BodyPartSource {
             try {
                 XContentBuilder builder = XContentBuilder.builder(type.xContent());
                 content.toXContent(builder, ToXContent.EMPTY_PARAMS);
-                return builder.bytes().array();
+                return builder.bytes().toBytes();
             } catch (IOException ioe) {
                 throw new EmailException("could not create an xcontent attachment [" + name + "]", ioe);
             }
@@ -218,10 +171,6 @@ public abstract class Attachment extends BodyPartSource {
                 super(id, name, content, XContentType.YAML);
             }
 
-            public Yaml(String id, String name, String description, ToXContent content) {
-                super(id, name, description, content, XContentType.YAML);
-            }
-
             @Override
             public String type() {
                 return "yaml";
@@ -238,10 +187,6 @@ public abstract class Attachment extends BodyPartSource {
                 super(id, name, content, XContentType.JSON);
             }
 
-            public Json(String id, String name, String description, ToXContent content) {
-                super(id, name, description, content, XContentType.JSON);
-            }
-
             @Override
             public String type() {
                 return "json";

src/main/java/org/elasticsearch/alerts/actions/email/service/Authentication.java

@@ -10,19 +10,39 @@ package org.elasticsearch.alerts.actions.email.service;
 */
 public class Authentication {
 
-    private final String username;
+    private final String user;
     private final String password;
 
-    public Authentication(String username, String password) {
+    public Authentication(String user, String password) {
-        this.username = username;
+        this.user = user;
         this.password = password;
     }
 
-    public String username() {
+    public String user() {
-        return username;
+        return user;
     }
 
     public String password() {
         return password;
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+
+        Authentication that = (Authentication) o;
+
+        if (password != null ? !password.equals(that.password) : that.password != null) return false;
+        if (user != null ? !user.equals(that.user) : that.user != null) return false;
+
+        return true;
+    }
+
+    @Override
+    public int hashCode() {
+        int result = user != null ? user.hashCode() : 0;
+        result = 31 * result + (password != null ? password.hashCode() : 0);
+        return result;
+    }
 }

src/main/java/org/elasticsearch/alerts/actions/email/service/Email.java

@@ -20,16 +20,14 @@ import javax.mail.internet.InternetAddress;
 import javax.mail.internet.MimeMessage;
 import java.io.IOException;
 import java.io.UnsupportedEncodingException;
-import java.util.ArrayList;
+import java.util.*;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Locale;
 
 /**
  *
  */
 public class Email implements ToXContent {
 
+    public static final ParseField ID_FIELD = new ParseField("id");
     public static final ParseField FROM_FIELD = new ParseField("from");
     public static final ParseField REPLY_TO_FIELD = new ParseField("reply_to");
     public static final ParseField PRIORITY_FIELD = new ParseField("priority");
@@ -40,8 +38,6 @@ public class Email implements ToXContent {
     public static final ParseField SUBJECT_FIELD = new ParseField("subject");
     public static final ParseField TEXT_BODY_FIELD = new ParseField("text_body");
     public static final ParseField HTML_BODY_FIELD = new ParseField("html_body");
-    public static final ParseField ATTACHMENTS_FIELD = new ParseField("attachments");
-    public static final ParseField INLINES_FIELD = new ParseField("inlines");
 
     final String id;
     final Address from;
@@ -65,7 +61,7 @@ public class Email implements ToXContent {
         this.from = from;
         this.replyTo = replyTo;
         this.priority = priority;
-        this.sentDate = sentDate;
+        this.sentDate = sentDate != null ? sentDate : new DateTime();
         this.to = to;
         this.cc = cc;
         this.bcc = bcc;
@@ -76,6 +72,10 @@ public class Email implements ToXContent {
         this.inlines = inlines;
     }
 
+    public String id() {
+        return id;
+    }
+
     public Address from() {
         return from;
     }
@@ -126,20 +126,46 @@ public class Email implements ToXContent {
 
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        return builder.startObject()
+        builder.startObject();
-                .field(FROM_FIELD.getPreferredName(), from)
+        builder.field(ID_FIELD.getPreferredName(), id);
-                .field(REPLY_TO_FIELD.getPreferredName(), (ToXContent) replyTo)
+        builder.field(FROM_FIELD.getPreferredName(), from);
-                .field(PRIORITY_FIELD.getPreferredName(), priority)
+        if (replyTo != null) {
-                .field(SENT_DATE_FIELD.getPreferredName(), sentDate)
+            builder.field(REPLY_TO_FIELD.getPreferredName(), (ToXContent) replyTo);
-                .field(TO_FIELD.getPreferredName(), (ToXContent) to)
+        }
-                .field(CC_FIELD.getPreferredName(), (ToXContent) cc)
+        if (priority != null) {
-                .field(BCC_FIELD.getPreferredName(), (ToXContent) bcc)
+            builder.field(PRIORITY_FIELD.getPreferredName(), priority);
-                .field(SUBJECT_FIELD.getPreferredName(), subject)
+        }
-                .field(TEXT_BODY_FIELD.getPreferredName(), textBody)
+        builder.field(SENT_DATE_FIELD.getPreferredName(), sentDate);
-                .field(HTML_BODY_FIELD.getPreferredName(), htmlBody)
+        builder.field(TO_FIELD.getPreferredName(), (ToXContent) to);
-                .field(ATTACHMENTS_FIELD.getPreferredName(), attachments)
+        if (cc != null) {
-                .field(INLINES_FIELD.getPreferredName(), inlines)
+            builder.field(CC_FIELD.getPreferredName(), (ToXContent) cc);
-                .endObject();
+        }
+        if (bcc != null) {
+            builder.field(BCC_FIELD.getPreferredName(), (ToXContent) bcc);
+        }
+        builder.field(SUBJECT_FIELD.getPreferredName(), subject);
+        builder.field(TEXT_BODY_FIELD.getPreferredName(), textBody);
+        if (htmlBody != null) {
+            builder.field(HTML_BODY_FIELD.getPreferredName(), htmlBody);
+        }
+        return builder.endObject();
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+
+        Email email = (Email) o;
+
+        if (!id.equals(email.id)) return false;
+
+        return true;
+    }
+
+    @Override
+    public int hashCode() {
+        return id.hashCode();
     }
 
     public static Builder builder() {
@@ -154,7 +180,9 @@ public class Email implements ToXContent {
             if (token == XContentParser.Token.FIELD_NAME) {
                 currentFieldName = parser.currentName();
             } else if ((token.isValue() || token == XContentParser.Token.START_OBJECT || token == XContentParser.Token.START_ARRAY) && currentFieldName != null) {
-                if (FROM_FIELD.match(currentFieldName)) {
+                if (ID_FIELD.match(currentFieldName)) {
+                    email.id(parser.text());
+                } else if (FROM_FIELD.match(currentFieldName)) {
                     email.from(Address.parse(currentFieldName, token, parser));
                 } else if (REPLY_TO_FIELD.match(currentFieldName)) {
                     email.replyTo(AddressList.parse(currentFieldName, token, parser));
@@ -174,10 +202,6 @@ public class Email implements ToXContent {
                     email.textBody(parser.text());
                 } else if (HTML_BODY_FIELD.match(currentFieldName)) {
                     email.htmlBody(parser.text());
-                } else if (ATTACHMENTS_FIELD.match(currentFieldName)) {
-                    //@TODO handle this
-                } else if (INLINES_FIELD.match(currentFieldName)) {
-                    //@TODO handle this
                 } else {
                     throw new EmailException("could not parse email. unrecognized field [" + currentFieldName + "]");
                 }
@@ -293,7 +317,6 @@ public class Email implements ToXContent {
 
         public Email build() {
             assert id != null : "email id should not be null (should be set to the alert id";
-            assert to != null && !to.isEmpty() : "email must have a [to] recipient";
             return new Email(id, from, replyTo, priority, sentDate, to, cc, bcc, subject, textBody, htmlBody, attachments.build(), inlines.build());
         }
 
@@ -427,6 +450,8 @@ public class Email implements ToXContent {
 
     public static class AddressList implements Iterable<Address>, ToXContent {
 
+        public static final AddressList EMPTY = new AddressList(Collections.<Address>emptyList());
+
         private final List<Address> addresses;
 
         public AddressList(List<Address> addresses) {
@@ -446,6 +471,10 @@ public class Email implements ToXContent {
             return addresses.toArray(new Address[addresses.size()]);
         }
 
+        public int size() {
+            return addresses.size();
+        }
+
         @Override
         public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
             builder.startArray();
@@ -499,6 +528,23 @@ public class Email implements ToXContent {
             throw new EmailException("could not parse [" + field + "] as address list. field must either be a string " +
                     "(comma-separated list of RFC822 encoded addresses) or an array of objects representing addresses");
         }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+
+            AddressList addresses1 = (AddressList) o;
+
+            if (!addresses.equals(addresses1.addresses)) return false;
+
+            return true;
+        }
+
+        @Override
+        public int hashCode() {
+            return addresses.hashCode();
+        }
     }
 
 }

src/main/java/org/elasticsearch/alerts/actions/email/service/EmailService.java

@@ -5,17 +5,11 @@
  */
 package org.elasticsearch.alerts.actions.email.service;
 
-import org.elasticsearch.cluster.ClusterState;
-
 /**
  *
  */
 public interface EmailService {
 
-    void start(ClusterState state);
-
-    void stop();
-
     EmailSent send(Email email, Authentication auth, Profile profile);
 
     EmailSent send(Email email, Authentication auth, Profile profile, String accountName);

src/main/java/org/elasticsearch/alerts/actions/email/service/Inline.java

@@ -6,6 +6,9 @@
 package org.elasticsearch.alerts.actions.email.service;
 
 import org.elasticsearch.alerts.actions.email.service.support.BodyPartSource;
+import org.elasticsearch.common.bytes.BytesArray;
+import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.inject.Provider;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 
 import javax.activation.DataHandler;
@@ -14,7 +17,10 @@ import javax.activation.FileDataSource;
 import javax.mail.MessagingException;
 import javax.mail.Part;
 import javax.mail.internet.MimeBodyPart;
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
 import java.nio.file.Path;
 
 /**
@@ -22,82 +28,22 @@ import java.nio.file.Path;
  */
 public abstract class Inline extends BodyPartSource {
 
-    public Inline(String id) {
+    protected Inline(String id, String name, String contentType) {
-        super(id);
+        super(id, name, contentType);
     }
 
-    public Inline(String id, String name) {
+    public abstract String type();
-        super(id, name);
-    }
-
-    public Inline(String id, String name, String description) {
-        super(id, name, description);
-    }
 
     @Override
     public final MimeBodyPart bodyPart() throws MessagingException {
         MimeBodyPart part = new MimeBodyPart();
         part.setDisposition(Part.INLINE);
+        part.setContentID(id);
+        part.setFileName(name);
         writeTo(part);
         return part;
     }
 
-    protected abstract void writeTo(MimeBodyPart part) throws MessagingException;
-
-    public static class File extends Inline {
-
-        static final String TYPE = "file";
-
-        private final Path path;
-        private DataSource dataSource;
-        private final String contentType;
-
-        public File(String id, Path path) {
-            this(id, path.getFileName().toString(), path);
-        }
-
-        public File(String id, Path path, String contentType) {
-            this(id, path.getFileName().toString(), path, contentType);
-        }
-
-        public File(String id, String name, Path path) {
-            this(id, name, name, path);
-        }
-        public File(String id, String name, Path path, String contentType) {
-            this(id, name, name, path, contentType);
-        }
-
-        public File(String id, String name, String description, Path path) {
-            this(id, name, description, path, null);
-        }
-
-        public File(String id, String name, String description, Path path, String contentType) {
-            super(id, name, description);
-            this.path = path;
-            this.dataSource = new FileDataSource(path.toFile());
-            this.contentType = contentType;
-        }
-
-        public Path path() {
-            return path;
-        }
-
-        public String type() {
-            return TYPE;
-        }
-
-        String contentType() {
-            return contentType != null ? contentType : dataSource.getContentType();
-        }
-
-        @Override
-        public void writeTo(MimeBodyPart part) throws MessagingException {
-            DataHandler handler = contentType != null ?
-                    new DataHandler(dataSource, contentType) :
-                    new DataHandler(dataSource);
-            part.setDataHandler(handler);
-        }
-
     /**
      * intentionally not emitting path as it may come as an information leak
      */
@@ -107,9 +53,134 @@ public abstract class Inline extends BodyPartSource {
                 .field("type", type())
                 .field("id", id)
                 .field("name", name)
-                    .field("description", description)
+                .field("content_type", contentType)
-                    .field("content_type", contentType())
                 .endObject();
     }
+
+    protected abstract void writeTo(MimeBodyPart part) throws MessagingException;
+
+    public static class File extends Inline {
+
+        static final String TYPE = "file";
+
+        private final Path path;
+        private DataSource dataSource;
+
+        public File(String id, Path path) {
+            this(id, path.getFileName().toString(), path);
+        }
+
+        public File(String id, String name, Path path) {
+            this(id, name, path, fileTypeMap.getContentType(path.toFile()));
+        }
+
+        public File(String id, String name, Path path, String contentType) {
+            super(id, name, contentType);
+            this.path = path;
+            this.dataSource = new FileDataSource(path.toFile());
+        }
+
+        public Path path() {
+            return path;
+        }
+
+        public String type() {
+            return TYPE;
+        }
+
+        @Override
+        public void writeTo(MimeBodyPart part) throws MessagingException {
+            part.setDataHandler(new DataHandler(dataSource, contentType));
+        }
+    }
+
+    public static class Stream extends Inline {
+
+        static final String TYPE = "stream";
+
+        private final Provider<InputStream> source;
+
+        public Stream(String id, String name, Provider<InputStream> source) {
+            this(id, name, fileTypeMap.getContentType(name), source);
+        }
+
+        public Stream(String id, String name, String contentType, Provider<InputStream> source) {
+            super(id, name, contentType);
+            this.source = source;
+        }
+
+        @Override
+        public String type() {
+            return TYPE;
+        }
+
+        @Override
+        protected void writeTo(MimeBodyPart part) throws MessagingException {
+            DataSource ds = new StreamDataSource(name, contentType, source);
+            DataHandler dh = new DataHandler(ds);
+            part.setDataHandler(dh);
+        }
+
+        static class StreamDataSource implements DataSource {
+
+            private final String name;
+            private final String contentType;
+            private final Provider<InputStream> source;
+
+            public StreamDataSource(String name, String contentType, Provider<InputStream> source) {
+                this.name = name;
+                this.contentType = contentType;
+                this.source = source;
+            }
+
+            @Override
+            public InputStream getInputStream() throws IOException {
+                return source.get();
+            }
+
+            @Override
+            public OutputStream getOutputStream() throws IOException {
+                throw new UnsupportedOperationException();
+            }
+
+            @Override
+            public String getContentType() {
+                return contentType;
+            }
+
+            @Override
+            public String getName() {
+                return name;
+            }
+        }
+    }
+
+    public static class Bytes extends Stream {
+
+        public Bytes(String id, String name, String contentType, byte[] bytes) {
+            super(id, name, contentType, new BytesStreamProvider(bytes));
+        }
+
+        public Bytes(String id, String name, String contentType, BytesReference bytes) {
+            super(id, name, contentType, new BytesStreamProvider(bytes));
+        }
+
+        static class BytesStreamProvider implements Provider<InputStream> {
+
+            private final BytesReference bytes;
+
+            public BytesStreamProvider(byte[] bytes) {
+                this(new BytesArray(bytes));
+            }
+
+            public BytesStreamProvider(BytesReference bytes) {
+                this.bytes = bytes;
+            }
+
+            @Override
+            public InputStream get() {
+                return new ByteArrayInputStream(bytes.array(), bytes.arrayOffset(), bytes.length());
+            }
+        }
     }
 }

src/main/java/org/elasticsearch/alerts/actions/email/service/InternalEmailService.java

@@ -5,25 +5,23 @@
  */
 package org.elasticsearch.alerts.actions.email.service;
 
-import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.ElasticsearchException;
-import org.elasticsearch.common.component.AbstractComponent;
+import org.elasticsearch.common.component.AbstractLifecycleComponent;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.logging.ESLogger;
 import org.elasticsearch.common.settings.ImmutableSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.node.settings.NodeSettingsService;
 
 import javax.mail.MessagingException;
-import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  *
  */
-public class InternalEmailService extends AbstractComponent implements EmailService {
+public class InternalEmailService extends AbstractLifecycleComponent<InternalEmailService> implements EmailService {
 
     private volatile Accounts accounts;
 
-    private final AtomicBoolean started = new AtomicBoolean(false);
-
     @Inject
     public InternalEmailService(Settings settings, NodeSettingsService nodeSettingsService) {
         super(settings);
@@ -35,6 +33,19 @@ public class InternalEmailService extends AbstractComponent implements EmailServ
         });
     }
 
+    @Override
+    protected void doStart() throws ElasticsearchException {
+        reset(settings);
+    }
+
+    @Override
+    protected void doStop() throws ElasticsearchException {
+    }
+
+    @Override
+    protected void doClose() throws ElasticsearchException {
+    }
+
     @Override
     public EmailSent send(Email email, Authentication auth, Profile profile) {
         return send(email, auth, profile, (String) null);
@@ -59,29 +70,16 @@ public class InternalEmailService extends AbstractComponent implements EmailServ
         return new EmailSent(account.name(), email);
     }
 
-    @Override
+    void reset(Settings nodeSettings) {
-    public synchronized void start(ClusterState state) {
-        if (started.get()) {
-            return;
-        }
-        reset(state.metaData().settings());
-        started.set(true);
-    }
-
-    @Override
-    public synchronized void stop() {
-        started.set(false);
-    }
-
-    synchronized void reset(Settings nodeSettings) {
-        if (!started.get()) {
-            return;
-        }
         Settings settings = ImmutableSettings.builder()
                 .put(componentSettings)
                 .put(nodeSettings.getComponentSettings(InternalEmailService.class))
                 .build();
-        accounts = new Accounts(settings, logger);
+        accounts = createAccounts(settings, logger);
+    }
+
+    protected Accounts createAccounts(Settings settings, ESLogger logger) {
+        return new Accounts(settings, logger);
     }
 
 }

src/main/java/org/elasticsearch/alerts/actions/email/service/Profile.java

@@ -16,7 +16,6 @@ import javax.mail.internet.MimeBodyPart;
 import javax.mail.internet.MimeMessage;
 import javax.mail.internet.MimeMultipart;
 import java.io.IOException;
-import java.util.Date;
 import java.util.Locale;
 
 /**
@@ -26,11 +25,51 @@ import java.util.Locale;
 public enum Profile implements ToXContent {
 
     STANDARD() {
+
+        @Override
+        public String textBody(MimeMessage msg) throws IOException, MessagingException {
+            MimeMultipart mixed = (MimeMultipart) msg.getContent();
+            MimeMultipart related = null;
+            for (int i = 0; i < mixed.getCount(); i++) {
+                MimeBodyPart part = (MimeBodyPart) mixed.getBodyPart(i);
+
+                if (part.getContentType().startsWith("multipart/related")) {
+                    related = (MimeMultipart) part.getContent();
+                    break;
+                }
+            }
+            if (related == null) {
+                throw new EmailException("could not extract body text from mime message");
+            }
+
+            MimeMultipart alternative = null;
+            for (int i = 0; i < related.getCount(); i++) {
+                MimeBodyPart part = (MimeBodyPart) related.getBodyPart(i);
+                if (part.getContentType().startsWith("multipart/alternative")) {
+                    alternative = (MimeMultipart) part.getContent();
+                    break;
+                }
+            }
+            if (alternative == null) {
+                throw new EmailException("could not extract body text from mime message");
+            }
+
+            for (int i = 0; i < alternative.getCount(); i++) {
+                MimeBodyPart part = (MimeBodyPart) alternative.getBodyPart(i);
+                if (part.getContentType().startsWith("text/plain")) {
+                    return (String) part.getContent();
+                }
+            }
+
+            throw new EmailException("could not extract body text from mime message");
+        }
+
         @Override
         public MimeMessage toMimeMessage(Email email, Session session) throws MessagingException {
             MimeMessage message = createCommon(email, session);
 
             MimeMultipart mixed = new MimeMultipart("mixed");
+            message.setContent(mixed);
 
             MimeMultipart related = new MimeMultipart("related");
             mixed.addBodyPart(wrap(related, null));
@@ -39,12 +78,16 @@ public enum Profile implements ToXContent {
             related.addBodyPart(wrap(alternative, "text/alternative"));
 
             MimeBodyPart text = new MimeBodyPart();
+            if (email.textBody != null) {
                 text.setText(email.textBody, Charsets.UTF_8.name());
+            } else {
+                text.setText("", Charsets.UTF_8.name());
+            }
             alternative.addBodyPart(text);
 
             if (email.htmlBody != null) {
                 MimeBodyPart html = new MimeBodyPart();
-                text.setText(email.textBody, Charsets.UTF_8.name(), "html");
+                html.setText(email.htmlBody, Charsets.UTF_8.name(), "html");
                 alternative.addBodyPart(html);
             }
 
@@ -65,18 +108,36 @@ public enum Profile implements ToXContent {
     },
 
     OUTLOOK() {
+
+        @Override
+        public String textBody(MimeMessage msg) throws IOException, MessagingException {
+            return STANDARD.textBody(msg);
+        }
+
         @Override
         public MimeMessage toMimeMessage(Email email, Session session) throws MessagingException {
             return STANDARD.toMimeMessage(email, session);
         }
     },
     GMAIL() {
+
+        @Override
+        public String textBody(MimeMessage msg) throws IOException, MessagingException {
+            return STANDARD.textBody(msg);
+        }
+
         @Override
         public MimeMessage toMimeMessage(Email email, Session session) throws MessagingException {
             return STANDARD.toMimeMessage(email, session);
         }
     },
     MAC() {
+
+        @Override
+        public String textBody(MimeMessage msg) throws IOException, MessagingException {
+            return STANDARD.textBody(msg);
+        }
+
         @Override
         public MimeMessage toMimeMessage(Email email, Session session) throws MessagingException {
             return STANDARD.toMimeMessage(email, session);
@@ -87,6 +148,8 @@ public enum Profile implements ToXContent {
 
     public abstract MimeMessage toMimeMessage(Email email, Session session) throws MessagingException ;
 
+    public abstract String textBody(MimeMessage msg) throws IOException, MessagingException;
+
     public static Profile resolve(String name) {
         Profile profile = resolve(name, null);
         if (profile == null) {
@@ -104,6 +167,7 @@ public enum Profile implements ToXContent {
             case "standard":    return STANDARD;
             case "outlook":     return OUTLOOK;
             case "gmail":       return GMAIL;
+            case "mac":         return MAC;
             default:
                 return defaultProfile;
         }
@@ -126,14 +190,19 @@ public enum Profile implements ToXContent {
         if (email.priority != null) {
             email.priority.applyTo(message);
         }
-        Date sentDate = email.sentDate != null ? email.sentDate.toDate() : new Date();
+        message.setSentDate(email.sentDate.toDate());
-        message.setSentDate(sentDate);
-
         message.setRecipients(Message.RecipientType.TO, email.to.toArray());
+        if (email.cc != null) {
             message.setRecipients(Message.RecipientType.CC, email.cc.toArray());
+        }
+        if (email.bcc != null) {
             message.setRecipients(Message.RecipientType.BCC, email.bcc.toArray());
-
+        }
+        if (email.subject != null) {
             message.setSubject(email.subject, Charsets.UTF_8.name());
+        } else {
+            message.setSubject("", Charsets.UTF_8.name());
+        }
 
         return message;
     }

src/main/java/org/elasticsearch/alerts/actions/email/service/support/BodyPartSource.java

@@ -7,6 +7,7 @@ package org.elasticsearch.alerts.actions.email.service.support;
 
 import org.elasticsearch.common.xcontent.ToXContent;
 
+import javax.activation.FileTypeMap;
 import javax.mail.MessagingException;
 import javax.mail.internet.MimeBodyPart;
 
@@ -15,22 +16,20 @@ import javax.mail.internet.MimeBodyPart;
  */
 public abstract class BodyPartSource implements ToXContent {
 
+    protected static FileTypeMap fileTypeMap = FileTypeMap.getDefaultFileTypeMap();
+
     protected final String id;
     protected final String name;
-    protected final String description;
+    protected final String contentType;
 
-    public BodyPartSource(String id) {
+    public BodyPartSource(String id, String contentType) {
-        this(id, id);
+        this(id, id, contentType);
     }
 
-    public BodyPartSource(String id, String name) {
+    public BodyPartSource(String id, String name, String contentType) {
-        this(id, name, name);
-    }
-
-    public BodyPartSource(String id, String name, String description) {
         this.id = id;
         this.name = name;
-        this.description = description;
+        this.contentType = contentType;
     }
 
     public String id() {
@@ -41,8 +40,8 @@ public abstract class BodyPartSource implements ToXContent {
         return name;
     }
 
-    public String description() {
+    public String contentType() {
-        return description;
+        return contentType;
     }
 
     public abstract MimeBodyPart bodyPart() throws MessagingException;

src/main/java/org/elasticsearch/alerts/support/template/ScriptTemplate.java

@@ -50,6 +50,10 @@ public class ScriptTemplate implements ToXContent, Template {
         this(service, text, DEFAULT_LANG, ScriptService.ScriptType.INLINE, Collections.<String, Object>emptyMap());
     }
 
+    public ScriptTemplate(ScriptServiceProxy service, String text, String lang, ScriptService.ScriptType type) {
+        this(service, text, lang, type, Collections.<String, Object>emptyMap());
+    }
+
     public ScriptTemplate(ScriptServiceProxy service, String text, String lang, ScriptService.ScriptType type, Map<String, Object> params) {
         this.service = service;
         this.text = text;
@@ -135,8 +139,12 @@ public class ScriptTemplate implements ToXContent, Template {
 
         @Override
         public ScriptTemplate parse(XContentParser parser) throws IOException {
-            assert parser.currentToken() == XContentParser.Token.START_OBJECT : "Expected START_OBJECT, but was " + parser.currentToken();
+            if (parser.currentToken() == XContentParser.Token.VALUE_STRING) {
-
+                return new ScriptTemplate(scriptService, parser.text());
+            }
+            if (parser.currentToken() != XContentParser.Token.START_OBJECT) {
+                throw new ParseException("expected either a string or an object, but found [" + parser.currentToken() + "] instead");
+            }
             String text = null;
             ScriptService.ScriptType type = ScriptService.ScriptType.INLINE;
             String lang = DEFAULT_LANG;

src/test/java/org/elasticsearch/alerts/AbstractAlertingTests.java

@@ -55,7 +55,7 @@ import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.util.*;
 
-import static org.elasticsearch.common.xcontent.XContentFactory.*;
+import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
 import static org.elasticsearch.index.query.QueryBuilders.*;
 import static org.elasticsearch.search.builder.SearchSourceBuilder.searchSource;
 import static org.hamcrest.Matchers.*;
@@ -174,12 +174,12 @@ public abstract class AbstractAlertingTests extends ElasticsearchIntegrationTest
         Email.AddressList to = new Email.AddressList(emailAddressList);
 
 
-        Email.Builder emailBuilder = Email.builder();
+        Email.Builder emailBuilder = Email.builder().id("prototype");
         emailBuilder.from(from);
         emailBuilder.to(to);
 
 
-        EmailAction emailAction = new EmailAction(logger, noopEmailService(), emailBuilder,
+        EmailAction emailAction = new EmailAction(logger, noopEmailService(), emailBuilder.build(),
                 new Authentication("testname", "testpassword"), Profile.STANDARD, "testaccount", body, body, null, true);
 
         actions.add(emailAction);
@@ -471,19 +471,10 @@ public abstract class AbstractAlertingTests extends ElasticsearchIntegrationTest
     }
 
     private static class NoopEmailService implements EmailService {
-        @Override
-        public void start(ClusterState state) {
-
-        }
-
-        @Override
-        public void stop() {
-
-        }
 
         @Override
         public EmailSent send(Email email, Authentication auth, Profile profile) {
-            return new EmailSent(auth.username(), email);
+            return new EmailSent(auth.user(), email);
         }
 
         @Override

src/test/java/org/elasticsearch/alerts/actions/email/EmailActionTest.java

@@ -1,116 +0,0 @@
-/*
- * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
- * or more contributor license agreements. Licensed under the Elastic License;
- * you may not use this file except in compliance with the Elastic License.
- */
-package org.elasticsearch.alerts.actions.email;
-
-import org.elasticsearch.alerts.Alert;
-import org.elasticsearch.alerts.ExecutionContext;
-import org.elasticsearch.alerts.Payload;
-import org.elasticsearch.alerts.actions.email.service.Authentication;
-import org.elasticsearch.alerts.actions.email.service.Email;
-import org.elasticsearch.alerts.actions.email.service.EmailService;
-import org.elasticsearch.alerts.actions.email.service.Profile;
-import org.elasticsearch.alerts.scheduler.schedule.CronSchedule;
-import org.elasticsearch.alerts.support.init.proxy.ScriptServiceProxy;
-import org.elasticsearch.alerts.support.template.ScriptTemplate;
-import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.common.joda.time.DateTime;
-import org.elasticsearch.common.settings.ImmutableSettings;
-import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.unit.TimeValue;
-import org.elasticsearch.env.Environment;
-import org.elasticsearch.script.ScriptEngineService;
-import org.elasticsearch.script.ScriptService;
-import org.elasticsearch.script.mustache.MustacheScriptEngineService;
-import org.elasticsearch.test.ElasticsearchTestCase;
-import org.elasticsearch.threadpool.ThreadPool;
-import org.elasticsearch.watcher.ResourceWatcherService;
-
-import javax.mail.MessagingException;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-
-/**
-*/
-public class EmailActionTest extends ElasticsearchTestCase {
-
-    public void testEmailTemplateRender() throws IOException, MessagingException {
-
-        Settings settings = ImmutableSettings.settingsBuilder().build();
-        MustacheScriptEngineService mustacheScriptEngineService = new MustacheScriptEngineService(settings);
-        ThreadPool threadPool = new ThreadPool(ThreadPool.Names.SAME);
-        Set<ScriptEngineService> engineServiceSet = new HashSet<>();
-        engineServiceSet.add(mustacheScriptEngineService);
-
-        ScriptServiceProxy scriptService = ScriptServiceProxy.of(new ScriptService(settings, new Environment(), engineServiceSet, new ResourceWatcherService(settings, threadPool)));
-
-        ScriptTemplate template = new ScriptTemplate(scriptService, "{{alert_name}} executed with {{response.hits.total}} hits");
-
-        EmailService emailService = new EmailServiceMock();
-
-        Email.Address from = new Email.Address("from@test.com");
-        List<Email.Address> emailAddressList = new ArrayList<>();
-        emailAddressList.add(new Email.Address("to@test.com"));
-        Email.AddressList to = new Email.AddressList(emailAddressList);
-
-
-        Email.Builder emailBuilder = Email.builder();
-        emailBuilder.from(from);
-        emailBuilder.to(to);
-
-        EmailAction emailAction = new EmailAction(logger, emailService, emailBuilder,
-                new Authentication("testname", "testpassword"), Profile.STANDARD, "testaccount", template, template, null, true);
-
-        //This is ok since the execution of the action only relies on the alert name
-        Alert alert = new Alert(
-                "test-serialization",
-                new CronSchedule("0/5 * * * * ? *"),
-                null,
-                null,
-                new TimeValue(0),
-                null,
-                null,
-                new Alert.Status()
-        );
-        ExecutionContext ctx = new ExecutionContext("test-serialization#1", alert, new DateTime(), new DateTime());
-        EmailAction.Result result = emailAction.execute(ctx, new Payload.Simple());
-
-        threadPool.shutdownNow();
-        
-        assertTrue(result.success());
-
-        EmailAction.Result.Success success = (EmailAction.Result.Success) result;
-        assertEquals(success.account(), "testaccount");
-        assertArrayEquals(success.email().to().toArray(), to.toArray() );
-        assertEquals(success.email().from(), from);
-        //@TODO add more here
-    }
-
-    static class EmailServiceMock implements EmailService {
-        @Override
-        public void start(ClusterState state) {
-
-        }
-
-        @Override
-        public void stop() {
-
-        }
-
-        @Override
-        public EmailSent send(Email email, Authentication auth, Profile profile) {
-            return new EmailSent(auth.username(), email);
-        }
-
-        @Override
-        public EmailSent send(Email email, Authentication auth, Profile profile, String accountName) {
-            return new EmailSent(accountName, email);
-        }
-    }
-
-}

src/test/java/org/elasticsearch/alerts/actions/email/EmailActionTests.java

@@ -0,0 +1,365 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.alerts.actions.email;
+
+import com.carrotsearch.randomizedtesting.annotations.Repeat;
+import org.elasticsearch.alerts.Alert;
+import org.elasticsearch.alerts.ExecutionContext;
+import org.elasticsearch.alerts.Payload;
+import org.elasticsearch.alerts.actions.ActionSettingsException;
+import org.elasticsearch.alerts.actions.email.service.*;
+import org.elasticsearch.alerts.support.init.proxy.ScriptServiceProxy;
+import org.elasticsearch.alerts.support.template.ScriptTemplate;
+import org.elasticsearch.alerts.support.template.Template;
+import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.collect.ImmutableMap;
+import org.elasticsearch.common.joda.time.DateTime;
+import org.elasticsearch.common.settings.ImmutableSettings;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.common.xcontent.json.JsonXContent;
+import org.elasticsearch.test.ElasticsearchTestCase;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
+import static org.hamcrest.Matchers.*;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ *
+ */
+public class EmailActionTests extends ElasticsearchTestCase {
+
+    @Test @Repeat(iterations = 20)
+    public void testExecute() throws Exception {
+        final String account = "account1";
+        EmailService service = new EmailService() {
+            @Override
+            public EmailSent send(Email email, Authentication auth, Profile profile) {
+                return new EmailSent(account, email);
+            }
+
+            @Override
+            public EmailSent send(Email email, Authentication auth, Profile profile, String accountName) {
+                return new EmailSent(account, email);
+            }
+        };
+        Email email = Email.builder().id("prototype").build();
+        Authentication auth = new Authentication("user", "passwd");
+        Profile profile = randomFrom(Profile.values());
+
+        Template subject = mock(Template.class);
+        Template textBody = mock(Template.class);
+        Template htmlBody = randomBoolean() ? null : mock(Template.class);
+        boolean attachPayload = randomBoolean();
+        EmailAction action = new EmailAction(logger, service, email, auth, profile, account, subject, textBody, htmlBody, attachPayload);
+
+        final Map<String, Object> data = new HashMap<>();
+        Payload payload = new Payload() {
+            @Override
+            public Map<String, Object> data() {
+                return data;
+            }
+
+            @Override
+            public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+                return builder.map(data);
+            }
+        };
+
+        String ctxId = randomAsciiOfLength(5);
+        ExecutionContext ctx = mock(ExecutionContext.class);
+        when(ctx.id()).thenReturn(ctxId);
+        Alert alert = mock(Alert.class);
+        when(alert.name()).thenReturn("alert1");
+        when(ctx.alert()).thenReturn(alert);
+
+        Map<String, Object> expectedModel = ImmutableMap.<String, Object>builder()
+                .put("alert_name", "alert1")
+                .put("payload", data)
+                .build();
+
+        when(subject.render(expectedModel)).thenReturn("_subject");
+        when(textBody.render(expectedModel)).thenReturn("_text_body");
+        if (htmlBody != null) {
+            when (htmlBody.render(expectedModel)).thenReturn("_html_body");
+        }
+
+        EmailAction.Result result = action.execute(ctx, payload);
+
+        assertThat(result, notNullValue());
+        assertThat(result, instanceOf(EmailAction.Result.Success.class));
+        assertThat(((EmailAction.Result.Success) result).account(), equalTo(account));
+        Email actualEmail = ((EmailAction.Result.Success) result).email();
+        assertThat(actualEmail.id(), is(ctxId));
+        assertThat(actualEmail, notNullValue());
+        assertThat(actualEmail.subject(), is("_subject"));
+        assertThat(actualEmail.textBody(), is("_text_body"));
+        if (htmlBody != null) {
+            assertThat(actualEmail.htmlBody(), is("_html_body"));
+        }
+        if (attachPayload) {
+            assertThat(actualEmail.attachments(), hasKey("payload"));
+        }
+    }
+
+    @Test @Repeat(iterations = 20)
+    public void testParser() throws Exception {
+        ScriptServiceProxy scriptService = mock(ScriptServiceProxy.class);
+        EmailService emailService = mock(EmailService.class);
+        Profile profile = randomFrom(Profile.values());
+        Email.Priority priority = randomFrom(Email.Priority.values());
+        Email.Address[] to = rarely() ? null : Email.AddressList.parse(randomBoolean() ? "to@domain" : "to1@domain,to2@domain").toArray();
+        Email.Address[] cc = rarely() ? null : Email.AddressList.parse(randomBoolean() ? "cc@domain" : "cc1@domain,cc2@domain").toArray();
+        Email.Address[] bcc = rarely() ? null : Email.AddressList.parse(randomBoolean() ? "bcc@domain" : "bcc1@domain,bcc2@domain").toArray();
+        Email.Address[] replyTo = rarely() ? null : Email.AddressList.parse(randomBoolean() ? "reply@domain" : "reply1@domain,reply2@domain").toArray();
+        ScriptTemplate subject = randomBoolean() ? new ScriptTemplate(scriptService, "_subject") : null;
+        ScriptTemplate textBody = randomBoolean() ? new ScriptTemplate(scriptService, "_text_body") : null;
+        ScriptTemplate htmlBody = randomBoolean() ? new ScriptTemplate(scriptService, "_text_html") : null;
+        boolean attachPayload = randomBoolean();
+        XContentBuilder builder = jsonBuilder().startObject()
+                .field("account", "_account")
+                .field("profile", profile.name())
+                .field("user", "_user")
+                .field("password", "_passwd")
+                .field("attach_payload", attachPayload)
+                .field("from", "from@domain")
+                .field("priority", priority.name());
+        if (to != null) {
+            if (to.length == 1) {
+                builder.field("to", to[0]);
+            } else {
+                builder.array("to", (Object[]) to);
+            }
+        }
+        if (cc != null) {
+            if (cc.length == 1) {
+                builder.field("cc", cc[0]);
+            } else {
+                builder.array("cc", (Object[]) cc);
+            }
+        }
+        if (bcc != null) {
+            if (bcc.length == 1) {
+                builder.field("bcc", bcc[0]);
+            } else {
+                builder.array("bcc", (Object[]) bcc);
+            }
+        }
+        if (replyTo != null) {
+            if (replyTo.length == 1) {
+                builder.field("reply_to", replyTo[0]);
+            } else {
+                builder.array("reply_to", (Object[]) replyTo);
+            }
+        }
+        if (subject != null) {
+            if (randomBoolean()) {
+                builder.field("subject", subject.text());
+            } else {
+                builder.field("subject", subject);
+            }
+        }
+        if (textBody != null) {
+            if (randomBoolean()) {
+                builder.field("text_body", textBody.text());
+            } else {
+                builder.field("text_body", textBody);
+            }
+        }
+        if (htmlBody != null) {
+            if (randomBoolean()) {
+                builder.field("html_body", htmlBody.text());
+            } else {
+                builder.field("html_body", htmlBody);
+            }
+        }
+        BytesReference bytes = builder.bytes();
+        XContentParser parser = JsonXContent.jsonXContent.createParser(bytes);
+        parser.nextToken();
+        EmailAction action = new EmailAction.Parser(ImmutableSettings.EMPTY, emailService,
+                new ScriptTemplate.Parser(ImmutableSettings.EMPTY, scriptService)).parse(parser);
+
+        assertThat(action, notNullValue());
+        assertThat(action.account, is("_account"));
+        assertThat(action.attachPayload, is(attachPayload));
+        assertThat(action.auth, notNullValue());
+        assertThat(action.auth.user(), is("_user"));
+        assertThat(action.auth.password(), is("_passwd"));
+        assertThat(action.subject, is((Template) subject));
+        assertThat(action.textBody, is((Template) textBody));
+        assertThat(action.htmlBody, is((Template) htmlBody));
+        assertThat(action.emailPrototype.priority(), is(priority));
+        if (to != null) {
+            assertThat(action.emailPrototype.to().toArray(), arrayContainingInAnyOrder(to));
+        } else {
+            assertThat(action.emailPrototype.to(), nullValue());
+        }
+        if (cc != null) {
+            assertThat(action.emailPrototype.cc().toArray(), arrayContainingInAnyOrder(cc));
+        } else {
+            assertThat(action.emailPrototype.cc(), nullValue());
+        }
+        if (bcc != null) {
+            assertThat(action.emailPrototype.bcc().toArray(), arrayContainingInAnyOrder(bcc));
+        } else {
+            assertThat(action.emailPrototype.bcc(), nullValue());
+        }
+        if (replyTo != null) {
+            assertThat(action.emailPrototype.replyTo().toArray(), arrayContainingInAnyOrder(replyTo));
+        } else {
+            assertThat(action.emailPrototype.replyTo(), nullValue());
+        }
+    }
+
+    @Test @Repeat(iterations = 20)
+    public void testParser_SelfGenerated() throws Exception {
+        EmailService service = mock(EmailService.class);
+        Email.Builder emailPrototypeBuilder = Email.builder().id("prototype");
+        if (randomBoolean()) {
+            emailPrototypeBuilder.from(new Email.Address("from@domain"));
+        }
+        if (randomBoolean()) {
+            emailPrototypeBuilder.to(Email.AddressList.parse(randomBoolean() ? "to@domain" : "to1@domain,to2@damain"));
+        }
+        if (randomBoolean()) {
+            emailPrototypeBuilder.cc(Email.AddressList.parse(randomBoolean() ? "cc@domain" : "cc1@domain,cc2@damain"));
+        }
+        if (randomBoolean()) {
+            emailPrototypeBuilder.bcc(Email.AddressList.parse(randomBoolean() ? "bcc@domain" : "bcc1@domain,bcc2@damain"));
+        }
+        if (randomBoolean()) {
+            emailPrototypeBuilder.replyTo(Email.AddressList.parse(randomBoolean() ? "reply@domain" : "reply1@domain,reply2@damain"));
+        }
+        Email email = emailPrototypeBuilder.build();
+        Authentication auth = randomBoolean() ? null : new Authentication("_user", "_passwd");
+        Profile profile = randomFrom(Profile.values());
+        String account = randomAsciiOfLength(6);
+        Template subject = new TemplateMock("_subject");
+        Template textBody = new TemplateMock("_text_body");
+        Template htmlBody = randomBoolean() ? null : new TemplateMock("_html_body");
+        boolean attachPayload = randomBoolean();
+
+        EmailAction action = new EmailAction(logger, service, email, auth, profile, account, subject, textBody, htmlBody, attachPayload);
+
+        XContentBuilder builder = jsonBuilder();
+        action.toXContent(builder, Attachment.XContent.EMPTY_PARAMS);
+        BytesReference bytes = builder.bytes();
+        XContentParser parser = JsonXContent.jsonXContent.createParser(bytes);
+        parser.nextToken();
+        EmailAction parsed = new EmailAction.Parser(ImmutableSettings.EMPTY, service, new TemplateMock.Parser()).parse(parser);
+        assertThat(parsed, equalTo(action));
+
+    }
+
+    @Test(expected = ActionSettingsException.class) @Repeat(iterations = 100)
+    public void testParser_Invalid() throws Exception {
+        ScriptServiceProxy scriptService = mock(ScriptServiceProxy.class);
+        EmailService emailService = mock(EmailService.class);
+        XContentBuilder builder = jsonBuilder().startObject()
+                .field("unknown_field", "value");
+        BytesReference bytes = builder.bytes();
+        XContentParser parser = JsonXContent.jsonXContent.createParser(bytes);
+        new EmailAction.Parser(ImmutableSettings.EMPTY, emailService,
+                new ScriptTemplate.Parser(ImmutableSettings.EMPTY, scriptService)).parse(parser);
+    }
+
+    @Test @Repeat(iterations = 20)
+    public void testParser_Result() throws Exception {
+        boolean success = randomBoolean();
+        Email email = Email.builder().id("_id")
+                .from(new Email.Address("from@domain"))
+                .to(Email.AddressList.parse("to@domain"))
+                .sentDate(new DateTime())
+                .subject("_subject")
+                .textBody("_text_body")
+                .build();
+        XContentBuilder builder = jsonBuilder().startObject()
+                .field("success", success);
+        if (success) {
+            builder.field("email", email);
+            builder.field("account", "_account");
+        } else {
+            builder.field("reason", "_reason");
+        }
+        builder.endObject();
+        BytesReference bytes = builder.bytes();
+        XContentParser parser = JsonXContent.jsonXContent.createParser(bytes);
+        parser.nextToken();
+        EmailAction.Result result = new EmailAction.Parser(ImmutableSettings.EMPTY, mock(EmailService.class), new TemplateMock.Parser())
+                .parseResult(parser);
+        assertThat(result.success(), is(success));
+        if (success) {
+            assertThat(result, instanceOf(EmailAction.Result.Success.class));
+            assertThat(((EmailAction.Result.Success) result).email(), equalTo(email));
+            assertThat(((EmailAction.Result.Success) result).account(), is("_account"));
+        } else {
+            assertThat(result, instanceOf(EmailAction.Result.Failure.class));
+            assertThat(((EmailAction.Result.Failure) result).reason(), is("_reason"));
+        }
+    }
+
+    @Test(expected = EmailException.class)
+    public void testParser_Result_Invalid() throws Exception {
+        XContentBuilder builder = jsonBuilder().startObject()
+                .field("unknown_field", "value")
+                .endObject();
+        BytesReference bytes = builder.bytes();
+        XContentParser parser = JsonXContent.jsonXContent.createParser(bytes);
+        parser.nextToken();
+        new EmailAction.Parser(ImmutableSettings.EMPTY, mock(EmailService.class), new TemplateMock.Parser())
+                .parseResult(parser);
+    }
+
+    static class TemplateMock implements Template {
+
+        private final String name;
+
+        public TemplateMock(String name) {
+            this.name = name;
+        }
+
+        @Override
+        public String render(Map<String, Object> model) {
+            return "";
+        }
+
+        @Override
+        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+            return builder.value(name);
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+
+            TemplateMock that = (TemplateMock) o;
+
+            if (!name.equals(that.name)) return false;
+
+            return true;
+        }
+
+        @Override
+        public int hashCode() {
+            return name.hashCode();
+        }
+
+        public static class Parser implements Template.Parser<TemplateMock> {
+
+            @Override
+            public TemplateMock parse(XContentParser parser) throws IOException, ParseException {
+                return new TemplateMock(parser.text());
+            }
+        }
+    }
+}

src/test/java/org/elasticsearch/alerts/actions/email/service/AccountTests.java

@@ -0,0 +1,269 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.alerts.actions.email.service;
+
+import org.elasticsearch.alerts.actions.email.service.support.EmailServer;
+import org.elasticsearch.common.settings.ImmutableSettings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.test.ElasticsearchTestCase;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import javax.mail.Address;
+import javax.mail.Message;
+import javax.mail.internet.InternetAddress;
+import javax.mail.internet.MimeMessage;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.hamcrest.Matchers.*;
+
+/**
+ *
+ */
+public class AccountTests extends ElasticsearchTestCase {
+
+    static final String USERNAME = "_user";
+    static final String PASSWORD = "_passwd";
+
+    private EmailServer server;
+
+    @Before
+    public void init() throws Exception {
+        server = new EmailServer("localhost", 2500, USERNAME, PASSWORD);
+        server.start();
+    }
+
+    @After
+    public void cleanup() throws Exception {
+        server.stop();
+    }
+
+    @Test
+    public void testConfig() throws Exception {
+
+        ImmutableSettings.Builder builder = ImmutableSettings.builder();
+
+        Profile profile = rarely() ? Profile.STANDARD : randomFrom(Profile.values());
+        if (profile != Profile.STANDARD) {
+            builder.put("profile", profile.name());
+        }
+
+        Account.Config.EmailDefaults emailDefaults;
+        if (randomBoolean()) {
+            ImmutableSettings.Builder sb = ImmutableSettings.builder();
+            if (randomBoolean()) {
+                sb.put(Email.FROM_FIELD.getPreferredName(), "from@domain");
+            }
+            if (randomBoolean()) {
+                sb.put(Email.REPLY_TO_FIELD.getPreferredName(), "replyto@domain");
+            }
+            if (randomBoolean()) {
+                sb.put(Email.PRIORITY_FIELD.getPreferredName(), randomFrom(Email.Priority.values()));
+            }
+            if (randomBoolean()) {
+                sb.put(Email.TO_FIELD.getPreferredName(), "to@domain");
+            }
+            if (randomBoolean()) {
+                sb.put(Email.CC_FIELD.getPreferredName(), "cc@domain");
+            }
+            if (randomBoolean()) {
+                sb.put(Email.BCC_FIELD.getPreferredName(), "bcc@domain");
+            }
+            if (randomBoolean()) {
+                sb.put(Email.SUBJECT_FIELD.getPreferredName(), "_subject");
+            }
+            Settings settings = sb.build();
+            emailDefaults = new Account.Config.EmailDefaults(settings);
+            for (String name : settings.names()) {
+                builder.put("email_defaults." + name, settings.get(name));
+            }
+        } else {
+            emailDefaults = new Account.Config.EmailDefaults(ImmutableSettings.EMPTY);
+        }
+
+        Properties smtpProps = new Properties();
+        ImmutableSettings.Builder smtpBuilder = ImmutableSettings.builder();
+        String host = "somehost";
+        String setting = randomFrom("host", "localaddress", "local_address");
+        smtpBuilder.put(setting, host);
+        if (setting.equals("local_address")) {
+            // we need to remove the `_`... we only added support for `_` for readability
+            // the actual properties (java mail properties) don't contain underscores
+            setting = "localaddress";
+        }
+        smtpProps.put("mail.smtp." + setting, host);
+        String user = null;
+        if (randomBoolean()) {
+            user = randomAsciiOfLength(5);
+            setting = randomFrom("user", "from");
+            smtpBuilder.put(setting, user);
+            smtpProps.put("mail.smtp." + setting, user);
+        }
+        int port = 25;
+        if (randomBoolean()) {
+            port = randomIntBetween(2000, 2500);
+            setting = randomFrom("port", "localport", "local_port");
+            smtpBuilder.put(setting, port);
+            if (setting.equals("local_port")) {
+                setting = "localport";
+            }
+            smtpProps.setProperty("mail.smtp." + setting, String.valueOf(port));
+        }
+        String password = null;
+        if (randomBoolean()) {
+            password = randomAsciiOfLength(8);
+            smtpBuilder.put("password", password);
+            smtpProps.put("mail.smtp.password", password);
+        }
+        for (int i = 0; i < 5; i++) {
+            String name = randomAsciiOfLength(5);
+            String value = randomAsciiOfLength(6);
+            smtpProps.put("mail.smtp." + name, value);
+            smtpBuilder.put(name, value);
+        }
+
+        Settings smtpSettings = smtpBuilder.build();
+        for (String name : smtpSettings.names()) {
+            builder.put("smtp." + name, smtpSettings.get(name));
+        }
+
+        Settings settings = builder.build();
+
+        Account.Config config = new Account.Config("_name", settings);
+
+        if (profile != null) {
+            assertThat(config.profile, is(profile));
+        }
+        assertThat(config.defaults, equalTo(emailDefaults));
+        assertThat(config.smtp, notNullValue());
+        assertThat(config.smtp.port, is(port));
+        assertThat(config.smtp.host, is(host));
+        assertThat(config.smtp.user, is(user));
+        assertThat(config.smtp.password, is(password));
+        assertThat(config.smtp.properties, equalTo(smtpProps));
+    }
+
+    @Test
+    public void testSend() throws Exception {
+        Account account = new Account(new Account.Config("default", ImmutableSettings.builder()
+                .put("smtp.host", "localhost")
+                .put("smtp.port", 2500)
+                .put("smtp.user", USERNAME)
+                .put("smtp.password", PASSWORD)
+                .build()), logger);
+
+        Email email = Email.builder()
+                .id("_id")
+                .from(new Email.Address("from@domain.com"))
+                .to(Email.AddressList.parse("To<to@domain.com>"))
+                .subject("_subject")
+                .textBody("_text_body")
+                .build();
+
+        final CountDownLatch latch = new CountDownLatch(1);
+        EmailServer.Listener.Handle handle = server.addListener(new EmailServer.Listener() {
+            @Override
+            public void on(MimeMessage message) throws Exception {
+                assertThat(message.getFrom().length, is(1));
+                assertThat((InternetAddress) message.getFrom()[0], equalTo(new InternetAddress("from@domain.com")));
+                assertThat(message.getRecipients(Message.RecipientType.TO).length, is(1));
+                assertThat((InternetAddress) message.getRecipients(Message.RecipientType.TO)[0], equalTo(new InternetAddress("to@domain.com", "To")));
+                assertThat(message.getSubject(), equalTo("_subject"));
+                assertThat(Profile.STANDARD.textBody(message), equalTo("_text_body"));
+                latch.countDown();
+            }
+        });
+
+        account.send(email, null, Profile.STANDARD);
+
+        if (!latch.await(5, TimeUnit.SECONDS)) {
+            fail("waiting for email too long");
+        }
+
+        handle.remove();
+    }
+
+    @Test
+    public void testSend_CC_BCC() throws Exception {
+        Account account = new Account(new Account.Config("default", ImmutableSettings.builder()
+                .put("smtp.host", "localhost")
+                .put("smtp.port", 2500)
+                .put("smtp.user", USERNAME)
+                .put("smtp.password", PASSWORD)
+                .build()), logger);
+
+        Email email = Email.builder()
+                .id("_id")
+                .from(new Email.Address("from@domain.com"))
+                .to(Email.AddressList.parse("TO<to@domain.com>"))
+                .cc(Email.AddressList.parse("CC1<cc1@domain.com>,cc2@domain.com"))
+                .bcc(Email.AddressList.parse("BCC1<bcc1@domain.com>,bcc2@domain.com"))
+                .replyTo(Email.AddressList.parse("noreply@domain.com"))
+                .build();
+
+        final CountDownLatch latch = new CountDownLatch(5);
+        EmailServer.Listener.Handle handle = server.addListener(new EmailServer.Listener() {
+            @Override
+            public void on(MimeMessage message) throws Exception {
+                assertThat(message.getFrom().length, is(1));
+                assertThat((InternetAddress) message.getFrom()[0], equalTo(new InternetAddress("from@domain.com")));
+                assertThat(message.getRecipients(Message.RecipientType.TO).length, is(1));
+                assertThat((InternetAddress) message.getRecipients(Message.RecipientType.TO)[0], equalTo(new InternetAddress("to@domain.com", "TO")));
+                assertThat(message.getRecipients(Message.RecipientType.CC).length, is(2));
+                assertThat(message.getRecipients(Message.RecipientType.CC), hasItemInArray((Address) new InternetAddress("cc1@domain.com", "CC1")));
+                assertThat(message.getRecipients(Message.RecipientType.CC), hasItemInArray((Address) new InternetAddress("cc2@domain.com")));
+                assertThat(message.getReplyTo(), arrayWithSize(1));
+                assertThat(message.getReplyTo(), hasItemInArray((Address) new InternetAddress("noreply@domain.com")));
+                // bcc should not be there... (it's bcc after all)
+                latch.countDown();
+            }
+        });
+
+        account.send(email, null, Profile.STANDARD);
+
+        if (!latch.await(5, TimeUnit.SECONDS)) {
+            fail("waiting for email too long");
+        }
+
+        handle.remove();
+    }
+
+    @Test
+    public void testSend_Authentication() throws Exception {
+        Account account = new Account(new Account.Config("default", ImmutableSettings.builder()
+                .put("smtp.host", "localhost")
+                .put("smtp.port", 2500)
+                .build()), logger);
+
+        Email email = Email.builder()
+                .id("_id")
+                .from(new Email.Address("from@domain.com"))
+                .to(Email.AddressList.parse("To<to@domain.com>"))
+                .subject("_subject")
+                .textBody("_text_body")
+                .build();
+
+        final CountDownLatch latch = new CountDownLatch(1);
+        EmailServer.Listener.Handle handle = server.addListener(new EmailServer.Listener() {
+            @Override
+            public void on(MimeMessage message) throws Exception {
+                latch.countDown();
+            }
+        });
+
+        account.send(email, new Authentication(USERNAME, PASSWORD), Profile.STANDARD);
+
+        if (!latch.await(5, TimeUnit.SECONDS)) {
+            fail("waiting for email too long");
+        }
+
+        handle.remove();
+    }
+
+}

src/test/java/org/elasticsearch/alerts/actions/email/service/AccountsTests.java

@@ -0,0 +1,116 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.alerts.actions.email.service;
+
+import org.elasticsearch.common.settings.ImmutableSettings;
+import org.elasticsearch.test.ElasticsearchTestCase;
+import org.junit.Test;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.isOneOf;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.core.IsNull.nullValue;
+
+/**
+ *
+ */
+public class AccountsTests extends ElasticsearchTestCase {
+
+    @Test
+    public void testSingleAccount() throws Exception {
+        ImmutableSettings.Builder builder = ImmutableSettings.builder()
+                .put("default_account", "account1");
+        addAccountSettings("account1", builder);
+
+        Accounts accounts = new Accounts(builder.build(), logger);
+        Account account = accounts.account("account1");
+        assertThat(account, notNullValue());
+        assertThat(account.name(), equalTo("account1"));
+        account = accounts.account(null); // falling back on the default
+        assertThat(account, notNullValue());
+        assertThat(account.name(), equalTo("account1"));
+    }
+
+    @Test
+    public void testSingleAccount_NoExplicitDefault() throws Exception {
+        ImmutableSettings.Builder builder = ImmutableSettings.builder();
+        addAccountSettings("account1", builder);
+
+        Accounts accounts = new Accounts(builder.build(), logger);
+        Account account = accounts.account("account1");
+        assertThat(account, notNullValue());
+        assertThat(account.name(), equalTo("account1"));
+        account = accounts.account(null); // falling back on the default
+        assertThat(account, notNullValue());
+        assertThat(account.name(), equalTo("account1"));
+    }
+
+    @Test
+    public void testMultipleAccounts() throws Exception {
+        ImmutableSettings.Builder builder = ImmutableSettings.builder()
+                .put("default_account", "account1");
+        addAccountSettings("account1", builder);
+        addAccountSettings("account2", builder);
+
+        Accounts accounts = new Accounts(builder.build(), logger);
+        Account account = accounts.account("account1");
+        assertThat(account, notNullValue());
+        assertThat(account.name(), equalTo("account1"));
+        account = accounts.account("account2");
+        assertThat(account, notNullValue());
+        assertThat(account.name(), equalTo("account2"));
+        account = accounts.account(null); // falling back on the default
+        assertThat(account, notNullValue());
+        assertThat(account.name(), equalTo("account1"));
+    }
+
+    @Test
+    public void testMultipleAccounts_NoExplicitDefault() throws Exception {
+        ImmutableSettings.Builder builder = ImmutableSettings.builder()
+                .put("default_account", "account1");
+        addAccountSettings("account1", builder);
+        addAccountSettings("account2", builder);
+
+        Accounts accounts = new Accounts(builder.build(), logger);
+        Account account = accounts.account("account1");
+        assertThat(account, notNullValue());
+        assertThat(account.name(), equalTo("account1"));
+        account = accounts.account("account2");
+        assertThat(account, notNullValue());
+        assertThat(account.name(), equalTo("account2"));
+        account = accounts.account(null);
+        assertThat(account, notNullValue());
+        assertThat(account.name(), isOneOf("account1", "account2"));
+    }
+
+    @Test(expected = EmailSettingsException.class)
+    public void testMultipleAccounts_UnknownDefault() throws Exception {
+        ImmutableSettings.Builder builder = ImmutableSettings.builder()
+                .put("default_account", "unknown");
+        addAccountSettings("account1", builder);
+        addAccountSettings("account2", builder);
+        new Accounts(builder.build(), logger);
+    }
+
+    @Test
+    public void testNoAccount() throws Exception {
+        ImmutableSettings.Builder builder = ImmutableSettings.builder();
+        Accounts accounts = new Accounts(builder.build(), logger);
+        Account account = accounts.account(null);
+        assertThat(account, nullValue());
+    }
+
+    @Test(expected = EmailSettingsException.class)
+    public void testNoAccount_WithDefaultAccount() throws Exception {
+        ImmutableSettings.Builder builder = ImmutableSettings.builder()
+                .put("default_account", "unknown");
+        new Accounts(builder.build(), logger);
+    }
+
+    private void addAccountSettings(String name, ImmutableSettings.Builder builder) {
+        builder.put("account." + name + ".smtp.host", "_host");
+    }
+}

src/test/java/org/elasticsearch/alerts/actions/email/service/InternalEmailServiceTests.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.alerts.actions.email.service;
+
+import org.elasticsearch.common.logging.ESLogger;
+import org.elasticsearch.common.settings.ImmutableSettings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.node.settings.NodeSettingsService;
+import org.elasticsearch.test.ElasticsearchTestCase;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.hamcrest.Matchers.*;
+import static org.mockito.Mockito.*;
+
+/**
+ *
+ */
+public class InternalEmailServiceTests extends ElasticsearchTestCase {
+
+    private InternalEmailService service;
+    private Accounts accounts;
+
+    @Before
+    public void init() throws Exception {
+        accounts = mock(Accounts.class);
+        service = new InternalEmailService(ImmutableSettings.EMPTY, new NodeSettingsService(ImmutableSettings.EMPTY)) {
+            @Override
+            protected Accounts createAccounts(Settings settings, ESLogger logger) {
+                return accounts;
+            }
+        };
+        service.start();
+    }
+
+    @After
+    public void cleanup() throws Exception {
+        service.stop();
+    }
+
+    @Test
+    public void testSend() throws Exception {
+        Account account = mock(Account.class);
+        when(account.name()).thenReturn("account1");
+        when(accounts.account("account1")).thenReturn(account);
+        Email email = mock(Email.class);
+        Authentication auth = new Authentication("user", "passwd");
+        Profile profile = randomFrom(Profile.values());
+        EmailService.EmailSent sent = service.send(email, auth, profile, "account1");
+        verify(account).send(email, auth, profile);
+        assertThat(sent, notNullValue());
+        assertThat(sent.email(), sameInstance(email));
+        assertThat(sent.account(), is("account1"));
+    }
+
+}

src/test/java/org/elasticsearch/alerts/actions/email/service/ManualPublicSmtpServersTests.java

@@ -0,0 +1,120 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.alerts.actions.email.service;
+
+import org.elasticsearch.common.cli.Terminal;
+import org.elasticsearch.common.inject.Provider;
+import org.elasticsearch.common.settings.ImmutableSettings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.ToXContent;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.node.settings.NodeSettingsService;
+import org.junit.Ignore;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ *
+ */
+@Ignore
+public class ManualPublicSmtpServersTests {
+
+    private static final Terminal terminal = Terminal.DEFAULT;
+
+    public static class Gmail {
+
+        public static void main(String[] args) throws Exception {
+            test(Profile.GMAIL, ImmutableSettings.builder()
+                    .put("alerts.actions.email.service.account.gmail.smtp.auth", true)
+                    .put("alerts.actions.email.service.account.gmail.smtp.starttls.enable", true)
+                    .put("alerts.actions.email.service.account.gmail.smtp.host", "smtp.gmail.com")
+                    .put("alerts.actions.email.service.account.gmail.smtp.port", 587)
+                    .put("alerts.actions.email.service.account.gmail.smtp.user", terminal.readText("username: "))
+                    .put("alerts.actions.email.service.account.gmail.smtp.password", new String(terminal.readSecret("password: ")))
+                    .put("alerts.actions.email.service.account.gmail.email_defaults.to", terminal.readText("to: "))
+            );
+
+        }
+    }
+
+    public static class OutlookDotCom {
+
+        public static void main(String[] args) throws Exception {
+            test(Profile.STANDARD, ImmutableSettings.builder()
+                    .put("alerts.actions.email.service.account.outlook.smtp.auth", true)
+                    .put("alerts.actions.email.service.account.outlook.smtp.starttls.enable", true)
+                    .put("alerts.actions.email.service.account.outlook.smtp.host", "smtp-mail.outlook.com")
+                    .put("alerts.actions.email.service.account.outlook.smtp.port", 587)
+                    .put("alerts.actions.email.service.account.outlook.smtp.user", "elastic.user@outlook.com")
+                    .put("alerts.actions.email.service.account.outlook.smtp.password", "fantastic42")
+                    .put("alerts.actions.email.service.account.outlook.email_defaults.to", "elastic.user@outlook.com")
+                    .put()
+            );
+        }
+    }
+
+    public static class YahooMail {
+
+        public static void main(String[] args) throws Exception {
+            test(Profile.STANDARD, ImmutableSettings.builder()
+                            .put("alerts.actions.email.service.account.yahoo.smtp.starttls.enable", true)
+                            .put("alerts.actions.email.service.account.yahoo.smtp.auth", true)
+                            .put("alerts.actions.email.service.account.yahoo.smtp.host", "smtp.mail.yahoo.com")
+                            .put("alerts.actions.email.service.account.yahoo.smtp.port", 587)
+                            .put("alerts.actions.email.service.account.yahoo.smtp.user", "elastic.user@yahoo.com")
+                            .put("alerts.actions.email.service.account.yahoo.smtp.password", "fantastic42")
+                            // note: from must be set to the same authenticated user account
+                            .put("alerts.actions.email.service.account.yahoo.email_defaults.from", "elastic.user@yahoo.com")
+                            .put("alerts.actions.email.service.account.yahoo.email_defaults.to", "elastic.user@yahoo.com")
+            );
+        }
+    }
+
+    static void test(Profile profile, Settings.Builder builder) throws Exception {
+        InternalEmailService service = startEmailService(builder);
+        try {
+
+            ToXContent content = new ToXContent() {
+                @Override
+                public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+                    return builder.startObject()
+                            .field("key1", "value1")
+                            .field("key2", "value2")
+                            .field("key3", "value3")
+                            .endObject();
+                }
+            };
+
+            Email email = Email.builder()
+                    .id("_id")
+                    .subject("_subject")
+                    .textBody("_text_body")
+                    .htmlBody("<b>html body</b><p/><p/><img src=\"cid:logo\"/>")
+                    .attach(new Attachment.XContent.Yaml("test.yml", content))
+                    .inline(new Inline.Stream("logo", "logo.jpg", new Provider<InputStream>() {
+                        @Override
+                        public InputStream get() {
+                            return InternalEmailServiceTests.class.getResourceAsStream("logo.jpg");
+                        }
+                    }))
+                    .build();
+
+            EmailService.EmailSent sent = service.send(email, null, profile);
+
+            terminal.println("email sent via account [%s]", sent.account());
+        } finally {
+            service.stop();
+        }
+    }
+
+    static InternalEmailService startEmailService(Settings.Builder builder) {
+        Settings settings = builder.build();
+        InternalEmailService service = new InternalEmailService(settings, new NodeSettingsService(settings));
+        service.start();
+        return service;
+    }
+}

src/test/java/org/elasticsearch/alerts/actions/email/service/support/EmailServer.java

@@ -0,0 +1,110 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.alerts.actions.email.service.support;
+
+import org.subethamail.smtp.TooMuchDataException;
+import org.subethamail.smtp.auth.EasyAuthenticationHandlerFactory;
+import org.subethamail.smtp.auth.LoginFailedException;
+import org.subethamail.smtp.auth.UsernamePasswordValidator;
+import org.subethamail.smtp.helper.SimpleMessageListener;
+import org.subethamail.smtp.helper.SimpleMessageListenerAdapter;
+import org.subethamail.smtp.server.SMTPServer;
+
+import javax.mail.MessagingException;
+import javax.mail.Session;
+import javax.mail.internet.MimeMessage;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.fail;
+
+/**
+ * An mini email smtp server that can be used for unit testing
+ *
+ *
+ */
+public class EmailServer {
+
+    private final List<Listener> listeners = new CopyOnWriteArrayList<>();
+
+    private final SMTPServer server;
+
+    public EmailServer(String host, int port, final String username, final String password) {
+        server = new SMTPServer(new SimpleMessageListenerAdapter(new SimpleMessageListener() {
+            @Override
+            public boolean accept(String from, String recipient) {
+                return true;
+            }
+
+            @Override
+            public void deliver(String from, String recipient, InputStream data) throws TooMuchDataException, IOException {
+                try {
+                    Session session = Session.getDefaultInstance(new Properties());
+                    MimeMessage msg = new MimeMessage(session, data);
+                    for (Listener listener : listeners) {
+                        try {
+                            listener.on(msg);
+                        } catch (Exception e) {
+                            fail(e.getMessage());
+                            e.printStackTrace();
+                        }
+                    }
+                } catch (MessagingException me) {
+                    throw new RuntimeException("could not create mime message", me);
+                }
+            }
+        }), new EasyAuthenticationHandlerFactory(new UsernamePasswordValidator() {
+            @Override
+            public void login(String user, String passwd) throws LoginFailedException {
+                assertThat(user, is(username));
+                assertThat(passwd, is(password));
+            }
+        }));
+        server.setHostName(host);
+        server.setPort(port);
+    }
+
+    public void start() {
+        server.start();
+    }
+
+    public void stop() {
+        server.stop();
+        listeners.clear();
+    }
+
+    public Listener.Handle addListener(Listener listener) {
+        listeners.add(listener);
+        return new Listener.Handle(listeners, listener);
+    }
+
+    public static interface Listener {
+
+        void on(MimeMessage message) throws Exception;
+
+        public static class Handle {
+
+            private final List<Listener> listeners;
+            private final Listener listener;
+
+            Handle(List<Listener> listeners, Listener listener) {
+                this.listeners = listeners;
+                this.listener = listener;
+            }
+
+            public void remove() {
+                listeners.remove(listener);
+            }
+        }
+
+    }
+
+}