mesh-catalog/modules/brother/cmd/brother/documents_test.go

package main

import (
	"bytes"
	"context"
	"encoding/base64"
	"encoding/binary"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"
)

// pdfOf is a document that is a PDF to the sniff and size bytes long, its bytes varying so that a part out of
// place changes the join.
func pdfOf(size int) []byte {
	b := make([]byte, size)
	copy(b, "%PDF-1.4\n")
	for i := 9; i < size; i++ {
		b[i] = byte(i*7 + i/251)
	}
	return b
}

// 1453/AC10: a document is cut into parts of at most PartSize, under the limit, and the parts joined in
// order are the document, with the SHA-256 checked; a wrong SHA-256 or a part missing is refused; a part out of
// range is refused naming the range.
func TestADocumentIsCutIntoPartsAndJoinedWithItsSha256Checked(t *testing.T) {
	doc := pdfOf(2*AnswerPartSize + 1234)
	n := answerPartCount(len(doc))
	if n != 3 {
		t.Fatalf("%d parts, want 3", n)
	}
	var parts [][]byte
	for i := 1; i <= n; i++ {
		p, total, err := cutPart(doc, i)
		if err != nil || total != n {
			t.Fatalf("part %d: %v (of %d)", i, err, total)
		}
		if len(p) > AnswerPartSize {
			t.Errorf("part %d is %d bytes, over %d", i, len(p), AnswerPartSize)
		}
		parts = append(parts, p)
	}
	got, err := joinParts(parts, sha256Hex(doc))
	if err != nil || !bytes.Equal(got, doc) {
		t.Fatalf("the join is not the document: %v", err)
	}
	if _, err := joinParts(parts, sha256Hex([]byte("other"))); err == nil || !strings.Contains(err.Error(), sha256Hex(doc)) {
		t.Errorf("a wrong SHA-256 was accepted or the error does not name the joined one: %v", err)
	}
	if _, err := joinParts(parts[:2], sha256Hex(doc)); err == nil {
		t.Errorf("a join with a part missing passed its SHA-256")
	}
	for _, bad := range []int{0, -1, n + 1} {
		if _, _, err := cutPart(doc, bad); err == nil || !strings.Contains(err.Error(), "out of range") {
			t.Errorf("part %d: error = %v", bad, err)
		}
	}
	if partCount(0) != 1 {
		t.Errorf("an empty document has one part")
	}
}

// 1453/AC10: scanner.document gives each part of a kept scan with the part count, the size and the SHA-256, each
// answer within the bus's 1 MiB; the parts joined are the scan. A part out of range, or a key that is no scan's, is
// refused, the first before anything is queued.
func TestScannerDocumentGivesEachPartOfAKeptScan(t *testing.T) {
	r := newRig(t, `{}`, "pw")
	doc := pdfOf(AnswerPartSize + 5000)
	key := sha256Hex(doc) + ".pdf"
	r.objects.Put(t.Context(), key, "application/pdf", doc)
	var joined []byte
	for i := 1; i <= 2; i++ {
		got, err := r.call(t, "scanner.document", map[string]any{"object": key, "part": float64(i)})
		if err != nil {
			t.Fatalf("part %d: %v", i, err)
		}
		m := got.(map[string]any)
		if m["parts"] != float64(2) || m["size"] != float64(len(doc)) || m["sha256"] != sha256Hex(doc) || m["part"] != float64(i) {
			t.Errorf("part %d answer = %v", i, m)
		}
		raw, err := base64.StdEncoding.DecodeString(m["data"].(string))
		if err != nil {
			t.Fatal(err)
		}
		joined = append(joined, raw...)
	}
	if !bytes.Equal(joined, doc) {
		t.Errorf("the parts fetched are not the scan")
	}
	before := r.bus.asks()
	if _, err := r.call(t, "scanner.document", map[string]any{"object": key, "part": float64(0)}); err == nil {
		t.Errorf("part 0 was asked for")
	}
	if r.bus.asks() != before {
		t.Errorf("an ask was queued for a part the tool could refuse")
	}
	if _, err := r.call(t, "scanner.document", map[string]any{"object": key, "part": float64(3)}); err == nil || !strings.Contains(err.Error(), "out of range") {
		t.Errorf("part 3: error = %v", err)
	}
	for _, k := range []string{"../etc/passwd", "uploads/x.pdf", strings.ToUpper(key)} {
		if _, err := r.call(t, "scanner.document", map[string]any{"object": k, "part": float64(1)}); err == nil || !strings.Contains(err.Error(), "not a scan's key") {
			t.Errorf("object %q: error = %v", k, err)
		}
	}
	if _, err := r.call(t, "scanner.document", map[string]any{"object": sha256Hex([]byte("none")) + ".pdf", "part": float64(1)}); err == nil {
		t.Errorf("a scan that is not kept was given")
	}
}

// 1453/AC10: a scan of more than one part is fetched whole by scanner.scan, joined, checked and filed.
func TestScanNowFetchesAScanOfSeveralParts(t *testing.T) {
	f := newFakeScanner(t, fixture(t, "scan.jpg"))
	r, _ := scanRig(t, f, Settings{Crop: false})
	// Everything a scan is made of but its size is the real one; the stored object is what the parts carry, so a
	// bigger object stands for a bigger scan: replace what the device keeps with a three-part document.
	big := pdfOf(2*AnswerPartSize + 99)
	bigKey := sha256Hex(big) + ".pdf"
	r.objects.Put(t.Context(), bigKey, "application/pdf", big)
	got := make([][]byte, 0, 3)
	for i := 1; i <= 3; i++ {
		p, err := r.worker.document(t.Context(), Ask{Op: OpDocument, Object: bigKey, Part: i})
		if err != nil {
			t.Fatal(err)
		}
		raw, _ := base64.StdEncoding.DecodeString(p["data"].(string))
		got = append(got, raw)
	}
	if joined, err := joinParts(got, sha256Hex(big)); err != nil || !bytes.Equal(joined, big) {
		t.Fatalf("three parts do not join: %v", err)
	}
	body, err := r.scan(t)
	if err != nil {
		t.Fatal(err)
	}
	if body["path"] == nil {
		t.Errorf("no path: %v", body)
	}
}

func upload(t *testing.T, r *printerRig, id string, doc []byte, sha string, part, parts int, piece []byte) (any, error) {
	t.Helper()
	return r.call(t, "printer.upload", map[string]any{"upload": id, "part": float64(part), "parts": float64(parts), "sha256": sha,
		"data": base64.StdEncoding.EncodeToString(piece)})
}

// 1453/AC11: an upload in parts, in any order, is stored as uploads/<id>.pdf once all have come and the joined
// SHA-256 matches; until then it says how many came and stores nothing.
func TestAnUploadIsStoredOnceAllItsPartsCameAndTheSha256Matches(t *testing.T) {
	r := newRig(t, `{}`, "pw")
	doc := pdfOf(PartSize + 700)
	sum := sha256Hex(doc)
	p1, _, _ := cutPartOf(doc, 1, PartSize)
	p2, _, _ := cutPartOf(doc, 2, PartSize)
	got, err := upload(t, r, "up-1", doc, sum, 2, 2, p2)
	if err != nil {
		t.Fatal(err)
	}
	if m := got.(map[string]any); m["complete"] != false || m["received"] != float64(1) {
		t.Errorf("after one part: %v", m)
	}
	if len(r.objects.keys()) != 0 {
		t.Errorf("a document was stored before its parts all came: %v", r.objects.keys())
	}
	got, err = upload(t, r, "up-1", doc, sum, 1, 2, p1)
	if err != nil {
		t.Fatal(err)
	}
	m := got.(map[string]any)
	if m["complete"] != true || m["object"] != "uploads/up-1.pdf" || m["sha256"] != sum || m["size"] != float64(len(doc)) {
		t.Errorf("after the last part: %v", m)
	}
	if kept, err := r.objects.Get(t.Context(), "uploads/up-1.pdf"); err != nil || !bytes.Equal(kept, doc) {
		t.Errorf("the stored document is not the upload (%v)", err)
	}
	// A JPEG is stored with its own extension.
	jpg := append([]byte{0xFF, 0xD8, 0xFF, 0xE0}, bytes.Repeat([]byte{7}, 100)...)
	if _, err := upload(t, r, "pic", jpg, sha256Hex(jpg), 1, 1, jpg); err != nil {
		t.Fatal(err)
	}
	if _, err := r.objects.Get(t.Context(), "uploads/pic.jpg"); err != nil {
		t.Errorf("the JPEG is not stored as uploads/pic.jpg: %v", err)
	}
}

// 1453/AC11: a wrong SHA-256 on the joined document stores nothing and is said; a document that is neither PDF nor
// JPEG stores nothing; a part out of range, a part larger than the limit, a part that is not base64 and parts that
// disagree are refused; the upload can then be sent again whole.
func TestAnUploadWithAWrongSha256OrAnOutOfRangePartStoresNothing(t *testing.T) {
	r := newRig(t, `{}`, "pw")
	doc := pdfOf(300)
	wrong := sha256Hex([]byte("not it"))
	if _, err := upload(t, r, "bad", doc, wrong, 1, 1, doc); err == nil || !strings.Contains(err.Error(), "refused") || !strings.Contains(err.Error(), sha256Hex(doc)) {
		t.Errorf("a wrong SHA-256: error = %v", err)
	}
	if len(r.objects.keys()) != 0 || r.objects.puts != 0 {
		t.Errorf("a document with a wrong SHA-256 was stored: %v", r.objects.keys())
	}
	text := []byte("plain text, no PDF")
	if _, err := upload(t, r, "txt", text, sha256Hex(text), 1, 1, text); err == nil || !strings.Contains(err.Error(), "neither a PDF nor a JPEG") {
		t.Errorf("a text upload: error = %v", err)
	}
	for _, c := range []struct{ part, parts int }{{0, 1}, {2, 1}, {1, 0}, {1, MostParts + 1}} {
		if _, err := upload(t, r, "range", doc, sha256Hex(doc), c.part, c.parts, doc); err == nil {
			t.Errorf("part %d of %d was taken", c.part, c.parts)
		}
	}
	if _, err := upload(t, r, "range", doc, sha256Hex(doc), 2, 1, doc); err == nil || !strings.Contains(err.Error(), "out of range") {
		t.Errorf("part 2 of 1: error = %v", err)
	}
	if _, err := r.call(t, "printer.upload", map[string]any{"upload": "x", "part": float64(1), "parts": float64(1), "sha256": sha256Hex(doc), "data": "!!"}); err == nil {
		t.Errorf("a part that is not base64 was taken")
	}
	if _, err := upload(t, r, "big", doc, sha256Hex(doc), 1, 1, make([]byte, MaxPart+1)); err == nil || !strings.Contains(err.Error(), "at most") {
		t.Errorf("a part over the limit: error = %v", err)
	}
	if r.bus.asks() != 2 {
		// Only the two uploads that were complete in form reached the device process; every refusal above them
		// was the tool's.
		t.Errorf("%d asks queued, want the 2 whose form was right", r.bus.asks())
	}
	// Parts that disagree about the whole.
	p := pdfOf(PartSize + 10)
	p1, _, _ := cutPartOf(p, 1, PartSize)
	if _, err := upload(t, r, "mix", p, sha256Hex(p), 1, 2, p1); err != nil {
		t.Fatal(err)
	}
	if _, err := upload(t, r, "mix", p, sha256Hex(p), 2, 3, p1); err == nil || !strings.Contains(err.Error(), "all say the same") {
		t.Errorf("parts that disagree: error = %v", err)
	}
	if len(r.objects.keys()) != 0 {
		t.Errorf("objects stored: %v", r.objects.keys())
	}
}

// 1453/AC11: an incomplete upload's parts are dropped after fifteen minutes.
func TestAnIncompleteUploadIsDroppedAfterAWhile(t *testing.T) {
	r := newRig(t, `{}`, "pw")
	doc := pdfOf(PartSize + 10)
	p1, _, _ := cutPartOf(doc, 1, PartSize)
	p2, _, _ := cutPartOf(doc, 2, PartSize)
	now := fixedNow
	r.worker.Now = func() time.Time { return now }
	if _, err := upload(t, r, "late", doc, sha256Hex(doc), 1, 2, p1); err != nil {
		t.Fatal(err)
	}
	now = now.Add(pendingFor + time.Minute)
	// Asked of the worker directly: the tool's own clock is the fixed one, and an ask this old would be stale.
	got, err := r.worker.upload(t.Context(), Ask{Op: OpUpload, Upload: "late", Part: 2, Parts: 2, SHA256: sha256Hex(doc),
		Data: base64.StdEncoding.EncodeToString(p2)})
	if err != nil {
		t.Fatal(err)
	}
	if got["complete"] != false {
		t.Errorf("the dropped part was still there: %v", got)
	}
}

// ippFake is a printer's IPP port: it records the request and answers a Print-Job answer with a job id and state.
type ippFake struct {
	url    string
	body   []byte
	status int
}

func newIPPFake(t *testing.T, jobID, jobState int) *ippFake {
	t.Helper()
	f := &ippFake{}
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		f.body, _ = io.ReadAll(r.Body)
		ans := []byte{0x02, 0x00, byte(f.status >> 8), byte(f.status), 0, 0, 0, 1, ippOperation}
		ans = appendAttr(ans, ippCharset, "attributes-charset", "utf-8")
		ans = append(ans, ippJob)
		ans = appendAttr(ans, ippInteger, "job-id", string(binary.BigEndian.AppendUint32(nil, uint32(jobID))))
		ans = appendAttr(ans, ippEnum, "job-state", string(binary.BigEndian.AppendUint32(nil, uint32(jobState))))
		ans = appendAttr(ans, ippKeyword, "job-state-reasons", "none")
		ans = append(ans, ippEnd)
		w.Header().Set("Content-Type", "application/ipp")
		w.Write(ans)
	}))
	t.Cleanup(srv.Close)
	f.url = srv.URL
	old := ippURL
	ippURL = func(string) string { return srv.URL }
	t.Cleanup(func() { ippURL = old })
	return f
}

// request splits what the fake was sent into its attributes and its document.
func (f *ippFake) request(t *testing.T) (IPPAnswer, []byte) {
	t.Helper()
	a, off, err := decodeIPP(f.body)
	if err != nil {
		t.Fatal(err)
	}
	return a, f.body[off:]
}

// 1453/AC12: printer.print sends a kept scan to the printer as an IPP Print-Job (operation 0x0002): the document
// bytes after the attributes, document-format by sniffing, copies, sides and media as asked, to the printer's own
// IPP address; the answer is the job's id and state.
func TestPrinterPrintSendsAPrintJobAndAnswersTheJob(t *testing.T) {
	fake := newIPPFake(t, 77, 5)
	r := newRig(t, `{"printer":"192.0.2.10"}`, "pw")
	doc := pdfOf(5000)
	key := sha256Hex(doc) + ".pdf"
	r.objects.Put(t.Context(), key, "application/pdf", doc)
	got, err := r.call(t, "printer.print", map[string]any{"object": key, "copies": float64(2), "sides": "two-sided-long-edge", "media": "na_letter_8.5x11in"})
	if err != nil {
		t.Fatal(err)
	}
	m := got.(map[string]any)
	if m["job-id"] != float64(77) || m["job-state"] != "processing" || m["format"] != "application/pdf" {
		t.Errorf("answer = %v", m)
	}
	if binary.BigEndian.Uint16(fake.body[2:4]) != 0x0002 {
		t.Errorf("operation = 0x%04x, want Print-Job 0x0002", binary.BigEndian.Uint16(fake.body[2:4]))
	}
	attrs, sent := fake.request(t)
	if !bytes.Equal(sent, doc) {
		t.Errorf("the document sent is not the kept one (%d bytes, want %d)", len(sent), len(doc))
	}
	text := func(n string) string { return strings.Join(attrs.texts(n), ",") }
	if text("document-format") != "application/pdf" || text("sides") != "two-sided-long-edge" || text("media") != "na_letter_8.5x11in" ||
		text("printer-uri") != "ipp://192.0.2.10/ipp/port1" {
		t.Errorf("attributes = %v", attrs.Attributes)
	}
	if c := attrs.Attributes["copies"]; len(c) != 1 || c[0].Integer != 2 {
		t.Errorf("copies = %v", c)
	}
}

// 1453/AC12: the defaults are one copy, one-sided, A4; a JPEG is sent as image/jpeg; an upload is printed by its id.
func TestPrinterPrintDefaultsAndFormatsAndUploads(t *testing.T) {
	fake := newIPPFake(t, 8, 3)
	r := newRig(t, `{"printer":"192.0.2.10"}`, "pw")
	jpg := append([]byte{0xFF, 0xD8, 0xFF, 0xE0}, bytes.Repeat([]byte{9}, 200)...)
	if _, err := upload(t, r, "photo", jpg, sha256Hex(jpg), 1, 1, jpg); err != nil {
		t.Fatal(err)
	}
	got, err := r.call(t, "printer.print", map[string]any{"upload": "photo"})
	if err != nil {
		t.Fatal(err)
	}
	if got.(map[string]any)["job-state"] != "pending" {
		t.Errorf("answer = %v", got)
	}
	attrs, sent := fake.request(t)
	if !bytes.Equal(sent, jpg) || strings.Join(attrs.texts("document-format"), "") != "image/jpeg" ||
		strings.Join(attrs.texts("sides"), "") != "one-sided" || strings.Join(attrs.texts("media"), "") != DefaultMedia || attrs.Attributes["copies"][0].Integer != 1 {
		t.Errorf("attributes = %v, %d bytes sent", attrs.Attributes, len(sent))
	}
	if _, err := r.call(t, "printer.print", map[string]any{"upload": "never-uploaded"}); err == nil || !strings.Contains(err.Error(), "not complete") {
		t.Errorf("an upload never made: error = %v", err)
	}
}

// 1453/AC12: a print that is not well formed is refused before anything is queued; the printer's refusal is an
// error naming its IPP status; a printer not set is said by name.
func TestPrinterPrintRefusesWhatIsNotWellFormed(t *testing.T) {
	fake := newIPPFake(t, 1, 3)
	r := newRig(t, `{"printer":"192.0.2.10"}`, "pw")
	key := sha256Hex([]byte("x")) + ".pdf"
	for name, args := range map[string]map[string]any{
		"neither":      {},
		"both":         {"object": key, "upload": "u"},
		"copies 0":     {"object": key, "copies": float64(-1)},
		"copies 100":   {"object": key, "copies": float64(100)},
		"sides":        {"object": key, "sides": "sideways"},
		"media":        {"object": key, "media": "A4 paper"},
		"not a key":    {"object": "../x.pdf"},
		"copies float": {"object": key, "copies": 1.5},
	} {
		if _, err := r.call(t, "printer.print", args); err == nil {
			t.Errorf("%s: accepted", name)
		}
	}
	if r.bus.asks() != 0 || fake.body != nil {
		t.Errorf("%d asks queued, the printer was sent %d bytes", r.bus.asks(), len(fake.body))
	}
	doc := pdfOf(100)
	dk := sha256Hex(doc) + ".pdf"
	r.objects.Put(t.Context(), dk, "application/pdf", doc)
	fake.status = 0x0400
	if _, err := r.call(t, "printer.print", map[string]any{"object": dk}); err == nil || !strings.Contains(err.Error(), "0x0400") {
		t.Errorf("a refused job: error = %v", err)
	}
	// A kept object that is not a document the printer takes is not sent.
	fake.status, fake.body = 0, nil
	r.objects.Put(t.Context(), key, "text/plain", []byte("hello"))
	if _, err := r.call(t, "printer.print", map[string]any{"object": key}); err == nil || !strings.Contains(err.Error(), "neither a PDF nor a JPEG") || fake.body != nil {
		t.Errorf("a text object: error = %v, sent %d bytes", err, len(fake.body))
	}
	un := newRig(t, `{}`, "pw")
	if _, err := un.call(t, "printer.print", map[string]any{"object": key}); err == nil || !strings.Contains(err.Error(), "printer is not set") || un.bus.asks() != 0 {
		t.Errorf("no printer: error = %v", err)
	}
}

// The Print-Job request is the one RFC 8011 gives: header, operation attributes in order, the job's own, the
// end-of-attributes tag, then the document.
func TestThePrintJobRequestIsWellFormed(t *testing.T) {
	doc := []byte("%PDF-1.4 hello")
	b := EncodePrintJob("ipp://192.0.2.10/ipp/port1", PrintOptions{Format: "application/pdf", Copies: 3, Sides: "one-sided", Media: DefaultMedia, Name: "n"}, doc)
	if !bytes.Equal(b[:8], []byte{2, 0, 0, 2, 0, 0, 0, 1}) || b[8] != ippOperation {
		t.Errorf("header = %x", b[:9])
	}
	a, off, err := decodeIPP(b)
	if err != nil || !bytes.Equal(b[off:], doc) {
		t.Fatalf("decode: %v, document %q", err, b[off:])
	}
	if first := string(b[9:]); !strings.HasPrefix(first[3:], "attributes-charset") {
		t.Errorf("the first attribute is not attributes-charset")
	}
	if len(a.Attributes["copies"]) != 1 || a.Attributes["copies"][0].Integer != 3 {
		t.Errorf("copies = %v", a.Attributes["copies"])
	}
}

// 1453: a scan and a print are acknowledged before the work (a slow one is not delivered again and done twice);
// the other operations are acknowledged after it.
func TestAScanAndAPrintAreAcknowledgedBeforeTheWork(t *testing.T) {
	f := newFakeScanner(t, fixture(t, "scan.jpg"))
	r, _ := scanRig(t, f, Settings{Crop: true})
	body, _ := json.Marshal(Ask{ID: "s1", Op: OpScan, At: fixedNow})
	calls := 0
	var postsAtAck int
	act, _ := r.worker.Handle(body, func() { calls++; postsAtAck = f.posts })
	if calls != 1 || postsAtAck != 0 || act != actionSettled || f.posts != 1 {
		t.Errorf("early called %d times with %d scans begun; action %v; %d scans in all", calls, postsAtAck, act, f.posts)
	}
	body, _ = json.Marshal(Ask{ID: "s2", Op: OpSettings, Page: DefaultPage, At: fixedNow})
	calls = 0
	if act, _ := r.worker.Handle(body, func() { calls++ }); calls != 0 || act != actionAck {
		t.Errorf("settings: early called %d times, action %v", calls, act)
	}
}

// 1453/AC11: a part of 700 KiB is taken and one of a byte more is refused, by the module's own message naming
// 700 KiB, before it is queued; the limit and the size the module cuts are one.
func TestAPartOverSevenHundredKiBIsRefusedByTheModulesOwnMessage(t *testing.T) {
	if MaxPart != PartSize || PartSize != 700<<10 {
		t.Fatalf("MaxPart %d, PartSize %d: both are 700 KiB", MaxPart, PartSize)
	}
	r := newRig(t, `{}`, "pw")
	big := make([]byte, 700<<10+1)
	_, err := upload(t, r, "over", big, sha256Hex(big), 1, 1, big)
	if err == nil || !strings.Contains(err.Error(), "700 KiB") {
		t.Fatalf("a part of 700 KiB + 1 byte: error = %v, want the module's own message naming 700 KiB", err)
	}
	if r.bus.asks() != 0 {
		t.Errorf("an oversized part was queued")
	}
	ok := pdfOf(700 << 10)
	if _, err := upload(t, r, "exact", ok, sha256Hex(ok), 1, 1, ok); err != nil {
		t.Errorf("a part of exactly 700 KiB: %v", err)
	}
}

// 1535: a scan larger than the state's value limit (256 KiB) is filed whole, its SHA-256 equal, and the answer to
// the scan itself is small: it carries the event's body and never the document. Every part answer fits the state.
func TestAScanLargerThanTheStatesValueLimitIsFiledWholeAndItsAnswerIsSmall(t *testing.T) {
	f := newFakeScanner(t, fixture(t, "scan.jpg"))
	r, folder := scanRig(t, f, Settings{Crop: false})
	// The real capture is small; the worker keeps a stand-in of 3 parts' size under the key the scan names.
	big := pdfOf(222388 + 2*AnswerPartSize)
	if len(big) <= StateMaxValue {
		t.Fatal("the stand-in is not larger than the state's value limit")
	}
	r.worker.Capture = func(ctx context.Context, req ScanRequest) (Capture, error) {
		return Capture{PDF: big, SHA256: sha256Hex(big), Resolution: 300, At: fixedNow}, nil
	}
	body, err := r.scan(t)
	if err != nil {
		t.Fatal(err)
	}
	path, _ := body["path"].(string)
	got, err := os.ReadFile(path)
	if err != nil || sha256Hex(got) != sha256Hex(big) || filepath.Dir(path) != folder {
		t.Fatalf("filed %s: %v, SHA-256 equal %v", path, err, err == nil && sha256Hex(got) == sha256Hex(big))
	}
	for id, raw := range r.answers.m {
		if len(raw) > StateMaxValue {
			t.Errorf("answer %s is %d bytes, over the state's %d", id, len(raw), StateMaxValue)
		}
	}
	for _, a := range r.bus.published {
		if a.Op == OpScan {
			if raw := r.answers.get(a.ID); len(raw) > 4<<10 || bytes.Contains(raw, []byte(`"data"`)) {
				t.Errorf("the scan's answer is %d bytes and may carry the document: %s", len(raw), raw)
			}
		}
	}
}

// 1535: an answer the state refuses ends the ask: a small error answer says why, and the queued message is
// terminated, never redelivered into the same failure; a scan's message, acknowledged early, stays settled.
func TestARefusedAnswerEndsTheAskWithAnErrorAndNoRedelivery(t *testing.T) {
	r := newRig(t, `{}`, "pw")
	big := pdfOf(300 << 10)
	key := sha256Hex(big) + ".pdf"
	r.objects.Put(t.Context(), key, "application/pdf", big)
	// A part answer of 700 KiB, as the old cut made it, is over the state's limit; the worker must not loop on it.
	r.worker.answerPart = PartSize
	ask, _ := json.Marshal(Ask{ID: "big1", Op: OpDocument, Object: key, Part: 1, At: fixedNow})
	act, _ := r.worker.Handle(ask, nil)
	if act != actionTerm {
		t.Errorf("action = %v, want term", act)
	}
	raw := r.answers.get("big1")
	if raw == nil || !bytes.Contains(raw, []byte(`"error"`)) || len(raw) > 4<<10 {
		t.Errorf("no small error answer: %s", raw)
	}
}

// 1535: only a refused value (error code 10054, message too large) ends the ask; a store that is away, a timeout
// or a cut connection is redelivered, a document fetch being safe to repeat.
func TestOnlyATooLargeAnswerEndsTheAskATransientFailureIsRetried(t *testing.T) {
	r := newRig(t, `{}`, "pw")
	big := pdfOf(300 << 10)
	key := sha256Hex(big) + ".pdf"
	r.objects.Put(t.Context(), key, "application/pdf", big)
	ask, _ := json.Marshal(Ask{ID: "t1", Op: OpDocument, Object: key, Part: 1, At: fixedNow})
	r.answers.fail = fmt.Errorf("nats: timeout")
	if act, d := r.worker.Handle(ask, nil); act != actionRetry || d <= 0 {
		t.Errorf("a timeout: action %v delay %v, want retry", act, d)
	}
	r.answers.fail = nil
	r.worker.answerPart = PartSize
	ask2, _ := json.Marshal(Ask{ID: "t2", Op: OpDocument, Object: key, Part: 1, At: fixedNow})
	if act, _ := r.worker.Handle(ask2, nil); act != actionTerm {
		t.Errorf("a too large answer: action %v, want term", act)
	}
}

// 1551: printer.print with a path reads the file, uploads it in parts and prints it in one call; a document of
// several parts arrives whole, and the answer has the job's id and state.
func TestPrinterPrintTakesAPathAndUploadsAndPrintsInOneCall(t *testing.T) {
	fake := newIPPFake(t, 31, 5)
	r := newRig(t, `{"printer":"192.0.2.10"}`, "pw")
	doc := pdfOf(2*PartSize + 1234)
	file := filepath.Join(t.TempDir(), "letter.bin") // the name says nothing: the bytes are read
	if err := os.WriteFile(file, doc, 0o600); err != nil {
		t.Fatal(err)
	}
	got, err := r.call(t, "printer.print", map[string]any{"path": file, "copies": float64(2)})
	if err != nil {
		t.Fatal(err)
	}
	m := got.(map[string]any)
	if m["job-id"] != float64(31) || m["job-state"] != "processing" || m["format"] != "application/pdf" || m["bytes"] != len(doc) {
		t.Errorf("answer = %v", m)
	}
	attrs, sent := fake.request(t)
	if !bytes.Equal(sent, doc) || attrs.Attributes["copies"][0].Integer != 2 {
		t.Errorf("sent %d bytes (want %d), attributes %v", len(sent), len(doc), attrs.Attributes)
	}
	ops := r.bus.ops()
	if len(ops) != 4 || ops[0] != OpUpload || ops[2] != OpUpload || ops[3] != OpPrint {
		t.Errorf("asks = %v, want three upload parts and a print", ops)
	}
}

// 1551: a file that is neither a PDF nor a JPEG by its bytes is refused whatever its name, and a path that is
// relative, missing, a folder, or named together with an object is refused; nothing is queued.
func TestPrinterPrintRefusesAPathThatIsNotAPDFOrJPEG(t *testing.T) {
	fake := newIPPFake(t, 1, 5)
	r := newRig(t, `{"printer":"192.0.2.10"}`, "pw")
	dir := t.TempDir()
	text := filepath.Join(dir, "fake.pdf")
	os.WriteFile(text, []byte("just text, named .pdf"), 0o600)
	for name, args := range map[string]map[string]any{
		"text named pdf": {"path": text},
		"relative":       {"path": "fake.pdf"},
		"missing":        {"path": filepath.Join(dir, "none.pdf")},
		"folder":         {"path": dir},
		"with object":    {"path": text, "object": sha256Hex([]byte("x")) + ".pdf"},
	} {
		if _, err := r.call(t, "printer.print", args); err == nil {
			t.Errorf("%s: printed", name)
		}
	}
	if _, err := r.call(t, "printer.print", map[string]any{"path": text}); err == nil || !strings.Contains(err.Error(), "neither a PDF nor a JPEG") {
		t.Errorf("error = %v", err)
	}
	if r.bus.asks() != 0 || fake.body != nil {
		t.Errorf("%d asks queued, printer sent %d bytes", r.bus.asks(), len(fake.body))
	}
}