<commit_before># -*- coding: UTF-8 -*-
# NOTE: Launch all tests with `nosetests` command from git repo root dir.

import unittest
import json
import os

from StringIO import StringIO
from tempfile import NamedTemporaryFile
from textwrap import dedent

from mock import patch
from nose.plugins.attrib import attr
import psycopg2

from load import main

env = json.load(open('.env.test.json'))


connection = psycopg2.connect(
    database=env['REDSHIFT_DBNAME'],
    user=env['REDSHIFT_USER'],
    password=env['<PASSWORD>PASSWORD'],
    host=env['REDSHIFT_HOST'],
    port=env['REDSHIFT_PORT']
)


class AggregationTestCase(unittest.TestCase):
    # Test aggregation functions by week, ip, place and risk.

    def setUp(self):
        # Patch main connection with the test one.
        patch('load.main.connection', connection).start()

        # Set isolation level to run CREATE DATABASE statement outside of transactions.
        main.connection.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)

        self.cursor = main.connection.cursor()

        # Recreate logentry table
        try:
            self.cursor.execute("DROP TABLE %s" % (main.tablename))
        except:
            pass
        main.create_table()

    def test_group_by_week(self):
        # GIVEN 3 entries of the same asn, risk and country, two of which within one week
        ntp_scan_csv = dedent('''\
        ts,ip,risk_id,asn,cc
        2016-09-20T00:00:01+00:00,172.16.31.10,2,12252,US
        2016-09-28T00:00:01+00:00,192.168.127.12,2,12252,US
        2016-09-29T00:00:01+00:00,192.168.127.12,2,12252,US
        ''')
        self.cursor.copy_expert("COPY logentry from STDIN csv header", StringIO(ntp_scan_csv))

        # WHEN grouped entries get created
        main.create_count()

        # THEN count table should have 2 entries which get grouped, and one entry which stands alone
        self.cursor.execute('select * from count;')
        self.assertEqual(
            self.cursor.fetchall(),
            [
                (2, 'US', 12252L, '2016-09-19', 'monthly', 1),
                (2, 'US', 12252L, '2016-09-26', 'monthly', 2)  # grouped two entries
            ])
    
    def test_group_by_distinct_ip(self):
        # GIVEN 7 entries of the same asn, risk and country from hostA (172.16.31.10) and hostB (190.81.134)
        # First week: 2 hostA entries, 1 hostB entry
        # Second week: 2 hostA entries, 2 hostB entries
        ntp_scan_csv = dedent('''\
        ts,ip,risk_id,asn,cc
        2016-09-20T00:00:01+00:00,172.16.31.10,2,12252,US
        2016-09-20T00:00:01+00:00,172.16.31.10,2,12252,US
        2016-09-20T00:00:01+00:00,192.168.127.12,2,12252,US
        2016-09-27T00:00:01+00:00,172.16.31.10,2,12252,US
        2016-09-28T00:00:01+00:00,172.16.31.10,2,12252,US
        2016-09-28T00:00:01+00:00,192.168.127.12,2,12252,US
        2016-09-29T00:00:01+00:00,192.168.127.12,2,12252,US
        ''')
        self.cursor.copy_expert("COPY logentry from STDIN csv header", StringIO(ntp_scan_csv))

        # WHEN grouped entries get created
        main.create_count()

        # THEN count table should have 2 rows corresponding to weeks, with properly grouped entries
        self.cursor.execute('select * from count;')
        self.assertEqual(
            self.cursor.fetchall(),
            [
                # First week: 2 entries from hostA count as one
                (2, 'US', 12252L, '2016-09-19', 'monthly', 2),

                # Second week: duplicated entries for hostA and hostB will merge to single one for each host
                (2, 'US', 12252L, '2016-09-26', 'monthly', 2)
            ])

    def test_group_by_ip_week_distinct_risk(self):
        # GIVEN 4 entries of the same asn, week and country from hostA (172.16.31.10) and hostB (190.81.134)
        # hostA: 2 entries of the same risk type
        # hostB: 2 entries of different risk type
        ntp_scan_csv = dedent('''\
        ts,ip,risk_id,asn,cc
        2016-09-28T00:00:01+00:00,172.16.31.10,2,12252,US
        2016-09-29T00:00:01+00:00,172.16.31.10,2,12252,US
        2016-09-28T00:00:01+00:00,192.168.127.12,2,12252,US
        2016-09-29T00:00:01+00:00,192.168.127.12,99,12252,US
        ''')
        self.cursor.copy_expert("COPY logentry from STDIN csv header", StringIO(ntp_scan_csv))

        # WHEN grouped entries get created
        main.create_count()

        # THEN count table should have 2 rows corresponding to different risks, with properly grouped entries
        self.cursor.execute('select * from count;')
        self.assertEqual(
            self.cursor.fetchall(),
            [

                # Risk type 99 - one entry from hostB
                (1L, 99, 'US', 12252L, '2016-09-26', 'monthly', 1),

                # Risk type 2 - 2 total count: one entry from hostB, and two from hostA, which grouped into one
                (2L, 2, 'US', 12252L, '2016-09-26', 'monthly', 2)
            ])

    def test_group_by_country(self):
        # GIVEN 3 entries of the same risk and week, two of which are from one country, but different asn
        ntp_scan_csv = dedent('''\<commit_msg>[test][s]: removed ids from expected results<commit_after># -*- coding: UTF-8 -*-
# NOTE: Launch all tests with `nosetests` command from git repo root dir.

import unittest
import json
import os

from StringIO import StringIO
from tempfile import NamedTemporaryFile
from textwrap import dedent

from mock import patch
from nose.plugins.attrib import attr
import psycopg2

from load import main

env = json.load(open('.env.test.json'))


connection = psycopg2.connect(
    database=env['REDSHIFT_DBNAME'],
    user=env['REDSHIFT_USER'],
    password=env['REDSHIFT_PASSWORD'],
    host=env['REDSHIFT_HOST'],
    port=env['REDSHIFT_PORT']
)


class AggregationTestCase(unittest.TestCase):
    # Test aggregation functions by week, ip, place and risk.

    def setUp(self):
        # Patch main connection with the test one.
        patch('load.main.connection', connection).start()

        # Set isolation level to run CREATE DATABASE statement outside of transactions.
        main.connection.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)

        self.cursor = main.connection.cursor()

        # Recreate logentry table
        try:
            self.cursor.execute("DROP TABLE %s" % (main.tablename))
        except:
            pass
        main.create_table()

    def test_group_by_week(self):
        # GIVEN 3 entries of the same asn, risk and country, two of which within one week
        ntp_scan_csv = dedent('''\
        ts,ip,risk_id,asn,cc
        2016-09-20T00:00:01+00:00,172.16.31.10,2,12252,US
        2016-09-28T00:00:01+00:00,192.168.127.12,2,12252,US
        2016-09-29T00:00:01+00:00,192.168.127.12,2,12252,US
        ''')
        self.cursor.copy_expert("COPY logentry from STDIN csv header", StringIO(ntp_scan_csv))

        # WHEN grouped entries get created
        main.create_count()

        # THEN count table should have 2 entries which get grouped, and one entry which stands alone
        self.cursor.execute('select * from count;')
        self.assertEqual(
            self.cursor.fetchall(),
            [
                (2, 'US', 12252L, '2016-09-19', 'monthly', 1),
                (2, 'US', 12252L, '2016-09-26', 'monthly', 2)  # grouped two entries
            ])
    
    def test_group_by_distinct_ip(self):
        # GIVEN 7 entries of the same asn, risk and country from hostA (172.16.31.10) and hostB (190.81.134)
        # First week: 2 hostA entries, 1 hostB entry
        # Second week: 2 hostA entries, 2 hostB entries
        ntp_scan_csv = dedent('''\
        ts,ip,risk_id,asn,cc
        2016-09-20T00:00:01+00:00,172.16.31.10,2,12252,US
        2016-09-20T00:00:01+00:00,172.16.31.10,2,12252,US
        2016-09-20T00:00:01+00:00,192.168.127.12,2,12252,US
        2016-09-27T00:00:01+00:00,172.16.31.10,2,12252,US
        2016-09-28T00:00:01+00:00,172.16.31.10,2,12252,US
        2016-09-28T00:00:01+00:00,192.168.127.12,2,12252,US
        2016-09-29T00:00:01+00:00,192.168.127.12,2,12252,US
        ''')
        self.cursor.copy_expert("COPY logentry from STDIN csv header", StringIO(ntp_scan_csv))

        # WHEN grouped entries get created
        main.create_count()

        # THEN count table should have 2 rows corresponding to weeks, with properly grouped entries
        self.cursor.execute('select * from count;')
        self.assertEqual(
            self.cursor.fetchall(),
            [
                # First week: 2 entries from hostA count as one
                (2, 'US', 12252L, '2016-09-19', 'monthly', 2),

                # Second week: duplicated entries for hostA and hostB will merge to single one for each host
                (2, 'US', 12252L, '2016-09-26', 'monthly', 2)
            ])

    def test_group_by_ip_week_distinct_risk(self):
        # GIVEN 4 entries of the same asn, week and country from hostA (172.16.31.10) and hostB (190.81.134)
        # hostA: 2 entries of the same risk type
        # hostB: 2 entries of different risk type
        ntp_scan_csv = dedent('''\
        ts,ip,risk_id,asn,cc
        2016-09-28T00:00:01+00:00,172.16.31.10,2,12252,US
        2016-09-29T00:00:01+00:00,172.16.31.10,2,12252,US
        2016-09-28T00:00:01+00:00,192.168.127.12,2,12252,US
        2016-09-29T00:00:01+00:00,192.168.127.12,99,12252,US
        ''')
        self.cursor.copy_expert("COPY logentry from STDIN csv header", StringIO(ntp_scan_csv))

        # WHEN grouped entries get created
        main.create_count()

        # THEN count table should have 2 rows corresponding to different risks, with properly grouped entries
        self.cursor.execute('select * from count;')
        self.assertEqual(
            self.cursor.fetchall(),
            [

                # Risk type 99 - one entry from hostB
                (99, 'US', 12252L, '2016-09-26', 'monthly', 1),

                # Risk type 2 - 2 total count: one entry from hostB, and two from hostA, which grouped into one
                (2, 'US', 12252L, '2016-09-26', 'monthly', 2)
            ])

    def test_group_by_country(self):
        # GIVEN 3 entries of the same risk and week, two of which are from one country, but different asn
        ntp_scan_csv = dedent('''\
{'fraction_non_alphanumeric': 0.11770995334370139, 'fraction_numerical': 0.12150077760497667, 'mean_word_length': 2.5092087312414733, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 32, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1179590', 'n_tokens_mistral': 4263, 'n_tokens_neox': 3404, 'n_words': 850}