Adding in pyarrow schema for jsonschema ingestion pw by ialarmedalien · Pull Request #302 · kbase/cdm-data-loaders

ialarmedalien · 2026-10-26T18:48:50Z

No description provided.

codecov · 2026-10-26T19:01:10Z

❌ 20 Tests Failed:

Tests completed	Failed	Passed	Skipped
5602	20	5582	245

View the top 3 failed test(s) by shortest run time

tests/cdm_data_loaders/pipelines/jsonlines/test_jsonschema_pipeline.py::test_convert_schema_to_pyarrow_pass_flat_and_nested_schema

Stack Traces | 0.014s run time

widget_schema = {'$schema': 'https://json-schema..../draft/2020-12/schema', 'type': 'object', 'required': ['widget_id', 'count'], 'pro...nimum': 0}, 'collected': {'type': 'string', 'format': 'date'}, 'label': {'type': 'string', 'default': 'not inserted'}}}

    def test_convert_schema_to_pyarrow_pass_flat_and_nested_schema(widget_schema: dict[str, Any]) -> None:
        """Validated JSON Schemas convert to PyArrow schemas with expected types and nullability."""
        pa_schema = convert_schema_to_pyarrow(widget_schema)
    
>       assert pa_schema == pa.schema(
            [
                pa.field("widget_id", pa.string(), nullable=False),
                pa.field("count", pa.int64(), nullable=False),
                pa.field("collected", pa.date32(), nullable=True),
                pa.field("label", pa.string(), nullable=True),
                pa.field(
                    "nested",
                    pa.struct([pa.field("tags", pa.list_(pa.string()), nullable=True)]),
                    nullable=True,
                ),
            ]
        )
E       assert widget_id: st...\nlabel: string == widget_id: st..., item: string
E         
E         Full diff:
E           widget_id: string not null
E           count: int64 not null
E           collected: date32[day]
E           label: string
E         - nested: struct<tags: list<item: string>>
E         -   child 0, tags: list<item: string>
E         -       child 0, item: string

.../pipelines/jsonlines/test_jsonschema_pipeline.py:162: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/test_jsonschema_pipeline.py::test_make_page_router_pass_preserves_records_and_empty_pages

Stack Traces | 0.014s run time

widget_schema = {'$schema': 'https://json-schema..../draft/2020-12/schema', 'type': 'object', 'required': ['widget_id', 'count'], 'pro...nimum': 0}, 'collected': {'type': 'string', 'format': 'date'}, 'label': {'type': 'string', 'default': 'not inserted'}}}

    def test_make_page_router_pass_preserves_records_and_empty_pages(widget_schema: dict[str, Any]) -> None:
        """Valid data is coerced into a PyArrow Table matching the schema without inserted defaults."""
        record = {
            "widget_id": "a",
            "count": 0,
            "collected": "2024-02-29",
            "nested": {"tags": ["one", "two"]},
            "undeclared": "ignored",
        }
        validate = make_router(widget_schema)
        results = list(validate([make_item(record)]))
        assert len(results) == 1
        assert results[0].meta.table_name == "widget"
        assert isinstance(results[0].data, pa.Table)
        assert results[0].data.schema == convert_schema_to_pyarrow(widget_schema)
>       assert results[0].data.to_pylist() == [
            {
                "widget_id": "a",
                "count": 0,
                "collected": date(2024, 2, 29),
                "label": None,
                "nested": {"tags": ["one", "two"]},
            }
        ]
E       AssertionError: assert [{'widget_id'...label': None}] == [{'widget_id'...': None, ...}]
E         
E         At index 0 diff: {'widget_id': 'a', 'count': 0, 'collected': datetime.date(2024, 2, 29), 'label': None} != {'widget_id': 'a', 'count': 0, 'collected': datetime.date(2024, 2, 29), 'label': None, 'nested': {'tags': ['one', 'two']}}
E         
E         Full diff:
E           [
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': datetime.date(2024, 2, 29),
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E           ]

.../pipelines/jsonlines/test_jsonschema_pipeline.py:487: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/test_jsonschema_pipeline.py::test_build_entity_resource_pass_reads_selected_files[plain]

Stack Traces | 0.059s run time

schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7f90e4e1f060>
widget_schema = {'$schema': 'https://json-schema..../draft/2020-12/schema', 'type': 'object', 'required': ['widget_id', 'count'], 'pro...nimum': 0}, 'collected': {'type': 'string', 'format': 'date'}, 'label': {'type': 'string', 'default': 'not inserted'}}}
compressed = False

    @pytest.mark.parametrize("compressed", [False, True], ids=["plain", "gzip"])
    def test_build_entity_resource_pass_reads_selected_files(
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        widget_schema: dict[str, Any],
        compressed: bool,
    ) -> None:
        """The real resource reads plain/gzip files, skips blanks, and honors the glob."""
        settings = schema_settings_factory()
        entity_dir = Path(settings.input_dir) / "widget"
        entity_dir.mkdir()
        record = {"widget_id": "a", "count": 0}
        contents = f"\n{json.dumps(record)}\n\n{{broken\n"
        filename = "data.jsonl.gz" if compressed else "data.jsonl"
        if compressed:
            with gzip.open(entity_dir / filename, "wt", encoding="utf-8") as stream:
                stream.write(contents)
        else:
            (entity_dir / filename).write_text(contents, encoding="utf-8")
        (entity_dir / "ignored.txt").write_text("not JSON", encoding="utf-8")
    
        resource = build_entity_resource("widget", widget_schema, settings)
        rows = list(resource)
    
        valid_tables = [item for item in rows if isinstance(item, pa.Table)]
        assert len(valid_tables) == 1
>       assert valid_tables[0].to_pylist() == [
            {"widget_id": "a", "count": 0, "collected": None, "label": None, "nested": None}
        ]
E       AssertionError: assert [{'widget_id'...label': None}] == [{'widget_id'...': None, ...}]
E         
E         At index 0 diff: {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None} != {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None, 'nested': None}
E         
E         Full diff:
E           [
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': None,
E               },
E           ]

.../pipelines/jsonlines/test_jsonschema_pipeline.py:585: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/test_jsonschema_pipeline.py::test_build_entity_resource_pass_reads_selected_files[gzip]

Stack Traces | 0.063s run time

schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7f90e4f0d8a0>
widget_schema = {'$schema': 'https://json-schema..../draft/2020-12/schema', 'type': 'object', 'required': ['widget_id', 'count'], 'pro...nimum': 0}, 'collected': {'type': 'string', 'format': 'date'}, 'label': {'type': 'string', 'default': 'not inserted'}}}
compressed = True

    @pytest.mark.parametrize("compressed", [False, True], ids=["plain", "gzip"])
    def test_build_entity_resource_pass_reads_selected_files(
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        widget_schema: dict[str, Any],
        compressed: bool,
    ) -> None:
        """The real resource reads plain/gzip files, skips blanks, and honors the glob."""
        settings = schema_settings_factory()
        entity_dir = Path(settings.input_dir) / "widget"
        entity_dir.mkdir()
        record = {"widget_id": "a", "count": 0}
        contents = f"\n{json.dumps(record)}\n\n{{broken\n"
        filename = "data.jsonl.gz" if compressed else "data.jsonl"
        if compressed:
            with gzip.open(entity_dir / filename, "wt", encoding="utf-8") as stream:
                stream.write(contents)
        else:
            (entity_dir / filename).write_text(contents, encoding="utf-8")
        (entity_dir / "ignored.txt").write_text("not JSON", encoding="utf-8")
    
        resource = build_entity_resource("widget", widget_schema, settings)
        rows = list(resource)
    
        valid_tables = [item for item in rows if isinstance(item, pa.Table)]
        assert len(valid_tables) == 1
>       assert valid_tables[0].to_pylist() == [
            {"widget_id": "a", "count": 0, "collected": None, "label": None, "nested": None}
        ]
E       AssertionError: assert [{'widget_id'...label': None}] == [{'widget_id'...': None, ...}]
E         
E         At index 0 diff: {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None} != {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None, 'nested': None}
E         
E         Full diff:
E           [
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': None,
E               },
E           ]

.../pipelines/jsonlines/test_jsonschema_pipeline.py:585: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonschema_end_to_end.py::test_cli_pass_loads_selected_entity[core-parquet]

Stack Traces | 0.642s run time

schema_input_dir = PosixPath('.../pytest-0/test_cli_pass_loads_selected_e3/input')
schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7f90384b0d60>
loader_file_format = <LoaderFileFormatEnum.PARQUET: 'parquet'>
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7f90383b4770>

    def test_cli_pass_loads_selected_entity(
        schema_input_dir: Path,
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        loader_file_format: str,
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """The real CTS CLI accepts a registry shortcut, table selection, and the requested file format."""
        settings = schema_settings_factory(
            input_dir=str(schema_input_dir),
            loader_file_format=loader_file_format,
            dataset_name="schema_cli",
            table_names=["gadget"],
        )
    
        load_info = assert_run_ok(
            pipeline_runner.run(
                JSONSCHEMA_PIPELINE_NAME,
                cli_args=[
                    *settings_to_argv(settings, exclude=["schema_files_module"]),
                    "-m",
                    settings.schema_files_module,
                ],
            )
        )
        assert load_info.dataset_name == "schema_cli"
        assert_pipeline_report_written(Path(settings.output_dir), load_info, metadata_moved=True)
>       assert read_data_tables(settings, load_info.dataset_name) == {"gadget": [EXPECTED_GADGET_ROW]}
E       AssertionError: assert {'gadget': [{...abel': None}]} == {'gadget': [{...: None, ...}]}
E         
E         Differing items:
E         {'gadget': [{'widget_id': 'g', 'count': 3, 'collected': None, 'label': None}]} != {'gadget': [{'widget_id': 'g', 'count': 3, 'collected': None, 'label': None, ...}]}
E         
E         Full diff:
E           {
E               'gadget': [
E                   {
E                       'widget_id': 'g',
E                       'count': 3,
E                       'collected': None,
E                       'label': None,
E         -             'nested': None,
E                   },
E               ],
E           }

.../jsonlines/integration/test_jsonschema_end_to_end.py:179: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonschema_end_to_end.py::test_cli_pass_loads_selected_entity[core-jsonl]

Stack Traces | 0.655s run time

schema_input_dir = PosixPath('.../pytest-0/test_cli_pass_loads_selected_e2/input')
schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7f90386faac0>
loader_file_format = <LoaderFileFormatEnum.JSONL: 'jsonl'>
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7f903872d850>

    def test_cli_pass_loads_selected_entity(
        schema_input_dir: Path,
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        loader_file_format: str,
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """The real CTS CLI accepts a registry shortcut, table selection, and the requested file format."""
        settings = schema_settings_factory(
            input_dir=str(schema_input_dir),
            loader_file_format=loader_file_format,
            dataset_name="schema_cli",
            table_names=["gadget"],
        )
    
        load_info = assert_run_ok(
            pipeline_runner.run(
                JSONSCHEMA_PIPELINE_NAME,
                cli_args=[
                    *settings_to_argv(settings, exclude=["schema_files_module"]),
                    "-m",
                    settings.schema_files_module,
                ],
            )
        )
        assert load_info.dataset_name == "schema_cli"
        assert_pipeline_report_written(Path(settings.output_dir), load_info, metadata_moved=True)
>       assert read_data_tables(settings, load_info.dataset_name) == {"gadget": [EXPECTED_GADGET_ROW]}
E       AssertionError: assert {'gadget': [{...abel': None}]} == {'gadget': [{...: None, ...}]}
E         
E         Differing items:
E         {'gadget': [{'widget_id': 'g', 'count': 3, 'collected': None, 'label': None}]} != {'gadget': [{'widget_id': 'g', 'count': 3, 'collected': None, 'label': None, ...}]}
E         
E         Full diff:
E           {
E               'gadget': [
E                   {
E                       'widget_id': 'g',
E                       'count': 3,
E                       'collected': None,
E                       'label': None,
E         -             'nested': None,
E                   },
E               ],
E           }

.../jsonlines/integration/test_jsonschema_end_to_end.py:179: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonschema_end_to_end.py::test_cli_pass_loads_selected_entity[cli-jsonl]

Stack Traces | 0.666s run time

schema_input_dir = PosixPath('.../pytest-0/test_cli_pass_loads_selected_e0/input')
schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7f907b5cff60>
loader_file_format = <LoaderFileFormatEnum.JSONL: 'jsonl'>
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7f907b9be510>

    def test_cli_pass_loads_selected_entity(
        schema_input_dir: Path,
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        loader_file_format: str,
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """The real CTS CLI accepts a registry shortcut, table selection, and the requested file format."""
        settings = schema_settings_factory(
            input_dir=str(schema_input_dir),
            loader_file_format=loader_file_format,
            dataset_name="schema_cli",
            table_names=["gadget"],
        )
    
        load_info = assert_run_ok(
            pipeline_runner.run(
                JSONSCHEMA_PIPELINE_NAME,
                cli_args=[
                    *settings_to_argv(settings, exclude=["schema_files_module"]),
                    "-m",
                    settings.schema_files_module,
                ],
            )
        )
        assert load_info.dataset_name == "schema_cli"
        assert_pipeline_report_written(Path(settings.output_dir), load_info, metadata_moved=True)
>       assert read_data_tables(settings, load_info.dataset_name) == {"gadget": [EXPECTED_GADGET_ROW]}
E       AssertionError: assert {'gadget': [{...abel': None}]} == {'gadget': [{...: None, ...}]}
E         
E         Differing items:
E         {'gadget': [{'widget_id': 'g', 'count': 3, 'collected': None, 'label': None}]} != {'gadget': [{'widget_id': 'g', 'count': 3, 'collected': None, 'label': None, ...}]}
E         
E         Full diff:
E           {
E               'gadget': [
E                   {
E                       'widget_id': 'g',
E                       'count': 3,
E                       'collected': None,
E                       'label': None,
E         -             'nested': None,
E                   },
E               ],
E           }

.../jsonlines/integration/test_jsonschema_end_to_end.py:179: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonschema_end_to_end.py::test_cli_pass_loads_selected_entity[cli-parquet]

Stack Traces | 0.68s run time

schema_input_dir = PosixPath('.../pytest-0/test_cli_pass_loads_selected_e1/input')
schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7f90521ccb80>
loader_file_format = <LoaderFileFormatEnum.PARQUET: 'parquet'>
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7f905208c0b0>

    def test_cli_pass_loads_selected_entity(
        schema_input_dir: Path,
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        loader_file_format: str,
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """The real CTS CLI accepts a registry shortcut, table selection, and the requested file format."""
        settings = schema_settings_factory(
            input_dir=str(schema_input_dir),
            loader_file_format=loader_file_format,
            dataset_name="schema_cli",
            table_names=["gadget"],
        )
    
        load_info = assert_run_ok(
            pipeline_runner.run(
                JSONSCHEMA_PIPELINE_NAME,
                cli_args=[
                    *settings_to_argv(settings, exclude=["schema_files_module"]),
                    "-m",
                    settings.schema_files_module,
                ],
            )
        )
        assert load_info.dataset_name == "schema_cli"
        assert_pipeline_report_written(Path(settings.output_dir), load_info, metadata_moved=True)
>       assert read_data_tables(settings, load_info.dataset_name) == {"gadget": [EXPECTED_GADGET_ROW]}
E       AssertionError: assert {'gadget': [{...abel': None}]} == {'gadget': [{...: None, ...}]}
E         
E         Differing items:
E         {'gadget': [{'widget_id': 'g', 'count': 3, 'collected': None, 'label': None}]} != {'gadget': [{'widget_id': 'g', 'count': 3, 'collected': None, 'label': None, ...}]}
E         
E         Full diff:
E           {
E               'gadget': [
E                   {
E                       'widget_id': 'g',
E                       'count': 3,
E                       'collected': None,
E                       'label': None,
E         -             'nested': None,
E                   },
E               ],
E           }

.../jsonlines/integration/test_jsonschema_end_to_end.py:179: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonschema_end_to_end.py::test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows[core-single-page-parquet]

Stack Traces | 0.712s run time

schema_input_dir = PosixPath('.../pytest-0/test_run_jsonlines_pipeline_pa18/input')
schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7f9038437420>
loader_file_format = <LoaderFileFormatEnum.PARQUET: 'parquet'>, buffer_size = 10
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7f90382fdfd0>

    @pytest.mark.parametrize("buffer_size", [1, 2, 10], ids=["single-row", "partial-pages", "single-page"])
    def test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows(
        schema_input_dir: Path,
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        loader_file_format: str,
        buffer_size: int,
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """Mixed files load every row exactly once with intact nested data and rejection provenance."""
        settings = schema_settings_factory(
            input_dir=str(schema_input_dir), loader_file_format=loader_file_format, buffer_size=buffer_size
        )
    
        load_info = assert_run_ok(pipeline_runner.run(JSONSCHEMA_PIPELINE_NAME, settings=settings))
        assert load_info.pipeline.pipeline_name == JSONSCHEMA_PIPELINE_NAME
        assert load_info.dataset_name == "schema_test"
        assert_pipeline_report_written(Path(settings.output_dir), load_info, metadata_moved=True)
        tables = read_data_tables(settings, load_info.dataset_name)
        assert set(tables) == {"widget", "widget_rejected", "gadget"}
>       assert [{key: parse_json_string(value) for key, value in row.items()} for row in tables["widget"]] == [
            EXPECTED_WIDGET_ROW
        ] * 3
E       AssertionError: assert [{'widget_id'...label': None}] == [{'widget_id'...': None, ...}]
E         
E         At index 0 diff: {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None} != {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None, 'nested': {'tags': ['one', 'two']}}
E         
E         Full diff:
E           [
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E           ]

.../jsonlines/integration/test_jsonschema_end_to_end.py:92: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonschema_end_to_end.py::test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows[core-single-page-jsonl]

Stack Traces | 0.716s run time

schema_input_dir = PosixPath('.../pytest-0/test_run_jsonlines_pipeline_pa6/input')
schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7f90459dc0e0>
loader_file_format = <LoaderFileFormatEnum.JSONL: 'jsonl'>, buffer_size = 10
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7f90524e7e30>

    @pytest.mark.parametrize("buffer_size", [1, 2, 10], ids=["single-row", "partial-pages", "single-page"])
    def test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows(
        schema_input_dir: Path,
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        loader_file_format: str,
        buffer_size: int,
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """Mixed files load every row exactly once with intact nested data and rejection provenance."""
        settings = schema_settings_factory(
            input_dir=str(schema_input_dir), loader_file_format=loader_file_format, buffer_size=buffer_size
        )
    
        load_info = assert_run_ok(pipeline_runner.run(JSONSCHEMA_PIPELINE_NAME, settings=settings))
        assert load_info.pipeline.pipeline_name == JSONSCHEMA_PIPELINE_NAME
        assert load_info.dataset_name == "schema_test"
        assert_pipeline_report_written(Path(settings.output_dir), load_info, metadata_moved=True)
        tables = read_data_tables(settings, load_info.dataset_name)
        assert set(tables) == {"widget", "widget_rejected", "gadget"}
>       assert [{key: parse_json_string(value) for key, value in row.items()} for row in tables["widget"]] == [
            EXPECTED_WIDGET_ROW
        ] * 3
E       AssertionError: assert [{'widget_id'...label': None}] == [{'widget_id'...': None, ...}]
E         
E         At index 0 diff: {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None} != {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None, 'nested': {'tags': ['one', 'two']}}
E         
E         Full diff:
E           [
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E           ]

.../jsonlines/integration/test_jsonschema_end_to_end.py:92: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonschema_end_to_end.py::test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows[core-single-row-jsonl]

Stack Traces | 0.74s run time

schema_input_dir = PosixPath('.../pytest-0/test_run_jsonlines_pipeline_pa19/input')
schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7f90381ea160>
loader_file_format = <LoaderFileFormatEnum.JSONL: 'jsonl'>, buffer_size = 1
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7f9038202c90>

    @pytest.mark.parametrize("buffer_size", [1, 2, 10], ids=["single-row", "partial-pages", "single-page"])
    def test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows(
        schema_input_dir: Path,
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        loader_file_format: str,
        buffer_size: int,
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """Mixed files load every row exactly once with intact nested data and rejection provenance."""
        settings = schema_settings_factory(
            input_dir=str(schema_input_dir), loader_file_format=loader_file_format, buffer_size=buffer_size
        )
    
        load_info = assert_run_ok(pipeline_runner.run(JSONSCHEMA_PIPELINE_NAME, settings=settings))
        assert load_info.pipeline.pipeline_name == JSONSCHEMA_PIPELINE_NAME
        assert load_info.dataset_name == "schema_test"
        assert_pipeline_report_written(Path(settings.output_dir), load_info, metadata_moved=True)
        tables = read_data_tables(settings, load_info.dataset_name)
        assert set(tables) == {"widget", "widget_rejected", "gadget"}
>       assert [{key: parse_json_string(value) for key, value in row.items()} for row in tables["widget"]] == [
            EXPECTED_WIDGET_ROW
        ] * 3
E       AssertionError: assert [{'widget_id'...label': None}] == [{'widget_id'...': None, ...}]
E         
E         At index 0 diff: {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None} != {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None, 'nested': {'tags': ['one', 'two']}}
E         
E         Full diff:
E           [
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E           ]

.../jsonlines/integration/test_jsonschema_end_to_end.py:92: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonschema_end_to_end.py::test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows[core-partial-pages-parquet]

Stack Traces | 0.75s run time

schema_input_dir = PosixPath('.../pytest-0/test_run_jsonlines_pipeline_pa4/input')
schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7f905227e840>
loader_file_format = <LoaderFileFormatEnum.PARQUET: 'parquet'>, buffer_size = 2
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7f90520be5d0>

    @pytest.mark.parametrize("buffer_size", [1, 2, 10], ids=["single-row", "partial-pages", "single-page"])
    def test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows(
        schema_input_dir: Path,
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        loader_file_format: str,
        buffer_size: int,
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """Mixed files load every row exactly once with intact nested data and rejection provenance."""
        settings = schema_settings_factory(
            input_dir=str(schema_input_dir), loader_file_format=loader_file_format, buffer_size=buffer_size
        )
    
        load_info = assert_run_ok(pipeline_runner.run(JSONSCHEMA_PIPELINE_NAME, settings=settings))
        assert load_info.pipeline.pipeline_name == JSONSCHEMA_PIPELINE_NAME
        assert load_info.dataset_name == "schema_test"
        assert_pipeline_report_written(Path(settings.output_dir), load_info, metadata_moved=True)
        tables = read_data_tables(settings, load_info.dataset_name)
        assert set(tables) == {"widget", "widget_rejected", "gadget"}
>       assert [{key: parse_json_string(value) for key, value in row.items()} for row in tables["widget"]] == [
            EXPECTED_WIDGET_ROW
        ] * 3
E       AssertionError: assert [{'widget_id'...label': None}] == [{'widget_id'...': None, ...}]
E         
E         At index 0 diff: {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None} != {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None, 'nested': {'tags': ['one', 'two']}}
E         
E         Full diff:
E           [
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E           ]

.../jsonlines/integration/test_jsonschema_end_to_end.py:92: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonschema_end_to_end.py::test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows[cli-partial-pages-jsonl]

Stack Traces | 0.754s run time

schema_input_dir = PosixPath('.../pytest-0/test_run_jsonlines_pipeline_pa17/input')
schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7f903856b560>
loader_file_format = <LoaderFileFormatEnum.JSONL: 'jsonl'>, buffer_size = 2
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7f90383db650>

    @pytest.mark.parametrize("buffer_size", [1, 2, 10], ids=["single-row", "partial-pages", "single-page"])
    def test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows(
        schema_input_dir: Path,
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        loader_file_format: str,
        buffer_size: int,
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """Mixed files load every row exactly once with intact nested data and rejection provenance."""
        settings = schema_settings_factory(
            input_dir=str(schema_input_dir), loader_file_format=loader_file_format, buffer_size=buffer_size
        )
    
        load_info = assert_run_ok(pipeline_runner.run(JSONSCHEMA_PIPELINE_NAME, settings=settings))
        assert load_info.pipeline.pipeline_name == JSONSCHEMA_PIPELINE_NAME
        assert load_info.dataset_name == "schema_test"
        assert_pipeline_report_written(Path(settings.output_dir), load_info, metadata_moved=True)
        tables = read_data_tables(settings, load_info.dataset_name)
        assert set(tables) == {"widget", "widget_rejected", "gadget"}
>       assert [{key: parse_json_string(value) for key, value in row.items()} for row in tables["widget"]] == [
            EXPECTED_WIDGET_ROW
        ] * 3
E       AssertionError: assert [{'widget_id'...label': None}] == [{'widget_id'...': None, ...}]
E         
E         At index 0 diff: {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None} != {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None, 'nested': {'tags': ['one', 'two']}}
E         
E         Full diff:
E           [
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E           ]

.../jsonlines/integration/test_jsonschema_end_to_end.py:92: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonschema_end_to_end.py::test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows[core-partial-pages-jsonl]

Stack Traces | 0.757s run time

schema_input_dir = PosixPath('.../pytest-0/test_run_jsonlines_pipeline_pa8/input')
schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7f9037897240>
loader_file_format = <LoaderFileFormatEnum.JSONL: 'jsonl'>, buffer_size = 2
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7f90378d1d90>

    @pytest.mark.parametrize("buffer_size", [1, 2, 10], ids=["single-row", "partial-pages", "single-page"])
    def test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows(
        schema_input_dir: Path,
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        loader_file_format: str,
        buffer_size: int,
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """Mixed files load every row exactly once with intact nested data and rejection provenance."""
        settings = schema_settings_factory(
            input_dir=str(schema_input_dir), loader_file_format=loader_file_format, buffer_size=buffer_size
        )
    
        load_info = assert_run_ok(pipeline_runner.run(JSONSCHEMA_PIPELINE_NAME, settings=settings))
        assert load_info.pipeline.pipeline_name == JSONSCHEMA_PIPELINE_NAME
        assert load_info.dataset_name == "schema_test"
        assert_pipeline_report_written(Path(settings.output_dir), load_info, metadata_moved=True)
        tables = read_data_tables(settings, load_info.dataset_name)
        assert set(tables) == {"widget", "widget_rejected", "gadget"}
>       assert [{key: parse_json_string(value) for key, value in row.items()} for row in tables["widget"]] == [
            EXPECTED_WIDGET_ROW
        ] * 3
E       AssertionError: assert [{'widget_id'...label': None}] == [{'widget_id'...': None, ...}]
E         
E         At index 0 diff: {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None} != {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None, 'nested': {'tags': ['one', 'two']}}
E         
E         Full diff:
E           [
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E           ]

.../jsonlines/integration/test_jsonschema_end_to_end.py:92: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonschema_end_to_end.py::test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows[cli-single-page-jsonl]

Stack Traces | 0.763s run time

schema_input_dir = PosixPath('.../pytest-0/test_run_jsonlines_pipeline_pa9/input')
schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7f90378971a0>
loader_file_format = <LoaderFileFormatEnum.JSONL: 'jsonl'>, buffer_size = 10
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7f903985d3d0>

    @pytest.mark.parametrize("buffer_size", [1, 2, 10], ids=["single-row", "partial-pages", "single-page"])
    def test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows(
        schema_input_dir: Path,
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        loader_file_format: str,
        buffer_size: int,
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """Mixed files load every row exactly once with intact nested data and rejection provenance."""
        settings = schema_settings_factory(
            input_dir=str(schema_input_dir), loader_file_format=loader_file_format, buffer_size=buffer_size
        )
    
        load_info = assert_run_ok(pipeline_runner.run(JSONSCHEMA_PIPELINE_NAME, settings=settings))
        assert load_info.pipeline.pipeline_name == JSONSCHEMA_PIPELINE_NAME
        assert load_info.dataset_name == "schema_test"
        assert_pipeline_report_written(Path(settings.output_dir), load_info, metadata_moved=True)
        tables = read_data_tables(settings, load_info.dataset_name)
        assert set(tables) == {"widget", "widget_rejected", "gadget"}
>       assert [{key: parse_json_string(value) for key, value in row.items()} for row in tables["widget"]] == [
            EXPECTED_WIDGET_ROW
        ] * 3
E       AssertionError: assert [{'widget_id'...label': None}] == [{'widget_id'...': None, ...}]
E         
E         At index 0 diff: {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None} != {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None, 'nested': {'tags': ['one', 'two']}}
E         
E         Full diff:
E           [
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E           ]

.../jsonlines/integration/test_jsonschema_end_to_end.py:92: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonschema_end_to_end.py::test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows[cli-single-page-parquet]

Stack Traces | 0.763s run time

schema_input_dir = PosixPath('.../pytest-0/test_run_jsonlines_pipeline_pa7/input')
schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7f9052117c40>
loader_file_format = <LoaderFileFormatEnum.PARQUET: 'parquet'>, buffer_size = 10
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7f905251eab0>

    @pytest.mark.parametrize("buffer_size", [1, 2, 10], ids=["single-row", "partial-pages", "single-page"])
    def test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows(
        schema_input_dir: Path,
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        loader_file_format: str,
        buffer_size: int,
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """Mixed files load every row exactly once with intact nested data and rejection provenance."""
        settings = schema_settings_factory(
            input_dir=str(schema_input_dir), loader_file_format=loader_file_format, buffer_size=buffer_size
        )
    
        load_info = assert_run_ok(pipeline_runner.run(JSONSCHEMA_PIPELINE_NAME, settings=settings))
        assert load_info.pipeline.pipeline_name == JSONSCHEMA_PIPELINE_NAME
        assert load_info.dataset_name == "schema_test"
        assert_pipeline_report_written(Path(settings.output_dir), load_info, metadata_moved=True)
        tables = read_data_tables(settings, load_info.dataset_name)
        assert set(tables) == {"widget", "widget_rejected", "gadget"}
>       assert [{key: parse_json_string(value) for key, value in row.items()} for row in tables["widget"]] == [
            EXPECTED_WIDGET_ROW
        ] * 3
E       AssertionError: assert [{'widget_id'...label': None}] == [{'widget_id'...': None, ...}]
E         
E         At index 0 diff: {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None} != {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None, 'nested': {'tags': ['one', 'two']}}
E         
E         Full diff:
E           [
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E           ]

.../jsonlines/integration/test_jsonschema_end_to_end.py:92: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonschema_end_to_end.py::test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows[core-single-row-parquet]

Stack Traces | 0.767s run time

schema_input_dir = PosixPath('.../pytest-0/test_run_jsonlines_pipeline_pa11/input')
schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7f90387694e0>
loader_file_format = <LoaderFileFormatEnum.PARQUET: 'parquet'>, buffer_size = 1
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7f9038767830>

    @pytest.mark.parametrize("buffer_size", [1, 2, 10], ids=["single-row", "partial-pages", "single-page"])
    def test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows(
        schema_input_dir: Path,
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        loader_file_format: str,
        buffer_size: int,
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """Mixed files load every row exactly once with intact nested data and rejection provenance."""
        settings = schema_settings_factory(
            input_dir=str(schema_input_dir), loader_file_format=loader_file_format, buffer_size=buffer_size
        )
    
        load_info = assert_run_ok(pipeline_runner.run(JSONSCHEMA_PIPELINE_NAME, settings=settings))
        assert load_info.pipeline.pipeline_name == JSONSCHEMA_PIPELINE_NAME
        assert load_info.dataset_name == "schema_test"
        assert_pipeline_report_written(Path(settings.output_dir), load_info, metadata_moved=True)
        tables = read_data_tables(settings, load_info.dataset_name)
        assert set(tables) == {"widget", "widget_rejected", "gadget"}
>       assert [{key: parse_json_string(value) for key, value in row.items()} for row in tables["widget"]] == [
            EXPECTED_WIDGET_ROW
        ] * 3
E       AssertionError: assert [{'widget_id'...label': None}] == [{'widget_id'...': None, ...}]
E         
E         At index 0 diff: {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None} != {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None, 'nested': {'tags': ['one', 'two']}}
E         
E         Full diff:
E           [
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E           ]

.../jsonlines/integration/test_jsonschema_end_to_end.py:92: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonschema_end_to_end.py::test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows[cli-partial-pages-parquet]

Stack Traces | 0.781s run time

schema_input_dir = PosixPath('.../pytest-0/test_run_jsonlines_pipeline_pa13/input')
schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7f90385fe020>
loader_file_format = <LoaderFileFormatEnum.PARQUET: 'parquet'>, buffer_size = 2
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7f9038635cd0>

    @pytest.mark.parametrize("buffer_size", [1, 2, 10], ids=["single-row", "partial-pages", "single-page"])
    def test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows(
        schema_input_dir: Path,
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        loader_file_format: str,
        buffer_size: int,
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """Mixed files load every row exactly once with intact nested data and rejection provenance."""
        settings = schema_settings_factory(
            input_dir=str(schema_input_dir), loader_file_format=loader_file_format, buffer_size=buffer_size
        )
    
        load_info = assert_run_ok(pipeline_runner.run(JSONSCHEMA_PIPELINE_NAME, settings=settings))
        assert load_info.pipeline.pipeline_name == JSONSCHEMA_PIPELINE_NAME
        assert load_info.dataset_name == "schema_test"
        assert_pipeline_report_written(Path(settings.output_dir), load_info, metadata_moved=True)
        tables = read_data_tables(settings, load_info.dataset_name)
        assert set(tables) == {"widget", "widget_rejected", "gadget"}
>       assert [{key: parse_json_string(value) for key, value in row.items()} for row in tables["widget"]] == [
            EXPECTED_WIDGET_ROW
        ] * 3
E       AssertionError: assert [{'widget_id'...label': None}] == [{'widget_id'...': None, ...}]
E         
E         At index 0 diff: {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None} != {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None, 'nested': {'tags': ['one', 'two']}}
E         
E         Full diff:
E           [
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E           ]

.../jsonlines/integration/test_jsonschema_end_to_end.py:92: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonschema_end_to_end.py::test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows[cli-single-row-jsonl]

Stack Traces | 0.793s run time

schema_input_dir = PosixPath('.../pytest-0/test_run_jsonlines_pipeline_pa16/input')
schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7f903856b9c0>
loader_file_format = <LoaderFileFormatEnum.JSONL: 'jsonl'>, buffer_size = 1
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7f90383da510>

    @pytest.mark.parametrize("buffer_size", [1, 2, 10], ids=["single-row", "partial-pages", "single-page"])
    def test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows(
        schema_input_dir: Path,
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        loader_file_format: str,
        buffer_size: int,
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """Mixed files load every row exactly once with intact nested data and rejection provenance."""
        settings = schema_settings_factory(
            input_dir=str(schema_input_dir), loader_file_format=loader_file_format, buffer_size=buffer_size
        )
    
        load_info = assert_run_ok(pipeline_runner.run(JSONSCHEMA_PIPELINE_NAME, settings=settings))
        assert load_info.pipeline.pipeline_name == JSONSCHEMA_PIPELINE_NAME
        assert load_info.dataset_name == "schema_test"
        assert_pipeline_report_written(Path(settings.output_dir), load_info, metadata_moved=True)
        tables = read_data_tables(settings, load_info.dataset_name)
        assert set(tables) == {"widget", "widget_rejected", "gadget"}
>       assert [{key: parse_json_string(value) for key, value in row.items()} for row in tables["widget"]] == [
            EXPECTED_WIDGET_ROW
        ] * 3
E       AssertionError: assert [{'widget_id'...label': None}] == [{'widget_id'...': None, ...}]
E         
E         At index 0 diff: {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None} != {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None, 'nested': {'tags': ['one', 'two']}}
E         
E         Full diff:
E           [
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E           ]

.../jsonlines/integration/test_jsonschema_end_to_end.py:92: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonschema_end_to_end.py::test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows[cli-single-row-parquet]

Stack Traces | 0.803s run time

schema_input_dir = PosixPath('.../pytest-0/test_run_jsonlines_pipeline_pa15/input')
schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7f90386f9ee0>
loader_file_format = <LoaderFileFormatEnum.PARQUET: 'parquet'>, buffer_size = 1
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7f903854cb90>

    @pytest.mark.parametrize("buffer_size", [1, 2, 10], ids=["single-row", "partial-pages", "single-page"])
    def test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows(
        schema_input_dir: Path,
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        loader_file_format: str,
        buffer_size: int,
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """Mixed files load every row exactly once with intact nested data and rejection provenance."""
        settings = schema_settings_factory(
            input_dir=str(schema_input_dir), loader_file_format=loader_file_format, buffer_size=buffer_size
        )
    
        load_info = assert_run_ok(pipeline_runner.run(JSONSCHEMA_PIPELINE_NAME, settings=settings))
        assert load_info.pipeline.pipeline_name == JSONSCHEMA_PIPELINE_NAME
        assert load_info.dataset_name == "schema_test"
        assert_pipeline_report_written(Path(settings.output_dir), load_info, metadata_moved=True)
        tables = read_data_tables(settings, load_info.dataset_name)
        assert set(tables) == {"widget", "widget_rejected", "gadget"}
>       assert [{key: parse_json_string(value) for key, value in row.items()} for row in tables["widget"]] == [
            EXPECTED_WIDGET_ROW
        ] * 3
E       AssertionError: assert [{'widget_id'...label': None}] == [{'widget_id'...': None, ...}]
E         
E         At index 0 diff: {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None} != {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None, 'nested': {'tags': ['one', 'two']}}
E         
E         Full diff:
E           [
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E           ]

.../jsonlines/integration/test_jsonschema_end_to_end.py:92: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonschema_end_to_end.py::test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows[docker-single-row-parquet]

Stack Traces | 1.73s run time

schema_input_dir = PosixPath('.../pytest-0/test_run_jsonlines_pipeline_pa5/input')
schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7fc113e4ede0>
loader_file_format = <LoaderFileFormatEnum.PARQUET: 'parquet'>, buffer_size = 1
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7fc113e18d10>

    @pytest.mark.parametrize("buffer_size", [1, 2, 10], ids=["single-row", "partial-pages", "single-page"])
    def test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows(
        schema_input_dir: Path,
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        loader_file_format: str,
        buffer_size: int,
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """Mixed files load every row exactly once with intact nested data and rejection provenance."""
        settings = schema_settings_factory(
            input_dir=str(schema_input_dir), loader_file_format=loader_file_format, buffer_size=buffer_size
        )
    
>       load_info = assert_run_ok(pipeline_runner.run(JSONSCHEMA_PIPELINE_NAME, settings=settings))
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.../jsonlines/integration/test_jsonschema_end_to_end.py:86: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

load_info = None

    def assert_run_ok(load_info: LoadInfo | None) -> LoadInfo:
        """Assert a pipeline run succeeded, and return the load info for chained use."""
>       assert load_info is not None
               ^^^^^^^^^^^^^^^^^^^^^
E       AssertionError

.../cdm_data_loaders/pipelines/helpers.py:159: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonschema_end_to_end.py::test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows[docker-partial-pages-jsonl]

Stack Traces | 1.83s run time

schema_input_dir = PosixPath('.../pytest-0/test_run_jsonlines_pipeline_pa2/input')
schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7fc113f9c900>
loader_file_format = <LoaderFileFormatEnum.JSONL: 'jsonl'>, buffer_size = 2
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7fc113f94c30>

    @pytest.mark.parametrize("buffer_size", [1, 2, 10], ids=["single-row", "partial-pages", "single-page"])
    def test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows(
        schema_input_dir: Path,
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        loader_file_format: str,
        buffer_size: int,
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """Mixed files load every row exactly once with intact nested data and rejection provenance."""
        settings = schema_settings_factory(
            input_dir=str(schema_input_dir), loader_file_format=loader_file_format, buffer_size=buffer_size
        )
    
>       load_info = assert_run_ok(pipeline_runner.run(JSONSCHEMA_PIPELINE_NAME, settings=settings))
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.../jsonlines/integration/test_jsonschema_end_to_end.py:86: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

load_info = None

    def assert_run_ok(load_info: LoadInfo | None) -> LoadInfo:
        """Assert a pipeline run succeeded, and return the load info for chained use."""
>       assert load_info is not None
               ^^^^^^^^^^^^^^^^^^^^^
E       AssertionError

.../cdm_data_loaders/pipelines/helpers.py:159: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonschema_end_to_end.py::test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows[docker-partial-pages-parquet]

Stack Traces | 1.91s run time

schema_input_dir = PosixPath('.../pytest-0/test_run_jsonlines_pipeline_pa1/input')
schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7fc113f83420>
loader_file_format = <LoaderFileFormatEnum.PARQUET: 'parquet'>, buffer_size = 2
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7fc120189bd0>

    @pytest.mark.parametrize("buffer_size", [1, 2, 10], ids=["single-row", "partial-pages", "single-page"])
    def test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows(
        schema_input_dir: Path,
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        loader_file_format: str,
        buffer_size: int,
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """Mixed files load every row exactly once with intact nested data and rejection provenance."""
        settings = schema_settings_factory(
            input_dir=str(schema_input_dir), loader_file_format=loader_file_format, buffer_size=buffer_size
        )
    
>       load_info = assert_run_ok(pipeline_runner.run(JSONSCHEMA_PIPELINE_NAME, settings=settings))
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.../jsonlines/integration/test_jsonschema_end_to_end.py:86: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

load_info = None

    def assert_run_ok(load_info: LoadInfo | None) -> LoadInfo:
        """Assert a pipeline run succeeded, and return the load info for chained use."""
>       assert load_info is not None
               ^^^^^^^^^^^^^^^^^^^^^
E       AssertionError

.../cdm_data_loaders/pipelines/helpers.py:159: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonschema_end_to_end.py::test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows[docker-single-row-jsonl]

Stack Traces | 1.93s run time

schema_input_dir = PosixPath('.../pytest-0/test_run_jsonlines_pipeline_pa6/input')
schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7fc113e4f740>
loader_file_format = <LoaderFileFormatEnum.JSONL: 'jsonl'>, buffer_size = 1
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7fc113e18af0>

    @pytest.mark.parametrize("buffer_size", [1, 2, 10], ids=["single-row", "partial-pages", "single-page"])
    def test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows(
        schema_input_dir: Path,
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        loader_file_format: str,
        buffer_size: int,
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """Mixed files load every row exactly once with intact nested data and rejection provenance."""
        settings = schema_settings_factory(
            input_dir=str(schema_input_dir), loader_file_format=loader_file_format, buffer_size=buffer_size
        )
    
>       load_info = assert_run_ok(pipeline_runner.run(JSONSCHEMA_PIPELINE_NAME, settings=settings))
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

.../jsonlines/integration/test_jsonschema_end_to_end.py:86: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

load_info = None

    def assert_run_ok(load_info: LoadInfo | None) -> LoadInfo:
        """Assert a pipeline run succeeded, and return the load info for chained use."""
>       assert load_info is not None
               ^^^^^^^^^^^^^^^^^^^^^
E       AssertionError

.../cdm_data_loaders/pipelines/helpers.py:159: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonschema_end_to_end.py::test_cli_pass_loads_selected_entity[docker-parquet]

Stack Traces | 2.07s run time

schema_input_dir = PosixPath('.../pytest-0/test_cli_pass_loads_selected_e1/input')
schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7fc113ea0040>
loader_file_format = <LoaderFileFormatEnum.PARQUET: 'parquet'>
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7fc113e21f50>

    def test_cli_pass_loads_selected_entity(
        schema_input_dir: Path,
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        loader_file_format: str,
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """The real CTS CLI accepts a registry shortcut, table selection, and the requested file format."""
        settings = schema_settings_factory(
            input_dir=str(schema_input_dir),
            loader_file_format=loader_file_format,
            dataset_name="schema_cli",
            table_names=["gadget"],
        )
    
        load_info = assert_run_ok(
            pipeline_runner.run(
                JSONSCHEMA_PIPELINE_NAME,
                cli_args=[
                    *settings_to_argv(settings, exclude=["schema_files_module"]),
                    "-m",
                    settings.schema_files_module,
                ],
            )
        )
        assert load_info.dataset_name == "schema_cli"
        assert_pipeline_report_written(Path(settings.output_dir), load_info, metadata_moved=True)
>       assert read_data_tables(settings, load_info.dataset_name) == {"gadget": [EXPECTED_GADGET_ROW]}
E       AssertionError: assert {'gadget': [{...abel': None}]} == {'gadget': [{...: None, ...}]}
E         
E         Differing items:
E         {'gadget': [{'widget_id': 'g', 'count': 3, 'collected': None, 'label': None}]} != {'gadget': [{'widget_id': 'g', 'count': 3, 'collected': None, 'label': None, ...}]}
E         
E         Full diff:
E           {
E               'gadget': [
E                   {
E                       'widget_id': 'g',
E                       'count': 3,
E                       'collected': None,
E                       'label': None,
E         -             'nested': None,
E                   },
E               ],
E           }

.../jsonlines/integration/test_jsonschema_end_to_end.py:179: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonschema_end_to_end.py::test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows[docker-single-page-parquet]

Stack Traces | 2.08s run time

schema_input_dir = PosixPath('.../pytest-0/test_run_jsonlines_pipeline_pa7/input')
schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7fc113ea0d60>
loader_file_format = <LoaderFileFormatEnum.PARQUET: 'parquet'>, buffer_size = 10
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7fc113e22a50>

    @pytest.mark.parametrize("buffer_size", [1, 2, 10], ids=["single-row", "partial-pages", "single-page"])
    def test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows(
        schema_input_dir: Path,
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        loader_file_format: str,
        buffer_size: int,
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """Mixed files load every row exactly once with intact nested data and rejection provenance."""
        settings = schema_settings_factory(
            input_dir=str(schema_input_dir), loader_file_format=loader_file_format, buffer_size=buffer_size
        )
    
        load_info = assert_run_ok(pipeline_runner.run(JSONSCHEMA_PIPELINE_NAME, settings=settings))
        assert load_info.pipeline.pipeline_name == JSONSCHEMA_PIPELINE_NAME
        assert load_info.dataset_name == "schema_test"
        assert_pipeline_report_written(Path(settings.output_dir), load_info, metadata_moved=True)
        tables = read_data_tables(settings, load_info.dataset_name)
        assert set(tables) == {"widget", "widget_rejected", "gadget"}
>       assert [{key: parse_json_string(value) for key, value in row.items()} for row in tables["widget"]] == [
            EXPECTED_WIDGET_ROW
        ] * 3
E       AssertionError: assert [{'widget_id'...label': None}] == [{'widget_id'...': None, ...}]
E         
E         At index 0 diff: {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None} != {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None, 'nested': {'tags': ['one', 'two']}}
E         
E         Full diff:
E           [
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E           ]

.../jsonlines/integration/test_jsonschema_end_to_end.py:92: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonschema_end_to_end.py::test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows[docker-single-page-jsonl]

Stack Traces | 2.08s run time

schema_input_dir = PosixPath('.../pytest-0/test_run_jsonlines_pipeline_pa4/input')
schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7fc113e4dc60>
loader_file_format = <LoaderFileFormatEnum.JSONL: 'jsonl'>, buffer_size = 10
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7fc113fa44d0>

    @pytest.mark.parametrize("buffer_size", [1, 2, 10], ids=["single-row", "partial-pages", "single-page"])
    def test_run_jsonlines_pipeline_pass_persists_valid_and_rejected_rows(
        schema_input_dir: Path,
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        loader_file_format: str,
        buffer_size: int,
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """Mixed files load every row exactly once with intact nested data and rejection provenance."""
        settings = schema_settings_factory(
            input_dir=str(schema_input_dir), loader_file_format=loader_file_format, buffer_size=buffer_size
        )
    
        load_info = assert_run_ok(pipeline_runner.run(JSONSCHEMA_PIPELINE_NAME, settings=settings))
        assert load_info.pipeline.pipeline_name == JSONSCHEMA_PIPELINE_NAME
        assert load_info.dataset_name == "schema_test"
        assert_pipeline_report_written(Path(settings.output_dir), load_info, metadata_moved=True)
        tables = read_data_tables(settings, load_info.dataset_name)
        assert set(tables) == {"widget", "widget_rejected", "gadget"}
>       assert [{key: parse_json_string(value) for key, value in row.items()} for row in tables["widget"]] == [
            EXPECTED_WIDGET_ROW
        ] * 3
E       AssertionError: assert [{'widget_id'...label': None}] == [{'widget_id'...': None, ...}]
E         
E         At index 0 diff: {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None} != {'widget_id': 'a', 'count': 0, 'collected': None, 'label': None, 'nested': {'tags': ['one', 'two']}}
E         
E         Full diff:
E           [
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E               {
E                   'widget_id': 'a',
E                   'count': 0,
E                   'collected': None,
E                   'label': None,
E         -         'nested': {
E         -             'tags': [
E         -                 'one',
E         -                 'two',
E         -             ],
E         -         },
E               },
E           ]

.../jsonlines/integration/test_jsonschema_end_to_end.py:92: AssertionError

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonschema_end_to_end.py::test_cli_pass_loads_selected_entity[docker-jsonl]

Stack Traces | 2.24s run time

schema_input_dir = PosixPath('.../pytest-0/test_cli_pass_loads_selected_e0/input')
schema_settings_factory = <function schema_settings_factory.<locals>._factory at 0x7fc120337b00>
loader_file_format = <LoaderFileFormatEnum.JSONL: 'jsonl'>
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7fc1201882d0>

    def test_cli_pass_loads_selected_entity(
        schema_input_dir: Path,
        schema_settings_factory: Callable[..., JsonlJsonschemaIngestSettings],
        loader_file_format: str,
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """The real CTS CLI accepts a registry shortcut, table selection, and the requested file format."""
        settings = schema_settings_factory(
            input_dir=str(schema_input_dir),
            loader_file_format=loader_file_format,
            dataset_name="schema_cli",
            table_names=["gadget"],
        )
    
        load_info = assert_run_ok(
            pipeline_runner.run(
                JSONSCHEMA_PIPELINE_NAME,
                cli_args=[
                    *settings_to_argv(settings, exclude=["schema_files_module"]),
                    "-m",
                    settings.schema_files_module,
                ],
            )
        )
        assert load_info.dataset_name == "schema_cli"
        assert_pipeline_report_written(Path(settings.output_dir), load_info, metadata_moved=True)
>       assert read_data_tables(settings, load_info.dataset_name) == {"gadget": [EXPECTED_GADGET_ROW]}
E       AssertionError: assert {'gadget': [{...abel': None}]} == {'gadget': [{...: None, ...}]}
E         
E         Differing items:
E         {'gadget': [{'widget_id': 'g', 'count': 3, 'collected': None, 'label': None}]} != {'gadget': [{'widget_id': 'g', 'count': 3, 'collected': None, 'label': None, ...}]}
E         
E         Full diff:
E           {
E               'gadget': [
E                   {
E                       'widget_id': 'g',
E                       'count': 3,
E                       'collected': None,
E                       'label': None,
E         -             'nested': None,
E                   },
E               ],
E           }

.../jsonlines/integration/test_jsonschema_end_to_end.py:179: AssertionError

View the full list of 1 ❄️ flaky test(s)

tests/cdm_data_loaders/pipelines/jsonlines/integration/test_jsonlines_end_to_end.py::test_run_jsonlines_pipeline_fail_unknown_table_name_raises[docker]

Flake rate in main: 100.00% (Passed 0 times, Failed 3 times)

Stack Traces | 1.07s run time

settings_factory = <function settings_factory.<locals>._factory at 0x7fc0b41d14e0>
pipeline_runner = <tests.cdm_data_loaders.pipelines.framework.PipelineTestRunner object at 0x7fc0a4f8ba70>

    def test_run_jsonlines_pipeline_fail_unknown_table_name_raises(
        settings_factory: Callable[..., JsonlPydanticIngestSettings],
        pipeline_runner: PipelineTestRunner,
    ) -> None:
        """A table name in table_names that is not in the entity model registry raises ValueError."""
        settings = settings_factory(table_names=["not_a_real_table"])
    
>       with pytest.raises(ValueError, match="Unknown table name"):
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E       Failed: DID NOT RAISE ValueError

.../jsonlines/integration/test_jsonlines_end_to_end.py:46: Failed

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}

ialarmedalien added 2 commits October 26, 2026 10:27

Adding in pyarrow schema for jsonschema ingestion pw

ba92285

Formatting; minor typo update

45243eb

Merge branch 'develop' into jsonschema_pyarrow

350139d

ialarmedalien closed this Oct 27, 2026

ialarmedalien deleted the jsonschema_pyarrow branch October 27, 2026 20:59

Navigation Menu

Repository navigation

Adding in pyarrow schema for jsonschema ingestion pw - #302

Adding in pyarrow schema for jsonschema ingestion pw#302
ialarmedalien wants to merge 3 commits into
developfrom
jsonschema_pyarrow

ialarmedalien commented Oct 26, 2026

Uh oh!

codecov Bot commented Oct 26, 2026 •

edited

Loading

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Conversation

ialarmedalien commented Oct 26, 2026

Uh oh!

codecov Bot commented Oct 26, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

❌ 20 Tests Failed:

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

codecov Bot commented Oct 26, 2026 •

edited

Loading