1791 lines
63 KiB
Dart
1791 lines
63 KiB
Dart
import 'dart:async';
|
|
import 'dart:convert';
|
|
import 'dart:io';
|
|
|
|
import 'package:http/http.dart' as http;
|
|
import 'package:test/test.dart';
|
|
import 'package:oto/oto/agent/agent_config.dart';
|
|
import 'package:oto/oto/agent/oto_server_job_client.dart';
|
|
import 'package:oto/oto/agent/registration_client.dart';
|
|
import 'package:oto/oto/agent/remote_run_executor.dart';
|
|
import 'package:oto/oto/application.dart';
|
|
import 'package:oto/oto/core/build_result.dart';
|
|
import 'package:oto/oto/core/execution_context.dart';
|
|
|
|
const _host = '127.0.0.1';
|
|
const _token = 'oto-smoke-token';
|
|
const _runnerId = 'oto-smoke-runner';
|
|
const _runnerAlias = 'oto-smoke-alias';
|
|
|
|
void main() {
|
|
group('socket-first transport smoke', () {
|
|
test(
|
|
'OtoServerSocketRegistrationClient socket session sends heartbeat and disconnects on close',
|
|
() async {
|
|
final port = await _freePort();
|
|
final serverAddr = '$_host:$port';
|
|
|
|
final (:process, :socketAddr) = await _startCoreWithSocket(serverAddr);
|
|
|
|
final output = StringBuffer();
|
|
final stdoutSub = process.stdout
|
|
.transform(systemEncoding.decoder)
|
|
.listen(output.write, onError: output.write);
|
|
final stderrSub = process.stderr
|
|
.transform(systemEncoding.decoder)
|
|
.listen(output.write, onError: output.write);
|
|
|
|
try {
|
|
await _waitForPort(_host, port, process, output);
|
|
|
|
final agentConfig = AgentConfig(
|
|
agent: const AgentIdentityConfig(
|
|
id: _runnerId,
|
|
alias: _runnerAlias,
|
|
enrollmentToken: _token,
|
|
),
|
|
server: ServerConnectionConfig(
|
|
url: 'http://$serverAddr',
|
|
socketUrl: 'tcp://$socketAddr',
|
|
),
|
|
runtime: const AgentRuntimeConfig(
|
|
installDir: '/tmp/install',
|
|
workspaceRoot: '/tmp/workspace',
|
|
logDir: '/tmp/log',
|
|
),
|
|
);
|
|
|
|
final client = OtoServerSocketRegistrationClient(
|
|
commandTypes: ['Shell', 'Git'],
|
|
heartbeatInterval: const Duration(milliseconds: 200),
|
|
);
|
|
|
|
final session = await client.openSession(agentConfig);
|
|
expect(session, isA<OtoServerPushJobSession>());
|
|
expect(session.result.accepted, isTrue);
|
|
expect(session.result.runnerId, _runnerId);
|
|
|
|
// Capture stream completion after close().
|
|
final pushSession = session as OtoServerPushJobSession;
|
|
final runRequestsDone = Completer<void>();
|
|
final cancelRequestsDone = Completer<void>();
|
|
final runSub = pushSession.runRequests.listen(
|
|
(_) {},
|
|
onDone: runRequestsDone.complete,
|
|
);
|
|
final cancelSub = pushSession.cancelRequests.listen(
|
|
(_) {},
|
|
onDone: cancelRequestsDone.complete,
|
|
);
|
|
|
|
final httpClient = http.Client();
|
|
try {
|
|
final isOnline = await _pollRunnerStatus(
|
|
httpClient,
|
|
serverAddr,
|
|
_runnerId,
|
|
'online',
|
|
);
|
|
expect(
|
|
isOnline,
|
|
isTrue,
|
|
reason: 'socket runner did not go online in registry',
|
|
);
|
|
|
|
// close() must cancel the heartbeat timer and close socket streams.
|
|
await session.close();
|
|
await runRequestsDone.future.timeout(
|
|
const Duration(seconds: 2),
|
|
onTimeout: () =>
|
|
fail('runRequests stream did not close after session.close()'),
|
|
);
|
|
await cancelRequestsDone.future.timeout(
|
|
const Duration(seconds: 2),
|
|
onTimeout: () => fail(
|
|
'cancelRequests stream did not close after session.close()',
|
|
),
|
|
);
|
|
await runSub.cancel();
|
|
await cancelSub.cancel();
|
|
|
|
final isDisconnected = await _pollRunnerStatus(
|
|
httpClient,
|
|
serverAddr,
|
|
_runnerId,
|
|
'disconnected',
|
|
timeout: const Duration(seconds: 5),
|
|
);
|
|
expect(
|
|
isDisconnected,
|
|
isTrue,
|
|
reason: 'socket runner did not transition to disconnected',
|
|
);
|
|
} finally {
|
|
httpClient.close();
|
|
}
|
|
} finally {
|
|
process.kill(ProcessSignal.sigterm);
|
|
await process.exitCode.timeout(
|
|
const Duration(seconds: 5),
|
|
onTimeout: () {
|
|
process.kill(ProcessSignal.sigkill);
|
|
return process.exitCode;
|
|
},
|
|
);
|
|
await stdoutSub.cancel();
|
|
await stderrSub.cancel();
|
|
}
|
|
},
|
|
timeout: const Timeout(Duration(seconds: 30)),
|
|
);
|
|
|
|
test(
|
|
'OtoServerSocketRegistrationClient receives socket cancel requests from HTTP cancel action',
|
|
() async {
|
|
final port = await _freePort();
|
|
final serverAddr = '$_host:$port';
|
|
const jobId = 'smoke-job-socket-cancel';
|
|
|
|
final (:process, :socketAddr) = await _startCoreWithSocket(serverAddr);
|
|
|
|
final output = StringBuffer();
|
|
final stdoutSub = process.stdout
|
|
.transform(systemEncoding.decoder)
|
|
.listen(output.write, onError: output.write);
|
|
final stderrSub = process.stderr
|
|
.transform(systemEncoding.decoder)
|
|
.listen(output.write, onError: output.write);
|
|
|
|
try {
|
|
await _waitForPort(_host, port, process, output);
|
|
|
|
final agentConfig = AgentConfig(
|
|
agent: const AgentIdentityConfig(
|
|
id: _runnerId,
|
|
alias: _runnerAlias,
|
|
enrollmentToken: _token,
|
|
),
|
|
server: ServerConnectionConfig(
|
|
url: 'http://$serverAddr',
|
|
socketUrl: 'tcp://$socketAddr',
|
|
),
|
|
runtime: const AgentRuntimeConfig(
|
|
installDir: '/tmp/install',
|
|
workspaceRoot: '/tmp/workspace',
|
|
logDir: '/tmp/log',
|
|
),
|
|
);
|
|
|
|
final client = OtoServerSocketRegistrationClient(
|
|
commandTypes: ['Shell'],
|
|
heartbeatInterval: const Duration(milliseconds: 200),
|
|
);
|
|
final session = await client.openSession(agentConfig);
|
|
expect(session.result.accepted, isTrue);
|
|
final pushSession = session as OtoServerPushJobSession;
|
|
|
|
final runReceived = Completer<String>();
|
|
final runSub = pushSession.runRequests.listen((request) {
|
|
if (!runReceived.isCompleted) {
|
|
expect(request.runnerId, _runnerId);
|
|
expect(request.jobId, jobId);
|
|
expect(request.executionId, isNotEmpty);
|
|
runReceived.complete(request.executionId);
|
|
}
|
|
});
|
|
var expectedExecutionId = '';
|
|
final cancelReceived = Completer<String>();
|
|
final cancelSub = pushSession.cancelRequests.listen((request) {
|
|
if (!cancelReceived.isCompleted) {
|
|
expect(request.runnerId, _runnerId);
|
|
expect(request.executionId, expectedExecutionId);
|
|
expect(request.reason, 'socket cancel smoke');
|
|
cancelReceived.complete(request.executionId);
|
|
}
|
|
});
|
|
|
|
final httpClient = http.Client();
|
|
final jobs = OtoServerJobClient(
|
|
serverUrl: 'http://$serverAddr',
|
|
runnerId: _runnerId,
|
|
client: httpClient,
|
|
);
|
|
try {
|
|
final createResp = await httpClient.post(
|
|
Uri.parse('http://$serverAddr/api/v1/jobs'),
|
|
headers: {'content-type': 'application/json'},
|
|
body: jsonEncode({
|
|
'id': jobId,
|
|
'name': 'socket cancel smoke',
|
|
'run_request': {
|
|
'pipeline_yaml': 'commands:\n - type: Shell',
|
|
'command_types': ['Shell'],
|
|
},
|
|
}),
|
|
);
|
|
expect(createResp.statusCode, equals(201));
|
|
|
|
expectedExecutionId = await runReceived.future.timeout(
|
|
const Duration(seconds: 3),
|
|
onTimeout: () => fail('socket RunRequest was not received'),
|
|
);
|
|
|
|
final cancelRes = await jobs.cancelRun(
|
|
executionId: expectedExecutionId,
|
|
reason: 'socket cancel smoke',
|
|
);
|
|
expect(cancelRes.success, isTrue);
|
|
|
|
await cancelReceived.future.timeout(
|
|
const Duration(seconds: 3),
|
|
onTimeout: () => fail('socket CancelRunRequest was not received'),
|
|
);
|
|
|
|
final execResp = await httpClient.get(
|
|
Uri.parse(
|
|
'http://$serverAddr/api/v1/executions/$expectedExecutionId',
|
|
),
|
|
);
|
|
final execJson = jsonDecode(execResp.body) as Map<String, dynamic>;
|
|
expect(execJson['state'], equals('canceled'));
|
|
} finally {
|
|
await runSub.cancel();
|
|
await cancelSub.cancel();
|
|
httpClient.close();
|
|
await session.close();
|
|
}
|
|
} finally {
|
|
process.kill(ProcessSignal.sigterm);
|
|
await process.exitCode.timeout(
|
|
const Duration(seconds: 5),
|
|
onTimeout: () {
|
|
process.kill(ProcessSignal.sigkill);
|
|
return process.exitCode;
|
|
},
|
|
);
|
|
await stdoutSub.cancel();
|
|
await stderrSub.cancel();
|
|
}
|
|
},
|
|
timeout: const Timeout(Duration(seconds: 30)),
|
|
);
|
|
|
|
test(
|
|
'OtoServerSocketRegistrationClient socket session close is idempotent after server disconnect',
|
|
() async {
|
|
final port = await _freePort();
|
|
final serverAddr = '$_host:$port';
|
|
|
|
final (:process, :socketAddr) = await _startCoreWithSocket(serverAddr);
|
|
|
|
final output = StringBuffer();
|
|
final stdoutSub = process.stdout
|
|
.transform(systemEncoding.decoder)
|
|
.listen(output.write, onError: output.write);
|
|
final stderrSub = process.stderr
|
|
.transform(systemEncoding.decoder)
|
|
.listen(output.write, onError: output.write);
|
|
|
|
try {
|
|
await _waitForPort(_host, port, process, output);
|
|
|
|
final agentConfig = AgentConfig(
|
|
agent: const AgentIdentityConfig(
|
|
id: _runnerId,
|
|
alias: _runnerAlias,
|
|
enrollmentToken: _token,
|
|
),
|
|
server: ServerConnectionConfig(
|
|
url: 'http://$serverAddr',
|
|
socketUrl: 'tcp://$socketAddr',
|
|
),
|
|
runtime: const AgentRuntimeConfig(
|
|
installDir: '/tmp/install',
|
|
workspaceRoot: '/tmp/workspace',
|
|
logDir: '/tmp/log',
|
|
),
|
|
);
|
|
|
|
final client = OtoServerSocketRegistrationClient(
|
|
commandTypes: ['Shell'],
|
|
// Short interval so in-flight heartbeats exercise the catch block.
|
|
heartbeatInterval: const Duration(milliseconds: 50),
|
|
);
|
|
|
|
final session = await client.openSession(agentConfig);
|
|
expect(session.result.accepted, isTrue);
|
|
|
|
// Graceful server shutdown: closes all proto-socket clients cleanly,
|
|
// sending EOF to the Dart side. In-flight heartbeat sendRequest calls
|
|
// receive StateError('connection closed'), which is swallowed by the
|
|
// _sendHeartbeat catch block.
|
|
process.kill(ProcessSignal.sigterm);
|
|
await process.exitCode.timeout(const Duration(seconds: 5));
|
|
|
|
// Allow the EOF to propagate and the transport to auto-close.
|
|
await Future<void>.delayed(const Duration(milliseconds: 300));
|
|
|
|
// session.close() must complete without throwing even though the
|
|
// proto-socket transport is already closed by the EOF handler.
|
|
await expectLater(session.close(), completes);
|
|
} finally {
|
|
process.kill(ProcessSignal.sigkill);
|
|
await process.exitCode.timeout(
|
|
const Duration(seconds: 2),
|
|
onTimeout: () => -1,
|
|
);
|
|
await stdoutSub.cancel();
|
|
await stderrSub.cancel();
|
|
}
|
|
},
|
|
timeout: const Timeout(Duration(seconds: 30)),
|
|
);
|
|
|
|
});
|
|
|
|
group('compatibility fallback HTTP', () {
|
|
test(
|
|
'OTO Dart runner registers with Go OTO Server via compatibility fallback HTTP, goes online, and disconnects on close',
|
|
() async {
|
|
final port = await _freePort();
|
|
final serverAddr = '$_host:$port';
|
|
|
|
// Start Go OTO Server in background
|
|
final serverProcess = await _startCore(serverAddr);
|
|
|
|
final output = StringBuffer();
|
|
final stdoutSub = serverProcess.stdout
|
|
.transform(systemEncoding.decoder)
|
|
.listen(output.write, onError: output.write);
|
|
final stderrSub = serverProcess.stderr
|
|
.transform(systemEncoding.decoder)
|
|
.listen(output.write, onError: output.write);
|
|
|
|
try {
|
|
// Wait for server port to listen
|
|
await _waitForPort(_host, port, serverProcess, output);
|
|
|
|
final agentConfig = AgentConfig(
|
|
agent: const AgentIdentityConfig(
|
|
id: _runnerId,
|
|
alias: _runnerAlias,
|
|
enrollmentToken: _token,
|
|
),
|
|
server: ServerConnectionConfig(url: 'http://$serverAddr'),
|
|
runtime: const AgentRuntimeConfig(
|
|
installDir: '/tmp/install',
|
|
workspaceRoot: '/tmp/workspace',
|
|
logDir: '/tmp/log',
|
|
),
|
|
);
|
|
|
|
// Open OTO Server session (with short heartbeat interval for testing)
|
|
final client = OtoServerRegistrationClient(
|
|
commandTypes: ['Shell', 'Git'],
|
|
heartbeatInterval: const Duration(milliseconds: 200),
|
|
);
|
|
|
|
final session = await client.openSession(agentConfig);
|
|
final result = session.result;
|
|
|
|
expect(result.accepted, isTrue);
|
|
expect(result.runnerId, _runnerId);
|
|
expect(result.alias, _runnerAlias);
|
|
|
|
// Poll the server's GET endpoint until status becomes 'online'
|
|
final httpClient = http.Client();
|
|
try {
|
|
final statusUrl = Uri.parse(
|
|
'http://$serverAddr/api/v1/runners/$_runnerId',
|
|
);
|
|
var isOnline = false;
|
|
final deadline = DateTime.now().add(const Duration(seconds: 10));
|
|
|
|
while (DateTime.now().isBefore(deadline)) {
|
|
final response = await httpClient.get(statusUrl);
|
|
if (response.statusCode == 200) {
|
|
final data = jsonDecode(response.body);
|
|
if (data['status'] == 'online') {
|
|
isOnline = true;
|
|
break;
|
|
}
|
|
}
|
|
await Future<void>.delayed(const Duration(milliseconds: 200));
|
|
}
|
|
|
|
expect(
|
|
isOnline,
|
|
isTrue,
|
|
reason: 'Runner did not transition to online state in registry',
|
|
);
|
|
|
|
// Close the session to trigger disconnect
|
|
await session.close();
|
|
|
|
// Verify status transitions to disconnected
|
|
var isDisconnected = false;
|
|
final disconnectDeadline = DateTime.now().add(
|
|
const Duration(seconds: 5),
|
|
);
|
|
|
|
while (DateTime.now().isBefore(disconnectDeadline)) {
|
|
final response = await httpClient.get(statusUrl);
|
|
if (response.statusCode == 200) {
|
|
final data = jsonDecode(response.body);
|
|
if (data['status'] == 'disconnected') {
|
|
isDisconnected = true;
|
|
break;
|
|
}
|
|
}
|
|
await Future<void>.delayed(const Duration(milliseconds: 200));
|
|
}
|
|
|
|
expect(
|
|
isDisconnected,
|
|
isTrue,
|
|
reason:
|
|
'Runner did not transition to disconnected state in registry',
|
|
);
|
|
} finally {
|
|
httpClient.close();
|
|
}
|
|
} finally {
|
|
serverProcess.kill(ProcessSignal.sigterm);
|
|
await serverProcess.exitCode.timeout(
|
|
const Duration(seconds: 5),
|
|
onTimeout: () {
|
|
serverProcess.kill(ProcessSignal.sigkill);
|
|
return serverProcess.exitCode;
|
|
},
|
|
);
|
|
await stdoutSub.cancel();
|
|
await stderrSub.cancel();
|
|
}
|
|
},
|
|
timeout: const Timeout(Duration(seconds: 30)),
|
|
);
|
|
|
|
test(
|
|
'OTO Server issues runner bootstrap command and serves bootstrap script',
|
|
() async {
|
|
final port = await _freePort();
|
|
final serverAddr = '$_host:$port';
|
|
|
|
// Start Go OTO Server in background
|
|
final serverProcess = await _startCore(
|
|
serverAddr,
|
|
releaseBaseUrl: 'https://example.com/releases',
|
|
);
|
|
|
|
final output = StringBuffer();
|
|
final stdoutSub = serverProcess.stdout
|
|
.transform(systemEncoding.decoder)
|
|
.listen(output.write, onError: output.write);
|
|
final stderrSub = serverProcess.stderr
|
|
.transform(systemEncoding.decoder)
|
|
.listen(output.write, onError: output.write);
|
|
|
|
try {
|
|
// Wait for server port to listen
|
|
await _waitForPort(_host, port, serverProcess, output);
|
|
|
|
final httpClient = http.Client();
|
|
try {
|
|
// 1. Verify bootstrap command endpoint
|
|
final bootstrapCmdUrl = Uri.parse(
|
|
'http://$serverAddr/api/v1/runners/bootstrap-command',
|
|
);
|
|
final response = await httpClient.post(
|
|
bootstrapCmdUrl,
|
|
headers: {'content-type': 'application/json'},
|
|
body: jsonEncode({
|
|
'runner_id': 'test-runner-id',
|
|
'token': 'test-token-123',
|
|
}),
|
|
);
|
|
|
|
expect(response.statusCode, equals(200));
|
|
final data = jsonDecode(response.body);
|
|
expect(data['bootstrap_command'], isNotNull);
|
|
final bootstrapCommand = data['bootstrap_command'] as String;
|
|
|
|
// Verify command contains server URL and token only. Runner ID is
|
|
// derived on the node by the bootstrap script.
|
|
expect(bootstrapCommand, contains('http://$serverAddr'));
|
|
expect(bootstrapCommand, isNot(contains('test-runner-id')));
|
|
expect(bootstrapCommand, contains('test-token-123'));
|
|
expect(bootstrapCommand, isNot(contains('--server-url')));
|
|
expect(bootstrapCommand, isNot(contains('--agent-id')));
|
|
expect(bootstrapCommand, contains('--token'));
|
|
expect(bootstrapCommand, isNot(contains('--enrollment-token')));
|
|
|
|
// 2. Full script execution test using fake curl/tar
|
|
final tempDir = await Directory.systemTemp.createTemp(
|
|
'oto_smoke_script_run_',
|
|
);
|
|
final tempHome = Directory('${tempDir.path}/home');
|
|
await tempHome.create(recursive: true);
|
|
// Create fake OTO executable source
|
|
final fakeOtoSource = Directory('${tempDir.path}/fake_oto_src');
|
|
await fakeOtoSource.create(recursive: true);
|
|
final fakeOtoFile = File('${fakeOtoSource.path}/oto');
|
|
await fakeOtoFile.writeAsString(
|
|
'#!/bin/sh\necho "fake-oto-started"\n',
|
|
);
|
|
await Process.run('chmod', ['+x', fakeOtoFile.path]);
|
|
|
|
// Create fake tar.gz archive
|
|
final fakeTarGz = File('${tempDir.path}/oto-linux-x64.tar.gz');
|
|
await Process.run('tar', [
|
|
'-czf',
|
|
fakeTarGz.path,
|
|
'-C',
|
|
fakeOtoSource.path,
|
|
'oto',
|
|
]);
|
|
|
|
// Create fake bin dir and fake curl
|
|
final fakeBinDir = Directory('${tempDir.path}/bin');
|
|
await fakeBinDir.create(recursive: true);
|
|
final realCurlResult = await Process.run('which', ['curl']);
|
|
final realCurlPath = realCurlResult.stdout.toString().trim();
|
|
expect(realCurlPath, isNotEmpty, reason: 'curl must be available');
|
|
final fakeCurl = File('${fakeBinDir.path}/curl');
|
|
await fakeCurl.writeAsString('''#!/bin/sh
|
|
out_file=""
|
|
url=""
|
|
while [ \$# -gt 0 ]; do
|
|
if [ "\$1" = "-o" ]; then
|
|
out_file="\$2"
|
|
shift 2
|
|
elif [ "\$1" = "-fsSL" ]; then
|
|
shift 1
|
|
else
|
|
url="\$1"
|
|
shift 1
|
|
fi
|
|
done
|
|
|
|
if [ -n "\$out_file" ]; then
|
|
cp "${fakeTarGz.path}" "\$out_file"
|
|
else
|
|
"$realCurlPath" -fsSL "\$url"
|
|
fi
|
|
''');
|
|
await Process.run('chmod', ['+x', fakeCurl.path]);
|
|
|
|
// Append custom paths and options to the command string
|
|
final customConfigPath = '${tempDir.path}/smoke-config.yaml';
|
|
final customInstallDir = '${tempDir.path}/install_dir';
|
|
final customWorkspaceRoot = '${tempDir.path}/workspace';
|
|
final customLogDir = '${tempDir.path}/log';
|
|
|
|
final fullCommand = [
|
|
bootstrapCommand,
|
|
"--config-path '$customConfigPath'",
|
|
"--install-dir '$customInstallDir'",
|
|
"--workspace-root '$customWorkspaceRoot'",
|
|
"--log-dir '$customLogDir'",
|
|
'--no-background',
|
|
].join(' ');
|
|
|
|
// Set up environment with fake curl path and custom HOME
|
|
final env = Map<String, String>.from(Platform.environment);
|
|
env['PATH'] = '${fakeBinDir.path}:${env['PATH']}';
|
|
env['HOME'] = tempHome.path;
|
|
|
|
// Run the full pipeline via bash
|
|
final scriptResult = await Process.run('bash', [
|
|
'-c',
|
|
fullCommand,
|
|
], environment: env);
|
|
|
|
expect(
|
|
scriptResult.exitCode,
|
|
equals(0),
|
|
reason:
|
|
'Script failed: ${scriptResult.stderr}\nStdout: ${scriptResult.stdout}',
|
|
);
|
|
|
|
// Assertions:
|
|
// - Generated config exists and has correct values
|
|
final configFile = File(customConfigPath);
|
|
expect(
|
|
await configFile.exists(),
|
|
isTrue,
|
|
reason: 'Config file not generated',
|
|
);
|
|
final configContent = await configFile.readAsString();
|
|
expect(configContent, contains('server:'));
|
|
expect(configContent, contains('url: "http://$serverAddr"'));
|
|
expect(configContent, matches(RegExp(r'id: "oto-[0-9a-f]{32}"')));
|
|
expect(configContent, contains('enrollment_token: "test-token-123"'));
|
|
|
|
// - Installed binary exists and has execution permissions
|
|
final installedOto = File('$customInstallDir/oto');
|
|
expect(
|
|
await installedOto.exists(),
|
|
isTrue,
|
|
reason: 'OTO binary not installed',
|
|
);
|
|
final stat = await installedOto.stat();
|
|
expect(
|
|
stat.mode & 0x49,
|
|
isNot(0),
|
|
reason: 'Installed binary not executable',
|
|
);
|
|
|
|
await tempDir.delete(recursive: true);
|
|
|
|
// 3. Verify bootstrap script hosting endpoint
|
|
final scriptUrl = Uri.parse(
|
|
'http://$serverAddr/bootstrap/oto-agent.sh',
|
|
);
|
|
final scriptResponse = await httpClient.get(scriptUrl);
|
|
expect(scriptResponse.statusCode, equals(200));
|
|
expect(
|
|
scriptResponse.headers['content-type'],
|
|
contains('application/x-sh'),
|
|
);
|
|
expect(scriptResponse.body, isNotEmpty);
|
|
expect(scriptResponse.body, contains('--server-url'));
|
|
} finally {
|
|
httpClient.close();
|
|
}
|
|
} finally {
|
|
serverProcess.kill(ProcessSignal.sigterm);
|
|
await serverProcess.exitCode.timeout(
|
|
const Duration(seconds: 5),
|
|
onTimeout: () {
|
|
serverProcess.kill(ProcessSignal.sigkill);
|
|
return serverProcess.exitCode;
|
|
},
|
|
);
|
|
await stdoutSub.cancel();
|
|
await stderrSub.cancel();
|
|
}
|
|
},
|
|
timeout: const Timeout(Duration(seconds: 30)),
|
|
);
|
|
|
|
test(
|
|
'OTO Server owns job execution logs and artifacts reported by runner via compatibility fallback HTTP',
|
|
() async {
|
|
final port = await _freePort();
|
|
final serverAddr = '$_host:$port';
|
|
const jobId = 'oto-smoke-job';
|
|
const executionId = 'oto-smoke-execution';
|
|
|
|
final serverProcess = await _startCore(serverAddr);
|
|
|
|
final output = StringBuffer();
|
|
final stdoutSub = serverProcess.stdout
|
|
.transform(systemEncoding.decoder)
|
|
.listen(output.write, onError: output.write);
|
|
final stderrSub = serverProcess.stderr
|
|
.transform(systemEncoding.decoder)
|
|
.listen(output.write, onError: output.write);
|
|
|
|
try {
|
|
await _waitForPort(_host, port, serverProcess, output);
|
|
|
|
final agentConfig = AgentConfig(
|
|
agent: const AgentIdentityConfig(
|
|
id: _runnerId,
|
|
alias: _runnerAlias,
|
|
enrollmentToken: _token,
|
|
),
|
|
server: ServerConnectionConfig(url: 'http://$serverAddr'),
|
|
runtime: const AgentRuntimeConfig(
|
|
installDir: '/tmp/install',
|
|
workspaceRoot: '/tmp/workspace',
|
|
logDir: '/tmp/log',
|
|
),
|
|
);
|
|
|
|
final registrationClient = OtoServerRegistrationClient(
|
|
commandTypes: ['Shell', 'Git'],
|
|
heartbeatInterval: const Duration(milliseconds: 200),
|
|
);
|
|
final session = await registrationClient.openSession(agentConfig);
|
|
try {
|
|
expect(session.result.accepted, isTrue);
|
|
expect(session, isA<OtoServerJobSession>());
|
|
final jobs = (session as OtoServerJobSession).jobs;
|
|
|
|
final httpClient = http.Client();
|
|
try {
|
|
const inlineYaml = 'commands:\n - type: Shell';
|
|
final createJobResponse = await httpClient.post(
|
|
Uri.parse('http://$serverAddr/api/v1/jobs'),
|
|
headers: {'content-type': 'application/json'},
|
|
body: jsonEncode({
|
|
'id': jobId,
|
|
'name': 'smoke build',
|
|
'run_request': {
|
|
'pipeline_yaml': inlineYaml,
|
|
'variables': {'FLAVOR': 'release'},
|
|
'command_types': ['Shell', 'Git'],
|
|
},
|
|
}),
|
|
);
|
|
expect(
|
|
createJobResponse.statusCode,
|
|
equals(201),
|
|
reason: 'Create job failed: ${createJobResponse.body}',
|
|
);
|
|
|
|
final claim = await jobs.claimJob(
|
|
jobId: jobId,
|
|
executionId: executionId,
|
|
);
|
|
expect(claim.accepted, isTrue);
|
|
expect(claim.state, 'running');
|
|
expect(claim.runRequest, isNotNull);
|
|
expect(claim.runRequest!.pipelineYaml, inlineYaml);
|
|
expect(claim.runRequest!.jobId, jobId);
|
|
expect(claim.runRequest!.executionId, executionId);
|
|
|
|
await jobs.appendLog(
|
|
executionId: executionId,
|
|
line: 'runner started smoke build',
|
|
);
|
|
await jobs.reportArtifact(
|
|
executionId: executionId,
|
|
name: 'smoke-report',
|
|
path: '/tmp/oto-smoke/report.txt',
|
|
);
|
|
final report = await jobs.reportExecution(
|
|
jobId: jobId,
|
|
executionId: executionId,
|
|
result: BuildResult.success(
|
|
stepEvents: [
|
|
StepEvent(
|
|
stepId: 0,
|
|
workflowIndex: 0,
|
|
stepType: 'command',
|
|
event: 'started',
|
|
timestamp: DateTime.now().toUtc().toIso8601String(),
|
|
commandId: 'smoke-build',
|
|
commandType: 'Shell',
|
|
),
|
|
StepEvent(
|
|
stepId: 0,
|
|
workflowIndex: 0,
|
|
stepType: 'command',
|
|
event: 'completed',
|
|
timestamp: DateTime.now().toUtc().toIso8601String(),
|
|
commandId: 'smoke-build',
|
|
commandType: 'Shell',
|
|
),
|
|
],
|
|
),
|
|
);
|
|
expect(report.accepted, isTrue);
|
|
expect(report.state, 'succeeded');
|
|
|
|
final job = await _getJson(
|
|
httpClient,
|
|
'http://$serverAddr/api/v1/jobs/$jobId',
|
|
);
|
|
expect(job['state'], 'succeeded');
|
|
expect(job['execution_id'], executionId);
|
|
|
|
final execution = await _getJson(
|
|
httpClient,
|
|
'http://$serverAddr/api/v1/executions/$executionId',
|
|
);
|
|
expect(execution['state'], 'succeeded');
|
|
expect(execution['job_id'], jobId);
|
|
|
|
final logs = await _getJson(
|
|
httpClient,
|
|
'http://$serverAddr/api/v1/executions/$executionId/logs',
|
|
);
|
|
final logLines = (logs['logs'] as List<dynamic>)
|
|
.map(
|
|
(entry) =>
|
|
(entry as Map<String, dynamic>)['Line'] ?? entry['line'],
|
|
)
|
|
.map((line) => line.toString())
|
|
.toList();
|
|
expect(logLines, contains('runner started smoke build'));
|
|
expect(logLines, contains('Build completed successfully.'));
|
|
|
|
final artifacts = await _getJson(
|
|
httpClient,
|
|
'http://$serverAddr/api/v1/executions/$executionId/artifacts',
|
|
);
|
|
final artifactRows = artifacts['artifacts'] as List<dynamic>;
|
|
expect(artifactRows, hasLength(1));
|
|
final artifact = artifactRows.single as Map<String, dynamic>;
|
|
expect(artifact['Name'] ?? artifact['name'], 'smoke-report');
|
|
expect(
|
|
artifact['Path'] ?? artifact['path'],
|
|
'/tmp/oto-smoke/report.txt',
|
|
);
|
|
} finally {
|
|
httpClient.close();
|
|
}
|
|
} finally {
|
|
await session.close();
|
|
}
|
|
} finally {
|
|
serverProcess.kill(ProcessSignal.sigterm);
|
|
await serverProcess.exitCode.timeout(
|
|
const Duration(seconds: 5),
|
|
onTimeout: () {
|
|
serverProcess.kill(ProcessSignal.sigkill);
|
|
return serverProcess.exitCode;
|
|
},
|
|
);
|
|
await stdoutSub.cancel();
|
|
await stderrSub.cancel();
|
|
}
|
|
},
|
|
timeout: const Timeout(Duration(seconds: 30)),
|
|
);
|
|
|
|
test(
|
|
'remoteRunExecutor runOnce executes job via Go OTO Server and reports step events (compatibility fallback HTTP)',
|
|
() async {
|
|
final port = await _freePort();
|
|
final serverAddr = '$_host:$port';
|
|
const jobId = 'remote-smoke-job';
|
|
const executionId = 'remote-smoke-exec';
|
|
|
|
final serverProcess = await _startCore(serverAddr);
|
|
|
|
final output = StringBuffer();
|
|
final stdoutSub = serverProcess.stdout
|
|
.transform(systemEncoding.decoder)
|
|
.listen(output.write, onError: output.write);
|
|
final stderrSub = serverProcess.stderr
|
|
.transform(systemEncoding.decoder)
|
|
.listen(output.write, onError: output.write);
|
|
|
|
try {
|
|
await _waitForPort(_host, port, serverProcess, output);
|
|
|
|
final agentConfig = AgentConfig(
|
|
agent: const AgentIdentityConfig(
|
|
id: _runnerId,
|
|
alias: _runnerAlias,
|
|
enrollmentToken: _token,
|
|
),
|
|
server: ServerConnectionConfig(url: 'http://$serverAddr'),
|
|
runtime: const AgentRuntimeConfig(
|
|
installDir: '/tmp/install',
|
|
workspaceRoot: '/tmp/workspace',
|
|
logDir: '/tmp/log',
|
|
),
|
|
);
|
|
|
|
final regClient = OtoServerRegistrationClient(
|
|
commandTypes: ['Shell', 'Git'],
|
|
);
|
|
final session = await regClient.openSession(agentConfig);
|
|
expect(
|
|
session.result.accepted,
|
|
isTrue,
|
|
reason: 'Registration was rejected',
|
|
);
|
|
final jobs = (session as OtoServerJobSession).jobs;
|
|
|
|
final httpClient = http.Client();
|
|
try {
|
|
// Create a job with valid pipeline YAML
|
|
final createResp = await httpClient.post(
|
|
Uri.parse('http://$serverAddr/api/v1/jobs'),
|
|
headers: {'content-type': 'application/json'},
|
|
body: jsonEncode({
|
|
'id': jobId,
|
|
'name': 'remote run smoke',
|
|
'run_request': {
|
|
'pipeline_yaml': '''
|
|
property:
|
|
workspace: .
|
|
commands:
|
|
- command: Print
|
|
id: say-hello
|
|
param:
|
|
message: hello from smoke
|
|
pipeline:
|
|
id: smoke-pipe
|
|
workflow:
|
|
- exe: say-hello
|
|
''',
|
|
'command_types': ['Print'],
|
|
},
|
|
}),
|
|
);
|
|
expect(
|
|
createResp.statusCode,
|
|
equals(201),
|
|
reason: 'Create job failed: ${createResp.body}',
|
|
);
|
|
|
|
final claim = await jobs.claimJob(
|
|
jobId: jobId,
|
|
executionId: executionId,
|
|
);
|
|
expect(claim.accepted, isTrue);
|
|
expect(claim.runRequest, isNotNull);
|
|
|
|
// Execute via RemoteRunExecutor.runOnce
|
|
final parser = RunRequestParser(workspaceRoot: '/tmp/workspace');
|
|
final executor = RemoteRunExecutor(
|
|
parser: parser,
|
|
jobClient: jobs,
|
|
onLog: (msg) {},
|
|
);
|
|
|
|
final buildResult = await executor.runOnce(
|
|
jobId: claim.jobId,
|
|
executionId: claim.executionId,
|
|
runRequest: claim.runRequest!,
|
|
);
|
|
|
|
expect(
|
|
buildResult.success,
|
|
isTrue,
|
|
reason: 'Execution failed: ${buildResult.message}',
|
|
);
|
|
expect(buildResult.exitCode, 0);
|
|
expect(buildResult.stepEvents, isNotEmpty);
|
|
|
|
final events = buildResult.stepEvents.map((e) => e.event).toList();
|
|
expect(events, contains('started'));
|
|
expect(events, contains('completed'));
|
|
|
|
// Verify logs on server
|
|
final logsResp = await httpClient.get(
|
|
Uri.parse('http://$serverAddr/api/v1/executions/$executionId/logs'),
|
|
);
|
|
final logsJson = jsonDecode(logsResp.body) as Map<String, dynamic>;
|
|
final logsList = logsJson['logs'] as List<dynamic>;
|
|
final lines = logsList
|
|
.map((e) => (e as Map<String, dynamic>)['Line'] ?? e['line'])
|
|
.map((l) => l.toString())
|
|
.toList();
|
|
expect(lines, contains('run started for job $jobId'));
|
|
expect(lines, contains('Build completed successfully.'));
|
|
|
|
// Verify execution state
|
|
final execResp = await httpClient.get(
|
|
Uri.parse('http://$serverAddr/api/v1/executions/$executionId'),
|
|
);
|
|
expect(execResp.statusCode, 200);
|
|
final execJson = jsonDecode(execResp.body) as Map<String, dynamic>;
|
|
expect(execJson['state'], equals('succeeded'));
|
|
expect(execJson['job_id'], equals(jobId));
|
|
|
|
httpClient.close();
|
|
} finally {
|
|
await session.close();
|
|
}
|
|
} finally {
|
|
serverProcess.kill(ProcessSignal.sigterm);
|
|
await serverProcess.exitCode.timeout(
|
|
const Duration(seconds: 5),
|
|
onTimeout: () {
|
|
serverProcess.kill(ProcessSignal.sigkill);
|
|
return serverProcess.exitCode;
|
|
},
|
|
);
|
|
await stdoutSub.cancel();
|
|
await stderrSub.cancel();
|
|
}
|
|
},
|
|
timeout: const Timeout(Duration(seconds: 30)),
|
|
);
|
|
|
|
test(
|
|
'remoteRunExecutor runOnce merges variables into pipeline property (compatibility fallback HTTP) (REVIEW_API-3)',
|
|
() async {
|
|
final port = await _freePort();
|
|
final serverAddr = '$_host:$port';
|
|
const jobId = 'var-merge-job';
|
|
const executionId = 'var-merge-exec';
|
|
|
|
final serverProcess = await _startCore(serverAddr);
|
|
|
|
final output = StringBuffer();
|
|
final stdoutSub = serverProcess.stdout
|
|
.transform(systemEncoding.decoder)
|
|
.listen(output.write, onError: output.write);
|
|
final stderrSub = serverProcess.stderr
|
|
.transform(systemEncoding.decoder)
|
|
.listen(output.write, onError: output.write);
|
|
|
|
try {
|
|
await _waitForPort(_host, port, serverProcess, output);
|
|
|
|
final agentConfig = AgentConfig(
|
|
agent: const AgentIdentityConfig(
|
|
id: _runnerId,
|
|
alias: _runnerAlias,
|
|
enrollmentToken: _token,
|
|
),
|
|
server: ServerConnectionConfig(url: 'http://$serverAddr'),
|
|
runtime: const AgentRuntimeConfig(
|
|
installDir: '/tmp/install',
|
|
workspaceRoot: '/tmp/workspace',
|
|
logDir: '/tmp/log',
|
|
),
|
|
);
|
|
|
|
final regClient = OtoServerRegistrationClient(commandTypes: ['Print']);
|
|
final session = await regClient.openSession(agentConfig);
|
|
expect(
|
|
session.result.accepted,
|
|
isTrue,
|
|
reason: 'Registration rejected',
|
|
);
|
|
final jobs = (session as OtoServerJobSession).jobs;
|
|
|
|
final httpClient = http.Client();
|
|
try {
|
|
// YAML uses <!property.FLAVOR> tag; variables must override it.
|
|
final createResp = await httpClient.post(
|
|
Uri.parse('http://$serverAddr/api/v1/jobs'),
|
|
headers: {'content-type': 'application/json'},
|
|
body: jsonEncode({
|
|
'id': jobId,
|
|
'name': 'variable merge smoke',
|
|
'run_request': {
|
|
'pipeline_yaml': '''
|
|
property:
|
|
workspace: .
|
|
FLAVOR: original
|
|
commands:
|
|
- command: Print
|
|
id: print-flavor
|
|
param:
|
|
message: "<!property.FLAVOR>"
|
|
pipeline:
|
|
id: var-pipe
|
|
workflow:
|
|
- exe: print-flavor
|
|
''',
|
|
'variables': {'FLAVOR': 'release'},
|
|
'command_types': ['Print'],
|
|
},
|
|
}),
|
|
);
|
|
expect(
|
|
createResp.statusCode,
|
|
equals(201),
|
|
reason: 'Create job failed: ${createResp.body}',
|
|
);
|
|
|
|
final claim = await jobs.claimJob(
|
|
jobId: jobId,
|
|
executionId: executionId,
|
|
);
|
|
expect(claim.accepted, isTrue);
|
|
expect(claim.runRequest, isNotNull);
|
|
|
|
final parser = RunRequestParser(workspaceRoot: '/tmp/workspace');
|
|
final capturedLogs = <String>[];
|
|
final executor = RemoteRunExecutor(
|
|
parser: parser,
|
|
jobClient: jobs,
|
|
onLog: capturedLogs.add,
|
|
);
|
|
|
|
final buildResult = await executor.runOnce(
|
|
jobId: claim.jobId,
|
|
executionId: claim.executionId,
|
|
runRequest: claim.runRequest!,
|
|
);
|
|
|
|
expect(
|
|
buildResult.success,
|
|
isTrue,
|
|
reason: 'Execution failed: ${buildResult.message}',
|
|
);
|
|
expect(
|
|
buildResult.stepEvents,
|
|
isNotEmpty,
|
|
reason: 'Expected step events from the build',
|
|
);
|
|
|
|
// REVIEW_API2-2: assert the remote variable actually overrode the
|
|
// YAML value. After a successful build, Application.instance.property
|
|
// holds the merged, tag-replaced property map that the pipeline used
|
|
// to resolve `<!property.FLAVOR>`. It must be the remote 'release',
|
|
// not the YAML default 'original'.
|
|
final mergedFlavor = Application.instance.property['FLAVOR'];
|
|
expect(
|
|
mergedFlavor,
|
|
equals('release'),
|
|
reason:
|
|
'remote variable FLAVOR was not substituted; got: $mergedFlavor',
|
|
);
|
|
expect(
|
|
mergedFlavor,
|
|
isNot(equals('original')),
|
|
reason:
|
|
'YAML default "original" leaked through instead of the remote override',
|
|
);
|
|
|
|
expect(
|
|
capturedLogs.any((l) => l.contains('success=true')),
|
|
isTrue,
|
|
reason: 'Expected successful execution report; logs: $capturedLogs',
|
|
);
|
|
} finally {
|
|
httpClient.close();
|
|
await session.close();
|
|
}
|
|
} finally {
|
|
serverProcess.kill(ProcessSignal.sigterm);
|
|
await serverProcess.exitCode.timeout(
|
|
const Duration(seconds: 5),
|
|
onTimeout: () {
|
|
serverProcess.kill(ProcessSignal.sigkill);
|
|
return serverProcess.exitCode;
|
|
},
|
|
);
|
|
await stdoutSub.cancel();
|
|
await stderrSub.cancel();
|
|
}
|
|
},
|
|
timeout: const Timeout(Duration(seconds: 30)),
|
|
);
|
|
|
|
test(
|
|
'remoteRunExecutor runOnce reports failure via compatibility fallback HTTP when property is invalid despite variables (REVIEW_API3-1)',
|
|
() async {
|
|
final port = await _freePort();
|
|
final serverAddr = '$_host:$port';
|
|
const jobId = 'bad-prop-job';
|
|
const executionId = 'bad-prop-exec';
|
|
|
|
final serverProcess = await _startCore(serverAddr);
|
|
|
|
final output = StringBuffer();
|
|
final stdoutSub = serverProcess.stdout
|
|
.transform(systemEncoding.decoder)
|
|
.listen(output.write, onError: output.write);
|
|
final stderrSub = serverProcess.stderr
|
|
.transform(systemEncoding.decoder)
|
|
.listen(output.write, onError: output.write);
|
|
|
|
try {
|
|
await _waitForPort(_host, port, serverProcess, output);
|
|
|
|
final agentConfig = AgentConfig(
|
|
agent: const AgentIdentityConfig(
|
|
id: _runnerId,
|
|
alias: _runnerAlias,
|
|
enrollmentToken: _token,
|
|
),
|
|
server: ServerConnectionConfig(url: 'http://$serverAddr'),
|
|
runtime: const AgentRuntimeConfig(
|
|
installDir: '/tmp/install',
|
|
workspaceRoot: '/tmp/workspace',
|
|
logDir: '/tmp/log',
|
|
),
|
|
);
|
|
|
|
final regClient = OtoServerRegistrationClient(commandTypes: ['Print']);
|
|
final session = await regClient.openSession(agentConfig);
|
|
expect(
|
|
session.result.accepted,
|
|
isTrue,
|
|
reason: 'Registration rejected',
|
|
);
|
|
final jobs = (session as OtoServerJobSession).jobs;
|
|
|
|
final httpClient = http.Client();
|
|
try {
|
|
// `property` is a list (invalid). Remote variables must NOT coerce it
|
|
// into a valid map; Application.build validation must still reject it.
|
|
final createResp = await httpClient.post(
|
|
Uri.parse('http://$serverAddr/api/v1/jobs'),
|
|
headers: {'content-type': 'application/json'},
|
|
body: jsonEncode({
|
|
'id': jobId,
|
|
'name': 'invalid property smoke',
|
|
'run_request': {
|
|
'pipeline_yaml': '''
|
|
property:
|
|
- bad
|
|
commands:
|
|
- command: Print
|
|
id: print-flavor
|
|
param:
|
|
message: "<!property.FLAVOR>"
|
|
pipeline:
|
|
id: bad-pipe
|
|
workflow:
|
|
- exe: print-flavor
|
|
''',
|
|
'variables': {'FLAVOR': 'release'},
|
|
'command_types': ['Print'],
|
|
},
|
|
}),
|
|
);
|
|
expect(
|
|
createResp.statusCode,
|
|
equals(201),
|
|
reason: 'Create job failed: ${createResp.body}',
|
|
);
|
|
|
|
final claim = await jobs.claimJob(
|
|
jobId: jobId,
|
|
executionId: executionId,
|
|
);
|
|
expect(claim.accepted, isTrue);
|
|
expect(claim.runRequest, isNotNull);
|
|
|
|
final parser = RunRequestParser(workspaceRoot: '/tmp/workspace');
|
|
final executor = RemoteRunExecutor(
|
|
parser: parser,
|
|
jobClient: jobs,
|
|
onLog: (msg) {},
|
|
);
|
|
|
|
final buildResult = await executor.runOnce(
|
|
jobId: claim.jobId,
|
|
executionId: claim.executionId,
|
|
runRequest: claim.runRequest!,
|
|
);
|
|
|
|
// Build must fail on the invalid property; the merge must not have
|
|
// rewritten it into a valid map.
|
|
expect(
|
|
buildResult.success,
|
|
isFalse,
|
|
reason: 'Expected failure from invalid property type',
|
|
);
|
|
|
|
// Server job should be in failed state because reportExecution ran.
|
|
final jobResp = await httpClient.get(
|
|
Uri.parse('http://$serverAddr/api/v1/jobs/$jobId'),
|
|
);
|
|
expect(jobResp.statusCode, 200);
|
|
final jobJson = jsonDecode(jobResp.body) as Map<String, dynamic>;
|
|
expect(
|
|
jobJson['state'],
|
|
equals('failed'),
|
|
reason:
|
|
'Server did not receive failure report; got: ${jobJson['state']}',
|
|
);
|
|
} finally {
|
|
httpClient.close();
|
|
await session.close();
|
|
}
|
|
} finally {
|
|
serverProcess.kill(ProcessSignal.sigterm);
|
|
await serverProcess.exitCode.timeout(
|
|
const Duration(seconds: 5),
|
|
onTimeout: () {
|
|
serverProcess.kill(ProcessSignal.sigkill);
|
|
return serverProcess.exitCode;
|
|
},
|
|
);
|
|
await stdoutSub.cancel();
|
|
await stderrSub.cancel();
|
|
}
|
|
},
|
|
timeout: const Timeout(Duration(seconds: 30)),
|
|
);
|
|
|
|
test(
|
|
'remoteRunExecutor runOnce sends failure report via compatibility fallback HTTP when parse fails (REVIEW_API-5)',
|
|
() async {
|
|
final port = await _freePort();
|
|
final serverAddr = '$_host:$port';
|
|
const jobId = 'parse-fail-job';
|
|
const executionId = 'parse-fail-exec';
|
|
|
|
final serverProcess = await _startCore(serverAddr);
|
|
|
|
final output = StringBuffer();
|
|
final stdoutSub = serverProcess.stdout
|
|
.transform(systemEncoding.decoder)
|
|
.listen(output.write, onError: output.write);
|
|
final stderrSub = serverProcess.stderr
|
|
.transform(systemEncoding.decoder)
|
|
.listen(output.write, onError: output.write);
|
|
|
|
try {
|
|
await _waitForPort(_host, port, serverProcess, output);
|
|
|
|
final agentConfig = AgentConfig(
|
|
agent: const AgentIdentityConfig(
|
|
id: _runnerId,
|
|
alias: _runnerAlias,
|
|
enrollmentToken: _token,
|
|
),
|
|
server: ServerConnectionConfig(url: 'http://$serverAddr'),
|
|
runtime: const AgentRuntimeConfig(
|
|
installDir: '/tmp/install',
|
|
workspaceRoot: '/tmp/workspace',
|
|
logDir: '/tmp/log',
|
|
),
|
|
);
|
|
|
|
final regClient = OtoServerRegistrationClient(commandTypes: ['Shell']);
|
|
final session = await regClient.openSession(agentConfig);
|
|
expect(
|
|
session.result.accepted,
|
|
isTrue,
|
|
reason: 'Registration rejected',
|
|
);
|
|
final jobs = (session as OtoServerJobSession).jobs;
|
|
|
|
final httpClient = http.Client();
|
|
try {
|
|
// Create a job with a path that escapes the workspace root
|
|
// (the parser will throw ArgumentError, triggering REVIEW_API-5 path).
|
|
final createResp = await httpClient.post(
|
|
Uri.parse('http://$serverAddr/api/v1/jobs'),
|
|
headers: {'content-type': 'application/json'},
|
|
body: jsonEncode({
|
|
'id': jobId,
|
|
'name': 'parse fail smoke',
|
|
'run_request': {
|
|
'pipeline_yaml_path': '/etc/passwd',
|
|
'command_types': ['Shell'],
|
|
},
|
|
}),
|
|
);
|
|
expect(
|
|
createResp.statusCode,
|
|
equals(201),
|
|
reason: 'Create job failed: ${createResp.body}',
|
|
);
|
|
|
|
final claim = await jobs.claimJob(
|
|
jobId: jobId,
|
|
executionId: executionId,
|
|
);
|
|
expect(claim.accepted, isTrue);
|
|
expect(claim.runRequest, isNotNull);
|
|
|
|
// workspaceRoot=/tmp/workspace; /etc/passwd is outside → parse error.
|
|
final parser = RunRequestParser(workspaceRoot: '/tmp/workspace');
|
|
final executor = RemoteRunExecutor(
|
|
parser: parser,
|
|
jobClient: jobs,
|
|
onLog: (msg) {},
|
|
);
|
|
|
|
final buildResult = await executor.runOnce(
|
|
jobId: claim.jobId,
|
|
executionId: claim.executionId,
|
|
runRequest: claim.runRequest!,
|
|
);
|
|
|
|
// Build should have failed.
|
|
expect(
|
|
buildResult.success,
|
|
isFalse,
|
|
reason: 'Expected failure from parse error',
|
|
);
|
|
|
|
// Server job should be in failed state because reportExecution was called.
|
|
final jobResp = await httpClient.get(
|
|
Uri.parse('http://$serverAddr/api/v1/jobs/$jobId'),
|
|
);
|
|
expect(jobResp.statusCode, 200);
|
|
final jobJson = jsonDecode(jobResp.body) as Map<String, dynamic>;
|
|
expect(
|
|
jobJson['state'],
|
|
equals('failed'),
|
|
reason:
|
|
'Server did not receive failure report; got: ${jobJson['state']}',
|
|
);
|
|
} finally {
|
|
httpClient.close();
|
|
await session.close();
|
|
}
|
|
} finally {
|
|
serverProcess.kill(ProcessSignal.sigterm);
|
|
await serverProcess.exitCode.timeout(
|
|
const Duration(seconds: 5),
|
|
onTimeout: () {
|
|
serverProcess.kill(ProcessSignal.sigkill);
|
|
return serverProcess.exitCode;
|
|
},
|
|
);
|
|
await stdoutSub.cancel();
|
|
await stderrSub.cancel();
|
|
}
|
|
},
|
|
timeout: const Timeout(Duration(seconds: 30)),
|
|
);
|
|
|
|
test(
|
|
'remoteRunExecutor runOnce reports declared artifacts via compatibility fallback HTTP (API-3)',
|
|
() async {
|
|
final port = await _freePort();
|
|
final serverAddr = '$_host:$port';
|
|
const jobId = 'artifact-smoke-job';
|
|
const executionId = 'artifact-smoke-exec';
|
|
|
|
final serverProcess = await _startCore(serverAddr);
|
|
|
|
final output = StringBuffer();
|
|
final stdoutSub = serverProcess.stdout
|
|
.transform(systemEncoding.decoder)
|
|
.listen(output.write, onError: output.write);
|
|
final stderrSub = serverProcess.stderr
|
|
.transform(systemEncoding.decoder)
|
|
.listen(output.write, onError: output.write);
|
|
|
|
try {
|
|
await _waitForPort(_host, port, serverProcess, output);
|
|
|
|
final agentConfig = AgentConfig(
|
|
agent: const AgentIdentityConfig(
|
|
id: _runnerId,
|
|
alias: _runnerAlias,
|
|
enrollmentToken: _token,
|
|
),
|
|
server: ServerConnectionConfig(url: 'http://$serverAddr'),
|
|
runtime: const AgentRuntimeConfig(
|
|
installDir: '/tmp/install',
|
|
workspaceRoot: '/tmp/workspace',
|
|
logDir: '/tmp/log',
|
|
),
|
|
);
|
|
|
|
final regClient = OtoServerRegistrationClient(commandTypes: ['Print']);
|
|
final session = await regClient.openSession(agentConfig);
|
|
expect(
|
|
session.result.accepted,
|
|
isTrue,
|
|
reason: 'Registration rejected',
|
|
);
|
|
final jobs = (session as OtoServerJobSession).jobs;
|
|
|
|
final httpClient = http.Client();
|
|
try {
|
|
// The pipeline declares an artifact via property.artifacts; the runner
|
|
// must report that metadata to the server after the execution report.
|
|
final createResp = await httpClient.post(
|
|
Uri.parse('http://$serverAddr/api/v1/jobs'),
|
|
headers: {'content-type': 'application/json'},
|
|
body: jsonEncode({
|
|
'id': jobId,
|
|
'name': 'artifact event smoke',
|
|
'run_request': {
|
|
'pipeline_yaml': '''
|
|
property:
|
|
workspace: .
|
|
artifacts:
|
|
- name: smoke-report
|
|
path: dist/report.json
|
|
commands:
|
|
- command: Print
|
|
id: say-hello
|
|
param:
|
|
message: hello from artifact smoke
|
|
pipeline:
|
|
id: artifact-pipe
|
|
workflow:
|
|
- exe: say-hello
|
|
''',
|
|
'command_types': ['Print'],
|
|
},
|
|
}),
|
|
);
|
|
expect(
|
|
createResp.statusCode,
|
|
equals(201),
|
|
reason: 'Create job failed: ${createResp.body}',
|
|
);
|
|
|
|
final claim = await jobs.claimJob(
|
|
jobId: jobId,
|
|
executionId: executionId,
|
|
);
|
|
expect(claim.accepted, isTrue);
|
|
expect(claim.runRequest, isNotNull);
|
|
|
|
// workspaceRoot confines the declared relative artifact path.
|
|
final parser = RunRequestParser(workspaceRoot: '/tmp/workspace');
|
|
final executor = RemoteRunExecutor(
|
|
parser: parser,
|
|
jobClient: jobs,
|
|
onLog: (msg) {},
|
|
);
|
|
|
|
final buildResult = await executor.runOnce(
|
|
jobId: claim.jobId,
|
|
executionId: claim.executionId,
|
|
runRequest: claim.runRequest!,
|
|
);
|
|
|
|
expect(
|
|
buildResult.success,
|
|
isTrue,
|
|
reason: 'Execution failed: ${buildResult.message}',
|
|
);
|
|
expect(buildResult.artifacts, hasLength(1));
|
|
expect(buildResult.artifacts.first.name, 'smoke-report');
|
|
|
|
// The declared artifact must be persisted on the server's artifacts
|
|
// endpoint by the remote loop, not only carried in the build result.
|
|
final artifacts = await _getJson(
|
|
httpClient,
|
|
'http://$serverAddr/api/v1/executions/$executionId/artifacts',
|
|
);
|
|
final artifactRows = artifacts['artifacts'] as List<dynamic>;
|
|
expect(artifactRows, hasLength(1));
|
|
final artifact = artifactRows.single as Map<String, dynamic>;
|
|
expect(artifact['Name'] ?? artifact['name'], 'smoke-report');
|
|
expect(artifact['Path'] ?? artifact['path'], 'dist/report.json');
|
|
} finally {
|
|
httpClient.close();
|
|
await session.close();
|
|
}
|
|
} finally {
|
|
serverProcess.kill(ProcessSignal.sigterm);
|
|
await serverProcess.exitCode.timeout(
|
|
const Duration(seconds: 5),
|
|
onTimeout: () {
|
|
serverProcess.kill(ProcessSignal.sigkill);
|
|
return serverProcess.exitCode;
|
|
},
|
|
);
|
|
await stdoutSub.cancel();
|
|
await stderrSub.cancel();
|
|
}
|
|
},
|
|
timeout: const Timeout(Duration(seconds: 30)),
|
|
);
|
|
|
|
test(
|
|
'OTO Server cancel, status, and self-update via compatibility fallback HTTP',
|
|
() async {
|
|
final port = await _freePort();
|
|
final serverAddr = '$_host:$port';
|
|
const jobId = 'smoke-job-cancel';
|
|
const executionId = 'smoke-exec-cancel';
|
|
|
|
final serverProcess = await _startCore(serverAddr);
|
|
|
|
final output = StringBuffer();
|
|
final stdoutSub = serverProcess.stdout
|
|
.transform(systemEncoding.decoder)
|
|
.listen(output.write, onError: output.write);
|
|
final stderrSub = serverProcess.stderr
|
|
.transform(systemEncoding.decoder)
|
|
.listen(output.write, onError: output.write);
|
|
|
|
try {
|
|
await _waitForPort(_host, port, serverProcess, output);
|
|
|
|
final agentConfig = AgentConfig(
|
|
agent: const AgentIdentityConfig(
|
|
id: _runnerId,
|
|
alias: _runnerAlias,
|
|
enrollmentToken: _token,
|
|
),
|
|
server: ServerConnectionConfig(url: 'http://$serverAddr'),
|
|
runtime: const AgentRuntimeConfig(
|
|
installDir: '/tmp/install',
|
|
workspaceRoot: '/tmp/workspace',
|
|
logDir: '/tmp/log',
|
|
),
|
|
);
|
|
|
|
final regClient = OtoServerRegistrationClient(
|
|
commandTypes: ['Shell', 'Git'],
|
|
);
|
|
final session = await regClient.openSession(agentConfig);
|
|
expect(session.result.accepted, isTrue);
|
|
final jobs = (session as OtoServerJobSession).jobs;
|
|
|
|
final httpClient = http.Client();
|
|
try {
|
|
// 1. Create a job
|
|
final createResp = await httpClient.post(
|
|
Uri.parse('http://$serverAddr/api/v1/jobs'),
|
|
headers: {'content-type': 'application/json'},
|
|
body: jsonEncode({
|
|
'id': jobId,
|
|
'name': 'smoke build',
|
|
'run_request': {
|
|
'pipeline_yaml': 'commands:\n - type: Shell',
|
|
'command_types': ['Shell'],
|
|
},
|
|
}),
|
|
);
|
|
expect(createResp.statusCode, equals(201));
|
|
|
|
// 2. Claim job
|
|
final claim = await jobs.claimJob(
|
|
jobId: jobId,
|
|
executionId: executionId,
|
|
);
|
|
expect(claim.accepted, isTrue);
|
|
|
|
// 3. Check status (should be running, and current execution should be set)
|
|
final statusRes = await jobs.fetchRunnerStatus();
|
|
expect(statusRes.accepted, isTrue);
|
|
expect(statusRes.runnerId, _runnerId);
|
|
expect(statusRes.currentExecutionId, executionId);
|
|
expect(statusRes.currentJobId, jobId);
|
|
|
|
// 4. Request self-update (should be deferred because execution is running)
|
|
final updateRes = await jobs.requestSelfUpdate(
|
|
version: 'v2.0.0',
|
|
downloadUrl: 'https://example.com/binary',
|
|
);
|
|
expect(updateRes.accepted, isFalse);
|
|
expect(updateRes.deferred, isTrue);
|
|
|
|
// 5. Cancel running job
|
|
final cancelRes = await jobs.cancelRun(
|
|
executionId: executionId,
|
|
reason: 'smoke testing cancel',
|
|
);
|
|
expect(cancelRes.success, isTrue);
|
|
|
|
// Verify execution is canceled in store
|
|
final execResp = await httpClient.get(
|
|
Uri.parse('http://$serverAddr/api/v1/executions/$executionId'),
|
|
);
|
|
final execJson = jsonDecode(execResp.body) as Map<String, dynamic>;
|
|
expect(execJson['state'], equals('canceled'));
|
|
|
|
// 6. Request self-update again (should be accepted now that execution is canceled/idle)
|
|
final updateRes2 = await jobs.requestSelfUpdate(
|
|
version: 'v2.0.0',
|
|
downloadUrl: 'https://example.com/binary',
|
|
);
|
|
expect(updateRes2.accepted, isTrue);
|
|
expect(updateRes2.deferred, isFalse);
|
|
} finally {
|
|
httpClient.close();
|
|
await session.close();
|
|
}
|
|
} finally {
|
|
serverProcess.kill(ProcessSignal.sigterm);
|
|
await serverProcess.exitCode.timeout(
|
|
const Duration(seconds: 5),
|
|
onTimeout: () {
|
|
serverProcess.kill(ProcessSignal.sigkill);
|
|
return serverProcess.exitCode;
|
|
},
|
|
);
|
|
await stdoutSub.cancel();
|
|
await stderrSub.cancel();
|
|
}
|
|
},
|
|
timeout: const Timeout(Duration(seconds: 30)),
|
|
);
|
|
|
|
});
|
|
}
|
|
|
|
// ─── Test helpers ──────────────────────────────────────────────────────────────
|
|
|
|
typedef _CoreWithSocket = ({Process process, String socketAddr});
|
|
|
|
Future<_CoreWithSocket> _startCoreWithSocket(String serverAddr) async {
|
|
final socketPort = await _freePort();
|
|
final socketAddr = '$_host:$socketPort';
|
|
final process = await Process.start(
|
|
'go',
|
|
['run', './cmd/oto-core'],
|
|
workingDirectory: '../../services/core',
|
|
environment: {
|
|
'OTO_CORE_ADDR': serverAddr,
|
|
'OTO_RUNNER_SOCKET_ADDR': socketAddr,
|
|
'OTO_RUNNER_SOCKET_PUBLIC_URL': 'tcp://$socketAddr',
|
|
},
|
|
);
|
|
return (process: process, socketAddr: socketAddr);
|
|
}
|
|
|
|
Future<bool> _pollRunnerStatus(
|
|
http.Client client,
|
|
String serverAddr,
|
|
String runnerId,
|
|
String expected, {
|
|
Duration timeout = const Duration(seconds: 10),
|
|
}) async {
|
|
final url = Uri.parse('http://$serverAddr/api/v1/runners/$runnerId');
|
|
final deadline = DateTime.now().add(timeout);
|
|
while (DateTime.now().isBefore(deadline)) {
|
|
final resp = await client.get(url);
|
|
if (resp.statusCode == 200) {
|
|
final data = jsonDecode(resp.body);
|
|
if (data['status'] == expected) return true;
|
|
}
|
|
await Future<void>.delayed(const Duration(milliseconds: 100));
|
|
}
|
|
return false;
|
|
}
|
|
|
|
Future<int> _freePort() async {
|
|
final socket = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0);
|
|
final port = socket.port;
|
|
await socket.close();
|
|
return port;
|
|
}
|
|
|
|
Future<Process> _startCore(String serverAddr, {String? releaseBaseUrl}) async {
|
|
final socketPort = await _freePort();
|
|
final environment = <String, String>{
|
|
'OTO_CORE_ADDR': serverAddr,
|
|
'OTO_RUNNER_SOCKET_ADDR': '$_host:$socketPort',
|
|
'OTO_RUNNER_SOCKET_PUBLIC_URL': 'tcp://$_host:$socketPort',
|
|
};
|
|
if (releaseBaseUrl != null) {
|
|
environment['OTO_RUNNER_RELEASE_BASE_URL'] = releaseBaseUrl;
|
|
}
|
|
return Process.start(
|
|
'go',
|
|
['run', './cmd/oto-core'],
|
|
workingDirectory: '../../services/core',
|
|
environment: environment,
|
|
);
|
|
}
|
|
|
|
Future<void> _waitForPort(
|
|
String host,
|
|
int port,
|
|
Process process,
|
|
StringBuffer output,
|
|
) async {
|
|
final deadline = DateTime.now().add(const Duration(seconds: 15));
|
|
while (DateTime.now().isBefore(deadline)) {
|
|
final exit = await _tryExitCode(process);
|
|
if (exit != null) {
|
|
fail('OTO Core exited before listening (code $exit):\n$output');
|
|
}
|
|
try {
|
|
final socket = await Socket.connect(
|
|
host,
|
|
port,
|
|
timeout: const Duration(milliseconds: 200),
|
|
);
|
|
await socket.close();
|
|
return;
|
|
} catch (_) {
|
|
await Future<void>.delayed(const Duration(milliseconds: 100));
|
|
}
|
|
}
|
|
fail('timed out waiting for OTO Core on $host:$port:\n$output');
|
|
}
|
|
|
|
Future<int?> _tryExitCode(Process process) {
|
|
return process.exitCode
|
|
.timeout(Duration.zero, onTimeout: () => -1)
|
|
.then((code) => code == -1 ? null : code);
|
|
}
|
|
|
|
Future<Map<String, dynamic>> _getJson(http.Client client, String url) async {
|
|
final response = await client.get(Uri.parse(url));
|
|
expect(response.statusCode, equals(200), reason: response.body);
|
|
return jsonDecode(response.body) as Map<String, dynamic>;
|
|
}
|