package tests::DlfConverterProcessTest;

use strict;

use base qw/ tests::DlfConverterProcessFixture /;

use Lire::DlfConverterProcess;
use Lire::Utils qw/tempfile/;
use Lire::I18N qw/set_fh_encoding/;

use Class::Inner;

my %schemas =
(
 'test1' => <<EOF,
<lire:dlf-schema superservice="test1" timestamp="time"
 xmlns:lire="http://www.logreport.org/LDSML/">
<lire:field name="time" type="timestamp"/>
<lire:field name="code" type="string"/>
<lire:field name="message" type="string"/>
</lire:dlf-schema>
EOF
 'test2' => <<EOF,
<lire:dlf-schema superservice="test2" timestamp="time_start"
 xmlns:lire="http://www.logreport.org/LDSML/">
<lire:field name="time_start" type="timestamp"/>
<lire:field name="file" type="filename"/>
<lire:field name="result" type="string"/>
</lire:dlf-schema>
EOF
);

my %logs =
  (
   'test_line' =>
   {
    'log' => <<EOF,
schema=test1;time=90000;code=500;message=This is a long message
schema=test1;
schema=test2;time_start=1023456;file=/var/tmp;result=OK
ignore=1;key=another line to be ignored
schema=test1;time=
schema=test1;time=100000;message=Another message
ignore=1;key=this line will be ignored
EOF
    'error_count' => 1,
    'line_count' => 7,
    'ignored_count' => 2,
    'saved_count' => 0,
    'dlf_count' => 4,
    'dlf' =>
    {
     'test1' =>
     [
      {
       'dlf_id'  => 1,
       'dlf_source' => 'myjob-20040311',
       'time'    => 90000,
       'code'    => 500,
       'message' => "This is a long message",
      },
      {
       'dlf_id'  => 2,
       'dlf_source' => 'myjob-20040311',
       'time'    => undef,
       'code'    => undef,
       'message' => undef,
      },
      {
       'dlf_id'  => 3,
       'dlf_source' => 'myjob-20040311',
       'time'    => 100000,
       'code'    => undef,
       'message' => "Another message",
      },
     ],
     'test2' =>
     [
      {
       'dlf_id'  => 1,
       'dlf_source' => 'myjob-20040311',
       'time_start' => 1023456,
       'file' => "/var/tmp",
       'result' => "OK",
      },
     ],
    },
   },

   'continuation' =>
   {
    'log' => <<EOF,
schema=test1;time=90000;code=500;message=This is a long message;save=1
schema=test2;time_start=1023456;file=/var/tmp;result=OK;save=1
EOF
    'error_count' => 0,
    'line_count' => 2,
    'ignored_count' => 0,
    'saved_count' => 2,
    'dlf_count' => 0,
    'dlf' => {},
   },
  );

# Modify for file test
$logs{'test_file'} = { %{$logs{'test_line'}} };
$logs{'test_file'}{'name'} = "test_file";
$logs{'test_file'}{'line_count'} = 0;
$logs{'test_file'}{'ignored_count'} = undef;
$logs{'test_file'}{'saved_count'} = undef;

# Continuation results
my $cont_results =
  {
    'log' => "",
    'error_count' => 0,
    'line_count' => 4, # 2 saved + 2 of original log
    'ignored_count' => 0,
    'saved_count' => 2, # 2 of the original log
    'dlf_count' => 2, # 2 saved
    'dlf' =>
    {
     'test1' => [
               {
                'dlf_id'  => 1,
                'dlf_source' => 'myjob-20040311',
                'time'    => 90000,
                'code'    => 500,
                'message' => "This is a long message",
               },
              ],
     'test2' => [
               {
                'dlf_id'  => 1,
                'dlf_source' => 'myjob-20040311',
                'time_start' => 1023456,
                'file' => "/var/tmp",
                'result' => "OK",
               },
              ],
    },
  };

sub schema_fixtures {
    \%schemas;
}

sub import_job_fixtures {
    my $r = { map { $_ => $logs{$_}{'log'} } keys %logs };

    return $r;
}

my @converters =
  (
   new tests::DlfConverterProcessTest::DlfConverter( "line", 1 ),
   new tests::DlfConverterProcessTest::DlfConverter( "file", 0 )
  );

sub converter_fixtures {
    \@converters;
}

sub test_new {
    my $self = $_[0];

    my $p = new Lire::DlfConverterProcess( $self->import_job( "test_line"),
                                           $self->dlf_store() );
    $self->assert_isa( 'Lire::DlfConverterProcess', $p );
    $self->assert_str_equals( $self->import_job( "test_line" ),
                              $p->import_job() );
    $self->assert_str_equals( $self->dlf_store(), $p->dlf_store() );
}

sub test_run_import_job_line {
    my $self = $_[0];

    $self->import_job( "test_line" )->converter( "line" );
    my $p = new Lire::DlfConverterProcess( $self->import_job( "test_line" ),
                                           $self->dlf_store() );
    $self->assert_isa( 'Lire::DlfConverterProcess', $p );
    $p->{'_job_id'} = 'myjob-20040311';
    $p->run_import_job();

    $self->assert_dlf_converter_match_results( $logs{'test_line'}, $p );

    eval { $p->run_import_job() };
    $self->assert_not_null( $@, "calling run_import_job twice should fail" );
}

sub test_run_import_job_file {
    my $self = $_[0];

    $self->import_job( "test_file" )->converter( "file" );
    my $p = new Lire::DlfConverterProcess( $self->import_job( "test_file" ),
                                           $self->dlf_store() );
    $self->assert_isa( 'Lire::DlfConverterProcess', $p );

    $p->{'_job_id'} = 'myjob-20040311';
    $p->run_import_job();
    $self->assert_dlf_converter_match_results( $logs{'test_file'}, $p );
}

sub test_convert_continuation {
    my $self = $_[0];

    $self->import_job( "continuation" )->converter( "line" );
    my $p = new Lire::DlfConverterProcess( $self->import_job( "continuation" ),
                                           $self->dlf_store() );
    $self->assert_isa( 'Lire::DlfConverterProcess', $p );

    $p->{'_job_id'} = 'myjob-20040311';
    $p->run_import_job();
    $self->assert_dlf_converter_match_results( $logs{'continuation'}, $p );

    # Should process saved lines now
    $p = new Lire::DlfConverterProcess( $self->import_job( "continuation" ),
                                        $self->dlf_store() );
    $p->{'_job_id'} = 'myjob-20040311';
    $p->run_import_job();
    $self->assert_dlf_converter_match_results( $cont_results, $p );
}

sub set_up_converter_process {
    my $self = $_[0];

    $self->{'process'} =
      new Lire::DlfConverterProcess( $self->import_job( "test_line" ),
                                     $self->dlf_store() );

    $self->import_job( "test_line" )->converter( "line" );
    $self->{'process'}->_init_converter();

    return;
}

sub test_init_streams {
    my $self = $_[0];

    $self->set_up_converter_process();
    $self->{'process'}->_init_streams();
    $self->assert_isa( 'Lire::DlfStream', $self->{'process'}{'_streams'}{'test1'} );
    $self->assert_isa( 'Lire::DlfStream', $self->{'process'}{'_streams'}{'test2'} );
    $self->assert_isa( 'Lire::DlfStream', $self->{'process'}{'_log_stream'} );
}

sub test_error {
    my $self = $_[0];

    $self->set_up_converter_process();
    $self->{'process'}{'_log_stream'} =
      $self->dlf_store()->open_dlf_stream( 'lire_import_log', 'w' );

    $self->{'process'}{'_job_id'} = 'My ID';
    $self->{'process'}{'_line_count'} = 54;
    $self->{'process'}{'_error_count'} = 0;

    $self->assert_dies( qr/missing 'error_msg' parameter/,
                        sub { $self->{'process'}->error() } );

    $self->{'process'}->error( 'Bad line', 'Test line' );
    $self->{'process'}{'_line_count'} = 55;
    $self->{'process'}->error( 'Unknown error' );
    $self->assert_num_equals( 2, $self->{'process'}{'_error_count'} );
    $self->{'process'}{'_log_stream'}->close();

    my $log = $self->dlf_store()->open_dlf_stream( 'lire_import_log', 'r',
                                                   'time' );
    my $dlf1 = $log->read_dlf();
    my $dlf2 = $log->read_dlf();
    $self->assert_null( $log->read_dlf(), "unexpected DLF" );
    $self->assert_not_null(  $dlf1->{'time'}, 'Time should be set' );
    $self->assert_str_equals( 'My ID', $dlf1->{'job_id'} );
    $self->assert_str_equals( 'test_line', $dlf1->{'job_name'} );
    $self->assert_num_equals( 54, $dlf1->{'line_no'} );
    $self->assert_str_equals(  'error', $dlf1->{'type'} );
    $self->assert_str_equals(  'Test line', $dlf1->{'line'} );
    $self->assert_str_equals(  'Bad line', $dlf1->{'msg'} );

    $self->assert_str_equals( '', $dlf2->{'line'} );
    $self->assert_str_equals(  'Unknown error', $dlf2->{'msg'} );
}

sub test_save_log_line_not_supported {
    my $self = $_[0];

    my $process =
      new Lire::DlfConverterProcess( $self->import_job( 'test_file' ),
                                     $self->dlf_store() );
    $self->import_job( "test_file" )->converter( "file" );
    $process->_init_converter();

    $self->assert_dies( qr/only DLF converter handling log lines can save log line/,
                        sub { $process->save_log_line( 'A line' ) } );
}

sub test_save_log_line {
    my $self = $_[0];

    $self->set_up_converter_process();
    $self->{'process'}{'_log_stream'} =
      $self->dlf_store()->open_dlf_stream( 'lire_import_log', 'w' );

    $self->{'process'}{'_job_id'} = 'My ID';
    $self->{'process'}{'_line_count'} = 54;
    $self->{'process'}{'_saved_count'} = 0;

    $self->assert_dies( qr/missing 'line' parameter/,
                        sub { $self->{'process'}->save_log_line() } );

    $self->{'process'}->save_log_line( 'My line');
    $self->{'process'}{'_line_count'} = 55;
    $self->assert_num_equals( 1, $self->{'process'}{'_saved_count'} );
    $self->{'process'}{'_log_stream'}->close();

    my $log = $self->dlf_store()->open_dlf_stream( 'lire_import_log', 'r',
                                                   'time' );
    my $dlf = $log->read_dlf();
    $self->assert_null( $log->read_dlf(), "unexpected DLF" );
    $self->assert_not_null(  $dlf->{'time'}, 'Time should be set' );
    $self->assert_str_equals( 'My ID', $dlf->{'job_id'} );
    $self->assert_str_equals( 'test_line', $dlf->{'job_name'} );
    $self->assert_num_equals( 54, $dlf->{'line_no'} );
    $self->assert_str_equals(  'continuation', $dlf->{'type'} );
    $self->assert_str_equals(  'My line', $dlf->{'line'} );
    $self->assert_null( $dlf->{'msg'}, 'msg should be null' );
}

sub test_ignore_log_line {
    my $self = $_[0];

    $self->set_up_converter_process();
    $self->{'process'}{'_log_stream'} =
      $self->dlf_store()->open_dlf_stream( 'lire_import_log', 'w' );

    $self->{'process'}{'_job_id'} = 'My ID';
    $self->{'process'}{'_line_count'} = 54;
    $self->{'process'}{'_ignored_count'} = 0;

    $self->assert_dies( qr/missing 'line' parameter/,
                        sub { $self->{'process'}->ignore_log_line() } );

    $self->{'process'}->ignore_log_line( 'My line' );
    $self->{'process'}{'_line_count'} = 55;
    $self->{'process'}->ignore_log_line( 'Another line', 'Whatever' );
    $self->{'process'}{'_line_count'} = 55;
    $self->assert_num_equals( 2, $self->{'process'}{'_ignored_count'} );
    $self->{'process'}{'_log_stream'}->close();

    my $log = $self->dlf_store()->open_dlf_stream( 'lire_import_log', 'r',
                                                   'time' );
    my @dlf = ( $log->read_dlf(), $log->read_dlf() );
    $self->assert_null( $log->read_dlf(), "unexpected DLF" );
    $self->assert_not_null(  $dlf[0]{'time'}, 'Time should be set' );
    $self->assert_str_equals( 'My ID', $dlf[0]{'job_id'} );
    $self->assert_str_equals( 'test_line', $dlf[0]{'job_name'} );
    $self->assert_num_equals( 54, $dlf[0]{'line_no'} );
    $self->assert_str_equals(  'ignored', $dlf[0]{'type'} );
    $self->assert_str_equals(  'My line', $dlf[0]{'line'} );
    $self->assert_str_equals( 'Unknown reason', $dlf[0]{'msg'} );

    $self->assert_str_equals(  'Another line', $dlf[1]{'line'} );
    $self->assert_str_equals( 'Whatever', $dlf[1]{'msg'} );
}

sub test_save_import_stats {
    my $self = $_[0];

    $self->set_up_converter_process();

    $self->{'process'}{'_job_id'} = 'My ID';
    $self->{'process'}{'_time_start'} = time - 1000;
    $self->{'process'}{'_line_count'} = 300;
    $self->{'process'}{'_dlf_count'} = 294;
    $self->{'process'}{'_ignored_count'} = 3;
    $self->{'process'}{'_saved_count'} = 1;
    $self->{'process'}{'_error_count'} = 2;

    $self->{'process'}->_save_import_stats();

    my $stats = $self->dlf_store()->open_dlf_stream( 'lire_import_stats', 'r');
    my $dlf = $stats->read_dlf();
    $self->assert_null( $stats->read_dlf(), "unexpected DLF" );
    $stats->close();

    $self->assert_str_equals(  $dlf->{'time_start'},
                               $self->{'process'}{'_time_start'} );
    $self->assert_str_equals( 'My ID', $dlf->{'job_id'} );
    $self->assert_str_equals( 'test_line', $dlf->{'job_name'} );
    $self->assert_num_equals( 300, $dlf->{'line_count'} );
    $self->assert_num_equals(  294, $dlf->{'dlf_count'} );
    $self->assert_num_equals(  3, $dlf->{'ignored_count'} );
    $self->assert_num_equals( 1, $dlf->{'saved_count'} );
    $self->assert_num_equals( 2, $dlf->{'error_count'} );

    $self->assert( $dlf->{'elapsed'} >= 1000, 
                   "elapsed should be 1000 or more");
}

sub test_handle_continuation {
    my $self = $_[0];

    my $process = new Lire::DlfConverterProcess($self->import_job('test_line'),
                                                $self->dlf_store() );
    $process->{'_time_start'} = time;
    $process->{'_line_count'} = 0;
    $process->{'_log_stream'} =
      $self->dlf_store()->open_dlf_stream( 'lire_import_log', 'w' );
    $process->{'_converter'} =
      new Class::Inner( 'parent' => 'Lire::DlfConverter',
                        'methods' =>
                        {
                         'new' => sub { return bless [], shift },
                         'name' => sub { return 'mock' },
                         'handle_log_lines' => sub { return 1 },
                         'process_log_line' => sub {
                             push @{$_[0]}, $_[2];
                         }
                        } );
    my $stream = $self->dlf_store()->open_dlf_stream( 'lire_import_log', 'w' );
    $stream->write_dlf( { 'time' => time - 100,
                          'type' => 'continuation',
                          'line_no' => 1,
                          'line' => 'My line 1',
                          'job_name' => 'test_line',
                          } );
    $stream->write_dlf( { 'time' => time - 50,
                          'type' => 'continuation',
                          'line_no' => 2,
                          'line' => 'My line 2',
                          'job_name' => 'test_line',
                          } );
    $stream->write_dlf( { 'line_no' => 1,
                          'type' => 'ignored',
                          'line' => 'An ignored line',
                          'job_name' => 'test_line',
                          } );
    $stream->write_dlf( {'line_no' => 2,
                          'type' => 'continuation',
                          'line' => 'A line',
                          'job_name' => 'another_job',
                          } );
    $stream->close();

    $process->_handle_continuation();
    $self->assert_num_equals( 2, $process->{'_line_count'} );
    my @lines = @{$process->{'_converter'}};
    $self->assert_deep_equals( [ 'My line 1', 'My line 2' ],
                               \@lines  );
    $stream = $self->dlf_store()->open_dlf_stream( 'lire_import_log', 'r',
                                                   'line_no' );
    my @dlf = ( $stream->read_dlf(), $stream->read_dlf() );
    $self->assert_null( $stream->read_dlf(), 'Stream should be empty' );
    $self->assert_str_equals( 'An ignored line', $dlf[0]{'line'} );
    $self->assert_str_equals( 'A line', $dlf[1]{'line'} );
    $stream->close();
}

sub test_init_converter {
    my $self = $_[0];

    my $process =
      new Lire::DlfConverterProcess( $self->import_job( "test_line" ),
                                     $self->dlf_store() );

    $self->import_job( "test_line" )->converter( "line" );
    my $cfg = { 'myconfig' => 1 };
    $self->import_job( "test_line" )->converter_config( $cfg  );

    $self->assert_null( $process->{'_converter'},
                        '_converter should be undef');
    $process->_init_converter();

    $self->assert_isa( 'Lire::DlfConverter', $process->{'_converter'} );
    $self->assert_str_equals( $process, $process->{'_converter'}{'process'} );
    $self->assert_str_equals( 'line', $process->{'_converter'}->name() );
    $self->assert_str_equals( $cfg, $process->{'_converter'}{'config'} );

    return;
}

package tests::DlfConverterProcessTest::DlfConverter;

use base qw/Lire::DlfConverter/;

sub new {
    bless { 'name' => $_[1],
            'handle_lines' => $_[2],
          }, $_[0];
}

sub name { $_[0]{'name'} };

sub handle_log_lines { $_[0]{'handle_lines'} }

sub schemas { return ("test1", "test2") };

sub init_dlf_converter {
    $_[0]{'process'} = $_[1];
    $_[0]{'config'} = $_[2];
    return;
}

sub process_log_file {
    my ( $self, $process, $fh ) = @_;
    while ( <$fh> ) {
        chomp $_;
        $self->process_log_line( $process, $_ );
    }
}

sub process_log_line {
    my ( $self, $process, $line ) = @_;

    my %r = ();
    my @fields = split /;/, $line;
    foreach my $f ( @fields ) {
        my ( $key, $value ) = $f =~ /^(\w+)=(.+)$/;

        if ( defined $key && defined $value ) {
            $r{$key} = $value;
        } else {
            $process->error( "contains an invalid field", $line );
            return;
        }
    }

    if ( $r{'ignore'}) {
        $process->ignore_log_line( $line )
          if $self->{'handle_lines'};
    } elsif ( $r{'save'}) {
        $line =~ s/save=1/save=0/;
        $process->save_log_line( $line )
          if $self->{'handle_lines'};
    } else {
        $process->write_dlf( $r{'schema'}, \%r );
    }
}

sub finish_conversion {
    delete $_[0]{'process'};
    delete $_[0]{'config'};
    return;
}


1;