#207 - Fix JsonStreamReaderTask / JsonStreamWriterTask

CHANGELOG.md

@@ -17,6 +17,7 @@ Latest
 * [#204](https://github.com/cleverage/process-bundle/issues/204) Fix CachedTransformer, SlugifyTransformer and TypeSetterTransformer edge cases: accept a non-string input in `cached` (key built after `key_transformers`), reject an invalid `transliterator` in `slugify` with an `InvalidOptionsException`, remove the unreachable error branch of `type_setter`. Update documentation, add tests.
 * [#205](https://github.com/cleverage/process-bundle/issues/205) Fix FileSplitterTask: produced files lost the first line, doubled line breaks and ended with an empty file; each file now contains exactly `max_lines` lines of the source. Update documentation, add tests.
 * [#206](https://github.com/cleverage/process-bundle/issues/206) Fix AdvancedStatCounterTask, IterableBatchTask, ConditionTrait and ColumnAggregatorTask edge cases: log the first counted execution with correct counts, accept a `null` `batch_count` (only flush at the end), allow scalar inputs in conditions (`''` path on the whole value), aggregate `null` column values. Update documentation, add tests.
+* [#207](https://github.com/cleverage/process-bundle/issues/207) Fix JsonStreamReaderTask / JsonStreamWriterTask: throw an explicit `\UnexpectedValueException` when a line decodes to a scalar, create the missing parent directory when writing. Update documentation, add tests.
 
 ## Deprecated
 * [#189](https://github.com/cleverage/process-bundle/issues/189) EventDispatcherTask: when `event_name` is set, listening to `CleverAge\ProcessBundle\Event\EventDispatcherTaskEvent` is deprecated (the event is still dispatched under its class name, with an `E_USER_DEPRECATED` error, if it has listeners). Listen to the configured `event_name` instead: the BC layer will be removed in v6.0.

docs/reference/tasks/json_stream_reader_task.md

@@ -23,7 +23,8 @@ Underlying methods are [SplFileObject::fgets](https://www.php.net/manual/en/splf
 [json_decode](https://www.php.net/manual/en/function.json-decode.php).
 
 When a line is empty or decodes to `null`, no output is produced and the task is skipped for this iteration. Each
-line must be a JSON object or array: a line decoding to a scalar (e.g. `42` or `"foo"`) raises a `\TypeError`.
+line must be a JSON object or array: a line decoding to a scalar (e.g. `42` or `"foo"`) raises an
+`\UnexpectedValueException` giving the line number and the file path.
 
 Options
 -------

docs/reference/tasks/json_stream_writer_task.md

@@ -26,7 +26,7 @@ Options
 
 | Code                    | Type          | Required | Default | Description                                                                                                                                                                                                                            |
 |-------------------------|---------------|:--------:|---------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| `file_path`             | `string`      |  **X**   |         | Path of the file to write to, opened in `wb` mode (overwritten).<br/>Placeholders `{date}`, `{date_time}`, `{timestamp}` and `{unique_token}` are replaced in the path                                                                 |
+| `file_path`             | `string`      |  **X**   |         | Path of the file to write to, opened in `wb` mode (overwritten); missing parent directories are created.<br/>Placeholders `{date}`, `{date_time}`, `{timestamp}` and `{unique_token}` are replaced in the path |
 | `spl_file_object_flags` | `array\|null` |          | `null`  | List of `SplFileObject` flags, summed and passed to [SplFileObject::setFlags](https://www.php.net/manual/en/splfileobject.setflags.php).<br/>`null` means `DROP_NEW_LINE`, `READ_AHEAD` and `SKIP_EMPTY`; an empty array means no flag |
 | `json_flags`            | `array\|null` |          | `null`  | List of JSON flags, summed and passed to [json_encode](https://www.php.net/manual/en/function.json-encode.php).<br/>`null` means `JSON_THROW_ON_ERROR`; an empty array means no flag                                                   |
 
@@ -59,6 +59,6 @@ Notes
 
 * `{date}` is replaced by `Ymd`, `{date_time}` by `Ymd_His`, `{timestamp}` by the Unix timestamp and `{unique_token}`
   by a `uniqid()` value.
-* The parent directory of the file must exist.
+* The parent directory of the file is created (recursively) if it does not exist.
 * Using `JSON_PRETTY_PRINT` produces multi-line documents: the file can no longer be read by
   [JsonStreamReaderTask](json_stream_reader_task.md).

src/Filesystem/JsonStreamFile.php

@@ -32,6 +32,13 @@ public function __construct(
         ?array $splFileObjectFlags = null,
         ?array $jsonFlags = null,
     ) {
+        if (!\in_array($filename, ['php://stdin', 'php://stdout', 'php://stderr'], true) && !str_starts_with($mode, 'r')) {
+            $dirname = \dirname($filename);
+            if (!@mkdir($dirname, 0o755, true) && !is_dir($dirname)) {
+                throw new \RuntimeException(\sprintf('Directory "%s" was not created', $dirname));
+            }
+        }
+
         $this->file = new \SplFileObject($filename, $mode);
 
         // Useful to skip empty trailing lines (doesn't work well on PHP 8, see readLine() code)
@@ -78,6 +85,8 @@ public function isEndOfFile(): bool
 
     /**
      * Return an array containing current data and moving the file pointer.
+     *
+     * @throws \UnexpectedValueException if the line decodes to a scalar value
      */
     public function readLine(?int $length = null): ?array
     {
@@ -90,9 +99,14 @@ public function readLine(?int $length = null): ?array
         if ('' === $rawLine) {
             return null;
         }
-        ++$this->lineNumber;
+        $currentLineNumber = $this->lineNumber++;
+
+        $data = json_decode($rawLine, true, 512, $this->jsonFlags);
+        if (null !== $data && !\is_array($data)) {
+            throw new \UnexpectedValueException(\sprintf('Line %d of file "%s" must be a JSON object or array, got %s', $currentLineNumber, $this->file->getPathname(), get_debug_type($data)));
+        }
 
-        return json_decode($rawLine, true, 512, $this->jsonFlags);
+        return $data;
     }
 
     public function writeLine(array $fields): int

tests/Filesystem/JsonStreamFileTest.php

@@ -0,0 +1,117 @@
+<?php
+
+declare(strict_types=1);
+
+/*
+ * This file is part of the CleverAge/ProcessBundle package.
+ *
+ * Copyright (c) Clever-Age
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace CleverAge\ProcessBundle\Tests\Filesystem;
+
+use CleverAge\ProcessBundle\Filesystem\JsonStreamFile;
+use PHPUnit\Framework\Attributes\DataProvider;
+use PHPUnit\Framework\TestCase;
+
+#[\PHPUnit\Framework\Attributes\CoversClass(JsonStreamFile::class)]
+class JsonStreamFileTest extends TestCase
+{
+    private string $tmpDir;
+
+    protected function setUp(): void
+    {
+        $this->tmpDir = sys_get_temp_dir().'/'.uniqid('json_stream_file_test_', true);
+        mkdir($this->tmpDir);
+    }
+
+    protected function tearDown(): void
+    {
+        $this->removeDirectory($this->tmpDir);
+    }
+
+    public function testReadLineDecodesObjectsAndArrays(): void
+    {
+        $file = new JsonStreamFile($this->createFile("{\"a\":1}\n[1,2]\n"));
+
+        self::assertSame(['a' => 1], $file->readLine());
+        self::assertSame([1, 2], $file->readLine());
+        self::assertNull($file->readLine());
+    }
+
+    public function testReadLineReturnsNullForNullLine(): void
+    {
+        $file = new JsonStreamFile($this->createFile("null\n"));
+
+        self::assertNull($file->readLine());
+    }
+
+    /**
+     * @return iterable<string, array{string, string}>
+     */
+    public static function scalarLineProvider(): iterable
+    {
+        yield 'int' => ['42', 'int'];
+        yield 'float' => ['4.2', 'float'];
+        yield 'string' => ['"foo"', 'string'];
+        yield 'bool' => ['true', 'bool'];
+    }
+
+    #[DataProvider('scalarLineProvider')]
+    public function testReadLineThrowsOnScalarLine(string $line, string $type): void
+    {
+        $filename = $this->createFile("{\"a\":1}\n{$line}\n");
+        $file = new JsonStreamFile($filename);
+        $file->readLine();
+
+        $this->expectException(\UnexpectedValueException::class);
+        $this->expectExceptionMessage(\sprintf('Line 2 of file "%s" must be a JSON object or array, got %s', $filename, $type));
+        $file->readLine();
+    }
+
+    public function testWriteCreatesMissingParentDirectory(): void
+    {
+        $filename = $this->tmpDir.'/sub/dir/out.jsonl';
+
+        $file = new JsonStreamFile($filename, 'wb');
+        $file->writeLine(['a' => 1]);
+        unset($file);
+
+        self::assertSame('{"a":1}'.\PHP_EOL, file_get_contents($filename));
+    }
+
+    public function testReadDoesNotCreateMissingParentDirectory(): void
+    {
+        $filename = $this->tmpDir.'/missing/in.jsonl';
+
+        try {
+            new JsonStreamFile($filename);
+            self::fail('Opening a missing file for reading should fail');
+        } catch (\RuntimeException) {
+        }
+
+        self::assertDirectoryDoesNotExist(\dirname($filename));
+    }
+
+    private function createFile(string $content): string
+    {
+        $filename = $this->tmpDir.'/in.jsonl';
+        file_put_contents($filename, $content);
+
+        return $filename;
+    }
+
+    private function removeDirectory(string $dir): void
+    {
+        foreach (new \RecursiveIteratorIterator(
+            new \RecursiveDirectoryIterator($dir, \FilesystemIterator::SKIP_DOTS),
+            \RecursiveIteratorIterator::CHILD_FIRST
+        ) as $item) {
+            $item->isDir() ? rmdir($item->getPathname()) : unlink($item->getPathname());
+        }
+        rmdir($dir);
+    }
+}

tests/Task/File/JsonStream/JsonStreamWriterTaskTest.php

@@ -0,0 +1,71 @@
+<?php
+
+declare(strict_types=1);
+
+/*
+ * This file is part of the CleverAge/ProcessBundle package.
+ *
+ * Copyright (c) Clever-Age
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace CleverAge\ProcessBundle\Tests\Task\File\JsonStream;
+
+use CleverAge\ProcessBundle\Configuration\ProcessConfiguration;
+use CleverAge\ProcessBundle\Configuration\TaskConfiguration;
+use CleverAge\ProcessBundle\Context\ContextualOptionResolver;
+use CleverAge\ProcessBundle\Filesystem\JsonStreamFile;
+use CleverAge\ProcessBundle\Model\AbstractConfigurableTask;
+use CleverAge\ProcessBundle\Model\ProcessHistory;
+use CleverAge\ProcessBundle\Model\ProcessState;
+use CleverAge\ProcessBundle\Task\File\JsonStream\JsonStreamWriterTask;
+use PHPUnit\Framework\TestCase;
+
+#[\PHPUnit\Framework\Attributes\CoversClass(JsonStreamWriterTask::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(JsonStreamFile::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(AbstractConfigurableTask::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(ProcessConfiguration::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(TaskConfiguration::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(ContextualOptionResolver::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(ProcessHistory::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(ProcessState::class)]
+class JsonStreamWriterTaskTest extends TestCase
+{
+    private string $tmpDir;
+
+    protected function setUp(): void
+    {
+        $this->tmpDir = sys_get_temp_dir().'/'.uniqid('json_stream_writer_test_', true);
+    }
+
+    protected function tearDown(): void
+    {
+        @unlink($this->tmpDir.'/sub/out.jsonl');
+        @rmdir($this->tmpDir.'/sub');
+        @rmdir($this->tmpDir);
+    }
+
+    public function testWritesInMissingParentDirectory(): void
+    {
+        $filePath = $this->tmpDir.'/sub/out.jsonl';
+
+        $processConfiguration = new ProcessConfiguration('test', []);
+        $state = new ProcessState($processConfiguration, new ProcessHistory($processConfiguration));
+        $state->setContextualOptionResolver(new ContextualOptionResolver());
+        $state->setContext([]);
+        $state->setTaskConfiguration(new TaskConfiguration('write', JsonStreamWriterTask::class, ['file_path' => $filePath]));
+
+        $task = new JsonStreamWriterTask();
+        $task->initialize($state);
+        foreach ([['a' => 1], ['b' => 2]] as $input) {
+            $state->setInput($input);
+            $task->execute($state);
+        }
+        $task->proceed($state);
+
+        self::assertSame($filePath, $state->getOutput());
+        self::assertSame('{"a":1}'.\PHP_EOL.'{"b":2}'.\PHP_EOL, file_get_contents($filePath));
+    }
+}