view release on metacpan or  search on metacpan

t/91-streamd.t  view on Meta::CPAN

#!perl
# 91-streamd.t
#
# Integration test for `iforest streamd`: starts the daemon in the
# foreground in a forked child on a temp Unix socket and drives it over
# the JSON-lines protocol.  Covers startup artefacts (socket, pid file),
# the request kinds (row / rows / cmd) and modes, client-tag echo on
# success and error, per-message error isolation, multiple concurrent
# connections, command and interval saves with the latest.json symlink,
# clean SIGTERM shutdown, resume-from-latest, and (when
# Algorithm::ToNumberMunger is installed) raw munged values that CSV
# framing could never carry.
#
# Skipped on Windows (Unix sockets + fork) and without JSON::MaybeXS.

use strict;
use warnings;
use Test::More;
use File::Temp qw(tempdir);
use File::Spec;
use IO::Select       ();
use IO::Socket::UNIX ();

my $bin = File::Spec->rel2abs('bin/iforest');
plan skip_all => 'bin/iforest not found' unless -x $bin;
plan skip_all => 'streamd needs Unix sockets and fork()' if $^O eq 'MSWin32';
plan skip_all => 'JSON::MaybeXS is not installed'
	unless eval { require JSON::MaybeXS; 1 };

my $JSON = JSON::MaybeXS->new( utf8 => 1 );

my $tmp    = tempdir( CLEANUP => 1 );
my $sock   = "$tmp/s.sock";
my $pidf   = "$tmp/s.pid";
my $mdir   = "$tmp/models";
my $logf   = "$tmp/streamd.log";
my $latest = "$mdir/latest.json";

# sun_path caps out around 104 bytes; a deep TMPDIR would make the whole
# run fail for reasons that have nothing to do with the daemon.
plan skip_all => 'temp socket path too long for a Unix socket'
	if length($sock) > 100;

my %BUF;         # per-connection read buffer, for line framing
my @ALL_PIDS;    # every daemon spawned, for the END sweep

# Fork+exec a daemon with the given argv and wait for its socket.
sub spawn_daemon {
	my ( $wait_sock, @argv ) = @_;
	my $pid = fork();
	die "fork failed: $!" unless defined $pid;
	if ( !$pid ) {
		open( STDOUT, '>>', $logf ) or die $!;
		open( STDERR, '>>', $logf ) or die $!;
		exec( $^X, '-Ilib', $bin, 'streamd', '-f', @argv ) or die "exec failed: $!";
	}
	push @ALL_PIDS, $pid;
	for ( 1 .. 100 ) {
		last if -S $wait_sock;
		select( undef, undef, undef, 0.1 );    ## no critic (ProhibitSleepViaSelect)
	}
	return $pid;
} ## end sub spawn_daemon

sub start_daemon {
	my (@extra) = @_;
	return spawn_daemon(
		$sock,
		'--socket'    => $sock,
		'--pid'       => $pidf,
		'--model-dir' => $mdir,
		'--log'       => $logf,
		@extra
	);
} ## end sub start_daemon

sub stop_daemon {
	my ($pid) = @_;
	kill( 'TERM', $pid );
	waitpid( $pid, 0 );
	return $? >> 8;
}

sub connect_client {
	my ($path) = @_;
	$path //= $sock;
	my $s = IO::Socket::UNIX->new( Peer => $path )
		or die "connect to $path failed: $!";
	$s->autoflush(1);
	$BUF{ fileno($s) } = '';