54 lines
1.2 KiB
Python
54 lines
1.2 KiB
Python
import re
|
|
|
|
|
|
class MockCursor:
|
|
def __init__(self, data):
|
|
self.cursor_pos = 0
|
|
self.data = data
|
|
|
|
def fetch(self, batch_size):
|
|
batch = self.data[self.cursor_pos: self.cursor_pos + batch_size]
|
|
self.cursor_pos += batch_size
|
|
return batch
|
|
|
|
|
|
class MockPlPy:
|
|
def __init__(self):
|
|
self._reset()
|
|
|
|
def _reset(self):
|
|
self.infos = []
|
|
self.notices = []
|
|
self.debugs = []
|
|
self.logs = []
|
|
self.warnings = []
|
|
self.errors = []
|
|
self.fatals = []
|
|
self.executes = []
|
|
self.results = []
|
|
self.prepares = []
|
|
self.results = {}
|
|
|
|
def _define_result(self, query, result):
|
|
pattern = re.compile(query, re.IGNORECASE | re.MULTILINE)
|
|
self.results[pattern] = result
|
|
|
|
def notice(self, msg):
|
|
self.notices.append(msg)
|
|
|
|
def debug(self, msg):
|
|
self.notices.append(msg)
|
|
|
|
def info(self, msg):
|
|
self.infos.append(msg)
|
|
|
|
def cursor(self, query):
|
|
data = self.execute(query)
|
|
return MockCursor(data)
|
|
|
|
def execute(self, query, rows=1):
|
|
for pattern, result in self.results.iteritems():
|
|
if pattern.search(query):
|
|
return result
|
|
return []
|