<commit_before>#!/usr/local/bin/python
# coding: utf-8

import datetime
import unittest

try:
    from unittest.mock import patch, Mock, MagicMock, call
except ImportError:
    from mock import patch, Mock, MagicMock, call

import pad.plugins.header_eval


class TestHeaderEvalBase(unittest.TestCase):
    def setUp(self):
        unittest.TestCase.setUp(self)
        self.local_data = {}
        self.global_data = {}
        self.mock_ctxt = MagicMock(**{
            "get_plugin_data.side_effect": lambda p, k: self.global_data[k],
            "set_plugin_data.side_effect": lambda p, k,
                                                  v: self.global_data.
                                   setdefault(k, v)
        })
        self.mock_msg = MagicMock()
        self.plugin = pad.plugins.header_eval.HeaderEval(self.mock_ctxt)
        self.plugin.set_local = lambda m, k, v: self.local_data.__setitem__(k,
                                                                            v)
        self.plugin.get_local = lambda m, k: self.local_data.__getitem__(k)
        self.plugin.set_global = self.global_data.__setitem__
        self.plugin.get_global = self.global_data.__getitem__
        self.mock_ruleset = MagicMock()
        self.mock_locale = patch("pad.plugins.header_eval."
                                 "pad.locales.charset_ok_for_locales").start()
        self.mock_is_domain_valid = patch("pad.plugins.header_eval."
                                          "HeaderEval.is_domain_valid").start()

    def tearDown(self):
        unittest.TestCase.tearDown(self)
        patch.stopall()


class TestHeaderEval(TestHeaderEvalBase):

    def test_check_for_fake_aol_relay_in_rcvd_spam(self):
        header = ("from unknown (HELO mta05bw.bigpond.com) (172.16.17.32) "
                  "by rly-xw01.mx.aol.com with QMQP; Sat, 15 Jun 2002 "
                  "23:37:16 -0000")
        self.mock_msg.get_decoded_header.return_value = [header]
        result = self.plugin.check_for_fake_aol_relay_in_rcvd(self.mock_msg)
        self.assertTrue(result)

    def test_check_for_fake_aol_relay_in_rcvd_not_spam1(self):
        header = ("from  rly-xj02.mx.aol.com (rly-xj02.mail.aol.com "
                  "[172.20.116.39]) by omr-r05.mx.aol.com (v83.35) with "
                  "ESMTP id RELAYIN7-0501132011; Wed, 01 May 2002 "
                  "13:20:11 -0400")
        self.mock_msg.get_decoded_header.return_value = [header]
        result = self.plugin.check_for_fake_aol_relay_in_rcvd(self.mock_msg)
        self.assertFalse(result)

    def test_check_for_fake_aol_relay_in_rcvd_not_spam2(self):
        header = ("from logs-tr.proxy.aol.com (logs-tr.proxy.aol.com "
                  "[172.16.31.10]) by rly-ip01.mx.aol.com "
                  "(8.8.8/8.8.8/AOL-5.0.0) with ESMTP id NAA08955 for "
                  "<<EMAIL>>; Thu, 4 Apr 2002 13:11:20 "
                  "-0500 (EST)")
        self.mock_msg.get_decoded_header.return_value = [header]
        result = self.plugin.check_for_fake_aol_relay_in_rcvd(self.mock_msg)
        self.assertFalse(result)

    def test_check_for_fake_aol_relay_in_rcvd_not_spam3_not_aol(self):
        header = ("by 10.28.54.13 with SMTP id d13csp1785386wma; Mon, "
                  "28 Nov 2016 07:40:07 -0800 (PST)")
        self.mock_msg.get_decoded_header.return_value = [header]
        result = self.plugin.check_for_fake_aol_relay_in_rcvd(self.mock_msg)
        self.assertFalse(result)

    def test_check_for_faraway_charset_in_headers_no_locale(self):
        self.mock_locale.return_value = False
        self.mock_ctxt.conf.get_global.return_value = ""
        self.mock_msg.get_raw_header.return_value = ["=?UTF8?B?dGVzdA==?="]
        result = self.plugin.check_for_faraway_charset_in_headers(self.mock_msg)
        self.assertFalse(result)

    def test_check_for_faraway_charset_in_headers_all_locale(self):
        self.mock_locale.return_value = False
        self.mock_ctxt.conf.get_global.return_value = "all"
        self.mock_msg.get_raw_header.return_value = ["=?UTF8?B?dGVzdA==?="]
        result = self.plugin.check_for_faraway_charset_in_headers(self.mock_msg)
        self.assertFalse(result)

    def test_check_for_faraway_charset_in_headers_spam(self):
        self.mock_locale.return_value = False
        self.mock_ctxt.conf.get_global.return_value = "ru"
        self.mock_msg.get_raw_header.return_value = ["=?UTF8?B?dGVzdA==?="]
        result = self.plugin.check_for_faraway_charset_in_headers(self.mock_msg)
        self.assertTrue(result)

    def test_check_for_faraway_charset_in_headers_ham(self):
        self.mock_locale.return_value = True
        self.mock_ctxt.conf.get_global.return_value = "ru"
        self.mock_msg.get_raw_header.return_value = ["=?UTF8?B?dGVzdA==?="]
        result = self.plugin.check_for_faraway_charset_in_headers(self.mock_msg)
        self.assertFalse(result)

    def test_check_for_faraway_charset_in_headers_invalid_header(self):
        self.mock_locale.return_value = False
        self.mock_ctxt.conf.get_global.return_value = "ru"
        patch("pad.plugins.header_eval.email.header.decode_header",
              side_effect=ValueError).start()
        self.mock_msg.get_raw_header.return_value = ["=?UTF8?B?dGVzdA==?="]
        result = self.plugin.check_for_faraway_charset_in_headers(self.mock_msg)
        self.assertFalse(result)

    def test_check_for_faraway_charset_in_headers_correct_call(self):
        self.mock_locale.return_value = False
        self.mock_ctxt.conf.get_global.return_value = "ru ko"
        self.mock_msg.get_raw_header.return_value = ["=?UTF8?B?dGVzdA==?="]
        result = self.plugin.check_for_faraway_charset_in_headers(self.mock_msg)
        self.mock_locale.assert_called_with("utf8", ["ru", "ko"])

    def test_check_header_count_range_match(self):
        self.mock_msg.get_raw_header.return_value = ["a", "b"]
        result = self.plugin.check_header_count_range(self.mock_msg, "Test",
                                                      "2", "3")
        self.mock_msg.get_raw_header.assert_called_with("Test")
        self.assertTrue(result)

    def test_check_header_count_range_no_match(self):
        self.mock_msg.get_raw_header.return_value = ["a", "b"]
        result = self.plugin.check_header_count_range(self.mock_msg, "Test",
                                                      "3", "4")
        self.mock_msg.get_raw_header.assert_called_with("Test")
        self.assertFalse(result)

    def test_check_for_missing_to_header_has_to(self):
        self.mock_msg.get_raw_header.side_effect = [["<EMAIL>"]]
        result = self.plugin.check_for_missing_to_header(self.mock_msg)
        self.assertFalse(result)

    def test_check_for_missing_to_header_has_apparently_to(self):
        self.mock_msg.get_raw_header.side_effect = [[], ["<EMAIL>"]]
        result = self.plugin.check_for_missing_to_header(self.mock_msg)
        self.assertFalse(result)

    def test_check_for_missing_to_header_match(self):
        self.mock_msg.get_raw_header.side_effect = [[], []]
        result = self.plugin.check_for_missing_to_header(self.mock_msg)
        self.assertTrue(result)

    def test_subject_is_all_caps(self):
        header = ("Re: THIS IS A CAPS HEADER")
        self.mock_msg.get_decoded_header.return_value = [header]
        result = self.plugin.subject_is_all_caps(self.mock_msg)
        self.assertTrue(result)

    def test_subject_is_all_caps_false(self):
        header = ("Fwd: Re: this is not a caps header")
        self.mock_msg.get_decoded_header.return_value = [header]
        result = self.plugin.subject_is_all_caps(self.mock_msg)
        self.assertFalse(result)

    def test_illegal_chars(self):
        self.mock_msg.get_raw_header.return_value = [u"ùTest"]
        result = self.plugin.check_illegal_chars(self.mock_msg, "Subject",
                                                 '0.1', '0')
        self.assertTrue(result)

    def test_illegal_chars_all(self):
        self.mock_msg.raw_headers = {"From": [u"ùTùùesùùùùtù"],
                                     "Subject": [u"ùTùùesùùùùtù"],
                                     "Another": [u"Normal Text"]}
        result = self.plugin.check_illegal_chars(self.mock_msg, "ALL", '0.1',
                                                 '0')
        self.assertFalse(result)

    def test_illegal_chars_all_true(self):
        self.mock_msg.raw_headers = {"From": [u"ùTùùesùùùùtù"],
                                     "Subject": [u"ùTùùesùùùùtù"],
                                     "Another": [u"ùùesùùù"]}
        result = self.plugin.check_illegal_chars(self.mock_msg, "ALL", '0.1',
                                                 '0')
        self.assertTrue(result)

    def test_illegal_chars_exempt(self):
        self.mock_msg.get_raw_header.return_value = [u"Test\xa2"]
        result = self.plugin.check_illegal_chars(self.mock_msg, "Subject",
                                                 '0.1', '0')
        self.assertFalse(result)

    def test_check_for_msn_groups_headers(self):
        received = "from mail pickup service by p23.groups.msn.com"
        message_id = "<<EMAIL>>"
        self.mock_msg.get_decoded_header.side_effect = [["<<EMAIL>>"],
                                                        [received],
                                                        [message_id]]
        result = self.plugin.check_for_msn_groups_headers(self.mock_msg)
        self.assertTrue(result)

    def test_gated_through_received_hdr_remover(self):
        mailing_list = "contact <EMAIL>; run by ezmlm"
        received = "(qmail 47240 invoked by uid 33); 01 Oct 2010 20:35:23 +0000"
        delivered_to = "mailing list <EMAIL>"
        self.mock_msg.get_decoded_header.side_effect = [[mailing_list],
                                                        [received],
                                                        [delivered_to]]
        result = self.plugin.gated_through_received_hdr_remover(self.mock_msg)
        self.assertTrue(result)

    def test_gated_thorugh_received_hdr_remover_no_received(self):<commit_msg>Refs #28 Fix unittests in order to receive correct values from 'raw_headers' and 'get_raw_header'<commit_after>#!/usr/local/bin/python
# coding: utf-8

import datetime
import unittest

try:
    from unittest.mock import patch, Mock, MagicMock, call
except ImportError:
    from mock import patch, Mock, MagicMock, call

import pad.plugins.header_eval


class TestHeaderEvalBase(unittest.TestCase):
    def setUp(self):
        unittest.TestCase.setUp(self)
        self.local_data = {}
        self.global_data = {}
        self.mock_ctxt = MagicMock(**{
            "get_plugin_data.side_effect": lambda p, k: self.global_data[k],
            "set_plugin_data.side_effect": lambda p, k,
                                                  v: self.global_data.
                                   setdefault(k, v)
        })
        self.mock_msg = MagicMock()
        self.plugin = pad.plugins.header_eval.HeaderEval(self.mock_ctxt)
        self.plugin.set_local = lambda m, k, v: self.local_data.__setitem__(k,
                                                                            v)
        self.plugin.get_local = lambda m, k: self.local_data.__getitem__(k)
        self.plugin.set_global = self.global_data.__setitem__
        self.plugin.get_global = self.global_data.__getitem__
        self.mock_ruleset = MagicMock()
        self.mock_locale = patch("pad.plugins.header_eval."
                                 "pad.locales.charset_ok_for_locales").start()
        self.mock_is_domain_valid = patch("pad.plugins.header_eval."
                                          "HeaderEval.is_domain_valid").start()

    def tearDown(self):
        unittest.TestCase.tearDown(self)
        patch.stopall()


class TestHeaderEval(TestHeaderEvalBase):

    def test_check_for_fake_aol_relay_in_rcvd_spam(self):
        header = ("from unknown (HELO mta05bw.bigpond.com) (172.16.17.32) "
                  "by rly-xw01.mx.aol.com with QMQP; Sat, 15 Jun 2002 "
                  "23:37:16 -0000")
        self.mock_msg.get_decoded_header.return_value = [header]
        result = self.plugin.check_for_fake_aol_relay_in_rcvd(self.mock_msg)
        self.assertTrue(result)

    def test_check_for_fake_aol_relay_in_rcvd_not_spam1(self):
        header = ("from  rly-xj02.mx.aol.com (rly-xj02.mail.aol.com "
                  "[172.20.116.39]) by omr-r05.mx.aol.com (v83.35) with "
                  "ESMTP id RELAYIN7-0501132011; Wed, 01 May 2002 "
                  "13:20:11 -0400")
        self.mock_msg.get_decoded_header.return_value = [header]
        result = self.plugin.check_for_fake_aol_relay_in_rcvd(self.mock_msg)
        self.assertFalse(result)

    def test_check_for_fake_aol_relay_in_rcvd_not_spam2(self):
        header = ("from logs-tr.proxy.aol.com (logs-tr.proxy.aol.com "
                  "[152.163.201.132]) by rly-ip01.mx.aol.com "
                  "(8.8.8/8.8.8/AOL-5.0.0) with ESMTP id NAA08955 for "
                  "<<EMAIL>>; Thu, 4 Apr 2002 13:11:20 "
                  "-0500 (EST)")
        self.mock_msg.get_decoded_header.return_value = [header]
        result = self.plugin.check_for_fake_aol_relay_in_rcvd(self.mock_msg)
        self.assertFalse(result)

    def test_check_for_fake_aol_relay_in_rcvd_not_spam3_not_aol(self):
        header = ("by 10.28.54.13 with SMTP id d13csp1785386wma; Mon, "
                  "28 Nov 2016 07:40:07 -0800 (PST)")
        self.mock_msg.get_decoded_header.return_value = [header]
        result = self.plugin.check_for_fake_aol_relay_in_rcvd(self.mock_msg)
        self.assertFalse(result)

    def test_check_for_faraway_charset_in_headers_no_locale(self):
        self.mock_locale.return_value = False
        self.mock_ctxt.conf.get_global.return_value = ""
        self.mock_msg.get_raw_header.return_value = ["=?UTF8?B?dGVzdA==?="]
        result = self.plugin.check_for_faraway_charset_in_headers(self.mock_msg)
        self.assertFalse(result)

    def test_check_for_faraway_charset_in_headers_all_locale(self):
        self.mock_locale.return_value = False
        self.mock_ctxt.conf.get_global.return_value = "all"
        self.mock_msg.get_raw_header.return_value = ["=?UTF8?B?dGVzdA==?="]
        result = self.plugin.check_for_faraway_charset_in_headers(self.mock_msg)
        self.assertFalse(result)

    def test_check_for_faraway_charset_in_headers_spam(self):
        self.mock_locale.return_value = False
        self.mock_ctxt.conf.get_global.return_value = "ru"
        self.mock_msg.get_raw_header.return_value = ["=?UTF8?B?dGVzdA==?="]
        result = self.plugin.check_for_faraway_charset_in_headers(self.mock_msg)
        self.assertTrue(result)

    def test_check_for_faraway_charset_in_headers_ham(self):
        self.mock_locale.return_value = True
        self.mock_ctxt.conf.get_global.return_value = "ru"
        self.mock_msg.get_raw_header.return_value = ["=?UTF8?B?dGVzdA==?="]
        result = self.plugin.check_for_faraway_charset_in_headers(self.mock_msg)
        self.assertFalse(result)

    def test_check_for_faraway_charset_in_headers_invalid_header(self):
        self.mock_locale.return_value = False
        self.mock_ctxt.conf.get_global.return_value = "ru"
        patch("pad.plugins.header_eval.email.header.decode_header",
              side_effect=ValueError).start()
        self.mock_msg.get_raw_header.return_value = ["=?UTF8?B?dGVzdA==?="]
        result = self.plugin.check_for_faraway_charset_in_headers(self.mock_msg)
        self.assertFalse(result)

    def test_check_for_faraway_charset_in_headers_correct_call(self):
        self.mock_locale.return_value = False
        self.mock_ctxt.conf.get_global.return_value = "ru ko"
        self.mock_msg.get_raw_header.return_value = ["=?UTF8?B?dGVzdA==?="]
        result = self.plugin.check_for_faraway_charset_in_headers(self.mock_msg)
        self.mock_locale.assert_called_with("utf8", ["ru", "ko"])

    def test_check_header_count_range_match(self):
        self.mock_msg.get_raw_header.return_value = ["a", "b"]
        result = self.plugin.check_header_count_range(self.mock_msg, "Test",
                                                      "2", "3")
        self.mock_msg.get_raw_header.assert_called_with("Test")
        self.assertTrue(result)

    def test_check_header_count_range_no_match(self):
        self.mock_msg.get_raw_header.return_value = ["a", "b"]
        result = self.plugin.check_header_count_range(self.mock_msg, "Test",
                                                      "3", "4")
        self.mock_msg.get_raw_header.assert_called_with("Test")
        self.assertFalse(result)

    def test_check_for_missing_to_header_has_to(self):
        self.mock_msg.get_raw_header.side_effect = [["<EMAIL>"]]
        result = self.plugin.check_for_missing_to_header(self.mock_msg)
        self.assertFalse(result)

    def test_check_for_missing_to_header_has_apparently_to(self):
        self.mock_msg.get_raw_header.side_effect = [[], ["<EMAIL>"]]
        result = self.plugin.check_for_missing_to_header(self.mock_msg)
        self.assertFalse(result)

    def test_check_for_missing_to_header_match(self):
        self.mock_msg.get_raw_header.side_effect = [[], []]
        result = self.plugin.check_for_missing_to_header(self.mock_msg)
        self.assertTrue(result)

    def test_subject_is_all_caps(self):
        header = ("Re: THIS IS A CAPS HEADER")
        self.mock_msg.get_decoded_header.return_value = [header]
        result = self.plugin.subject_is_all_caps(self.mock_msg)
        self.assertTrue(result)

    def test_subject_is_all_caps_false(self):
        header = ("Fwd: Re: this is not a caps header")
        self.mock_msg.get_decoded_header.return_value = [header]
        result = self.plugin.subject_is_all_caps(self.mock_msg)
        self.assertFalse(result)

    def test_illegal_chars(self):
        self.mock_msg.get_raw_header.return_value = ["ùTest"]
        result = self.plugin.check_illegal_chars(self.mock_msg, "Subject",
                                                 '0.1', '0')
        self.assertTrue(result)

    def test_illegal_chars_all(self):
        self.mock_msg.raw_headers = {"From": ["ùTùùesùùùùtù"],
                                     "Subject": ["ùTùùesùùùùtù"],
                                     "Another": ["Normal Text"]}
        result = self.plugin.check_illegal_chars(self.mock_msg, "ALL", '0.1',
                                                 '0')
        self.assertFalse(result)

    def test_illegal_chars_all_true(self):
        self.mock_msg.raw_headers = {"From": ["ùTùùesùùùùtù"],
                                     "Subject": ["ùTùùesùùùùtù"],
                                     "Another": ["ùùesùùù"]}
        result = self.plugin.check_illegal_chars(self.mock_msg, "ALL", '0.1',
                                                 '0')
        self.assertTrue(result)

    def test_illegal_chars_exempt(self):
        self.mock_msg.get_raw_header.return_value = ["Test\\xa2"]
        result = self.plugin.check_illegal_chars(self.mock_msg, "Subject",
                                                 '0.1', '0')
        self.assertFalse(result)

    def test_check_for_msn_groups_headers(self):
        received = "from mail pickup service by p23.groups.msn.com"
        message_id = "<<EMAIL>>"
        self.mock_msg.get_decoded_header.side_effect = [["<<EMAIL>>"],
                                                        [received],
                                                        [message_id]]
        result = self.plugin.check_for_msn_groups_headers(self.mock_msg)
        self.assertTrue(result)

    def test_gated_through_received_hdr_remover(self):
        mailing_list = "contact <EMAIL>; run by ezmlm"
        received = "(qmail 47240 invoked by uid 33); 01 Oct 2010 20:35:23 +0000"
        delivered_to = "mailing list <EMAIL>"
        self.mock_msg.get_decoded_header.side_effect = [[mailing_list],
                                                        [received],
                                                        [delivered_to]]
        result = self.plugin.gated_through_received_hdr_remover(self.mock_msg)
        self.assertTrue(result)

    def test_gated_thorugh_received_hdr_remover_no_received(self):
{'fraction_non_alphanumeric': 0.10633284241531664, 'fraction_numerical': 0.020078546882670596, 'mean_word_length': 2.3676640767068937, 'pattern_counts': {'":': 16, '<': 23, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 8, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 66, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7547910', 'n_tokens_mistral': 6962, 'n_tokens_neox': 6633, 'n_words': 862}