Mercurial > sqlpython

comparison pexpecter.py @ 0:9c87fa772ec1

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
before big refactor
author catherine@serenity.wpafb.af.mil
date Fri, 30 Nov 2007 13:04:51 -0500
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:9c87fa772ec1
1 """
2 pexpecter
3
4 Uses pexpect to handle interactive sessions
5 Create subclass of Session for each type of program to be used
6 """
7 import re, os
8
9 try:
10 import pexpect
11
12 class Session(object):
13 available = True
14 call = 'theprogram %s'
15 errPattern = re.compile('.')
16 validPattern = re.compile('Connected to:')
17 promptstub = '>'
18 def __init__(self, argstring):
19 self.argstring = argstring
20 self.sess = pexpect.spawn("%s %s" % (self.call, self.argstring))
21 try:
22 self.sess.expect(self.promptstub)
23 self.valid = self.validPattern.search(self.sess.before)
24 self.prompt = '[\r\n]%s%s' % (self.sess.before.splitlines()[-1], self.promptstub)
25 except:
26 self.valid = False
27 def success(self, result):
28 return not self.errPattern.search(result)
29 def attempt(self, command, timeout=30):
30 self.sess.sendline(self._pre_attempt(command))
31 try:
32 self.sess.expect(self.prompt, timeout=timeout)
33 except pexpect.TIMEOUT:
34 return (False, """Errror: Waited %d seconds with no response from %s.
35 To wait longer, set timeout.""" % (timeout, str(self.__class__)))
36 result = self.sess.before
37 success = self.success(result)
38 if success:
39 print 'Executed through %s' % (str(self.__class__))
40 return (success, result)
41 def _pre_attempt(self, command):
42 return command
43
44 class YASQLSession(Session):
45 errPattern = re.compile('\n[A-Z2]{3,4}-\d{4}:\s')
46 terminatorPattern = re.compile('(;|\\g|\\i|\/|\\G|\\s|\\S)\s*\d*\s*$')
47 call = os.popen('locate -r /yasql$').readline().strip()
48 if not call:
49 print 'yasql not found; commands cannot failover to YASQL'
50 available = False
51 def _pre_attempt(self, command):
52 if not self.terminatorPattern.search(command):
53 return '%s;' % (command)
54 return command
55
56 class SQLSession(Session):
57 def _pre_attempt(self, command):
58 if command.strip()[-1] != ';':
59 return '%s;' % (command)
60 return command
61
62 class SqlPlusSession(SQLSession):
63 call = r'sqlplus'
64 errPattern = re.compile('\n[A-Z2]{3,4}-\d{4}:\s')
65 """ def _pre_attempt(self, command):
66 if command.strip()[-1] != ';':
67 return '%s;' % (command)
68 return command"""
69 # can still trip on: apparent error messages listed as data
70
71 class SenoraSession(SQLSession):
72 errPattern = re.compile('(\n[A-Z2]{3,4}-\d{4}:\s)|(\nwhat ? )')
73 call = os.popen('locate -r Senora\.pm$').readline().strip()
74 if call:
75 call = 'perl %s' % (call)
76 else:
77 print 'Senora.pm not found; commands cannot failover to Senora'
78 available = False
79
80 except ImportError:
81 print '''Python's pexpect module is not installed; cannot pass
82 commands through to sqlplus, etc.'''
83 class Session(object):
84 valid = False
85 available = False
86 class YASQLSession(Session):
87 pass
88 class SqlPlusSession(Session):
89 pass
90 class SenoraSession(Session):
91 pass
92
93 available = [s for s in [SenoraSession, YASQLSession, SqlPlusSession] if s.available]