Whole file
nedbat/cog
The author described this change as “feat: --check-fail-msg”. It counts as a record because the check below fails on the code as it stood at a53a8e4bf and passes on ee6d1ea00, with nothing else changed between the two runs.
Projectnedbat/cog
Fix saved2025-09-21
Sharing licenceMIT · LICENSE.txt
Change size+753 −743
What the code was meant to do, written into the code itself as a save note
feat: --check-fail-msg
The change
| 52 | 52 | -z The end-output marker can be omitted, and is assumed at eof. | |
| 53 | 53 | -v Print the version of cog and exit. | |
| 54 | 54 | --check Check that the files would not change if run again. | |
| 55 | - | --diff With --check, show a diff of what failed the check. | |
| 56 | - | --markers='START END END-OUTPUT' | |
| 57 | - | The patterns surrounding cog inline instructions. Should | |
| 58 | - | include three values separated by spaces, the start, end, | |
| 59 | - | and end-output markers. Defaults to '[[[cog ]]] [[[end]]]'. | |
| 60 | - | --verbosity=VERBOSITY | |
| 61 | - | Control the amount of output. 2 (the default) lists all files, | |
| 62 | - | 1 lists only changed files, 0 lists no files. | |
| 63 | - | -h, --help Print this help. | |
| 64 | - | """ | |
| 65 | - | ||
| 66 | - | ||
| 67 | - | class CogError(Exception): | |
| 68 | - | """Any exception raised by Cog.""" | |
| 69 | - | ||
| 70 | - | def __init__(self, msg, file="", line=0): | |
| 71 | - | if file: | |
| 72 | - | super().__init__(f"{file}({line}): {msg}") | |
| 73 | - | else: | |
| 74 | - | super().__init__(msg) | |
| 75 | - | ||
| 76 | - | ||
| 77 | - | class CogUsageError(CogError): | |
| 78 | - | """An error in usage of command-line arguments in cog.""" | |
| 79 | - | ||
| 80 | - | pass | |
| 81 | - | ||
| 82 | - | ||
| 83 | - | class CogInternalError(CogError): | |
| 84 | - | """An error in the coding of Cog. Should never happen.""" | |
| 85 | - | ||
| 86 | - | pass | |
| 87 | - | ||
| 88 | - | ||
| 89 | - | class CogGeneratedError(CogError): | |
| 90 | - | """An error raised by a user's Python code.""" | |
| 91 | - | ||
| 92 | - | pass | |
| 93 | - | ||
| 94 | - | ||
| 95 | - | class CogUserException(CogError): | |
| 96 | - | """An exception caught when running a user's Python code. | |
| 97 | - | ||
| 98 | - | The argument is the traceback message to print. | |
| 99 | - | ||
| 100 | - | """ | |
| 101 | - | ||
| 102 | - | pass | |
| 103 | - | ||
| 104 | - | ||
| 105 | - | class CogCheckFailed(CogError): | |
| 106 | - | """A --check failed.""" | |
| 107 | - | ||
| 108 | - | pass | |
| 109 | - | ||
| 110 | - | ||
| 111 | - | class CogGenerator(Redirectable): | |
| 112 | - | """A generator pulled from a source file.""" | |
| 113 | - | ||
| 114 | - | def __init__(self, options=None): | |
| 115 | - | super().__init__() | |
| 116 | - | self.markers = [] | |
| 117 | - | self.lines = [] | |
| 118 | - | self.options = options or CogOptions() | |
| 119 | - | ||
| 120 | - | def parse_marker(self, line): | |
| 121 | - | self.markers.append(line) | |
| 122 | - | ||
| 123 | - | def parse_line(self, line): | |
| 124 | - | self.lines.append(line.strip("\n")) | |
| 125 | - | ||
| 126 | - | def get_code(self): | |
| 127 | - | """Extract the executable Python code from the generator.""" | |
| 128 | - | # If the markers and lines all have the same prefix | |
| 129 | - | # (end-of-line comment chars, for example), | |
| 130 | - | # then remove it from all the lines. | |
| 131 | - | pref_in = common_prefix(self.markers + self.lines) | |
| 132 | - | if pref_in: | |
| 133 | - | self.markers = [line.replace(pref_in, "", 1) for line in self.markers] | |
| 134 | - | self.lines = [line.replace(pref_in, "", 1) for line in self.lines] | |
| 135 | - | ||
| 136 | - | return reindent_block(self.lines, "") | |
| 137 | - | ||
| 138 | - | def evaluate(self, cog, globals, fname): | |
| 139 | - | # figure out the right whitespace prefix for the output | |
| 140 | - | pref_out = white_prefix(self.markers) | |
| 141 | - | ||
| 142 | - | intext = self.get_code() | |
| 143 | - | if not intext: | |
| 144 | - | return "" | |
| 145 | - | ||
| 146 | - | prologue = "import " + cog.cogmodulename + " as cog\n" | |
| 147 | - | if self.options.prologue: | |
| 148 | - | prologue += self.options.prologue + "\n" | |
| 149 | - | code = compile(prologue + intext, str(fname), "exec") | |
| 150 | - | ||
| 151 | - | # Make sure the "cog" module has our state. | |
| 152 | - | cog.cogmodule.msg = self.msg | |
| 153 | - | cog.cogmodule.out = self.out | |
| 154 | - | cog.cogmodule.outl = self.outl | |
| 155 | - | cog.cogmodule.error = self.error | |
| 156 | - | ||
| 157 | - | real_stdout = sys.stdout | |
| 158 | - | if self.options.print_output: | |
| 159 | - | sys.stdout = captured_stdout = io.StringIO() | |
| 160 | - | ||
| 161 | - | self.outstring = "" | |
| 162 | - | try: | |
| 163 | - | eval(code, globals) | |
| 164 | - | except CogError: | |
| 165 | - | raise | |
| 166 | - | except: # noqa: E722 (we're just wrapping in CogUserException and rethrowing) | |
| 167 | - | typ, err, tb = sys.exc_info() | |
| 168 | - | frames = (tuple(fr) for fr in traceback.extract_tb(tb.tb_next)) | |
| 169 | - | frames = find_cog_source(frames, prologue) | |
| 170 | - | msg = "".join(traceback.format_list(frames)) | |
| 171 | - | msg += f"{typ.__name__}: {err}" | |
| 172 | - | raise CogUserException(msg) | |
| 173 | - | finally: | |
| 174 | - | sys.stdout = real_stdout | |
| 175 | - | ||
| 176 | - | if self.options.print_output: | |
| 177 | - | self.outstring = captured_stdout.getvalue() | |
| 178 | - | ||
| 179 | - | # We need to make sure that the last line in the output | |
| 180 | - | # ends with a newline, or it will be joined to the | |
| 181 | - | # end-output line, ruining cog's idempotency. | |
| 182 | - | if self.outstring and self.outstring[-1] != "\n": | |
| 183 | - | self.outstring += "\n" | |
| 184 | - | ||
| 185 | - | return reindent_block(self.outstring, pref_out) | |
| 186 | - | ||
| 187 | - | def msg(self, s): | |
| 188 | - | self.prout("Message: " + s) | |
| 189 | - | ||
| 190 | - | def out(self, sOut="", dedent=False, trimblanklines=False): | |
| 191 | - | """The cog.out function.""" | |
| 192 | - | if trimblanklines and ("\n" in sOut): | |
| 193 | - | lines = sOut.split("\n") | |
| 194 | - | if lines[0].strip() == "": | |
| 195 | - | del lines[0] | |
| 196 | - | if lines and lines[-1].strip() == "": | |
| 197 | - | del lines[-1] | |
| 198 | - | sOut = "\n".join(lines) + "\n" | |
| 199 | - | if dedent: | |
| 200 | - | sOut = reindent_block(sOut) | |
| 201 | - | self.outstring += sOut | |
| 202 | - | ||
| 203 | - | def outl(self, sOut="", **kw): | |
| 204 | - | """The cog.outl function.""" | |
| 205 | - | self.out(sOut, **kw) | |
| 206 | - | self.out("\n") | |
| 207 | - | ||
| 208 | - | def error(self, msg="Error raised by cog generator."): | |
| 209 | - | """The cog.error function. | |
| 210 | - | ||
| 211 | - | Instead of raising standard python errors, cog generators can use | |
| 212 | - | this function. It will display the error without a scary Python | |
| 213 | - | traceback. | |
| 214 | - | ||
| 215 | - | """ | |
| 216 | - | raise CogGeneratedError(msg) | |
| 217 | - | ||
| 218 | - | ||
| 219 | - | class CogOptions: | |
| 220 | - | """Options for a run of cog.""" | |
| 221 | - | ||
| 222 | - | def __init__(self): | |
| 223 | - | # Defaults for argument values. | |
| 224 | - | self.args = [] | |
| 225 | - | self.include_path = [] | |
| 226 | - | self.defines = {} | |
| 227 | - | self.show_version = False | |
| 228 | - | self.make_writable_cmd = None | |
| 229 | - | self.replace = False | |
| 230 | - | self.no_generate = False | |
| 231 | - | self.output_name = None | |
| 232 | - | self.warn_empty = False | |
| 233 | - | self.hash_output = False | |
| 234 | - | self.delete_code = False | |
| 235 | - | self.eof_can_be_end = False | |
| 236 | - | self.suffix = None | |
| 237 | - | self.newlines = False | |
| 238 | - | self.begin_spec = "[[[cog" | |
| 239 | - | self.end_spec = "]]]" | |
| 240 | - | self.end_output = "[[[end]]]" | |
| 241 | - | self.encoding = "utf-8" | |
| 242 | - | self.verbosity = 2 | |
| 243 | - | self.prologue = "" | |
| 244 | - | self.print_output = False | |
| 245 | - | self.check = False | |
| 246 | - | self.diff = False | |
| 247 | - | ||
| 248 | - | def __eq__(self, other): | |
| 249 | - | """Comparison operator for tests to use.""" | |
| 250 | - | return self.__dict__ == other.__dict__ | |
| 251 | - | ||
| 252 | - | def clone(self): | |
| 253 | - | """Make a clone of these options, for further refinement.""" | |
| 254 | - | return copy.deepcopy(self) | |
| 255 | - | ||
| 256 | - | def add_to_include_path(self, dirs): | |
| 257 | - | """Add directories to the include path.""" | |
| 258 | - | dirs = dirs.split(os.pathsep) | |
| 259 | - | self.include_path.extend(dirs) | |
| 260 | - | ||
| 261 | - | def parse_args(self, argv): | |
| 262 | - | # Parse the command line arguments. | |
| 263 | - | try: | |
| 264 | - | opts, self.args = getopt.getopt( | |
| 265 | - | argv, | |
| 266 | - | "cdD:eI:n:o:rs:p:PUvw:xz", | |
| 267 | - | [ | |
| 268 | - | "check", | |
| 269 | - | "diff", | |
| 270 | - | "markers=", | |
| 271 | - | "verbosity=", | |
| 272 | - | ], | |
| 273 | - | ) | |
| 274 | - | except getopt.error as msg: | |
| 275 | - | raise CogUsageError(msg) | |
| 276 | - | ||
| 277 | - | # Handle the command line arguments. | |
| 278 | - | for o, a in opts: | |
| 279 | - | if o == "-c": | |
| 280 | - | self.hash_output = True | |
| 281 | - | elif o == "-d": | |
| 282 | - | self.delete_code = True | |
| 283 | - | elif o == "-D": | |
| 284 | - | if a.count("=") < 1: | |
| 285 | - | raise CogUsageError("-D takes a name=value argument") | |
| 286 | - | name, value = a.split("=", 1) | |
| 287 | - | self.defines[name] = value | |
| 288 | - | elif o == "-e": | |
| 289 | - | self.warn_empty = True | |
| 290 | - | elif o == "-I": | |
| 291 | - | self.add_to_include_path(os.path.abspath(a)) | |
| 292 | - | elif o == "-n": | |
| 293 | - | self.encoding = a | |
| 294 | - | elif o == "-o": | |
| 295 | - | self.output_name = a | |
| 296 | - | elif o == "-r": | |
| 297 | - | self.replace = True | |
| 298 | - | elif o == "-s": | |
| 299 | - | self.suffix = a | |
| 300 | - | elif o == "-p": | |
| 301 | - | self.prologue = a | |
| 302 | - | elif o == "-P": | |
| 303 | - | self.print_output = True | |
| 304 | - | elif o == "-U": | |
| 305 | - | self.newlines = True | |
| 306 | - | elif o == "-v": | |
| 307 | - | self.show_version = True | |
| 308 | - | elif o == "-w": | |
| 309 | - | self.make_writable_cmd = a | |
| 310 | - | elif o == "-x": | |
| 311 | - | self.no_generate = True | |
| 312 | - | elif o == "-z": | |
| 313 | - | self.eof_can_be_end = True | |
| 314 | - | elif o == "--check": | |
| 315 | - | self.check = True | |
| 316 | - | elif o == "--diff": | |
| 317 | - | self.diff = True | |
| 318 | - | elif o == "--markers": | |
| 319 | - | self._parse_markers(a) | |
| 320 | - | elif o == "--verbosity": | |
| 321 | - | self.verbosity = int(a) | |
| 322 | - | else: | |
| 323 | - | # Since getopt.getopt is given a list of possible flags, | |
| 324 | - | # this is an internal error. | |
| 325 | - | raise CogInternalError(f"Don't understand argument {o}") | |
| 326 | - | ||
| 327 | - | def _parse_markers(self, val): | |
| 328 | - | try: | |
| 329 | - | self.begin_spec, self.end_spec, self.end_output = val.split(" ") | |
| 330 | - | except ValueError: | |
| 331 | - | raise CogUsageError( | |
| 332 | - | f"--markers requires 3 values separated by spaces, could not parse {val!r}" | |
| 333 | - | ) | |
| 334 | - | ||
| 335 | - | def validate(self): | |
| 336 | - | """Does nothing if everything is OK, raises CogError's if it's not.""" | |
| 337 | - | if self.replace and self.delete_code: | |
| 338 | - | raise CogUsageError( | |
| 339 | - | "Can't use -d with -r (or you would delete all your source!)" | |
| 340 | - | ) | |
| 341 | - | ||
| 342 | - | if self.replace and self.output_name: | |
| 343 | - | raise CogUsageError("Can't use -o with -r (they are opposites)") | |
| 344 | - | ||
| 345 | - | if self.diff and not self.check: | |
| 346 | - | raise CogUsageError("Can't use --diff without --check") | |
| 347 | - | ||
| 348 | - | ||
| 349 | - | class Cog(Redirectable): | |
| 350 | - | """The Cog engine.""" | |
| 351 | - | ||
| 352 | - | def __init__(self): | |
| 353 | - | super().__init__() | |
| 354 | - | self.options = CogOptions() | |
| 355 | - | self.cogmodulename = "cog" | |
| 356 | - | self.create_cog_module() | |
| 357 | - | self.check_failed = False | |
| 358 | - | self.hash_handler = None | |
| 359 | - | self._fix_end_output_patterns() | |
| 360 | - | ||
| 361 | - | def _fix_end_output_patterns(self): | |
| 362 | - | self.hash_handler = HashHandler(self.options.end_output) | |
| 363 | - | ||
| 364 | - | def show_warning(self, msg): | |
| 365 | - | self.prout(f"Warning: {msg}") | |
| 366 | - | ||
| 367 | - | def is_begin_spec_line(self, s): | |
| 368 | - | return self.options.begin_spec in s | |
| 369 | - | ||
| 370 | - | def is_end_spec_line(self, s): | |
| 371 | - | return self.options.end_spec in s and not self.is_end_output_line(s) | |
| 372 | - | ||
| 373 | - | def is_end_output_line(self, s): | |
| 374 | - | return self.options.end_output in s | |
| 375 | - | ||
| 376 | - | def create_cog_module(self): | |
| 377 | - | """Make a cog "module" object. | |
| 378 | - | ||
| 379 | - | Imported Python modules can use "import cog" to get our state. | |
| 380 | - | ||
| 381 | - | """ | |
| 382 | - | self.cogmodule = types.SimpleNamespace() | |
| 383 | - | self.cogmodule.path = [] | |
| 384 | - | ||
| 385 | - | def open_output_file(self, fname): | |
| 386 | - | """Open an output file, taking all the details into account.""" | |
| 387 | - | opts = {} | |
| 388 | - | mode = "w" | |
| 389 | - | opts["encoding"] = self.options.encoding | |
| 390 | - | if self.options.newlines: | |
| 391 | - | opts["newline"] = "\n" | |
| 392 | - | fdir = os.path.dirname(fname) | |
| 393 | - | if os.path.dirname(fdir) and not os.path.exists(fdir): | |
| 394 | - | os.makedirs(fdir) | |
| 395 | - | return open(fname, mode, **opts) | |
| 396 | - | ||
| 397 | - | def open_input_file(self, fname): | |
| 398 | - | """Open an input file.""" | |
| 399 | - | if fname == "-": | |
| 400 | - | return sys.stdin | |
| 401 | - | else: | |
| 402 | - | return open(fname, encoding=self.options.encoding) | |
| 403 | - | ||
| 404 | - | def process_file(self, file_in, file_out, fname=None, globals=None): | |
| 405 | - | """Process an input file object to an output file object. | |
| 406 | - | ||
| 407 | - | `fileIn` and `fileOut` can be file objects, or file names. | |
| 408 | - | ||
| 409 | - | """ | |
| 410 | - | file_name_in = fname or "" | |
| 411 | - | file_name_out = fname or "" | |
| 412 | - | file_in_to_close = file_out_to_close = None | |
| 413 | - | # Convert filenames to files. | |
| 414 | - | if isinstance(file_in, (bytes, str)): | |
| 415 | - | # Open the input file. | |
| 416 | - | file_name_in = file_in | |
| 417 | - | file_in = file_in_to_close = self.open_input_file(file_in) | |
| 418 | - | if isinstance(file_out, (bytes, str)): | |
| 419 | - | # Open the output file. | |
| 420 | - | file_name_out = file_out | |
| 421 | - | file_out = file_out_to_close = self.open_output_file(file_out) | |
| 422 | - | ||
| 423 | - | start_dir = os.getcwd() | |
| 424 | - | ||
| 425 | - | try: | |
| 426 | - | file_in = NumberedFileReader(file_in) | |
| 427 | - | ||
| 428 | - | saw_cog = False | |
| 429 | - | ||
| 430 | - | self.cogmodule.inFile = file_name_in | |
| 431 | - | self.cogmodule.outFile = file_name_out | |
| 432 | - | self.cogmodulename = "cog_" + md5(file_name_out.encode()).hexdigest() | |
| 433 | - | sys.modules[self.cogmodulename] = self.cogmodule | |
| 434 | - | # if "import cog" explicitly done in code by user, note threading will cause clashes. | |
| 435 | - | sys.modules["cog"] = self.cogmodule | |
| 436 | - | ||
| 437 | - | # The globals dict we'll use for this file. | |
| 438 | - | if globals is None: | |
| 439 | - | globals = {} | |
| 440 | - | ||
| 441 | - | # If there are any global defines, put them in the globals. | |
| 442 | - | globals.update(self.options.defines) | |
| 443 | - | ||
| 444 | - | # loop over generator chunks | |
| 445 | - | line = file_in.readline() | |
| 446 | - | while line: | |
| 447 | - | # Find the next spec begin | |
| 448 | - | while line and not self.is_begin_spec_line(line): | |
| 449 | - | if self.is_end_spec_line(line): | |
| 450 | - | raise CogError( | |
| 451 | - | f"Unexpected {self.options.end_spec!r}", | |
| 1102 further changed lines not shown | |||
The check that tells the two apart
fail→pass·cogapp/test_cogapp.py::CheckTests::test_check_bad_with_message
Check file cogapp/test_cogapp.py, taken without changes from the fix and copied onto the older code, so the exact same check runs against both versions.
Origin and history
The code before ita53a8e4bf47759d1bd4e3c11124ecdfdb80b72bc
Broken version dated2025-09-19
Modulecogapp.cogapp
Units changedCog, CogOptions
Fingerprint1866dd512dfe7001
Checked2026-08-18 by goldset/0.1
Every field above is generated by our program. None of it is written by hand.
Other bugs found in nedbat/cog
- 2025-06-02fix: cog restores the current directory