ID2T-toolkit/code/Test/test_SQLQueries.py

import unittest, sqlite3

from definitions import ROOT_DIR
import Core.Controller as Ctrl

pcap = ROOT_DIR + "/../resources/test/reference_1998.pcap"

controller = Ctrl.Controller(pcap_file_path=pcap, do_extra_tests=False, non_verbose=False)
controller.load_pcap_statistics(flag_write_file=False, flag_recalculate_stats=True, flag_print_statistics=False)


class UnitTestPyparsing(unittest.TestCase):

    def test_apostrophe(self):
        query = "Select ipAddress from ip_Statistics where pktsSent = '5'"
        query2 = "Select ipAddress from ip_Statistics where pktsSent = 5"
        self.assertEqual(controller.statistics.stats_db.process_db_query(query),
                         controller.statistics.stats_db.process_db_query(query2))

    def test_paranthesis(self):
        query = "Select (ipAddress) from (ip_Statistics) where (pktsSent) = (2 + (3))"
        self.assertEqual("72.247.178.67", controller.statistics.stats_db.process_db_query(query))

    def test_noresult(self):
        query = "Select ipAddress from ip_statistics where ipaddress = 'abc'"
        self.assertEqual([], controller.statistics.stats_db.process_db_query(query))

    def test_severaloperator(self):
        query1 = "Select ipAddress from ip_Statistics where pktsSent = '5'"
        query2 = "Select ipAddress from ip_Statistics where pktsSent < '5'"
        query3 = "Select ipAddress from ip_Statistics where pktsSent <= '5'"
        query4 = "Select ipAddress from ip_Statistics where pktsSent > '356'"
        query5 = "Select ipAddress from ip_Statistics where pktsSent >= '356'"

        self.assertEqual("72.247.178.67", controller.statistics.stats_db.process_db_query(query1))
        self.assertEqual("72.247.178.113", controller.statistics.stats_db.process_db_query(query2))
        self.assertEqual(["72.247.178.67", "72.247.178.113"], controller.statistics.stats_db.process_db_query(query3))
        self.assertEqual("10.0.2.15", controller.statistics.stats_db.process_db_query(query4))
        self.assertEqual(["10.0.2.15", "172.217.23.174"], controller.statistics.stats_db.process_db_query(query5))

        # too long query (stackoverflow)
        with self.assertRaises(sqlite3.OperationalError):
            query = "Select ipAddress from ip_statistics where pktsSent = "
            i = 0
            while (i < 15):
                query += "(Select pktsSent from ip_statistics where pktsSent ="
                i += 1
            query += "5"
            while (i > 0):
                query += ")"
                i -= 1
            controller.statistics.stats_db.process_db_query(query)
        # compare of tables with different dimension
        with self.assertRaises(sqlite3.OperationalError):
            controller.statistics.stats_db.process_db_query('Select ipAddress from ip_Statistics where pktsSent'
                                                            '= (Select * from ip_Statistics)')