ww/lib/bufio/stream_test.ww

// Row ownership mirrors ref/hare/bufio/stream_test+test.ha.
// Streams are built as `let st = memio.fixed(buf); &st.vt` — the #94
// unified value-return io surface (read/write/close return size/io.error).

package bufio_test;

import bufio;
import errors;
import io;
import memio;

fn sputstr(s: str, into: []u8, off: i32) i32 = {
	let i: i32 = 0;
	for (i < s.len) {
		into[off + i] = s[i];
		i += 1;
	};
	return off + s.len;
};

// closevt forwards writes to closesink (so a flush is observable) and
// records whether its closer fired in srcclosed.
let srcclosed: bool = false;
let closevt: io.vtable;
let closesink: memio.stream;

fn closewrite(s: io.stream, buf: []u8) (size | io.error) = {
	return io.write(&closesink.vt, buf);
};
fn closeclose(s: io.stream) (void | io.error) = {
	srcclosed = true;
	return void;
};
fn closesource() io.stream = {
	closevt.writer = (&closewrite): *io.writer;
	closevt.closer = (&closeclose): *io.closer;
	return &closevt;
};

type failstream = struct {
	vt: io.vtable,
	out: [16]u8,
	pos: i32,
	calls: i32,
	zero: bool,
};

fn failwrite(s: io.stream, buf: []u8) (size | io.error) = {
	let f: *failstream = s: *failstream;
	f.calls += 1;
	if (f.calls == 2) {
		if (f.zero) { return 0: size; };
		let nm: nomem;
		let e: io.error = nm;
		return e;
	};
	let n: i32 = buf.len;
	if (f.calls == 1 && n > 2) { n = 2; };
	let i: i32 = 0;
	for (i < n) {
		f.out[f.pos + i] = buf[i];
		i += 1;
	};
	f.pos += n;
	return n: size;
};

fn checkpartialflush(zero: bool) void = {
	let sink: failstream;
	sink.vt.writer = (&failwrite): *io.writer;
	sink.pos = 0;
	sink.calls = 0;
	sink.zero = zero;
	let rb: [1]u8;
	let wb: [4]u8;
	let b: bufio.stream = bufio.newstream(&sink.vt, rb[0:1], wb[0:4]);
	let src: [4]u8;
	let _: i32 = sputstr("ABCD", src[0:4], 0);
	let wr: (size | io.error) = io.write(&b.vt, src[0:4]);
	match (wr) { case let n: size => {}; case let e: io.error => abort(); };
	let first: (void | io.error) = bufio.flush(&b);
	match (first) {
	case void => abort();
	case let e: io.error => assert(e is nomem);
	};
	assert(sink.pos == 2);
	assert(b.wend == 2);
	assert(wb[0] == 'C' && wb[1] == 'D');
	let second: (void | io.error) = bufio.flush(&b);
	match (second) { case void => {}; case let e: io.error => abort(); };
	assert(b.wend == 0);
	assert(sink.pos == 4);
	assert(sink.out[0] == 'A' && sink.out[1] == 'B');
	assert(sink.out[2] == 'C' && sink.out[3] == 'D');
};

@test fn streampartialflusherror() void = {
	checkpartialflush(false);
};

@test fn streampartialflushzero() void = {
	checkpartialflush(true);
};

@test fn streamsmallwrite() void = {
	let raw: [16]u8;
	let mem: memio.stream = memio.fixed(raw[0:16]);
	let m:   io.stream = &mem.vt;

	let rb: [8]u8;
	let wb: [8]u8;
	let b: bufio.stream = bufio.newstream(m, rb[0:8], wb[0:8]);

	// Write under the buffer limit — nothing reaches src.
	let buf: [5]u8;
	let z: i32 = sputstr("hello", buf[0:5], 0);
	let p: io.stream = &b.vt;
	let r: (size | io.error) = io.write(p, buf[0:5]);
	match (r) {
	case let n: size => { assert(!(n: i32 != 5)); };
	case let e: io.error => abort();
	};
	assert(!(mem.pos != 0));

	let fr: (void | io.error) = bufio.flush(&b);
	match (fr) {
	case void => { };
	case let e: io.error => abort();
	};
	assert(!(mem.pos != 5));
	assert(!(raw[0] != 104u8));
	assert(!(raw[4] != 111u8));
};

@test fn streamautoflush() void = {
	let raw: [32]u8;
	let mem: memio.stream = memio.fixed(raw[0:32]);
	let m:   io.stream = &mem.vt;

	let rb: [4]u8;
	let wb: [4]u8;
	let b: bufio.stream = bufio.newstream(m, rb[0:4], wb[0:4]);

	// 10B payload through a 4B wbuf: bwrite must flush twice mid-write.
	let buf: [10]u8;
	let z: i32 = sputstr("abcdefghij", buf[0:10], 0);
	let p: io.stream = &b.vt;
	let r: (size | io.error) = io.write(p, buf[0:10]);
	match (r) {
	case let n: size => { assert(!(n: i32 != 10)); };
	case let e: io.error => abort();
	};
	// Two full flushes happened (8B); 2B still pending.
	assert(!(mem.pos != 8));

	let fr: (void | io.error) = bufio.flush(&b);
	match (fr) {
	case void => { };
	case let e: io.error => abort();
	};
	assert(!(mem.pos != 10));
	assert(!(raw[0] != 97u8));	// 'a'
	assert(!(raw[9] != 106u8));	// 'j'
};

@test fn streamlineflush() void = {
	let raw: [32]u8;
	let mem: memio.stream = memio.fixed(raw[0:32]);
	let m:   io.stream = &mem.vt;

	let rb: [4]u8;
	let wb: [16]u8;
	let b: bufio.stream = bufio.newstream(m, rb[0:4], wb[0:16]);

	let p: io.stream = &b.vt;

	// First write: no '\n', stays buffered.
	let h: [5]u8;
	let zh: i32 = sputstr("hello", h[0:5], 0);
	let r1: (size | io.error) = io.write(p, h[0:5]);
	match (r1) {
	case let n: size => { assert(!(n: i32 != 5)); };
	case let e: io.error => abort();
	};
	assert(!(mem.pos != 0));

	// Second write: '\n' present, triggers flush of everything (6B).
	let nl: [1]u8;
	nl[0] = 10u8;
	let r2: (size | io.error) = io.write(p, nl[0:1]);
	match (r2) {
	case let n: size => { assert(!(n: i32 != 1)); };
	case let e: io.error => abort();
	};
	assert(!(mem.pos != 6));
	assert(!(raw[0] != 104u8));
	assert(!(raw[5] != 10u8));
};

@test fn streamflushonwrite() void = {
	let raw: [32]u8;
	let mem: memio.stream = memio.fixed(raw[0:32]);
	let m:   io.stream = &mem.vt;

	let rb: [4]u8;
	let wb: [16]u8;
	let b: bufio.stream = bufio.newstream(m, rb[0:4], wb[0:16]);
	let nilbs: [1]u8;
	bufio.setflush(&b, nilbs[0:0]);

	let p: io.stream = &b.vt;
	let h: [2]u8;
	h[0] = 65u8; h[1] = 66u8;	// "AB"
	let r1: (size | io.error) = io.write(p, h[0:2]);
	match (r1) {
	case let n: size => { assert(!(n: i32 != 2)); };
	case let e: io.error => abort();
	};
	let f1: (void | io.error) = bufio.flush(&b);
	match (f1) {
	case void => { };
	case let e: io.error => abort();
	};
	assert(!(mem.pos != 2));
	assert(!(raw[0] != 65u8));

	let r2: (size | io.error) = io.write(p, h[0:1]);
	match (r2) {
	case let n: size => { assert(!(n: i32 != 1)); };
	case let e: io.error => abort();
	};
	let f2: (void | io.error) = bufio.flush(&b);
	match (f2) {
	case void => { };
	case let e: io.error => abort();
	};
	assert(!(mem.pos != 3));
	assert(!(raw[2] != 65u8));
};

@test fn streamisbuffered() void = {
	let raw: [8]u8;
	let mem: memio.stream = memio.fixed(raw[0:8]);
	let m:   io.stream = &mem.vt;

	let rb: [4]u8;
	let wb: [4]u8;
	let b: bufio.stream = bufio.newstream(m, rb[0:4], wb[0:4]);

	assert(!(!bufio.isbuffered(&b.vt)));
	assert(!(bufio.isbuffered(m)));
};

// rbuf holds SOME pending bytes (fewer than requested) AND has spare
// capacity: Hare shifts the pending region to the front and reads more
// into the tail before serving (ref/hare/bufio/stream.ha:209). Pre-fix
// ww served only the short pending count and never topped up.
@test fn streamreadtopup() void = {
	let raw: [16]u8;
	let n: i32 = sputstr("ABCDEFGHIJKLMNOP", raw[0:16], 0);

	let mem: memio.stream = memio.fixed(raw[0:n]);
	let m:   io.stream = &mem.vt;

	let rb: [8]u8;
	let wb: [4]u8;
	let b: bufio.stream = bufio.newstream(m, rb[0:8], wb[0:4]);

	let p: io.stream = &b.vt;

	// First read fills rbuf (8B "ABCDEFGH"), serves 3 — 5 pending remain.
	let out: [10]u8;
	let r1: (size | io.eof | io.error) = io.read(p, out[0:3]);
	match (r1) {
	case let z: size => { assert(!(z: i32 != 3)); };
	case io.eof => abort();
	case let e: io.error => abort();
	};

	// Second read wants 10 > 5 pending: Hare shifts "DEFGH" to the front,
	// reads 3 more ("IJK") into the tail, serves 8. Pre-fix ww returned 5.
	let r2: (size | io.eof | io.error) = io.read(p, out[0:10]);
	match (r2) {
	case let z: size => { assert(!(z: i32 != 8)); };
	case io.eof => abort();
	case let e: io.error => abort();
	};
	assert(!(out[0] != 68u8));	// 'D'
	assert(!(out[7] != 75u8));	// 'K'
};

// The load-bearing "EOF fatal only when avail==0" path
// (ref/hare/bufio/stream.ha:216): rbuf still holds pending bytes when
// the top-up io.read returns EOF — Hare drains the pending region, it
// does NOT propagate eof. Source is shorter than two reads so the
// second read's top-up io.read EOFs with bytes still buffered.
@test fn streamreadtopupeof() void = {
	let raw: [5]u8;
	let n: i32 = sputstr("ABCDE", raw[0:5], 0);

	let mem: memio.stream = memio.fixed(raw[0:n]);
	let m:   io.stream = &mem.vt;

	let rb: [8]u8;
	let wb: [4]u8;
	let b: bufio.stream = bufio.newstream(m, rb[0:8], wb[0:4]);

	let p: io.stream = &b.vt;

	// First read fills rbuf (5B "ABCDE", source drained), serves 3 —
	// 2 pending remain.
	let out: [10]u8;
	let r1: (size | io.eof | io.error) = io.read(p, out[0:3]);
	match (r1) {
	case let z: size => { assert(!(z: i32 != 3)); };
	case io.eof => abort();
	case let e: io.error => abort();
	};

	// Second read wants 10 > 2 pending: shift "DE" to the front, top-up
	// io.read hits EOF (source drained). avail==2 != 0 so Hare drains the
	// pending region — serves 2, NOT eof. "always fatal" would return eof.
	let r2: (size | io.eof | io.error) = io.read(p, out[0:10]);
	match (r2) {
	case let z: size => { assert(!(z: i32 != 2)); };
	case io.eof => abort();
	case let e: io.error => abort();
	};
	assert(!(out[0] != 68u8));	// 'D'
	assert(!(out[1] != 69u8));	// 'E'
};

@test fn streamunread() void = {
	let raw: [8]u8;
	let n: i32 = sputstr("ABCDEFGH", raw[0:8], 0);

	let mem: memio.stream = memio.fixed(raw[0:n]);
	let m:   io.stream = &mem.vt;

	let rb: [8]u8;
	let wb: [4]u8;
	let b: bufio.stream = bufio.newstream(m, rb[0:8], wb[0:4]);

	let p: io.stream = &b.vt;

	// Pull 3 bytes through bread — fills rbuf from src (8B), serves 3.
	let out: [4]u8;
	let r1: (size | io.eof | io.error) = io.read(p, out[0:3]);
	match (r1) {
	case let z: size => { assert(!(z: i32 != 3)); };
	case io.eof => abort();
	case let e: io.error => abort();
	};
	assert(!(out[0] != 65u8));	// 'A'
	assert(!(out[2] != 67u8));	// 'C'

	// Push back 2 bytes; next read returns them first.
	let push: [2]u8;
	push[0] = 88u8; push[1] = 89u8;	// "XY"
	bufio.unread(&b, push[0:2]);

	let r2: (size | io.eof | io.error) = io.read(p, out[0:2]);
	match (r2) {
	case let z: size => { assert(!(z: i32 != 2)); };
	case io.eof => abort();
	case let e: io.error => abort();
	};
	assert(!(out[0] != 88u8));	// 'X'
	assert(!(out[1] != 89u8));	// 'Y'

	// Subsequent read returns the original tail.
	let r3: (size | io.eof | io.error) = io.read(p, out[0:4]);
	match (r3) {
	case let z: size => { assert(!(z: i32 != 4)); };
	case io.eof => abort();
	case let e: io.error => abort();
	};
	assert(!(out[0] != 68u8));	// 'D'
	assert(!(out[3] != 71u8));	// 'G'
};

@test fn streamunreadoverlap() void = {
	let raw: [1]u8;
	let mem: memio.stream = memio.fixed(raw[0:0]);
	let rb: [4]u8;
	let _: i32 = sputstr("ABCD", rb[0:4], 0);
	let wb: [1]u8;
	let b: bufio.stream = bufio.newstream(&mem.vt, rb[0:4], wb[0:1]);
	bufio.unread(&b, rb[0:3]);
	let out: [3]u8;
	let r: (size | io.eof | io.error) = io.read(&b.vt, out[0:3]);
	match (r) {
	case let n: size => assert(n: i32 == 3);
	case io.eof => abort();
	case let e: io.error => abort();
	};
	assert(out[0] == 'A' && out[1] == 'B' && out[2] == 'C');
};

@test fn streamscannerunread() void = {
	let raw: [16]u8;
	let n: i32 = sputstr("hello\n", raw[0:16], 0);

	let mem: memio.stream = memio.fixed(raw[0:n]);
	let m:   io.stream = &mem.vt;

	let rb: [16]u8;
	let wb: [4]u8;
	let b: bufio.stream = bufio.newstream(m, rb[0:16], wb[0:4]);

	let push: [3]u8;
	push[0] = 88u8; push[1] = 89u8; push[2] = 90u8;	// "XYZ"
	bufio.unread(&b, push[0:3]);

	let p: io.stream = &b.vt;
	let sbuf: [32]u8;
	let sc:   bufio.scanner = bufio.newscannerbuf(p, sbuf[0:32]);

	let lr: (str | io.eof | io.error | errors.overflow) = bufio.scanline(&sc);
	match (lr) {
	case let v: str => {
		assert(!(v.len != 8));	// "XYZhello"
		assert(!(v[0] != 88u8));	// 'X'
		assert(!(v[7] != 111u8));	// 'o'
	};
	case io.eof => abort();
	case let e: io.error => abort();
	case errors.overflow => abort();
	};
	bufio.finish(&sc);
};

@test fn streamflushempty() void = {
	let raw: [8]u8;
	let mem: memio.stream = memio.fixed(raw[0:8]);
	let m:   io.stream = &mem.vt;

	let rb: [4]u8;
	let wb: [4]u8;
	let b: bufio.stream = bufio.newstream(m, rb[0:4], wb[0:4]);

	let r: (void | io.error) = bufio.flush(&b);
	match (r) {
	case void => { };
	case let e: io.error => abort();
	};
	assert(!(mem.pos != 0));
};

@test fn streamcloseflushes() void = {
	let raw: [8]u8;
	let mem: memio.stream = memio.fixed(raw[0:8]);
	let m:   io.stream = &mem.vt;

	let rb: [4]u8;
	let wb: [4]u8;
	let b: bufio.stream = bufio.newstream(m, rb[0:4], wb[0:4]);
	let nilbs: [1]u8;
	bufio.setflush(&b, nilbs[0:0]);	// suppress the default '\n' detector

	let p: io.stream = &b.vt;
	let h: [2]u8;
	h[0] = 80u8; h[1] = 81u8;	// "PQ"
	let r: (size | io.error) = io.write(p, h[0:2]);
	match (r) {
	case let n: size => { assert(!(n: i32 != 2)); };
	case let e: io.error => abort();
	};
	assert(!(mem.pos != 0));	// still buffered

	let cr: (void | io.error) = io.close(p);
	match (cr) {
	case void => { };
	case let e: io.error => abort();
	};
	assert(!(mem.pos != 2));
	assert(!(raw[0] != 80u8));
	assert(!(raw[1] != 81u8));
};

// Hare's close_buffered closes src only under flag::MANAGED_HANDLE
// (ref/hare/bufio/stream.ha:194); init's default flag::NONE flushes
// only. The close-recording closesource lets us assert both: the
// buffered byte reaches the sink (flush) AND the closer never fires.
@test fn streamclosenopropagate() void = {
	srcclosed = false;
	let raw: [8]u8;
	closesink = memio.fixed(raw[0:8]);
	let src: io.stream = closesource();

	let rb: [4]u8;
	let wb: [4]u8;
	let b: bufio.stream = bufio.newstream(src, rb[0:4], wb[0:4]);
	let nilbs: [1]u8;
	bufio.setflush(&b, nilbs[0:0]);	// suppress the default '\n' detector

	let p: io.stream = &b.vt;
	let h: [2]u8;
	h[0] = 80u8; h[1] = 81u8;	// "PQ"
	let r: (size | io.error) = io.write(p, h[0:2]);
	match (r) {
	case let n: size => { assert(!(n: i32 != 2)); };
	case let e: io.error => abort();
	};
	assert(!(closesink.pos != 0));	// still buffered

	let cr: (void | io.error) = io.close(p);
	match (cr) {
	case void => { };
	case let e: io.error => abort();
	};
	// Flush happened: the 2 buffered bytes reached the sink.
	assert(!(closesink.pos != 2));
	assert(!(raw[0] != 80u8));
	assert(!(raw[1] != 81u8));
	// But the underlying was NOT closed (Hare flag::NONE default).
	assert(!srcclosed);
};

@test fn streamsetflushcustom() void = {
	let raw: [16]u8;
	let mem: memio.stream = memio.fixed(raw[0:16]);
	let m:   io.stream = &mem.vt;

	let rb: [4]u8;
	let wb: [16]u8;
	let b: bufio.stream = bufio.newstream(m, rb[0:4], wb[0:16]);
	let bs: [1]u8;
	bs[0] = 32u8;	// ' '
	bufio.setflush(&b, bs[0:1]);

	let p: io.stream = &b.vt;
	let buf: [5]u8;
	let z: i32 = sputstr("ab cd", buf[0:5], 0);
	let r: (size | io.error) = io.write(p, buf[0:5]);
	match (r) {
	case let n: size => { assert(!(n: i32 != 5)); };
	case let e: io.error => abort();
	};
	// Space at index 2 triggers post-copy flush of all 5 bytes.
	assert(!(mem.pos != 5));
	assert(!(raw[2] != 32u8));
	assert(!(raw[4] != 100u8));	// 'd'
};