test_pxlog2sql.py #3

#!/usr/bin/env python3
#  -*- encoding: UTF8 -*-
# Test harness for JobsCmdFilter.py

from __future__ import print_function

import sys
import unittest
import os
import io
import logging
import re
import sqlite3
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from pxlog2sql import PxLog2sql

python3 = sys.version_info[0] >= 3

if python3:
    from io import StringIO
else:
    from StringIO import StringIO

fileHeader = """CREATE DATABASE IF NOT EXISTS testdb;
USE testdb;
DROP TABLE IF EXISTS syncs;
CREATE TABLE syncs (lineNumber INT NOT NULL, pid INT NOT NULL, endTime DATETIME NULL, completedLapse FLOAT NULL, uCpu INT NULL, sCpu INT NULL, diskIn INT NULL, diskOut INT NULL, ipcIn INT NULL, ipcOut INT NULL, maxRss INT NULL, pageFaults INT NULL, rpcMsgsIn INT NULL, rpcMsgsOut INT NULL, rpcSizeIn INT NULL, rpcSizeOut INT NULL, rpcHimarkFwd INT NULL, rpcHimarkRev INT NULL, rpcSnd FLOAT NULL, rpcRcv FLOAT NULL, filesServer int NULL, filesCache int NULL, MBServer int NULL, MBCache int NULL, PRIMARY KEY (lineNumber));
BEGIN TRANSACTION;
"""

def compressGzip(tmpFile, input):
    import gzip
    with gzip.open(tmpFile, 'wb') as f:
        if python3:
            f.write(bytes(input.encode()))
        else:
            f.write(input)

def compressZip(tmpFile, input):
    from zipfile import ZipFile
    tmpContentsFile = "~tmpContents.log"
    with open(tmpContentsFile, "w") as f:
        f.write(input)
    with ZipFile(tmpFile, 'w') as myzip:
        myzip.write(tmpContentsFile)

class MyOptions():
    def __init__(self, dbname, logname, sql=False, csv=True, verbosity=logging.DEBUG):
        self.dbname = dbname
        self.logname = logname
        self.logfile = [logname]
        self.sql = sql
        self.no_sql = not sql
        self.verbosity = verbosity
        self.interval = 10
        self.outlog = None
        self.reset = False
        self.fileInterval = None
        self.csv = csv

class TestPxLogParser(unittest.TestCase):

    def __init__(self, methodName='runTest'):
        super(TestPxLogParser, self).__init__(methodName=methodName)
        self.maxDiff = None

    def assertRegex(self, *args, **kwargs):
        if python3:
            return super(TestPxLogParser, self).assertRegex(*args, **kwargs)
        else:
            return super(TestPxLogParser, self).assertRegexpMatches(*args, **kwargs)

    def setUp(self):
        pass

    def tearDown(self):
        pass

    def runTestText(self, expected, expectedCsv, input, longlines=False, compressed=False):
        if not compressed:
            optionsText = MyOptions("testdb", "")
        else:
            optionsText = MyOptions("testdb", input)
        self.saved_stdoutput = StringIO()
        self.saved_stderr = StringIO()
        self.saved_csv = StringIO()
        if not compressed:
            parser = PxLog2sql(optionsText, inStream=StringIO(input), outStream=self.saved_stdoutput,
                                errStream=self.saved_stderr, csvStream=self.saved_csv)
        else:
            optionsText.logname = input
            parser = PxLog2sql(optionsText, outstream=self.saved_stdoutput, errstream=self.saved_stderr)
        parser.processLogs()
        actual = self.saved_stdoutput.getvalue()
        actualCsv = self.saved_csv.getvalue()
        self.assertMultiLineEqual(fileHeader, actual[:len(fileHeader)])
        if not longlines:
            exp = "\n".join(sorted(expected.split("\n")))
            act = "\n".join(sorted(actual[len(fileHeader):].split("\n")))
            self.assertMultiLineEqual(exp, act)
            expCsv = "\n".join(sorted(expectedCsv.split("\n")))
            actCsv = "\n".join(sorted(actualCsv.split("\n")[1:-1]))
            self.assertMultiLineEqual(expCsv, actCsv)
        else:
            rest = actual[len(fileHeader):]
            for expline in expected.split("\n"):
                reLine = re.compile(expline.replace("(", "\("), re.MULTILINE)
                self.assertRegex(rest, reLine, re.MULTILINE)
        reUnrecognised = re.compile("Unrecognised track:", re.MULTILINE)
        if re.search(reUnrecognised, self.saved_stderr.getvalue()):
            self.assertTrue(False, "Didn't process all track records: %s" % self.saved_stderr.getvalue())

    def runTestSQL(self, input, compressed=False):
        if not compressed:
            optionsSQL = MyOptions("testdb", "", sql=True, csv=None)
        else:
            optionsSQL = MyOptions("testdb", input, sql=True, csv=None)
        dbname = "%s.db" % optionsSQL.dbname
        if os.path.exists(dbname):
            os.remove(dbname)
        saved_stdoutput = StringIO()
        saved_stderr = StringIO()
        if not compressed:
            parser = PxLog2sql(optionsSQL, inStream=StringIO(input), outStream=saved_stdoutput, errStream=saved_stderr)
        else:
            optionsSQL.logname = input
            parser = PxLog2sql(optionsSQL, outStream=saved_stdoutput, errStream=saved_stderr)
        parser.processLogs()
        return sqlite3.connect(dbname)

    def testProxyLog(self):
        "Process standard proxy logs"

        input = r"""
Perforce proxy info:
	2018/04/06 13:46:19 pid 51449 completed .246s
--- lapse .246s
--- usage 122+92us 0+0io 6+10143net 1101824k 0pf
--- rpc msgs/size in+out 5+10010/0mb+39mb himarks 2000/2000 snd/rcv .005s/.009s
--- proxy faults 0 MB 0 other 0 flushes 2 cached 2
--- proxytotals files/size svr+cache 0+2/0mb+40mb
Perforce proxy info:
	2018/04/13 09:08:13 pid 99941 completed 1.61s
--- lapse 1.61s
--- usage 255+279us 0+8io 2458+10147net 1622016k 64pf
--- rpc msgs/size in+out 6+10011/0mb+39mb himarks 2000/2000 snd/rcv .000s/.010s
--- proxy faults 0 MB 0 other 0 flushes 2 cached 0
--- proxytotals files/size svr+cache 2+0/40mb+0mb
Perforce proxy info:
	2018/04/13 09:08:14 pid 99948 completed .268s
--- lapse .268s
--- usage 100+126us 0+0io 6+10143net 1019904k 1pf
--- rpc msgs/size in+out 5+10010/0mb+39mb himarks 2000/2000 snd/rcv .012s/.008s
--- proxy faults 0 MB 0 other 0 flushes 2 cached 2
--- proxytotals files/size svr+cache 0+2/0mb+40mb

"""
        expected = """INSERT IGNORE INTO syncs VALUES (2,51449,"2018-04-06 13:46:19",0.246,5,10010,0,39,2000,2000,.005,.009,122,92,0,0,6,10143,1101824,0,0,0,2,40);
INSERT IGNORE INTO syncs VALUES (9,99941,"2018-04-13 09:08:13",1,6,10011,0,39,2000,2000,.000,.010,255,279,0,8,2458,10147,1622016,64,2,40,0,0);
INSERT IGNORE INTO syncs VALUES (16,99948,"2018-04-13 09:08:14",0.268,5,10010,0,39,2000,2000,.012,.008,100,126,0,0,6,10143,1019904,1,0,0,2,40);
COMMIT;
BEGIN TRANSACTION;
"""
        expectedCsv = """2,51449,"2018-04-06 13:46:19",0.246,0,0,2,40
9,99941,"2018-04-13 09:08:13",1,2,40,0,0
16,99948,"2018-04-13 09:08:14",0.268,0,0,2,40"""
        self.runTestText(expected, expectedCsv, input)
        conn = self.runTestSQL(input)
        c = conn.cursor()
        c.execute('SELECT * from Syncs;')
        rows = c.fetchall()
        self.assertEqual(3, len(rows))
        row = rows[0]
        self.assertEqual(row[0], 2)
        self.assertEqual(row[1], 51449)
        self.assertEqual(row[2], u'2018-04-06 13:46:19')
        conn.close()

if __name__ == '__main__':
    unittest.main()