def setUp(self):
# Connect to the GPRSDB.
self.gprs_db = gprs_database.GPRSDB()
# Add some records to the GPRSDB.
self.now = time.time()
records = [
(psycopg2.TimestampFromTicks(self.now - 60), 'IMSI901550000000084',
'192.168.99.1', 100, 200, 100, 200),
(psycopg2.TimestampFromTicks(self.now - 30), 'IMSI901550000000084',
'192.168.99.1', 300, 500, 200, 300),
(psycopg2.TimestampFromTicks(self.now - 10), 'IMSI901550000000084',
'192.168.99.1', 700, 600, 400, 100),
(psycopg2.TimestampFromTicks(self.now - 5), 'IMSI901550000000084',
'192.168.99.1', 750, 625, 50, 25),
]
schema = ('record_timestamp, imsi, ipaddr, uploaded_bytes,'
' downloaded_bytes, uploaded_bytes_delta,'
' downloaded_bytes_delta')
connection = psycopg2.connect(host='localhost', database='endaga',
user=PG_USER, password=PG_PASSWORD)
with connection.cursor() as cursor:
for record in records:
values = "%s, '%s', '%s', %s, %s, %s, %s" % record
command = 'insert into gprs_records (%s) values(%s)' % (
schema, values)
cursor.execute(command)
connection.commit()
gprs_tests.py 文件源码
python
阅读 30
收藏 0
点赞 0
评论 0
评论列表
文章目录