"""cobol-string-unstring.py - the expected output of cobol-string-unstring.cob What it does: copies the GnuCOBOL 3.2 runtime routines the COBOL program relies on (STRING and UNSTRING from libcob/strings.c, the LINE SEQUENTIAL read from libcob/fileio.c), checks those copies against the cases in the GnuCOBOL 3.2 test suite (tests/testsuite.src/run_misc.at), and then walks the CSV file the same way the COBOL program does, printing what the program should display. Why it exists: the COBOL listing was desk-checked, not compiled. This file is how the output shown in the lesson was worked out. When the program is compiled and run, its output can be compared with this one line by line. Inputs : datasets/cobol-string-unstring.csv (or a path given as argument 1) Run : python code/cobol-string-unstring.py (from the site folder) python code/cobol-string-unstring.py --selftest (the suite checks only) Needs : Python 3.8 or later, standard library only """ import sys from pathlib import Path RECORD_MAX = 120 # FD Csv-File: 01 Csv-Line PIC X(120) # ---------------------------------------------------------------------------- # Ports of libcob/strings.c (GnuCOBOL 3.2). Fields are Python bytearrays that # are changed in place, as the C code changes cob_field data. # ---------------------------------------------------------------------------- def move_alnum(dst, src): """cob_str_memcpy for an alphanumeric receiver: left-justify, pad with spaces or truncate on the right, as MOVE does.""" n = len(dst) data = bytes(src[:n]) + b" " * max(0, n - len(src)) dst[:] = data class StringStmt: """STRING ... INTO dst [WITH POINTER ptr]: cob_string_init/_append/_finish.""" def __init__(self, dst, ptr=None): self.dst = dst self.ptr = ptr # a one-element list, the POINTER item self.offset = 0 self.overflow = False if ptr is not None: self.offset = ptr[0] - 1 if self.offset < 0 or self.offset >= len(dst): self.overflow = True # COB_EC_OVERFLOW_STRING, nothing moves def append(self, src, delim=None): """One source item; delim None = DELIMITED BY SIZE.""" if self.overflow: return src = bytes(src) size = len(src) if not size: return if delim is not None: for i in range(size - len(delim) + 1): if src[i:i + len(delim)] == delim: size = i break room = len(self.dst) - self.offset if size <= room: self.dst[self.offset:self.offset + size] = src[:size] self.offset += size else: self.dst[self.offset:self.offset + room] = src[:room] self.offset += room self.overflow = True def finish(self): if self.ptr is not None: self.ptr[0] = self.offset + 1 return self.overflow def unstring(src, delims, intos, ptr=None, tally=None): """UNSTRING src DELIMITED BY delims INTO intos [POINTER ptr] [TALLYING tally]. delims: list of (bytes, all_flag). intos: list of dicts with key 'dst' (bytearray) and optional 'dlm' (bytearray) and 'cnt' (one-element list). ptr and tally are one-element lists. Returns True on an overflow condition. Follows cob_unstring_init/_into/_tallying/_finish line by line.""" src = bytes(src) srsize = len(src) offset = 0 count = 0 exception = False if ptr is not None: offset = ptr[0] - 1 if offset < 0 or offset >= srsize: exception = True for target in intos: if exception: break if offset >= srsize: continue # nothing left: field untouched start = offset dlm_data = None match_size = 0 if not delims: match_size = min(len(target["dst"]), srsize - offset) move_alnum(target["dst"], src[start:start + match_size]) offset += match_size else: found = False if len(delims) == 1: dp, all_flag = delims[0] dl = len(dp) s = srsize - dl + 1 p = start while p < s: if src[p:p + dl] == dp: match_size = p - start move_alnum(target["dst"], src[start:start + match_size]) offset += match_size + dl dlm_data = dp if all_flag: p += dl while p < s: if src[p:p + dl] != dp: break offset += dl p += dl found = True break p += 1 else: p = start while p < srsize and not found: for dp, all_flag in delims: dl = len(dp) s2 = srsize - dl + 1 if p > s2: continue if src[p:p + dl] == dp: match_size = p - start move_alnum(target["dst"], src[start:start + match_size]) offset += match_size + dl dlm_data = dp if all_flag: q = p + dl while q < s2: if src[q:q + dl] != dp: break offset += dl q += dl found = True break p += 1 if not found: match_size = srsize - offset move_alnum(target["dst"], src[start:start + match_size]) offset = srsize count += 1 if target.get("dlm") is not None: if dlm_data is not None: move_alnum(target["dlm"], dlm_data) else: target["dlm"][:] = b" " * len(target["dlm"]) if target.get("cnt") is not None: target["cnt"][0] = match_size if tally is not None: tally[0] += count if offset < srsize: exception = True if ptr is not None: ptr[0] = offset + 1 return exception # ---------------------------------------------------------------------------- # libcob/fileio.c lineseq_read: strip the LF (and a CR before it), pad the # record area with spaces to RECORD_MAX. All lines here are shorter than 120 # bytes, so the split/truncate statuses (06/04) cannot arise. # ---------------------------------------------------------------------------- def read_lines(path): raw = Path(path).read_bytes() lines = raw.split(b"\n") if lines and lines[-1] == b"": lines.pop() for line in lines: if line.endswith(b"\r"): line = line[:-1] assert len(line) < RECORD_MAX, "line longer than the record" assert all(32 <= c < 127 for c in line), "control character in line" yield bytearray(line + b" " * (RECORD_MAX - len(line))) # ---------------------------------------------------------------------------- # The COBOL program, paragraph by paragraph. # ---------------------------------------------------------------------------- HEADER = b"cust_id,last_name,first_name,city,tags" def zz9(n): return str(n).rjust(3) def run(path): out = [] line_no = blank = accepted = rejected = 0 for csv_line in read_lines(path): line_no += 1 ed = zz9(line_no) if line_no == 1: if bytes(csv_line).rstrip(b" ") != HEADER: out.append("LINE 1 IS NOT THE EXPECTED HEADER; STOPPING") return out, 16 out.append("LINE 1 HEADER") continue # Find-Length trail = len(csv_line) - len(bytes(csv_line).rstrip(b" ")) line_len = RECORD_MAX - trail if line_len == 0: blank += 1 out.append(f"LINE {ed} BLANK, SKIPPED") continue source = bytes(csv_line[:line_len]) # Split-Line cust = {"Id": bytearray(b" " * 6), "Last": bytearray(b" " * 15), "First": bytearray(b" " * 12), "City": bytearray(b" " * 15), "Tags": bytearray(b" " * 40)} lens = {k: [0] for k in cust} fields = [0] tags_delim = bytearray(b" ") extra = trailing = False quotes = source.count(b'"') if quotes == 0: intos = [{"dst": cust[k], "cnt": lens[k]} for k in ("Id", "Last", "First", "City")] intos.append({"dst": cust["Tags"], "dlm": tags_delim, "cnt": lens["Tags"]}) extra = unstring(source, [(b",", False)], intos, tally=fields) if source[line_len - 1:line_len] == b"," and fields[0] < 5: fields[0] += 1 trailing = True # Check-Fields L = {k: v[0] for k, v in lens.items()} cid = bytes(cust["Id"]) if quotes > 0: reason = "QUOTED FIELD; THIS PROGRAM DOES NOT PARSE QUOTES" elif extra: reason = ("MORE THAN 5 FIELDS (THE TAGS FIELD ENDED AT '" + tags_delim.decode() + "')") elif fields[0] < 5: reason = f"ONLY {fields[0] % 10} FIELDS, EXPECTED 5" elif L["Id"] != 6 or not cid.isdigit(): reason = "CUSTOMER ID IS NOT SIX DIGITS" elif L["Last"] == 0: reason = "LAST NAME IS MISSING" elif L["Last"] > 15: reason = f"LAST NAME HAS {L['Last'] % 100:02d} CHARACTERS, THE FIELD HOLDS 15" elif L["First"] == 0: reason = "FIRST NAME IS MISSING" elif L["First"] > 12: reason = f"FIRST NAME HAS {L['First'] % 100:02d} CHARACTERS, THE FIELD HOLDS 12" elif L["City"] == 0: reason = "CITY IS MISSING" elif L["City"] > 15: reason = f"CITY HAS {L['City'] % 100:02d} CHARACTERS, THE FIELD HOLDS 15" elif L["Tags"] > 40: reason = f"TAGS HAVE {L['Tags'] % 100:02d} CHARACTERS, THE FIELD HOLDS 40" else: reason = "" assert len(reason) <= 60, "WS-Reason is X(60)" if reason: rejected += 1 out.append(f"LINE {ed} REJECT {reason}") continue # Build-Label label = bytearray(b" " * 32) lptr = [1] st = StringStmt(label, lptr) for piece in (cust["Last"][:L["Last"]], b", ", cust["First"][:L["First"]], b" (", cust["City"][:L["City"]], b")"): st.append(piece) label_cut = st.finish() proper = bytearray(b" " * 30) st = StringStmt(proper) for piece in (cust["Last"][:L["Last"]], b", ", cust["First"][:L["First"]]): st.append(piece) st.finish() naive = bytearray(b" " * 30) st = StringStmt(naive) st.append(cust["Last"], b" ") st.append(b", ") st.append(cust["First"], b" ") st.finish() # Split-Tags tag_count = empty_tags = 0 tag_list = bytearray(b" " * 40) list_ptr = [1] if L["Tags"] > 0: tsrc = bytes(cust["Tags"][:L["Tags"]]) tag_ptr = [1] guard = 0 while not tag_ptr[0] > L["Tags"]: guard += 1 assert guard < 100, "tag loop does not advance" one = bytearray(b" " * 40) one_len = [0] before = tag_ptr[0] unstring(tsrc, [(b"|", False)], [{"dst": one, "cnt": one_len}], ptr=tag_ptr) assert tag_ptr[0] > before if one_len[0] == 0: empty_tags += 1 else: tag_count += 1 st = StringStmt(tag_list, list_ptr) st.append(b"[") st.append(one[:one_len[0]]) st.append(b"]") assert not st.finish(), "tag list overflow (not handled)" if tag_count == 0: move_alnum(tag_list, b"(none)") # Show-Accepted accepted += 1 out.append(f"LINE {ed} OK {label.decode()} TAGS={tag_count % 10} " + tag_list.decode().rstrip(" ")) pad = " " * 13 if label_cut: out.append(f"{pad}NOTE: LABEL CUT AT 32 CHARACTERS, POINTER = " + str(lptr[0] % 100).rjust(2)) if naive != proper: out.append(f"{pad}NOTE: DELIMITED BY SPACE WOULD GIVE [" + naive.decode().rstrip(" ") + "]") if trailing: out.append(f"{pad}NOTE: TRAILING COMMA, TAGS FIELD IS EMPTY") if empty_tags > 0: out.append(f"{pad}NOTE: {empty_tags % 10} EMPTY TAG(S) SKIPPED") # Show-Totals out.append("----") out.append("LINES READ " + zz9(line_no)) out.append("ACCEPTED " + zz9(accepted)) out.append("REJECTED " + zz9(rejected)) out.append("BLANK " + zz9(blank)) return out, (4 if rejected else 0) # ---------------------------------------------------------------------------- # Self-test: the GnuCOBOL 3.2 test-suite cases for STRING and UNSTRING # (tests/testsuite.src/run_misc.at), run through the ports above. # ---------------------------------------------------------------------------- def selftest(): checks = [] def ok(name, cond): checks.append((name, bool(cond))) # "STRING WITH POINTER ON OVERFLOW with DELIMITER", cases A-D t = bytearray(b" "); p = [1] st = StringStmt(t, p); [st.append(x) for x in (b"A", b"B", b"C")] ovf = st.finish() ok("STRING A target ABC", t == b"ABC"); ok("STRING A pointer 4", p[0] == 4) ok("STRING A no overflow", not ovf) t = bytearray(b" "); p = [0] st = StringStmt(t, p); [st.append(x) for x in (b"A", b"B", b"C")] ovf = st.finish() ok("STRING B overflow", ovf); ok("STRING B target spaces", t == b" ") ok("STRING B pointer 0", p[0] == 0) t = bytearray(b" "); p = [4] st = StringStmt(t, p); [st.append(x) for x in (b"A", b"B", b"C")] ovf = st.finish() ok("STRING C overflow", ovf); ok("STRING C target spaces", t == b" ") ok("STRING C pointer 4", p[0] == 4) t = bytearray(b" "); p = [1] st = StringStmt(t, p) for x in (b"1|2", b"A|B", b"C|D"): st.append(x, b"|") st.finish() ok("STRING D target 1AC", t == b"1AC") # "STRING / UNSTRING [NOT] ON OVERFLOW", cases 1-4 t = bytearray(b" " * 20); st = StringStmt(t); st.append(b"data") ok("ON OVERFLOW 1", not st.finish() and t.rstrip() == b"data") t = bytearray(b" " * 20); st = StringStmt(t); st.append(b"data is too big here...") ok("ON OVERFLOW 2", st.finish() and t == b"data is too big here") t = bytearray(b" " * 20) ok("ON OVERFLOW 3", not unstring(b"data", [], [{"dst": t}]) and t.rstrip() == b"data") t = bytearray(b" " * 20) ok("ON OVERFLOW 4", unstring(b"data is too big here...", [], [{"dst": t}]) and t == b"data is too big here") # "UNSTRING DELIMITED ALL SPACE-2" rec = b" " * 4 + b"ABC AND DE" + b" " * 7 + b"FG AND HIJ" + b" " * 8 d = bytearray(b" " * 15); p = [1] unstring(rec, [(b" ", True)], [{"dst": d}], ptr=p) ok("ALL SPACE-2 #1 spaces", d == b" " * 15); ok("ALL SPACE-2 #1 ptr 5", p[0] == 5) unstring(rec, [(b" ", True)], [{"dst": d}], ptr=p) ok("ALL SPACE-2 #2 text", d.rstrip() == b"ABC AND DE"); ok("ALL SPACE-2 #2 ptr 21", p[0] == 21) unstring(rec, [(b" ", True)], [{"dst": d}], ptr=p) ok("ALL SPACE-2 #3 text", d.rstrip() == b" FG AND HIJ"); ok("ALL SPACE-2 #3 ptr 40", p[0] == 40) # "UNSTRING DELIMITED POINTER" lay = b" 10 AF-RECORD-TYPE-SEQUENCE-04 PIC 9(05) COMP-3." lay = lay + b" " * (66 - len(lay)) dl = [(b" PIC ", False), (b" COMP-3", False), (b".", False)] d = bytearray(b" " * 50); k = bytearray(b" " * 32); p = [1] unstring(lay, dl, [{"dst": d, "dlm": k}], ptr=p); ok("POINTER #1 48", p[0] == 48) p[0] += 7 unstring(lay, dl, [{"dst": d, "dlm": k}], ptr=p); ok("POINTER #2 62", p[0] == 62) unstring(lay, dl, [{"dst": d, "dlm": k}], ptr=p); ok("POINTER #3 63", p[0] == 63) # "UNSTRING DELIMITER IN" w = [bytearray(b" ") for _ in range(3)]; dm = [bytearray(b" ") for _ in range(2)] unstring(b"WWADDBCC", [(b"A", False), (b"B", False)], [{"dst": w[0], "dlm": dm[0]}, {"dst": w[1], "dlm": dm[1]}, {"dst": w[2]}]) ok("DELIMITER IN data", [bytes(x) for x in w] == [b"WW", b"DD", b"CC"]) ok("DELIMITER IN delims", [bytes(x) for x in dm] == [b"A", b"B"]) # "UNSTRING combined", cases A and B for case, dls, exp in ( ("A", [(b"|", False)], (b"ABC1", b"|", 4, b"DEF--GHI", b" ", 8, 14, 2, False)), ("B", [(b"|", False), (b"-", True)], (b"ABC1", b"|", 4, b"DEF", b"-", 3, 11, 2, True))): t1 = bytearray(b" " * 20); d1 = bytearray(b" "); c1 = [0] t2 = bytearray(b" " * 20); d2 = bytearray(b" "); c2 = [0] p = [1]; tal = [0] ovf = unstring(b"ABC1|DEF--GHI", dls, [{"dst": t1, "dlm": d1, "cnt": c1}, {"dst": t2, "dlm": d2, "cnt": c2}], ptr=p, tally=tal) got = (bytes(t1).rstrip(), bytes(d1), c1[0], bytes(t2).rstrip(), bytes(d2), c2[0], p[0], tal[0], ovf) ok(f"combined {case}", got == exp) # The trailing-delimiter rule this lesson depends on (strings.c returns # early from cob_unstring_into once the offset reaches the source size). a = bytearray(b" "); b = bytearray(b" "); c = bytearray(b"zz"); tal = [0] ovf = unstring(b"a,b,", [(b",", False)], [{"dst": a}, {"dst": b}, {"dst": c}], tally=tal) ok("trailing comma: third field untouched", c == b"zz") ok("trailing comma: tally 2", tal[0] == 2) ok("trailing comma: no overflow", not ovf) return checks if __name__ == "__main__": results = selftest() failed = [n for n, good in results if not good] if "--selftest" in sys.argv: for n, good in results: print(("ok " if good else "FAIL ") + n) if failed: print("SELF-TEST FAILED:", ", ".join(failed)) sys.exit(1) if "--selftest" in sys.argv: print(f"{len(results)} test-suite checks passed") sys.exit(0) args = [a for a in sys.argv[1:] if not a.startswith("--")] here = Path(__file__).resolve().parent.parent path = args[0] if args else here / "datasets" / "cobol-string-unstring.csv" lines, rc = run(path) print("\n".join(lines)) print(f"(return code {rc})")