Quick answer: Test JSON text sequences at six boundaries: encoded framing, multiline values, fragmented UTF-8, incomplete scalars, malformed-record recovery, and consumer resource limits. Use byte fixtures with expected accepted and rejected records. A parser accepting individual JSON objects does not prove that a stream preserves their boundaries or delivers a complete dataset.
A streaming import works with two compact objects, then fails when an upstream service enables pretty printing. Before debugging the data fields, check the framing contract. A line reader and a record-separator reader are answering different questions.
This guide addresses application/json-seq, separately from our NDJSON and JSON Lines guide and AI response streaming guide. The Python fixtures are focused acceptance tests, not a production network parser. Run equivalent cases through your deployed library and transport adapter before claiming end-to-end compatibility.
1. Does the producer emit the agreed record framing?
Assert the producer's output bytes rather than checking only the decoded objects. A JSON text sequence encoder prefixes each UTF-8 JSON value with record separator and appends line feed. Confirm the configured media type and test several top-level value types, because object-only fixtures can hide assumptions about strings, numbers, booleans and null.
RFC 7464 is the foundational February 2015 specification for this format. Its encoder framing uses byte 0x1E before a value and byte 0x0A after it. That framing contract differs from an ordinary JSON array and from simply concatenating serialized objects without delimiters.
import json
RS = b"\x1e"
def encode_record(value, pretty=False):
text = json.dumps(value, ensure_ascii=False, allow_nan=False,
indent=2 if pretty else None)
return RS + text.encode("utf-8") + b"\n"
assert encode_record({"id": 1}) == b'\x1e{"id": 1}\n'
for value in [{"id": 1}, [1, 2], "hello", 12, True, None]:
record = encode_record(value)
assert record.startswith(RS) and record.endswith(b"\n")
assert json.loads(record[1:]) == value
Keep a captured byte fixture alongside a human-readable representation. A terminal or log viewer can make a control byte invisible, leaving two different streams looking alike. Inspect escaped bytes or hexadecimal output when diagnosing failures; do not infer the presence of separators from how lines happen to appear on screen.
Verify your response configuration too. An endpoint sending this representation should identify the agreed format, and its client must select a compatible reader. Our JSON content-type guide covers the broader metadata boundary. Changing a response header alone does not convert the bytes already produced by an NDJSON serializer.
For a third-party source, preserve a short sanitized sample in the compatibility suite. Recreate its exact serializer options, including non-ASCII output, rather than testing a cleaner locally invented substitute. Record which producer configuration supplied that fixture so a future formatting change can be evaluated deliberately.
2. Can a completed record contain formatting newlines?
Test a pretty-printed object as one record, then verify that its internal newlines do not become extra records. Record-separated framing and newline-delimited framing have different boundaries. Extract a completed JSON value before inspecting its structure, and do not remove all newlines as a shortcut because that can conceal a reader using the wrong protocol.
The NDJSON specification prohibits literal newlines inside each serialized JSON text. In contrast, the record separator supplies the boundary for a JSON text sequence. That difference matters when an exporter changes indentation without changing the logical fields or values it sends.
pretty = encode_record({"id": 1, "details": {"ok": True}}, pretty=True)
assert pretty.count(b"\n") > 1
assert pretty.count(RS) == 1
assert json.loads(pretty[1:]) == {"id": 1, "details": {"ok": True}}
escaped = encode_record({"message": "first\nsecond", "marker": "\x1e"})
assert escaped.count(RS) == 1
assert json.loads(escaped[1:])["marker"] == "\x1e"
The escaped marker fixture checks a second boundary: a separator-looking character inside a string must remain data after decoding. Compare the original bytes and the decoded string separately. A displayed escape sequence is not the same representation as a raw control byte placed between records.
Use the JSON Viewer on an extracted completed record to inspect nested fields or copy a path. The viewer expects a JSON document and does not certify record-stream framing. Keep the framed source separately so a successful inspection cannot be mistaken for proof that the entire sequence was parsed correctly.
Wikipedia's JSON streaming overview gives background on the different framing families. For implementation behavior, use the selected library's documentation and your executable corpus. Similar names such as JSON stream and JSON Lines are insufficient evidence that two endpoints can exchange the same bytes.
3. Do fragmented reads preserve UTF-8 and record identity?
Feed the same stream through many artificial read boundaries and require identical decoded records. Include splits inside a multibyte character, between a separator and its JSON text, and between adjacent records. Reassembly tests expose byte handling mistakes; the deployed incremental parser still needs its own integration test with the same fragmentation corpus.
Python's JSON module documentation describes decoding JSON documents, not a complete record-stream transport. Keep framing responsibility outside the single-document decoder. The following test deliberately reassembles bytes before decoding; it demonstrates why independently decoding each network read is an unsafe assumption, not how to implement bounded incremental buffering.
wire = encode_record({"name": "Maya 馃榾"}) + encode_record({"id": 2})
expected = [{"name": "Maya 馃榾"}, {"id": 2}]
for split in range(len(wire) + 1):
chunks = [wire[:split], wire[split:]]
reassembled = b"".join(chunks)
values = [json.loads(part) for part in reassembled.split(RS)[1:]]
assert values == expected
emoji_start = wire.index("馃榾".encode("utf-8"))
try:
wire[:emoji_start + 1].decode("utf-8")
except UnicodeDecodeError:
pass
else:
raise AssertionError("expected an incomplete UTF-8 character")
Extend the integration corpus beyond two chunks. Test one byte per read, multiple records in one read, and a separator at the end of one read with the payload in the next. Keep the expected order and record count explicit so dropped or duplicated records cannot disappear behind a final aggregate total.
Do not use the join-based fixture as a memory-bounded production design. It retains the whole stream and is suitable only for these small test inputs. Your actual reader should maintain the documented incremental state and deliver completed values without requiring the upstream connection to close first.
Record the transport adapter and decoder configuration with the test results. A standalone decoder can pass every fixture while the HTTP adapter decodes bytes prematurely or retries a partial read incorrectly. Put the same byte corpus through that adapter before treating local parser success as service compatibility.
4. Does the consumer distinguish complete scalars from truncated input?
Test incomplete top-level numbers and literals separately from malformed objects. A prefix can resemble a complete scalar even when more bytes were intended. Preserve the framing completion policy and verify it before delivering a value; a single-document decoder accepting a numeric prefix does not establish that the producer finished writing the record.
The sequence specification treats trailing whitespace as important for detecting possible scalar truncation. For the fixtures below, an intentionally strict application profile requires every completed frame to end in line feed, including object records. This profile is narrower than the RFC parser's recovery grammar; it is not a claim that every conforming parser must reject every missing-LF object.
def decode_complete_frame(frame):
if not frame.endswith(b"\n"):
raise ValueError("incomplete application frame")
return json.loads(frame.decode("utf-8"))
assert decode_complete_frame(b"12\n") == 12
assert decode_complete_frame(b"true\n") is True
for frame in [b"12", b"tru", b'{"id":', b'{"id": 1}']:
try:
decode_complete_frame(frame)
except ValueError:
pass
else:
raise AssertionError("incomplete frame accepted")
Document the chosen completion rule in the endpoint contract and test the receiving library against it. Avoid silently assuming that a permissive parser enforces your stricter ingestion profile. A wrapper may need an explicit completion check, and that wrapper needs regression fixtures of its own.
Keep disconnect handling observable. Report whether the final record was accepted, rejected or left incomplete, and do not acknowledge it upstream merely because some prefix decoded successfully. An import that delivers earlier records and then disconnects requires a defined partial-success outcome rather than an invented all-or-nothing interpretation.
5. Does malformed-record recovery match the application's data policy?
Place a malformed record between two valid records and assert both processing results and rejection reporting. Continuing after an error can be useful for independent events, but it can hide missing data in an import that requires completeness. Make recovery a named policy instead of treating later valid output as proof that the stream succeeded.
The jq manual's sequence mode documents warning about malformed input and discarding input until the next record separator. That is a tool behavior to account for when building pipelines. It does not decide whether your application should commit a partially processed dataset or report the overall operation as successful.
parts = [b'{"id": 1}\n', b'{broken}\n', b'{"id": 3}\n']
accepted, rejected = [], []
for index, frame in enumerate(parts):
try:
accepted.append(decode_complete_frame(frame))
except ValueError:
rejected.append(index)
assert accepted == [{"id": 1}, {"id": 3}]
assert rejected == [1]
assert len(accepted) + len(rejected) == len(parts)
This loop tests a recovery decision after framing has already identified the records. It does not implement resynchronization on a live byte stream. Run the malformed-middle fixture through the actual sequence reader too, and verify that its recovery boundary agrees with the application report.
Retain rejected-record counts and safe identifiers, with payload logging governed by your data policy. A quiet skip can turn an apparently healthy dashboard into an incomplete dataset. Conversely, stopping an entire telemetry feed on one bad record may be inappropriate. Review the consequence of loss before choosing which failure should terminate processing.
Include an operational test where the consumer finishes with one rejected record and the reporting layer receives that count. A parser-only assertion cannot prove an alert, job result or administrative screen exposes partial failure. Require the visible result to distinguish a clean run from a recovered run, and verify that automated callers can make the same distinction without scraping a warning message from logs.
6. Do size limits and downstream gates apply before records are committed?
Test an exact record-byte limit, a value above it, and a valid JSON value with the wrong application shape. Stream framing proves neither resource safety nor domain validity. Apply limits during accumulation, then enforce the record contract before side effects, and test that rejected records never reach persistence or event publication.
The example below checks the size of an already completed frame. A production reader must additionally stop oversized accumulation before buffering an unbounded record. Keep the byte limit separate from a count of decoded characters, and distinguish per-record limits from limits on the whole import or connection lifetime.
def accept_event(frame, max_bytes):
if len(frame) > max_bytes:
return False
value = decode_complete_frame(frame)
return (isinstance(value, dict)
and type(value.get("id")) is int)
frame = b'{"id": 1}\n'
assert accept_event(frame, len(frame))
assert not accept_event(frame, len(frame) - 1)
assert not accept_event(b"null\n", 100)
assert not accept_event(b'{"id": true}\n', 100)
These examples use Python's standard library and check only the shown fixtures. Add cancellation, backpressure and persistence tests in the deployed application. Specifically, confirm that a slow consumer does not let an upstream reader build an unlimited queue, and that cancellation releases the connection and pending buffers according to the chosen client library's lifecycle.
Test retries with stable event identifiers when repeated delivery is possible. A correctly framed record can still represent an event already processed during an earlier connection. Define the application's duplicate policy and confirm it at the side-effect boundary; changing the delimiter cannot provide exactly-once processing. Keep this delivery concern separate from syntax recovery so a replay is not mislabeled as a malformed record.
Keep a release artifact containing expected accepted records, rejected records, final completion status and configured limits. The six checks separate encoding, framing, decoding and business acceptance, making failures easier to locate. Passing them locally supplies focused evidence; it does not certify arbitrary stream libraries, prove complete delivery, or replace testing the actual producer-to-consumer path.