libbe/arch.py

   1 from popen2 import Popen3
   2 import os
   3 import config
   4 client = config.get_val("arch_client")
   5 if client is None:
   6     client = "tla"
   7     config.set_val("arch_client", client)
   8
   9 def invoke(args):
  10     q=Popen3(args, True)
  11     output = q.fromchild.read()
  12     error = q.childerr.read()
  13     status = q.wait()
  14     if os.WIFEXITED(status):
  15         return os.WEXITSTATUS(status), output, error
  16     raise Exception("Command failed: %s" % error)
  17
  18 def invoke_client(*args, **kwargs):
  19     cl_args = [client]
  20     cl_args.extend(args)
  21     status,output,error = invoke(cl_args)
  22     if status not in (0,):
  23         raise Exception("Command failed: %s" % error)
  24     return output
  25
  26 def add_id(filename):
  27     invoke_client("add-id", filename)
  28
  29 def delete_id(filename):
  30     invoke_client("delete-id", filename)
  31
  32 def mkdir(path):
  33     os.mkdir(path)
  34     add_id(path)
  35
  36 def set_file_contents(path, contents):
  37     add = not os.path.exists(path)
  38     file(path, "wb").write(contents)
  39     if add:
  40         add_id(path)
  41
  42
  43 def path_in_reference(bug_dir, spec):
  44     if spec is not None:
  45         return invoke_client("file-find", bug_dir, spec).rstrip('\n')
  46     return invoke_client("file-find", bug_dir).rstrip('\n')
  47
  48
  49 def unlink(path):
  50     try:
  51         os.unlink(path)
  52         delete_id(path)
  53     except OSError, e:
  54         if e.errno != 2:
  55             raise
  56
  57
  58 def detect(path):
  59     """Detect whether a directory is revision-controlled using Arch"""
  60     path = os.path.realpath(path)
  61     while True:
  62         if os.path.exists(os.path.join(path, "{arch}")):
  63             return True
  64         if path == "/":
  65             return False
  66         path = os.path.dirname(path)
  67
  68
  69 name = "Arch"