test_log2sql.py #1

# -*- encoding: UTF8 -*-
# Test harness for JobsCmdFilter.py

from __future__ import print_function

import sys
import unittest
import os
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from log2sql import Log2sql

python3 = sys.version_info[0] >= 3

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

saved_stdoutput = StringIO()

fileHeader = """CREATE DATABASE IF NOT EXISTS testdb;
USE testdb;
DROP TABLE IF EXISTS process;
CREATE TABLE process (processkey CHAR(32) NOT NULL, lineNumber INT NOT NULL, pid INT NOT NULL, startTime DATETIME NOT NULL,endTime DATETIME NULL, computedLapse FLOAT NULL,completedLapse FLOAT NULL, user TEXT NOT NULL, workspace TEXT NOT NULL, ip TEXT NOT NULL, app TEXT NOT NULL, cmd TEXT NOT NULL, args TEXT 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, running INT NULL, error TEXT NULL, PRIMARY KEY (processkey, lineNumber));
DROP TABLE IF EXISTS tableUse;
CREATE TABLE tableUse (processkey CHAR(32) NOT NULL, lineNumber INT NOT NULL, tableName VARCHAR(255) NOT NULL, pagesIn INT NULL, pagesOut INT NULL, pagesCached INT NULL, readLocks INT NULL, writeLocks INT NULL, getRows INT NULL, posRows INT NULL, scanRows INT NULL, putRows int NULL, delRows INT NULL, totalReadWait INT NULL, totalReadHeld INT NULL, totalWriteWait INT NULL, totalWriteHeld INT NULL, maxReadWait INT NULL, maxReadHeld INT NULL, maxWriteWait INT NULL, maxWriteHeld INT NULL, PRIMARY KEY (processkey, lineNumber, tableName));
SET autocommit=0;
"""

class TestLogParser(unittest.TestCase):

    def __init__(self, methodName='runTest'):
        super(TestLogParser, self).__init__(methodName=methodName)

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

    def setUp(self):
        pass

    def tearDown(self):
        pass

    def testBasicParsing(self):
        "simple parsing"
        input = """
Perforce server info:
	2015/09/02 15:23:09 pid 1616 robert@robert-test 127.0.0.1 [Microsoft� Visual Studio� 2013/12.0.21005.1] 'user-sync //...'
Perforce server info:
	2015/09/02 15:23:09 pid 1616 compute end .031s
Perforce server info:
	2015/09/02 15:23:09 pid 1616 completed .031s
"""
        expected = """INSERT IGNORE INTO process VALUES ("5abd121d62ac8d3a07beffd9ff723f72",3,1616,"2015-09-02 15:23:09",NULL,NULL,NULL,"robert","robert-test","127.0.0.1","Microsoft\xef\xbf\xbd Visual Studio\xef\xbf\xbd 2013/12.0.21005.1","user-sync","//...",NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,0,NULL);
COMMIT;
"""
        log2sql = Log2sql("", "testdb", instream=StringIO(input), outstream=saved_stdoutput)
        log2sql.processLog()
        self.assertEqual("%s%s" % (fileHeader, expected),
                         saved_stdoutput.getvalue())

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