Implement a mail server to help with 3pid testing

cpanfile

@@ -23,6 +23,8 @@ requires 'DBI';
 requires 'DBD::Pg';
 requires 'Digest::HMAC_SHA1';
 requires 'Digest::SHA';
+requires 'Email::Address::XS';
+requires 'Email::MIME';
 requires 'File::Basename';
 requires 'File::Path';
 requires 'File::Slurper';

lib/SyTest/MailServer.pm

@@ -0,0 +1,124 @@
+package SyTest::MailServer;
+
+use strict;
+use warnings;
+use Carp;
+
+use SyTest::MailServer::Protocol;
+
+use base qw( IO::Async::Listener );
+
+=head1 NAME
+
+C<SyTest::MailServer> - serve SMTP with C<IO::Async>
+
+=head1 SYNOPSIS
+
+ use SyTest::MailServer;
+ use IO::Async::Loop;
+
+ my $loop = IO::Async::Loop->new();
+
+ my $mailserver = Net::Async::HTTP::Server->new(
+    on_mail => sub {
+       my $self = shift;
+       my ( $to, $from, $data ) = @_;
+    },
+ );
+
+ $loop->add( $mailserver );
+
+ $mailserver->listen(
+    addr => { family => "inet6", socktype => "stream", port => 2525 },
+ )->get
+
+ $loop->run;
+
+=head1 DESCRIPTION
+
+This module allows a program to respond asynchronously to SMTP requests, as
+part of a program based on L<IO::Async>. An object in this class listens on a
+single port and invokes the C<on_mail> callback or subclass method whenever
+a mail is received over SMTP.
+
+=cut
+
+=head1 EVENTS
+
+=head2 on_mail( $from, $to, $data )
+
+Invoked when a new mail is received.
+
+=head1 METHODS
+
+As a small subclass of L<IO::Async::Listener>, this class does not provide many
+new methods of its own. The superclass provides useful methods to control the
+basic operation of this server.
+
+Specifically, see the L<IO::Async::Listener/listen> method on how to actually
+bind the server to a listening socket to make it accept requests.
+
+=cut
+
+sub _init
+{
+   my $self = shift;
+   my ( $args ) = @_;
+
+   $args->{handle_class} = "SyTest::MailServer::Protocol";
+
+   $self->SUPER::_init( $args );
+}
+
+sub configure
+{
+   my $self = shift;
+   my %params = @_;
+
+   foreach ( qw( on_mail ) ) {
+      $self->{$_} = delete $params{$_} if $params{$_};
+   }
+
+   $self->SUPER::configure( %params );
+}
+
+sub _add_to_loop
+{
+   my $self = shift;
+
+   $self->can_event( "on_mail" ) or croak "Expected either a on_mail callback or an ->on_mail method";
+
+   $self->SUPER::_add_to_loop( @_ );
+}
+
+sub on_accept
+{
+   my $self = shift;
+   my ( $conn ) = @_;
+
+   $conn->configure(
+      on_closed => sub {
+         my $conn = shift;
+         $conn->on_closed();
+
+         $conn->remove_from_parent;
+      },
+   );
+
+   $self->add_child( $conn );
+
+   $conn->send_reply( 220, "Sytest test server" );
+
+   return $conn;
+}
+
+sub _received_mail
+{
+   my $self = shift;
+   my ( $from, $to, $data ) = @_;
+
+   $self->invoke_event( 'on_mail', $from, $to, $data );
+}
+
+1;
+

lib/SyTest/MailServer/Protocol.pm

@@ -0,0 +1,134 @@
+package SyTest::MailServer::Protocol;
+
+use strict;
+use warnings;
+use base qw( IO::Async::Stream );
+
+my $CRLF = "\x0d\x0a";
+
+sub on_read
+{
+   my $self = shift;
+   my ( $buffref, $eof ) = @_;
+
+   return 0 if $eof;
+   return 0 unless $$buffref =~ s/^(.*?$CRLF)//s;
+
+   my ( $verb, $params ) = $self->tokenize_command( $1 );
+   return $self->process_command( $verb, $params );
+} 
+
+sub on_closed
+{
+   my $self = shift;
+}
+
+sub tokenize_command {
+    my ( $self, $line ) = @_;
+    $line =~ s/\r?\n$//s;
+    $line =~ s/^\s+|\s+$//g;
+    my ( $verb, $params ) = split ' ', $line, 2;
+    $verb = uc($verb) if defined($verb);
+    return ( $verb, $params );
+}
+
+sub process_command
+{
+   my $self = shift;
+   my ( $verb, $params ) = @_;
+
+   $self->debug_printf( "COMMAND %s %s", $verb, $params );
+
+   if( my $code = $self->can( "on_" . $verb )) {
+      return $code->( $self, $params ) // 1;
+   } else {
+      $self->send_reply( 500, 'Syntax error: unrecognized command' );
+      return 1;
+   }
+}
+
+sub send_reply
+{
+   my ( $self ) = shift;
+   my ( $code, $msg ) = @_;
+
+   $self->write( "$code $msg\r\n" );
+}
+
+sub on_HELO
+{
+   my ( $self ) = shift;
+   my ( $params ) = @_;
+
+   $self->send_reply( 250, "hi" );
+}
+
+sub on_MAIL
+{
+   my ( $self ) = shift;
+   my ( $params ) = @_;
+
+   if( defined $self->{mail_from} ) {
+      $self->send_reply( 503, 'Bad sequence of commands' );
+      return;
+   }
+
+   unless ( $params =~ s/^from:\s*//i ) {
+      $self->send_reply( 501, 'Syntax error in parameters or arguments' );
+      return;
+   }
+
+   $self->{mail_from} = $params;
+   $self->send_reply( 250, "ok" );
+}
+
+sub on_RCPT
+{
+   my ( $self ) = shift;
+   my ( $params ) = @_;
+
+   if( defined $self->{rcpt_to} ) {
+      $self->send_reply( 503, 'Bad sequence of commands' );
+      return;
+   }
+
+   unless ( $params =~ s/^to:\s*//i ) {
+      $self->send_reply( 501, 'Syntax error in parameters or arguments' );
+      return;
+   }
+
+   $self->{rcpt_to} = $params;
+   $self->send_reply( 250, "ok" );
+}
+
+sub on_DATA
+{
+   my ( $self ) = shift;
+   my ( $params ) = @_;
+
+   if( not defined $self->{rcpt_to} or not defined $self->{mail_from} ) {
+      $self->send_reply( 503, 'Bad sequence of commands' );
+      return;
+   }
+
+   if ( $params ) {
+      $self->send_reply( 501, 'Syntax error in parameters or arguments' );
+      return;
+   }
+
+   $self->send_reply( 354, "send message" );
+
+   return sub {
+      my ( undef, $buffref, $eof ) = @_;
+      return 0 unless $$buffref =~ s/(^.*$CRLF)\.$CRLF//s;
+
+      $self->parent->_received_mail( $self->{mail_from}, $self->{rcpt_to}, $1 );
+      $self->send_reply( 250, "ok" );
+      $self->{rcpt_to} = undef;
+      $self->{mail_from} = undef;
+      return undef;
+   }
+}
+
+1;
+

tests/04mail-server.pl

@@ -0,0 +1,114 @@
+use SyTest::MailServer;
+use Email::Address::XS;
+use Email::MIME;
+use List::UtilsBy qw( extract_first_by );
+
+
+=head2 MAIL_SERVER_INFO
+
+A fixture which starts a test SMTP server.
+
+The result is a hash with the following members:
+
+=over
+
+=item host
+
+hostname where this server can be reached
+
+=item port
+
+port where this server can be reached
+
+=back
+
+=cut
+
+our $MAIL_SERVER_INFO = fixture(
+   requires => [],
+   setup => sub {
+      my $mail_server = SyTest::MailServer->new(
+         on_mail => \&_on_mail,
+      );
+      $loop->add( $mail_server );
+
+      $mail_server->listen(
+         host     => $BIND_HOST,
+         service  => 0,
+         socktype => 'stream',
+      )->then( sub {
+         my ( $listener ) = @_;
+         my $sockport = $listener->read_handle->sockport;
+         my $sockname = "$BIND_HOST:$sockport";
+
+         $OUTPUT->diag( "Started test SMTP Server at $sockname" );
+         Future->done({
+            host => $BIND_HOST,
+            port => $sockport,
+         });
+      });
+   },
+);
+
+push our @EXPORT, qw( MAIL_SERVER_INFO );
+
+struct MailAwaiter => [qw( future rcpt_match )];
+
+my @pending_awaiters;
+
+sub _on_mail {
+   my ( undef, $from, $to, $data ) = @_;
+
+   if( $CLIENT_LOG ) {
+      my $green = -t STDOUT ? "\e[1;32m" : "";
+      my $reset = -t STDOUT ? "\e[m" : "";
+      print "${green}Received mail${reset} from $from to $to:\n";
+      print "  $_\n" for split m/\n/, $data;
+      print "-- \n";
+   }
+
+   $to = Email::Address::XS->parse( $to )->address;
+   $from = Email::Address::XS->parse( $from )->address;
+   my $email = Email::MIME->new( $data );
+
+   my $awaiter = extract_first_by {
+      return $to eq $_->rcpt_match;
+   } @pending_awaiters;
+
+   if( $awaiter ) {
+      $awaiter->future->done( $from, $email );
+   } else {
+      warn "Received spurious email from $from to $to\n";
+   }
+}
+
+=head2 await_email_to
+
+   await_email( $rcpt )->then( sub {
+       my ( $from, $email ) = @_;
+       print $email->body;
+   });
+
+<$email> is an C<Email::MIME> instance.
+
+=cut
+
+sub await_email_to {
+   my ( $rcpt, %args ) = @_;
+   my $timeout = $args{timeout} // 10;
+
+   my $f = $loop->new_future;
+   my $awaiter = MailAwaiter( $f, $rcpt );
+   push @pending_awaiters, $awaiter;
+
+   $f->on_cancel( sub {
+      extract_first_by { $_ == $awaiter } @pending_awaiters;
+   });
+
+   return Future->wait_any(
+      $f,
+      delay( $timeout )->then_fail( "Timed out waiting for email" ),
+   );
+}
+
+push @EXPORT, qw( await_email_to );

tests/05homeserver.pl

@@ -25,10 +25,10 @@
    fixture(
       name => "HOMESERVER_$idx",
 
-      requires => [ $main::TEST_SERVER_INFO, @main::AS_INFO ],
+      requires => [ $main::TEST_SERVER_INFO, $main::MAIL_SERVER_INFO, @main::AS_INFO ],
 
       setup => sub {
-         my ( $test_server_info, @as_infos ) = @_;
+         my ( $test_server_info, $mail_server_info, @as_infos ) = @_;
 
          $OUTPUT->diag( "Starting Homeserver using $HS_FACTORY" );