view release on metacpan or  search on metacpan

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

	like( rt( $c, { cmd => 'frobnicate' } )->{error},          qr/unknown cmd/,  'unknown cmd errors' );
	like( rt( $c, { row => [ 1, 2 ], mode => 'x' } )->{error}, qr/mode must be/, 'bad mode errors' );

	is_deeply( rt( $c, { cmd => 'ping' } ), { ok => 'pong' }, 'connection is alive after all of it' );
}; ## end 'errors are per-message and carry the tag' => sub

subtest 'multiple concurrent connections' => sub {
	my $c2 = connect_client();
	is( rt( $c2, { cmd => 'stats' } )->{ok}{connections}, 2, 'stats sees both connections' );

	# Per-connection mode: c2 goes learn-only, c stays prequential.
	is_deeply( rt( $c2, { cmd => 'mode', mode => 'learn' } ), { ok => { mode => 'learn' } }, 'mode command' );
	is_deeply( rt( $c2, { row => [ 0.3, 0.3 ] } ), { ok => { learned => 1 } },
		'c2 rows now learn without scoring' );
	ok( defined rt( $c, { row => [ 0.3, 0.3 ] } )->{score}, 'c still gets scores' );

	close $c2;
	# Give the daemon a beat to notice the close.
	select( undef, undef, undef, 0.3 );    ## no critic (ProhibitSleepViaSelect)
	is( rt( $c, { cmd => 'stats' } )->{ok}{connections}, 1, 'closed connection is reaped' );
}; ## end 'multiple concurrent connections' => sub

subtest 'saves: command, interval, symlink' => sub {
	my $r = rt( $c, { cmd => 'save', tag => 's1' } );
	like( $r->{ok}{saved}, qr/\Aoiforest-\d{8}-\d{6}(?:-\d+)?\.json\z/, 'save returns the file name' );
	is( $r->{tag}, 's1', 'save reply carries the tag' );
	ok( -f "$mdir/$r->{ok}{saved}", 'the timestamped file exists' );
	ok( -l $latest,                 'latest.json is a symlink' );
	is( readlink($latest), $r->{ok}{saved}, 'and points at the newest save (relative target)' );

	# Interval saves happen only after learning; learn then wait past the
	# 1s interval.
	my $count_before = () = glob("$mdir/oiforest-*.json");
	rt( $c, { row => [ 0.2, 0.8 ] } );
	select( undef, undef, undef, 2.5 );    ## no critic (ProhibitSleepViaSelect)
	rt( $c, { cmd => 'ping' } );           # tick the loop
	my $count_after = () = glob("$mdir/oiforest-*.json");
	cmp_ok( $count_after, '>', $count_before, 'a periodic save fired after learning' );
}; ## end 'saves: command, interval, symlink' => sub

my $seen_at_shutdown = rt( $c, { cmd => 'stats' } )->{ok}{seen};

subtest 'clean shutdown and resume' => sub {
	is( stop_daemon($daemon), 0, 'SIGTERM exits 0' );
	undef $daemon;
	ok( !-e $sock, 'socket removed' );
	ok( !-e $pidf, 'pid file removed' );

	# latest.json is a complete model; the parent class loads it and it
	# carries everything learned (the shutdown save flushed the tail).
	require Algorithm::Classifier::IsolationForest;
	my $m = Algorithm::Classifier::IsolationForest->load($latest);
	isa_ok( $m, 'Algorithm::Classifier::IsolationForest::Online', 'latest.json' );
	is( $m->seen, $seen_at_shutdown, 'shutdown save captured everything learned' );

	# Restart with no creation knobs: resumes from latest.json.
	$daemon = start_daemon( '--save-interval' => 60 );
	my $c3 = connect_client();
	is( rt( $c3, { cmd => 'stats' } )->{ok}{seen}, $seen_at_shutdown, 'restart resumed from latest.json' );
	close $c3;
	is( stop_daemon($daemon), 0, 'second daemon exits 0' );
	undef $daemon;
}; ## end 'clean shutdown and resume' => sub

subtest 'raw munged values that CSV could never carry' => sub {
	plan skip_all => 'Algorithm::ToNumberMunger is not installed'
		unless eval { require Algorithm::ToNumberMunger; 1 };

	# A munged online model via a prototype, in its own model dir.
	my $mdir2 = "$tmp/models2";
	my $proto = "$tmp/proto.json";
	{
		open my $fh, '>', $proto or die $!;
		print {$fh} $JSON->encode(
			{
				format             => 'Algorithm::Classifier::IsolationForest::Prototype',
				class              => 'online',
				schema_version     => 'd1',
				schema_description => 'streamd munger test',
				schema             => {
					feature_names => [ 'method', 'path_len' ],
					mungers       => {
						method   => { munger => 'http_method_enum', default => -1 },
						path_len => { munger => 'length',           from    => 'path' },
					},
				},
				params => { n_trees => 20, window_size => 64, max_leaf_samples => 8 },
			}
		);
		close $fh;
	}

	$daemon = start_daemon(
		'--save-interval' => 60,
		'--model-dir'     => $mdir2,
		'-s'              => 7,
		'--prototype'     => $proto
	);
	my $c4 = connect_client();

	# Raw values with embedded commas, quotes, newline escapes, and
	# unicode -- the reason the protocol is JSON.
	my @raw = map { { method => 'GET', path => '/a,b/"c"/' . ( 'p' x ( 3 + $_ % 15 ) ) . "\x{2603}" } } 1 .. 60;
	is_deeply(
		rt( $c4, { rows => \@raw, mode => 'learn' } ),
		{ ok => { learned => 60 } },
		'raw tagged rows with hostile content learn cleanly'
	);
	my $r = rt( $c4, { row => { method => 'BREW', path => '/' . ( 'a' x 90 ) . ',oh no' }, tag => "t,\x{2603}" } );
	like( $r->{score}, qr/\A[\d.eE+-]+\z/, 'raw anomalous row scores' );
	is( $r->{tag}, "t,\x{2603}", 'unicode/comma tag round-trips' );

	close $c4;
	is( stop_daemon($daemon), 0, 'munged daemon exits 0' );
	undef $daemon;
}; ## end 'raw munged values that CSV could never carry' => sub

subtest '--set runs named instances side by side' => sub {
	my $sdir  = "$tmp/sets";
	my @knobs = ( '--save-interval' => 60, '-n' => 10, '--window' => 64, '--eta' => 8 );
	my $alpha = spawn_daemon(