diff --git a/.gitignore b/.gitignore index 82d7bb9..c1ea49a 100644 --- a/.gitignore +++ b/.gitignore @@ -10,3 +10,4 @@ composer.phar *.env.json /.phpunit.result.cache /phpunit.xml +/docs/superpowers diff --git a/README.md b/README.md index f95a596..6376fbf 100644 --- a/README.md +++ b/README.md @@ -488,6 +488,57 @@ Behavior: - Prints a summary: projects checked/disabled/errored, configurations scanned/touched, tasks migrated/skipped (unsupported vs. unresolvable). +### Migrate keboola.orchestrator configurations to keboola.flow +Batch driver for the automated `keboola.orchestrator` → `keboola.flow` migration +(see [AJDA-3117](https://linear.app/keboola/issue/AJDA-3117)). All migration logic lives in the +`keboola.flow-migration-tool` component; this command only creates one migration job per project +and supervises the batch. Safe to re-run with the same list: already-migrated orchestrations are +reported as skipped by the component, and projects with a live migration job are skipped here. + +``` +php cli.php manage:migrate-orchestrations-to-flow [-f|--force] [] \ + [--projects-file=PATH] [--concurrency=10] [--poll-interval=5] [--report=PATH] +``` + +Arguments: +- `token` (required): Manage API token. +- `url` (required): Stack URL, including `https://` (e.g. `https://connection.north-europe.azure.keboola.com`). +- `projects` (optional): Comma-separated project IDs (e.g. `1,7,146`), or `@path/to/file` with one ID + per line (blank lines and `#` comments are ignored). Exactly one of `projects`/`--projects-file` + must be given. + +Options: +- `--force` / `-f`: Run the real migration. Without it, jobs are created with `dryRun: true`. + **Note:** even without `--force` a real `keboola.flow-migration-tool` job and a real ephemeral + storage token are created in every eligible project — on PAYGO stacks mind the billing. +- `--projects-file=PATH`: File with one project ID per line (alternative to `@file` in the argument). +- `--concurrency=N` (default 10): Max migration jobs in flight at once. +- `--poll-interval=N` (default 5): Seconds between job status polls. +- `--report=PATH` (default `flow-migration--.csv`): CSV report path. + +Behavior: +- For each project: skips disabled/deleted projects; creates an ephemeral 1h storage token with full + project rights (`canManageBuckets`, `canManageTokens`, `canReadAllFileUploads`, `canPurgeTrash`) so + the component cannot be short of a permission mid-migration; skips projects with no + `keboola.orchestrator` configurations (no empty jobs in customers' job history). +- Creates the migration job via `configData` (no stored configuration is left behind) with + `parameters: {mode: "project", orchestrationIds: [], skipBroken: true, dryRun: }`. +- Keeps at most `--concurrency` jobs in flight, polls each job and refills the window as jobs finish. + A transient poll failure is tolerated up to 3 consecutive times per job; after that the project is + reported as failed and the job is left to finish server-side (its job ID stays in the report). +- Appends a CSV row (`projectId;jobId;status;durationSeconds;error`) the moment each project + resolves, so an interrupted run is still auditable. Every input project gets a row; skipped + projects carry the skip reason in `status`/`error` and an empty `jobId`. Re-running with the same + `--report` path appends to the existing file without repeating the header. +- A failing project never aborts the batch. Exit code is `1` if at least one project failed + (job `error`/`terminated`/`cancelled` or a driver-side error), `0` otherwise. +- Final summary: projects attempted / migrated / migrated with warning / skipped (no + orchestrations, disabled) / failed. +- Re-running the same project list is the intended recovery path: the component reports + already-migrated orchestrations as skipped. The command does **not** check for a migration job + already running in the project, so before re-running an interrupted batch let the jobs it already + created finish - two concurrent migrations of one project can both create the same flows. + ### Mass enablement of dynamic backends for multiple projects Prerequisities: https://keboola.atlassian.net/wiki/spaces/KB/pages/2135982081/Enable+Dynamic+Backends#Enable-for-project diff --git a/cli.php b/cli.php index 2908e23..4fd4f42 100644 --- a/cli.php +++ b/cli.php @@ -21,6 +21,7 @@ use Keboola\Console\Command\MassProjectEnableDynamicBackends; use Keboola\Console\Command\MassProjectExtendExpiration; use Keboola\Console\Command\MigrateDataAppsOrchestratorTasks; +use Keboola\Console\Command\MigrateOrchestrationsToFlow; use Keboola\Console\Command\OrganizationIntoMaintenanceMode; use Keboola\Console\Command\OrganizationResetWorkspacePasswords; use Keboola\Console\Command\OrganizationsAddFeature; @@ -48,6 +49,7 @@ $application->add(new MassProjectExtendExpiration()); $application->add(new MassProjectEnableDynamicBackends()); $application->add(new MigrateDataAppsOrchestratorTasks()); +$application->add(new MigrateOrchestrationsToFlow()); $application->add(new AddFeature()); $application->add(new CleanupLeakedTestFeatures()); $application->add(new AllStacksIterator()); diff --git a/src/Keboola/Console/Command/FlowMigration/BatchRunner.php b/src/Keboola/Console/Command/FlowMigration/BatchRunner.php new file mode 100644 index 0000000..9681546 --- /dev/null +++ b/src/Keboola/Console/Command/FlowMigration/BatchRunner.php @@ -0,0 +1,416 @@ += 1, validated by the calling command + */ + public function __construct( + ProjectClientsFactory $clientsFactory, + int $concurrency, + int $pollIntervalSeconds, + ?callable $sleep = null + ) { + $this->clientsFactory = $clientsFactory; + $this->concurrency = $concurrency; + $this->pollIntervalSeconds = $pollIntervalSeconds; + $this->sleep = $sleep ?? function (int $seconds): void { + sleep($seconds); + }; + } + + /** + * @param array $projectIds + * @param callable(ProjectResult): void $onProjectFinished invoked once per input project + * @return BatchSummary + */ + public function run(array $projectIds, bool $force, OutputInterface $output, callable $onProjectFinished): array + { + $pending = array_values(array_unique($projectIds)); + $summary = [ + 'attempted' => count($pending), + 'migrated' => 0, + 'migratedWithWarning' => 0, + 'skippedNoOrchestrations' => 0, + 'skippedDisabled' => 0, + 'failed' => 0, + ]; + /** @var array $inFlight */ + $inFlight = []; + + while ($pending !== [] || $inFlight !== []) { + while (count($inFlight) < $this->concurrency && $pending !== []) { + $submission = $this->submitProject(array_shift($pending), $force, $output); + if ($submission instanceof ProjectResult) { + $this->recordResult($submission, $summary, $output, $onProjectFinished); + continue; + } + $inFlight[] = $submission; + } + + if ($inFlight === []) { + continue; + } + + ($this->sleep)($this->pollIntervalSeconds); + $inFlight = $this->pollInFlightJobs($inFlight, $summary, $output, $onProjectFinished); + } + + return $summary; + } + + /** + * Runs the per-project pipeline up to job creation. Returns an in-flight slot on success, + * or an immediately-final ProjectResult (skip or driver-side error). + * + * Order matters: the disabled/deleted check runs before any token is created, and the + * configuration check runs before the queue guard so empty projects never appear in + * customers' job history. + * + * @return ProjectResult|InFlightJob + */ + private function submitProject(string $projectId, bool $force, OutputInterface $output) + { + try { + $resolved = $this->checkProjectIsActive($projectId); + if ($resolved !== null) { + return $resolved; + } + + $clients = $this->clientsFactory->createProjectClients($projectId); + + $resolved = $this->checkProjectNeedsMigration($projectId, $clients); + if ($resolved !== null) { + return $resolved; + } + + $job = $clients->queueClient->createJob($this->buildMigrationJobData($force)); + } catch (Throwable $e) { + return new ProjectResult( + $projectId, + null, + ProjectResult::STATUS_ERROR, + null, + $e->getMessage() + ); + } + + $this->writeLine($output, sprintf('Project %s: created job %s (%s)', $projectId, $job->id, $job->url)); + + return [ + 'projectId' => $projectId, + 'jobId' => $job->id, + 'queueClient' => $clients->queueClient, + 'startedAt' => microtime(true), + 'pollFailures' => 0, + ]; + } + + /** + * Runs before any ephemeral token is created. A deleted project (Manage API 404) is reported + * together with disabled ones - the migration has nothing to do in either case. Any other + * failure is rethrown so the caller turns it into a per-project error result. + * + * @return ProjectResult|null non-null when the project must not be migrated + */ + private function checkProjectIsActive(string $projectId): ?ProjectResult + { + try { + $project = $this->clientsFactory->getProject($projectId); + } catch (ManageClientException $e) { + if ($e->getCode() !== 404) { + throw $e; + } + + return new ProjectResult( + $projectId, + null, + ProjectResult::STATUS_SKIPPED_DISABLED, + null, + 'project is deleted' + ); + } + + if (isset($project['isDisabled']) && $project['isDisabled']) { + return new ProjectResult( + $projectId, + null, + ProjectResult::STATUS_SKIPPED_DISABLED, + null, + 'project is disabled' + ); + } + + return null; + } + + /** + * A project with no keboola.orchestrator configurations gets no job at all, so hundreds of + * empty jobs never show up in customers' job history. + * + * @return ProjectResult|null non-null when no new migration job should be created + */ + private function checkProjectNeedsMigration( + string $projectId, + ProjectClients $clients + ): ?ProjectResult { + $configurations = $clients->components->listComponentConfigurations( + (new ListComponentConfigurationsOptions()) + ->setComponentId(self::ORCHESTRATOR_COMPONENT_ID) + ->setIsDeleted(false) + ); + if (count($configurations) === 0) { + return new ProjectResult( + $projectId, + null, + ProjectResult::STATUS_SKIPPED_NO_ORCHESTRATIONS, + null, + 'no keboola.orchestrator configurations' + ); + } + + return null; + } + + /** + * The migration is requested through configData so no stored configuration is left behind + * in the customer's project. orchestrationIds and skipBroken are required by the component's + * config definition in "project" mode. + */ + private function buildMigrationJobData(bool $force): JobData + { + return new JobData( + self::MIGRATION_COMPONENT_ID, + null, + [ + 'parameters' => [ + 'mode' => 'project', + 'orchestrationIds' => [], + 'skipBroken' => true, + 'dryRun' => !$force, + ], + ] + ); + } + + /** + * Polls every in-flight job once and returns the slots that are still running. + * + * @param array $inFlight + * @param BatchSummary $summary + * @param callable(ProjectResult): void $onProjectFinished + * @return array + */ + private function pollInFlightJobs( + array $inFlight, + array &$summary, + OutputInterface $output, + callable $onProjectFinished + ): array { + $stillRunning = []; + foreach ($inFlight as $slot) { + $polled = $this->pollOneJob($slot, $summary, $output, $onProjectFinished); + if ($polled !== null) { + $stillRunning[] = $polled; + } + } + + return $stillRunning; + } + + /** + * @param InFlightJob $slot + * @param BatchSummary $summary + * @param callable(ProjectResult): void $onProjectFinished + * @return InFlightJob|null null once the project is resolved and its slot is freed + */ + private function pollOneJob( + array $slot, + array &$summary, + OutputInterface $output, + callable $onProjectFinished + ): ?array { + try { + $job = $slot['queueClient']->getJob($slot['jobId']); + } catch (Throwable $e) { + return $this->handlePollFailure($slot, $e, $summary, $output, $onProjectFinished); + } + + $slot['pollFailures'] = 0; + + if (!$job->isFinished) { + return $slot; + } + + $this->recordResult($this->buildFinishedJobResult($slot, $job), $summary, $output, $onProjectFinished); + + return null; + } + + /** + * Keeps the slot in flight until MAX_CONSECUTIVE_POLL_FAILURES is reached, then resolves the + * project as failed while preserving the job id - the job itself may still finish server-side. + * + * @param InFlightJob $slot + * @param BatchSummary $summary + * @param callable(ProjectResult): void $onProjectFinished + * @return InFlightJob|null + */ + private function handlePollFailure( + array $slot, + Throwable $error, + array &$summary, + OutputInterface $output, + callable $onProjectFinished + ): ?array { + $slot['pollFailures']++; + $this->writeLine($output, sprintf( + 'Project %s: polling job %s failed (%d/%d): %s', + $slot['projectId'], + $slot['jobId'], + $slot['pollFailures'], + self::MAX_CONSECUTIVE_POLL_FAILURES, + $error->getMessage() + )); + + if ($slot['pollFailures'] < self::MAX_CONSECUTIVE_POLL_FAILURES) { + return $slot; + } + + $this->recordResult( + new ProjectResult( + $slot['projectId'], + $slot['jobId'], + ProjectResult::STATUS_ERROR, + null, + sprintf( + 'polling gave up after %d consecutive failures, job may still be running: %s', + self::MAX_CONSECUTIVE_POLL_FAILURES, + $error->getMessage() + ) + ), + $summary, + $output, + $onProjectFinished + ); + + return null; + } + + /** + * @param InFlightJob $slot + */ + private function buildFinishedJobResult(array $slot, Job $job): ProjectResult + { + return new ProjectResult( + $slot['projectId'], + $slot['jobId'], + $job->status, + $job->durationSeconds ?? (int) round(microtime(true) - $slot['startedAt']), + $this->extractJobError($job) + ); + } + + private function extractJobError(Job $job): ?string + { + if ($job->isSuccess()) { + return null; + } + $result = $job->result; + if (is_array($result) && isset($result['message']) && is_scalar($result['message'])) { + return (string) $result['message']; + } + + return null; + } + + /** + * @param BatchSummary $summary + * @param callable(ProjectResult): void $onProjectFinished + */ + private function recordResult( + ProjectResult $result, + array &$summary, + OutputInterface $output, + callable $onProjectFinished + ): void { + $summary[$this->summaryKeyFor($result)]++; + + $this->writeLine($output, sprintf( + 'Project %s: %s%s%s', + $result->projectId, + $result->status, + $result->durationSeconds !== null ? sprintf(' in %d s', $result->durationSeconds) : '', + $result->error !== null ? sprintf(' (%s)', $result->error) : '' + )); + + $onProjectFinished($result); + } + + /** + * @return key-of + */ + private function summaryKeyFor(ProjectResult $result): string + { + return match ($result->status) { + JobStatuses::SUCCESS->value => 'migrated', + JobStatuses::WARNING->value => 'migratedWithWarning', + ProjectResult::STATUS_SKIPPED_NO_ORCHESTRATIONS => 'skippedNoOrchestrations', + ProjectResult::STATUS_SKIPPED_DISABLED => 'skippedDisabled', + default => 'failed', + }; + } + + private function writeLine(OutputInterface $output, string $message): void + { + $output->writeln(sprintf('[%s] %s', date('H:i:s'), $message)); + } +} diff --git a/src/Keboola/Console/Command/FlowMigration/ProjectClients.php b/src/Keboola/Console/Command/FlowMigration/ProjectClients.php new file mode 100644 index 0000000..fb9b7d6 --- /dev/null +++ b/src/Keboola/Console/Command/FlowMigration/ProjectClients.php @@ -0,0 +1,23 @@ +components = $components; + $this->queueClient = $queueClient; + } +} diff --git a/src/Keboola/Console/Command/FlowMigration/ProjectClientsFactory.php b/src/Keboola/Console/Command/FlowMigration/ProjectClientsFactory.php new file mode 100644 index 0000000..e335411 --- /dev/null +++ b/src/Keboola/Console/Command/FlowMigration/ProjectClientsFactory.php @@ -0,0 +1,69 @@ +manageClient = $manageClient; + $this->connectionUrl = $connectionUrl; + $this->queueApiUrl = $queueApiUrl; + } + + /** + * @return array Manage API project detail + */ + public function getProject(string $projectId): array + { + return $this->manageClient->getProject($projectId); + } + + public function createProjectClients(string $projectId): ProjectClients + { + // Full project rights on purpose. The migration writes configurations, tables, triggers + // (with a runWithTokenId copied from the source trigger, i.e. another token) and + // notification subscriptions, and a migration failing halfway through is worse than a + // short-lived privileged token. Restricting this only risks the component missing something. + $tokenInfo = $this->manageClient->createProjectStorageToken($projectId, [ + 'description' => self::TOKEN_DESCRIPTION, + 'expiresIn' => self::TOKEN_EXPIRES_IN_SECONDS, + 'canManageBuckets' => true, + 'canManageTokens' => true, + 'canReadAllFileUploads' => true, + 'canPurgeTrash' => true, + ]); + + $storageClient = new StorageClient([ + 'url' => $this->connectionUrl, + 'token' => $tokenInfo['token'], + ]); + + return new ProjectClients( + new Components($storageClient), + new JobQueueClient($this->queueApiUrl, $tokenInfo['token']) + ); + } +} diff --git a/src/Keboola/Console/Command/FlowMigration/ProjectResult.php b/src/Keboola/Console/Command/FlowMigration/ProjectResult.php new file mode 100644 index 0000000..6c855f0 --- /dev/null +++ b/src/Keboola/Console/Command/FlowMigration/ProjectResult.php @@ -0,0 +1,54 @@ +projectId = $projectId; + $this->jobId = $jobId; + $this->status = $status; + $this->durationSeconds = $durationSeconds; + $this->error = $error; + } + + public function isSkipped(): bool + { + return in_array($this->status, [ + self::STATUS_SKIPPED_DISABLED, + self::STATUS_SKIPPED_NO_ORCHESTRATIONS, + ], true); + } + + public function isFailed(): bool + { + // Anything that is not a skip and not a successful terminal job status is a failure - + // unexpected statuses fail loud rather than passing silently. + return !$this->isSkipped() + && !in_array($this->status, [JobStatuses::SUCCESS->value, JobStatuses::WARNING->value], true); + } +} diff --git a/src/Keboola/Console/Command/MigrateOrchestrationsToFlow.php b/src/Keboola/Console/Command/MigrateOrchestrationsToFlow.php new file mode 100644 index 0000000..e2ed09b --- /dev/null +++ b/src/Keboola/Console/Command/MigrateOrchestrationsToFlow.php @@ -0,0 +1,405 @@ + keboola.flow migration (AJDA-3117). + * All migration logic lives in the keboola.flow-migration-tool component; this command only + * creates and supervises its jobs across a list of projects on one stack. + */ +class MigrateOrchestrationsToFlow extends Command +{ + const ARG_TOKEN = 'token'; + const ARG_URL = 'url'; + const ARG_PROJECTS = 'projects'; + const OPT_FORCE = 'force'; + const OPT_PROJECTS_FILE = 'projects-file'; + const OPT_CONCURRENCY = 'concurrency'; + const OPT_POLL_INTERVAL = 'poll-interval'; + const OPT_REPORT = 'report'; + + private const ERROR_ONE_PROJECT_SOURCE = + 'Provide exactly one source of project IDs: the argument or --projects-file'; + + private const CSV_HEADER = ['projectId', 'jobId', 'status', 'durationSeconds', 'error']; + private const CSV_DELIMITER = ';'; + private const CSV_ENCLOSURE = '"'; + // No proprietary escaping (same default as keboola/csv): quotes are doubled, so an API error + // message containing \" cannot break the row for Excel/Sheets or any RFC-4180 parser. + private const CSV_ESCAPE = ''; + + protected function configure(): void + { + $this + ->setName('manage:migrate-orchestrations-to-flow') + ->setDescription( + 'Run the automated keboola.orchestrator -> keboola.flow migration for a batch of projects' + ) + ->addArgument(self::ARG_TOKEN, InputArgument::REQUIRED, 'Manage API token') + ->addArgument( + self::ARG_URL, + InputArgument::REQUIRED, + 'Stack URL, e.g. https://connection.north-europe.azure.keboola.com' + ) + ->addArgument( + self::ARG_PROJECTS, + InputArgument::OPTIONAL, + 'Comma-separated project IDs, or @path/to/file with one ID per line' + ) + ->addOption( + self::OPT_FORCE, + 'f', + InputOption::VALUE_NONE, + 'Run the real migration; without it jobs are created with dryRun: true' + ) + ->addOption( + self::OPT_PROJECTS_FILE, + null, + InputOption::VALUE_REQUIRED, + 'File with one project ID per line (alternative to @file in the argument)' + ) + ->addOption( + self::OPT_CONCURRENCY, + null, + InputOption::VALUE_REQUIRED, + 'Max migration jobs in flight at once', + '10' + ) + ->addOption( + self::OPT_POLL_INTERVAL, + null, + InputOption::VALUE_REQUIRED, + 'Seconds between job status polls', + '5' + ) + ->addOption( + self::OPT_REPORT, + null, + InputOption::VALUE_REQUIRED, + 'CSV report path (default: flow-migration--.csv)' + ); + } + + protected function execute(InputInterface $input, OutputInterface $output): int + { + $token = $input->getArgument(self::ARG_TOKEN); + assert(is_string($token)); + $url = $input->getArgument(self::ARG_URL); + assert(is_string($url)); + $force = (bool) $input->getOption(self::OPT_FORCE); + + $hostnameSuffix = $this->hostnameSuffixFromUrl($url); + if ($hostnameSuffix === null) { + $output->writeln(sprintf( + 'Invalid stack URL "%s": expected a URL like https://connection.keboola.com', + $url + )); + return 1; + } + + $projectIds = $this->resolveProjectIds($input, $output); + if ($projectIds === null) { + return 1; + } + + $concurrency = $this->parsePositiveIntOption($input, self::OPT_CONCURRENCY); + $pollInterval = $this->parsePositiveIntOption($input, self::OPT_POLL_INTERVAL); + if ($concurrency === null || $pollInterval === null) { + $output->writeln('Options --concurrency and --poll-interval must be positive integers'); + return 1; + } + + $reportPath = $this->resolveReportPath($input, $hostnameSuffix); + $this->printRunNotice($output, $force, count($projectIds), $concurrency, $pollInterval, $reportPath); + + $reportHandle = $this->openReport($reportPath, $output); + if ($reportHandle === null) { + return 1; + } + + $manageClient = new ManageClient(['url' => $url, 'token' => $token]); + $clientsFactory = new ProjectClientsFactory( + $manageClient, + $url, + (new ServiceClient($hostnameSuffix))->getQueueUrl() + ); + + $runner = new BatchRunner($clientsFactory, $concurrency, $pollInterval); + $summary = $runner->run( + $projectIds, + $force, + $output, + function (ProjectResult $result) use ($reportHandle): void { + $this->appendReportRow($reportHandle, $result); + } + ); + fclose($reportHandle); + + $this->printSummary($output, $summary); + + return $summary['failed'] > 0 ? 1 : 0; + } + + /** + * Derives the ServiceClient hostname suffix from a full connection URL, e.g. + * "https://connection.north-europe.azure.keboola.com" -> "north-europe.azure.keboola.com". + * Returns null when the URL does not look like a stack connection URL. + * + * @return non-empty-string|null + */ + private function hostnameSuffixFromUrl(string $url): ?string + { + $host = parse_url($url, PHP_URL_HOST); + if (!is_string($host) || !str_starts_with($host, 'connection.')) { + return null; + } + $suffix = substr($host, strlen('connection.')); + + return $suffix === '' ? null : $suffix; + } + + /** + * Resolves the project ID list from exactly one source: the argument + * (inline list or @file) or --projects-file. Prints an error and returns null otherwise. + * + * @return array|null + */ + private function resolveProjectIds(InputInterface $input, OutputInterface $output): ?array + { + $inlineList = $this->optionalStringInput($input->getArgument(self::ARG_PROJECTS)); + $filePath = $this->optionalStringInput($input->getOption(self::OPT_PROJECTS_FILE)); + + if ($inlineList !== null && $filePath !== null) { + $output->writeln(self::ERROR_ONE_PROJECT_SOURCE); + return null; + } + + // "@path" in the argument is shorthand for --projects-file=path. + if ($inlineList !== null && str_starts_with($inlineList, '@')) { + $filePath = substr($inlineList, 1); + $inlineList = null; + } + + if ($filePath !== null) { + return $this->readProjectIdsFile($filePath, $output); + } + + if ($inlineList === null) { + $output->writeln(self::ERROR_ONE_PROJECT_SOURCE); + return null; + } + + return $this->rejectEmptyProjectIds($this->parseProjectIdList($inlineList), $output); + } + + /** + * @return non-empty-string|null null for a missing or empty console input value + */ + private function optionalStringInput(mixed $value): ?string + { + return is_string($value) && $value !== '' ? $value : null; + } + + /** + * @return array|null + */ + private function readProjectIdsFile(string $path, OutputInterface $output): ?array + { + $contents = @file_get_contents($path); + if ($contents === false) { + $output->writeln(sprintf('Cannot read projects file "%s"', $path)); + return null; + } + + return $this->rejectEmptyProjectIds($this->parseProjectIdsFile($contents), $output); + } + + /** + * @param array|null $projectIds null when parsing found a non-numeric ID + * @return array|null + */ + private function rejectEmptyProjectIds(?array $projectIds, OutputInterface $output): ?array + { + if ($projectIds === null || $projectIds === []) { + $output->writeln('Projects list is empty or contains a non-numeric ID'); + return null; + } + + return $projectIds; + } + + /** + * @return array|null null when any entry is not a plain non-negative integer + */ + private function parseProjectIdList(string $raw): ?array + { + return $this->validateAndDeduplicate(array_map('trim', explode(',', $raw))); + } + + /** + * One ID per line; blank lines and lines starting with "#" are ignored. + * + * @return array|null null when any remaining line is not a plain non-negative integer + */ + private function parseProjectIdsFile(string $contents): ?array + { + $lines = preg_split('/\R/', $contents); + $ids = []; + foreach ($lines === false ? [] : $lines as $line) { + $line = trim($line); + if ($line === '' || str_starts_with($line, '#')) { + continue; + } + $ids[] = $line; + } + + return $this->validateAndDeduplicate($ids); + } + + /** + * @param array $ids + * @return array|null + */ + private function validateAndDeduplicate(array $ids): ?array + { + foreach ($ids as $id) { + if (!ctype_digit($id)) { + return null; + } + } + + return array_values(array_unique($ids)); + } + + private function parsePositiveIntOption(InputInterface $input, string $name): ?int + { + $value = $input->getOption($name); + if (!is_string($value) || !ctype_digit($value) || (int) $value < 1) { + return null; + } + + return (int) $value; + } + + private function resolveReportPath(InputInterface $input, string $hostnameSuffix): string + { + $reportPath = $input->getOption(self::OPT_REPORT); + if (is_string($reportPath) && $reportPath !== '') { + return $reportPath; + } + + return sprintf('flow-migration-%s-%s.csv', $hostnameSuffix, date('Ymd-His')); + } + + private function printRunNotice( + OutputInterface $output, + bool $force, + int $projectCount, + int $concurrency, + int $pollInterval, + string $reportPath + ): void { + if ($force) { + $output->writeln('Running in FORCE mode: migration jobs run with dryRun: false.'); + } else { + $output->writeln( + 'Running in dry-run mode: migration jobs run with dryRun: true. Use -f for the real migration.' + ); + // The usual cli-utils dry-run changes nothing at all - this one still spends a job slot + // (and on PAYGO stacks, credits) in every eligible project, so say it out loud. + $output->writeln('NOTE: even in dry-run mode a real keboola.flow-migration-tool job and a real' + . ' ephemeral storage token are created in every eligible project.'); + } + $output->writeln(sprintf( + 'Projects: %d, concurrency: %d, poll interval: %d s', + $projectCount, + $concurrency, + $pollInterval + )); + $output->writeln(sprintf('Report: %s', $reportPath)); + $output->writeln(''); + } + + /** + * Opens the CSV report in append mode and writes the header only for a new or empty file, so + * re-running with the same --report path keeps one continuous, valid CSV. + * + * @return resource|null null when the file cannot be opened + */ + private function openReport(string $reportPath, OutputInterface $output) + { + $needsHeader = !is_file($reportPath) || filesize($reportPath) === 0; + + $reportHandle = fopen($reportPath, 'a'); + if ($reportHandle === false) { + $output->writeln(sprintf('Cannot open report file "%s" for writing', $reportPath)); + return null; + } + + if ($needsHeader) { + fputcsv($reportHandle, self::CSV_HEADER, self::CSV_DELIMITER, self::CSV_ENCLOSURE, self::CSV_ESCAPE); + } + + return $reportHandle; + } + + /** + * @param resource $reportHandle + */ + private function appendReportRow($reportHandle, ProjectResult $result): void + { + fputcsv( + $reportHandle, + [ + $result->projectId, + $result->jobId ?? '', + $result->status, + $result->durationSeconds !== null ? (string) $result->durationSeconds : '', + $result->error ?? '', + ], + self::CSV_DELIMITER, + self::CSV_ENCLOSURE, + self::CSV_ESCAPE + ); + // Flush per row so an interrupted run still leaves an auditable report. + fflush($reportHandle); + } + + /** + * @param array{ + * attempted: int, + * migrated: int, + * migratedWithWarning: int, + * skippedNoOrchestrations: int, + * skippedDisabled: int, + * failed: int + * } $summary + */ + private function printSummary(OutputInterface $output, array $summary): void + { + $output->writeln(''); + $output->writeln(sprintf( + "DONE\nProjects attempted: %d\nMigrated (job success): %d\nMigrated with warning: %d\n" + . "Skipped (no orchestrations): %d\nSkipped (disabled/deleted): %d\nFailed: %d", + $summary['attempted'], + $summary['migrated'], + $summary['migratedWithWarning'], + $summary['skippedNoOrchestrations'], + $summary['skippedDisabled'], + $summary['failed'] + )); + } +} diff --git a/tests/FlowMigration/BatchRunnerTest.php b/tests/FlowMigration/BatchRunnerTest.php new file mode 100644 index 0000000..d3fc1b2 --- /dev/null +++ b/tests/FlowMigration/BatchRunnerTest.php @@ -0,0 +1,347 @@ + */ + private array $results = []; + + /** @var array */ + private array $sleeps = []; + + private function collector(): Closure + { + return function (ProjectResult $result): void { + $this->results[] = $result; + }; + } + + private function sleepRecorder(): Closure + { + return function (int $seconds): void { + $this->sleeps[] = $seconds; + }; + } + + /** + * @return array enabled project detail as returned by the Manage API + */ + private static function enabledProject(string $id): array + { + return ['id' => $id, 'name' => 'Project ' . $id, 'isDisabled' => false]; + } + + private static function clientsWith( + FakeJobQueueClient $queueClient, + bool $hasOrchestrations = true + ): ProjectClients { + $configs = $hasOrchestrations + ? ['keboola.orchestrator' => [['id' => 'orch-1', 'name' => 'Daily load', 'configuration' => []]]] + : []; + + return new ProjectClients(new FakeComponents($configs), $queueClient); + } + + public function testHappyPathCreatesJobAndReportsSuccess(): void + { + $queueClient = new FakeJobQueueClient( + [FakeJobQueueClient::makeJob('job-1', 'created')], + ['job-1' => [ + FakeJobQueueClient::makeJob('job-1', 'processing'), + FakeJobQueueClient::makeJob('job-1', 'success', 42), + ]] + ); + $factory = new FakeProjectClientsFactory( + ['100' => self::enabledProject('100')], + ['100' => self::clientsWith($queueClient)] + ); + $runner = new BatchRunner($factory, 10, 5, $this->sleepRecorder()); + + $summary = $runner->run(['100'], true, new BufferedOutput(), $this->collector()); + + // Exact job payload - this is the whole contract with keboola.flow-migration-tool. + $this->assertCount(1, $queueClient->createdJobs); + $this->assertSame('keboola.flow-migration-tool', $queueClient->createdJobs[0]['component']); + $this->assertNull($queueClient->createdJobs[0]['config']); + $this->assertSame('run', $queueClient->createdJobs[0]['mode']); + $this->assertSame( + ['parameters' => [ + 'mode' => 'project', + 'orchestrationIds' => [], + 'skipBroken' => true, + 'dryRun' => false, + ]], + $queueClient->createdJobs[0]['configData'] + ); + + $this->assertCount(1, $this->results); + $this->assertSame('100', $this->results[0]->projectId); + $this->assertSame('job-1', $this->results[0]->jobId); + $this->assertSame('success', $this->results[0]->status); + $this->assertSame(42, $this->results[0]->durationSeconds); + $this->assertNull($this->results[0]->error); + + // Two poll sweeps (processing, then success), each preceded by one poll-interval sleep. + $this->assertSame([5, 5], $this->sleeps); + + $this->assertSame(1, $summary['attempted']); + $this->assertSame(1, $summary['migrated']); + $this->assertSame(0, $summary['failed']); + } + + public function testWithoutForceJobRunsWithDryRunTrue(): void + { + $queueClient = new FakeJobQueueClient( + [FakeJobQueueClient::makeJob('job-1', 'created')], + ['job-1' => [FakeJobQueueClient::makeJob('job-1', 'success', 1)]] + ); + $factory = new FakeProjectClientsFactory( + ['100' => self::enabledProject('100')], + ['100' => self::clientsWith($queueClient)] + ); + $runner = new BatchRunner($factory, 10, 5, $this->sleepRecorder()); + + $runner->run(['100'], false, new BufferedOutput(), $this->collector()); + + $configData = $queueClient->createdJobs[0]['configData']; + $this->assertIsArray($configData); + $this->assertTrue($configData['parameters']['dryRun']); + } + + public function testJobEndingInErrorMarksProjectFailedButBatchContinues(): void + { + $queueClient1 = new FakeJobQueueClient( + [FakeJobQueueClient::makeJob('job-1', 'created')], + ['job-1' => [FakeJobQueueClient::makeJob('job-1', 'error', 10, ['message' => 'boom'])]] + ); + $queueClient2 = new FakeJobQueueClient( + [FakeJobQueueClient::makeJob('job-2', 'created')], + ['job-2' => [FakeJobQueueClient::makeJob('job-2', 'success', 20)]] + ); + $factory = new FakeProjectClientsFactory( + ['100' => self::enabledProject('100'), '200' => self::enabledProject('200')], + ['100' => self::clientsWith($queueClient1), '200' => self::clientsWith($queueClient2)] + ); + $runner = new BatchRunner($factory, 10, 5, $this->sleepRecorder()); + + $summary = $runner->run(['100', '200'], true, new BufferedOutput(), $this->collector()); + + $this->assertCount(2, $this->results); + $byProject = []; + foreach ($this->results as $result) { + $byProject[$result->projectId] = $result; + } + $this->assertSame('error', $byProject['100']->status); + $this->assertSame('boom', $byProject['100']->error); + $this->assertTrue($byProject['100']->isFailed()); + $this->assertSame('success', $byProject['200']->status); + + $this->assertSame(2, $summary['attempted']); + $this->assertSame(1, $summary['migrated']); + $this->assertSame(1, $summary['failed']); + } + + public function testSkipsDisabledAndDeletedProjectsWithoutCreatingTokenOrJob(): void + { + $factory = new FakeProjectClientsFactory([ + '100' => ['id' => '100', 'name' => 'Off', 'isDisabled' => true], + // A deleted project surfaces as a Manage API 404 and is reported the same way. + '200' => new ManageClientException('Project not found', 404), + ]); + $runner = new BatchRunner($factory, 10, 5, $this->sleepRecorder()); + + $summary = $runner->run(['100', '200'], true, new BufferedOutput(), $this->collector()); + + // No ephemeral token may be created for a project that will not be migrated. + $this->assertSame([], $factory->createClientsCalls); + $this->assertSame(ProjectResult::STATUS_SKIPPED_DISABLED, $this->results[0]->status); + $this->assertSame(ProjectResult::STATUS_SKIPPED_DISABLED, $this->results[1]->status); + $this->assertNull($this->results[0]->jobId); + $this->assertSame(2, $summary['skippedDisabled']); + $this->assertSame(0, $summary['failed']); + $this->assertSame([], $this->sleeps); + } + + public function testManageErrorOtherThan404MarksProjectFailed(): void + { + $factory = new FakeProjectClientsFactory( + ['100' => new ManageClientException('Internal error', 500)] + ); + $runner = new BatchRunner($factory, 10, 5, $this->sleepRecorder()); + + $summary = $runner->run(['100'], true, new BufferedOutput(), $this->collector()); + + $this->assertSame(ProjectResult::STATUS_ERROR, $this->results[0]->status); + $this->assertSame('Internal error', $this->results[0]->error); + $this->assertSame(1, $summary['failed']); + } + + public function testTransportFailureOnProjectLookupDoesNotAbortTheBatch(): void + { + // A DNS/connect failure surfaces as a Guzzle ConnectException, not a Manage API + // ClientException - it must still resolve as a per-project error. + $queueClient = new FakeJobQueueClient( + [FakeJobQueueClient::makeJob('job-2', 'created')], + ['job-2' => [FakeJobQueueClient::makeJob('job-2', 'success', 2)]] + ); + $factory = new FakeProjectClientsFactory( + [ + '100' => new ConnectException('cURL error 6: Could not resolve host', new Request('GET', '/')), + '200' => self::enabledProject('200'), + ], + ['200' => self::clientsWith($queueClient)] + ); + $runner = new BatchRunner($factory, 10, 5, $this->sleepRecorder()); + + $summary = $runner->run(['100', '200'], true, new BufferedOutput(), $this->collector()); + + $byProject = []; + foreach ($this->results as $result) { + $byProject[$result->projectId] = $result; + } + $this->assertSame(ProjectResult::STATUS_ERROR, $byProject['100']->status); + $this->assertSame('success', $byProject['200']->status); + $this->assertSame(1, $summary['failed']); + $this->assertSame(1, $summary['migrated']); + } + + public function testSkipsProjectWithoutOrchestratorConfigurations(): void + { + $queueClient = new FakeJobQueueClient(); + $factory = new FakeProjectClientsFactory( + ['100' => self::enabledProject('100')], + ['100' => self::clientsWith($queueClient, false)] + ); + $runner = new BatchRunner($factory, 10, 5, $this->sleepRecorder()); + + $summary = $runner->run(['100'], true, new BufferedOutput(), $this->collector()); + + // No job for a project with nothing to migrate. + $this->assertSame([], $queueClient->createdJobs); + $this->assertSame(ProjectResult::STATUS_SKIPPED_NO_ORCHESTRATIONS, $this->results[0]->status); + $this->assertSame(1, $summary['skippedNoOrchestrations']); + } + + public function testDeduplicatesInputProjectIds(): void + { + $queueClient = new FakeJobQueueClient( + [FakeJobQueueClient::makeJob('job-1', 'created')], + ['job-1' => [FakeJobQueueClient::makeJob('job-1', 'success', 1)]] + ); + $factory = new FakeProjectClientsFactory( + ['100' => self::enabledProject('100')], + ['100' => self::clientsWith($queueClient)] + ); + $runner = new BatchRunner($factory, 10, 5, $this->sleepRecorder()); + + $summary = $runner->run(['100', '100', '100'], true, new BufferedOutput(), $this->collector()); + + $this->assertSame(1, $summary['attempted']); + $this->assertCount(1, $queueClient->createdJobs); + $this->assertCount(1, $this->results); + } + + public function testTwoConsecutivePollFailuresAreToleratedAndJobFinishes(): void + { + $queueClient = new FakeJobQueueClient( + [FakeJobQueueClient::makeJob('job-1', 'created')], + ['job-1' => [ + new RuntimeException('blip 1'), + new RuntimeException('blip 2'), + FakeJobQueueClient::makeJob('job-1', 'success', 7), + ]] + ); + $factory = new FakeProjectClientsFactory( + ['100' => self::enabledProject('100')], + ['100' => self::clientsWith($queueClient)] + ); + $runner = new BatchRunner($factory, 10, 5, $this->sleepRecorder()); + + $summary = $runner->run(['100'], true, new BufferedOutput(), $this->collector()); + + $this->assertSame('success', $this->results[0]->status); + $this->assertSame(1, $summary['migrated']); + $this->assertSame(0, $summary['failed']); + $this->assertSame([5, 5, 5], $this->sleeps); + } + + public function testThreeConsecutivePollFailuresMarkProjectFailedWithJobIdKept(): void + { + $queueClient = new FakeJobQueueClient( + [FakeJobQueueClient::makeJob('job-1', 'created')], + ['job-1' => [ + new RuntimeException('down 1'), + new RuntimeException('down 2'), + new RuntimeException('down 3'), + ]] + ); + $factory = new FakeProjectClientsFactory( + ['100' => self::enabledProject('100')], + ['100' => self::clientsWith($queueClient)] + ); + $runner = new BatchRunner($factory, 10, 5, $this->sleepRecorder()); + + $summary = $runner->run(['100'], true, new BufferedOutput(), $this->collector()); + + $this->assertSame(ProjectResult::STATUS_ERROR, $this->results[0]->status); + // The job id must survive into the report - the job may still be running server-side. + $this->assertSame('job-1', $this->results[0]->jobId); + $this->assertIsString($this->results[0]->error); + $this->assertStringContainsString('polling gave up', $this->results[0]->error); + $this->assertSame(1, $summary['failed']); + } + + public function testConcurrencyWindowCapsInFlightJobsAndRefills(): void + { + // One shared fake for both projects makes the cross-project call order observable. + $queueClient = new FakeJobQueueClient( + [ + FakeJobQueueClient::makeJob('job-1', 'created'), + FakeJobQueueClient::makeJob('job-2', 'created'), + ], + [ + 'job-1' => [ + FakeJobQueueClient::makeJob('job-1', 'processing'), + FakeJobQueueClient::makeJob('job-1', 'success', 1), + ], + 'job-2' => [FakeJobQueueClient::makeJob('job-2', 'success', 1)], + ] + ); + $clients = self::clientsWith($queueClient); + $factory = new FakeProjectClientsFactory( + ['100' => self::enabledProject('100'), '200' => self::enabledProject('200')], + ['100' => $clients, '200' => $clients] + ); + $runner = new BatchRunner($factory, 1, 5, $this->sleepRecorder()); + + $summary = $runner->run(['100', '200'], true, new BufferedOutput(), $this->collector()); + + // With concurrency=1, job-2 must not be created until job-1 has finished. + $this->assertSame( + [ + ['createJob', 'job-1'], + ['getJob', 'job-1'], + ['getJob', 'job-1'], + ['createJob', 'job-2'], + ['getJob', 'job-2'], + ], + $queueClient->calls + ); + $this->assertSame(2, $summary['migrated']); + } +} diff --git a/tests/FlowMigration/FakeJobQueueClient.php b/tests/FlowMigration/FakeJobQueueClient.php new file mode 100644 index 0000000..cf50dcb --- /dev/null +++ b/tests/FlowMigration/FakeJobQueueClient.php @@ -0,0 +1,122 @@ +> recorded JobData->getArray() payloads */ + public array $createdJobs = []; + + /** @var array> ordered call log: [method, id] */ + public array $calls = []; + + /** @var array */ + private array $createJobReturns; + + /** @var array> */ + private array $getJobSequences; + + /** + * @param array $createJobReturns successive createJob() returns + * @param array> $getJobSequences jobId => successive getJob() outcomes + */ + public function __construct(array $createJobReturns = [], array $getJobSequences = []) + { + $this->createJobReturns = $createJobReturns; + $this->getJobSequences = $getJobSequences; + } + + public function createJob(JobData $jobData): Job + { + $this->createdJobs[] = $jobData->getArray(); + $job = array_shift($this->createJobReturns); + if ($job === null) { + throw new RuntimeException('FakeJobQueueClient: no scripted createJob return left'); + } + $this->calls[] = ['createJob', $job->id]; + + return $job; + } + + public function getJob(string $jobId): Job + { + $this->calls[] = ['getJob', $jobId]; + $sequence = $this->getJobSequences[$jobId] ?? []; + if ($sequence === []) { + throw new RuntimeException( + sprintf('FakeJobQueueClient: no scripted getJob outcome left for "%s"', $jobId) + ); + } + $outcome = array_shift($sequence); + $this->getJobSequences[$jobId] = $sequence; + if ($outcome instanceof Throwable) { + throw $outcome; + } + + return $outcome; + } + + /** + * Builds a real DTO\Job through its public factory so the fixture stays in sync with the SDK. + * + * @param array|null $result + */ + public static function makeJob( + string $id, + string $status, + ?int $durationSeconds = null, + ?array $result = null + ): Job { + $terminalStatuses = ['success', 'error', 'warning', 'terminated', 'cancelled']; + + return Job::fromApiResponse([ + 'id' => $id, + 'runId' => $id, + 'parentRunId' => '', + 'project' => ['id' => '123'], + 'token' => ['id' => '456', 'description' => 'test token'], + 'status' => $status, + 'desiredStatus' => 'processing', + 'mode' => 'run', + 'component' => 'keboola.flow-migration-tool', + 'config' => null, + 'configData' => null, + 'configRowIds' => null, + 'tag' => null, + 'createdTime' => '2026-08-10T10:00:00+00:00', + 'startTime' => null, + 'endTime' => null, + 'durationSeconds' => $durationSeconds, + 'result' => $result, + 'usageData' => null, + 'isFinished' => in_array($status, $terminalStatuses, true), + 'url' => sprintf('https://queue.example.com/jobs/%s', $id), + 'branchId' => null, + 'variableValuesId' => null, + 'variableValuesData' => [], + 'backend' => [], + 'behavior' => [], + 'executor' => null, + 'metrics' => null, + 'parallelism' => null, + 'type' => 'standard', + 'orchestrationJobId' => null, + 'orchestrationTaskId' => null, + 'onlyOrchestrationTaskIds' => null, + 'previousJobId' => null, + ]); + } +} diff --git a/tests/FlowMigration/FakeProjectClientsFactory.php b/tests/FlowMigration/FakeProjectClientsFactory.php new file mode 100644 index 0000000..4fd91e7 --- /dev/null +++ b/tests/FlowMigration/FakeProjectClientsFactory.php @@ -0,0 +1,67 @@ + projectIds passed to createProjectClients() */ + public array $createClientsCalls = []; + + /** @var array|Throwable> */ + private array $projects; + + /** @var array */ + private array $projectClients; + + /** + * @param array|Throwable> $projects projectId => project detail, or Throwable to throw + * @param array $projectClients projectId => clients, or Throwable + */ + public function __construct(array $projects, array $projectClients = []) + { + $this->projects = $projects; + $this->projectClients = $projectClients; + } + + public function getProject(string $projectId): array + { + if (!array_key_exists($projectId, $this->projects)) { + throw new RuntimeException( + sprintf('FakeProjectClientsFactory: unknown project "%s"', $projectId) + ); + } + $project = $this->projects[$projectId]; + if ($project instanceof Throwable) { + throw $project; + } + + return $project; + } + + public function createProjectClients(string $projectId): ProjectClients + { + $this->createClientsCalls[] = $projectId; + if (!array_key_exists($projectId, $this->projectClients)) { + throw new RuntimeException( + sprintf('FakeProjectClientsFactory: no clients for project "%s"', $projectId) + ); + } + $clients = $this->projectClients[$projectId]; + if ($clients instanceof Throwable) { + throw $clients; + } + + return $clients; + } +} diff --git a/tests/FlowMigration/ProjectResultTest.php b/tests/FlowMigration/ProjectResultTest.php new file mode 100644 index 0000000..693c9d5 --- /dev/null +++ b/tests/FlowMigration/ProjectResultTest.php @@ -0,0 +1,39 @@ +assertSame($expectedSkipped, $result->isSkipped()); + $this->assertSame($expectedFailed, $result->isFailed()); + } + + /** + * @return iterable + */ + public static function provideStatuses(): iterable + { + yield 'job success is neither skipped nor failed' => ['success', false, false]; + yield 'job warning counts as migrated, not failed' => ['warning', false, false]; + yield 'job error is failed' => ['error', false, true]; + yield 'job terminated is failed' => ['terminated', false, true]; + yield 'job cancelled is failed' => ['cancelled', false, true]; + yield 'skipped disabled' => [ProjectResult::STATUS_SKIPPED_DISABLED, true, false]; + yield 'skipped no orchestrations' => [ + ProjectResult::STATUS_SKIPPED_NO_ORCHESTRATIONS, + true, + false, + ]; + } +} diff --git a/tests/MigrateOrchestrationsToFlowTest.php b/tests/MigrateOrchestrationsToFlowTest.php new file mode 100644 index 0000000..1b7fb01 --- /dev/null +++ b/tests/MigrateOrchestrationsToFlowTest.php @@ -0,0 +1,95 @@ +|string|null + */ + private function invokePrivate(string $method, string $argument): array|string|null + { + $command = new MigrateOrchestrationsToFlow(); + $reflection = (new ReflectionClass($command))->getMethod($method); + $reflection->setAccessible(true); + + /** @var array|string|null $result */ + $result = $reflection->invoke($command, $argument); + + return $result; + } + + #[DataProvider('provideUrls')] + public function testHostnameSuffixFromUrl(string $url, ?string $expected): void + { + $this->assertSame($expected, $this->invokePrivate('hostnameSuffixFromUrl', $url)); + } + + /** + * @return iterable + */ + public static function provideUrls(): iterable + { + yield 'azure ne stack' => [ + 'https://connection.north-europe.azure.keboola.com', + 'north-europe.azure.keboola.com', + ]; + yield 'aws us stack' => ['https://connection.keboola.com', 'keboola.com']; + yield 'trailing slash is fine' => ['https://connection.keboola.com/', 'keboola.com']; + yield 'missing connection prefix' => ['https://queue.keboola.com', null]; + yield 'not a url' => ['not-a-url', null]; + yield 'bare connection host' => ['https://connection.', null]; + } + + /** + * @param array|null $expected + */ + #[DataProvider('provideProjectLists')] + public function testParseProjectIdList(string $input, ?array $expected): void + { + $this->assertSame($expected, $this->invokePrivate('parseProjectIdList', $input)); + } + + /** + * @return iterable|null}> + */ + public static function provideProjectLists(): iterable + { + yield 'plain list' => ['1,2,3', ['1', '2', '3']]; + yield 'whitespace is trimmed' => ['1, 2 ,3', ['1', '2', '3']]; + yield 'duplicates are removed' => ['1,2,1', ['1', '2']]; + yield 'non-numeric entry invalidates the list' => ['1,foo', null]; + yield 'decimal is rejected' => ['1.2', null]; + yield 'negative is rejected' => ['-1', null]; + yield 'empty string is rejected' => ['', null]; + } + + /** + * @param array|null $expected + */ + #[DataProvider('provideProjectFiles')] + public function testParseProjectIdsFile(string $contents, ?array $expected): void + { + $this->assertSame($expected, $this->invokePrivate('parseProjectIdsFile', $contents)); + } + + /** + * @return iterable|null}> + */ + public static function provideProjectFiles(): iterable + { + yield 'one id per line' => ["100\n200\n", ['100', '200']]; + yield 'blank lines and comments are ignored' => ["100\n\n# staging batch\n200\n", ['100', '200']]; + yield 'windows line endings' => ["100\r\n200\r\n", ['100', '200']]; + yield 'duplicates are removed' => ["100\n200\n100\n", ['100', '200']]; + yield 'non-numeric line invalidates the file' => ["100\nfoo\n", null]; + yield 'empty file is a valid empty list' => ['', []]; + } +}