aboutsummaryrefslogtreecommitdiffstats
ModeNameSize
-rw-r--r--.gitignore165logstatsplain
-rw-r--r--.templateconf60logstatsplain
-rw-r--r--LICENSE515logstatsplain
-rw-r--r--README503logstatsplain
d---------meta-selftest177logstatsplain
d---------meta-skeleton236logstatsplain
d---------meta850logstatsplain
-rwxr-xr-xoe-init-build-env2000logstatsplain
-rwxr-xr-xoe-init-build-env-memres2432logstatsplain
d---------scripts2695logstatsplain
generated by cgit 1.2.3-korg (git 2.39.0) at 2024-06-14 21:36:01 +0000
er data? if data_buf: yield data_buf if code_max > 0: raise MonotoneException("error code %d in automate packet." % (code_max)) class Standalone(Runner): """Runs commands by running monotone. One monotone process per command""" def run(self, command, args): # as we pass popen3 as sequence, it executes monotone with these # arguments - and does not pass them through the shell according # to help(os.popen3) # print(("standalone is running:", command, args)) to_run = self.base_command + [command] + args process = popen2.Popen3(to_run, capturestderr=True) for line in process.fromchild: yield line stderr_data = process.childerr.read() if len(stderr_data) > 0: raise MonotoneException("data on stderr for command '%s': %s" % (command, stderr_data)) terminate_popen3(process) class MtnObject: def __init__(self, obj_type): self.obj_type = obj_type class Tag(MtnObject): def __init__(self, name, revision, author, branches): MtnObject.__init__(self, "tag") self.name, self.revision, self.author, self.branches = name, Revision(revision), author, branches class Branch(MtnObject): def __init__(self, name): MtnObject.__init__(self, "branch") self.name = name class File(MtnObject): def __init__(self, name, in_revision): MtnObject.__init__(self, "file") self.name = name self.in_revision = in_revision class Dir(MtnObject): def __init__(self, name, in_revision): MtnObject.__init__(self, "dir") self.name = name self.in_revision = in_revision basic_io_name_tok = re.compile(r'^(\S+)') def basic_io_from_stream(gen): # all of these x_consume functions return parsed string # token to add to stanza, name of next consume function to call # new value of line (eg. with consumed tokens removed) def hex_consume(line): m = hex_re_c.match(line[1:]) if line[0] != '[' or not m: raise MonotoneException("This is not a hex token: %s" % line) end_of_match = m.end(m.lastindex) if line[end_of_match+1] != ']': raise MonotoneException("Hex token ends in character other than ']': %s" % line) return Revision(m.groups()[0]), choose_consume, line[end_of_match+2:] def name_consume(line): m = name_re_c.match(line) if not m: raise MonotoneException("Not a name: %s" % line) end_of_match = m.end(m.lastindex) return m.groups()[0], choose_consume, line[end_of_match:] def choose_consume(line): line = line.lstrip() if line == '': consumer = choose_consume elif line[0] == '[': consumer = hex_consume elif line[0] == '"': consumer = string_consume else: consumer = name_consume return None, consumer, line class StringState: def __init__(self): self.in_escape = False self.has_started = False self.has_ended = False self.value = '' def string_consume(line, state=None): if not state: state = StringState() if not state.has_started: if line[0] != '"': raise MonotoneException("Not a string: %s" % line) line = line[1:] state.has_started = True idx = 0 for idx, c in enumerate(line): if state.in_escape: if c != '\\' and c != '"': raise MonotoneException("Invalid escape code: %s in %s\n" % (c, line)) state.value += c state.in_escape = False else: if c == '\\': state.in_escape = True elif c == '"': state.has_ended = True break else: state.value += c if state.has_ended: return state.value, choose_consume, line[idx+1:] else: return (None, lambda s: string_consume(s, state), line[idx+1:]) consumer = choose_consume current_stanza = [] for line in gen: # if we're not in an actual consumer (which we shouldn't be, unless # we're parsing some sort of multi-line token) and we have a blank # line, it indicates the end of any current stanza if (consumer == choose_consume) and (line == '' or line == '\n') and current_stanza: yield current_stanza current_stanza = [] continue while line != '' and line != '\n': new_token, consumer, line = consumer(line) if new_token != None: current_stanza.append(new_token) if current_stanza: yield current_stanza class Operations: def __init__(self, runner_args): self.standalone = apply(Standalone, runner_args) self.automate = apply(Automate, runner_args) def tags(self): for stanza in basic_io_from_stream(self.automate.run('tags', [])): if stanza[0] == 'tag': branches = [] for branch in stanza[7:]: branches.append(Branch(branch)) yield Tag(stanza[1], stanza[3], stanza[5], branches) def branches(self): for line in (t.strip() for t in self.automate.run('branches', [])): if not line: continue yield apply(Branch, (line,)) def graph(self): for line in self.automate.run('graph', []): yield line def parents(self, revision): if revision != "": for line in (t.strip() for t in self.automate.run('parents', [revision])): if not line: continue yield apply(Revision, (line,)) def ancestry_difference(self, new_rev, old_revs): """ new_rev a single new revision number old_revs a list of revisions """ if new_rev != "": for line in (t.strip() for t in self.automate.run('ancestry_difference', [new_rev]+old_revs)): if not line: continue yield apply(Revision, (line,)) def children(self, revision): if revision != "": for line in (t.strip() for t in self.automate.run('children', [revision])): if not line: continue yield apply(Revision, (line,)) def toposort(self, revisions): for line in (t.strip() for t in self.automate.run('toposort', revisions)): if not line: continue yield apply(Revision, (line,)) def heads(self, branch): for line in (t.strip() for t in self.automate.run('heads', [branch])): if not line: continue yield apply(Revision, (line,)) def get_content_changed(self, revision, path): for stanza in basic_io_from_stream(self.automate.run('get_content_changed', [revision, path])): yield stanza def get_revision(self, revision): for stanza in basic_io_from_stream(self.automate.run('get_revision', [revision])): yield stanza def get_manifest_of(self, revision): for stanza in basic_io_from_stream(self.automate.run('get_manifest_of', [revision])): yield stanza def get_file(self, fileid): for stanza in self.automate.run('get_file', [fileid]): yield stanza def certs(self, revision): for stanza in basic_io_from_stream(self.automate.run('certs', [revision])): yield stanza def diff(self, revision_from, revision_to, files=[]): args = ['-r', revision_from, '-r', revision_to] + files for line in self.standalone.run('diff', args): yield line