vkgs-sim / tests /test_jsonstream.py

Add files using upload-large-folder tool

ceea884 verified 8 days ago

3.26 kB

"""스트리밍 JSON 리더 — 경계 조건 위주.
이 파서는 1.35 GB 파일을 다루므로 '버퍼 경계에 원소가 걸치는' 경우가 반드시 생긴다.
청크 크기를 아주 작게 강제해 그 경로를 일부러 밟는다.
"""

import json

import pytest

from nusim.io.jsonstream import index_table, iter_json_array, load_table


def _write(tmp_path, obj, name="t.json"):
    p = tmp_path / name
    p.write_text(json.dumps(obj, ensure_ascii=False))
    return p


def test_roundtrip_simple(tmp_path):
    data = [{"token": f"t{i}", "v": i} for i in range(100)]
    assert list(iter_json_array(_write(tmp_path, data))) == data


def test_empty_array(tmp_path):
    assert list(iter_json_array(_write(tmp_path, []))) == []


def test_single_element(tmp_path):
    assert list(iter_json_array(_write(tmp_path, [{"a": 1}]))) == [{"a": 1}]


@pytest.mark.parametrize("chunk", [1, 2, 3, 7, 16, 64, 4096])
def test_tiny_chunks_force_boundary_splits(tmp_path, chunk):
    """청크가 1바이트여도 결과가 같아야 한다. 경계 재시도 로직의 핵심 검증."""
    data = [{"token": f"tok{i}", "nested": {"x": [1, 2, 3]}, "s": "a" * i} for i in range(40)]
    p = _write(tmp_path, data)
    assert list(iter_json_array(p, chunk_size=chunk)) == data


def test_unicode_and_escapes(tmp_path):
    data = [
        {"desc": "싱가포르 onenorth, 우천"},
        {"desc": 'quote " and backslash \\ and bracket ] inside'},
        {"desc": "새\n줄\t탭"},
    ]
    p = _write(tmp_path, data)
    # ']' 가 문자열 안에 있어도 배열 끝으로 오인하면 안 된다
    assert list(iter_json_array(p, chunk_size=3)) == data


def test_whitespace_and_pretty_printed(tmp_path):
    data = [{"a": 1}, {"b": 2}]
    p = tmp_path / "pretty.json"
    p.write_text("\n\n  " + json.dumps(data, indent=4) + "\n\n")
    assert list(iter_json_array(p, chunk_size=5)) == data


def test_not_an_array(tmp_path):
    p = tmp_path / "obj.json"
    p.write_text('{"a": 1}')
    with pytest.raises(ValueError, match="배열"):
        list(iter_json_array(p))


def test_truncated_file_raises(tmp_path):
    p = tmp_path / "cut.json"
    p.write_text('[{"a": 1}, {"b": ')
    with pytest.raises(ValueError, match="잘렸"):
        list(iter_json_array(p, chunk_size=4))


def test_load_table_filters_rows_and_fields(tmp_path):
    data = [{"token": str(i), "keep": i % 2 == 0, "big": "x" * 50} for i in range(10)]
    p = _write(tmp_path, data)
    got = load_table(p, keep=lambda r: r["keep"], fields=("token",))
    assert got == [{"token": str(i)} for i in range(0, 10, 2)]


def test_index_table(tmp_path):
    data = [{"token": "a", "v": 1}, {"token": "b", "v": 2}]
    got = index_table(_write(tmp_path, data), fields=("v",))
    assert got == {"a": {"v": 1}, "b": {"v": 2}}


def test_memory_stays_bounded(tmp_path):
    """원소를 다 모으지 않고 흘려보내는지. 제너레이터가 리스트로 퇴화하면 실패한다."""
    import sys

    data = [{"i": i, "pad": "y" * 200} for i in range(5000)]
    p = _write(tmp_path, data)
    it = iter_json_array(p, chunk_size=1024)
    first = next(it)
    assert first["i"] == 0
    # 제너레이터 자체는 작아야 한다
    assert sys.getsizeof(it) < 1000

	"""스트리밍 JSON 리더 — 경계 조건 위주.

	이 파서는 1.35 GB 파일을 다루므로 '버퍼 경계에 원소가 걸치는' 경우가 반드시 생긴다.
	청크 크기를 아주 작게 강제해 그 경로를 일부러 밟는다.
	"""

	import json

	import pytest

	from nusim.io.jsonstream import index_table, iter_json_array, load_table


	def _write(tmp_path, obj, name="t.json"):
	p = tmp_path / name
	p.write_text(json.dumps(obj, ensure_ascii=False))
	return p


	def test_roundtrip_simple(tmp_path):
	data = [{"token": f"t{i}", "v": i} for i in range(100)]
	assert list(iter_json_array(_write(tmp_path, data))) == data


	def test_empty_array(tmp_path):
	assert list(iter_json_array(_write(tmp_path, []))) == []


	def test_single_element(tmp_path):
	assert list(iter_json_array(_write(tmp_path, [{"a": 1}]))) == [{"a": 1}]


	@pytest.mark.parametrize("chunk", [1, 2, 3, 7, 16, 64, 4096])
	def test_tiny_chunks_force_boundary_splits(tmp_path, chunk):
	"""청크가 1바이트여도 결과가 같아야 한다. 경계 재시도 로직의 핵심 검증."""
	data = [{"token": f"tok{i}", "nested": {"x": [1, 2, 3]}, "s": "a" * i} for i in range(40)]
	p = _write(tmp_path, data)
	assert list(iter_json_array(p, chunk_size=chunk)) == data


	def test_unicode_and_escapes(tmp_path):
	data = [
	{"desc": "싱가포르 onenorth, 우천"},
	{"desc": 'quote " and backslash \\ and bracket ] inside'},
	{"desc": "새\n줄\t탭"},
	]
	p = _write(tmp_path, data)
	# ']' 가 문자열 안에 있어도 배열 끝으로 오인하면 안 된다
	assert list(iter_json_array(p, chunk_size=3)) == data


	def test_whitespace_and_pretty_printed(tmp_path):
	data = [{"a": 1}, {"b": 2}]
	p = tmp_path / "pretty.json"
	p.write_text("\n\n " + json.dumps(data, indent=4) + "\n\n")
	assert list(iter_json_array(p, chunk_size=5)) == data


	def test_not_an_array(tmp_path):
	p = tmp_path / "obj.json"
	p.write_text('{"a": 1}')
	with pytest.raises(ValueError, match="배열"):
	list(iter_json_array(p))


	def test_truncated_file_raises(tmp_path):
	p = tmp_path / "cut.json"
	p.write_text('[{"a": 1}, {"b": ')
	with pytest.raises(ValueError, match="잘렸"):
	list(iter_json_array(p, chunk_size=4))


	def test_load_table_filters_rows_and_fields(tmp_path):
	data = [{"token": str(i), "keep": i % 2 == 0, "big": "x" * 50} for i in range(10)]
	p = _write(tmp_path, data)
	got = load_table(p, keep=lambda r: r["keep"], fields=("token",))
	assert got == [{"token": str(i)} for i in range(0, 10, 2)]


	def test_index_table(tmp_path):
	data = [{"token": "a", "v": 1}, {"token": "b", "v": 2}]
	got = index_table(_write(tmp_path, data), fields=("v",))
	assert got == {"a": {"v": 1}, "b": {"v": 2}}


	def test_memory_stays_bounded(tmp_path):
	"""원소를 다 모으지 않고 흘려보내는지. 제너레이터가 리스트로 퇴화하면 실패한다."""
	import sys

	data = [{"i": i, "pad": "y" * 200} for i in range(5000)]
	p = _write(tmp_path, data)
	it = iter_json_array(p, chunk_size=1024)
	first = next(it)
	assert first["i"] == 0
	# 제너레이터 자체는 작아야 한다
	assert sys.getsizeof(it) < 1000