<commit_before>"""
    test_integration.py

    Copyright (c) 2013-2014 Snowplow Analytics Ltd. All rights reserved.

    This program is licensed to you under the Apache License Version 2.0,
    and you may not use this file except in compliance with the Apache License
    Version 2.0. You may obtain a copy of the Apache License Version 2.0 at
    http://www.apache.org/licenses/LICENSE-2.0.

    Unless required by applicable law or agreed to in writing,
    software distributed under the Apache License Version 2.0 is distributed on
    an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
    express or implied. See the Apache License Version 2.0 for the specific
    language governing permissions and limitations there under.

    Authors: <NAME>, <NAME>, <NAME>
    Copyright: Copyright (c) 2013-2014 Snowplow Analytics Ltd
    License: Apache License Version 2.0
"""

import unittest
import time
import re
import redis
import json
import base64
from snowplow_tracker import tracker, _version, emitters, subject
from httmock import all_requests, HTTMock

try:
    from urllib.parse import unquote_plus  # Python 3

except ImportError:
    from urllib import unquote_plus        # Python 2

querystrings = [""]

default_emitter = emitters.Emitter("localhost", protocol="http", port=80)

default_subject = subject.Subject()

def from_querystring(field, url):
    pattern = re.compile("^[^#]*[?&]" + field + "=([^&#]*)")
    match = pattern.match(url)
    if match:
        return match.groups()[0]

@all_requests
def pass_response_content(url, request):
    querystrings.append(request.url)
    return {
        "url": request.url,
        "status_code": 200
    }

@all_requests
def fail_response_content(url, request):
    return {
        "url": request.url,
        "status_code": 501
    }


class IntegrationTest(unittest.TestCase):

    def test_integration_page_view(self):
        t = tracker.Tracker(default_emitter, default_subject)
        with HTTMock(pass_response_content):
            t.track_page_view("http://savethearctic.org", "Save The Arctic", "http://referrer.com")
        expected_fields = {"e": "pv", "page": "Save+The+Arctic", "url": "http%3A%2F%2Fsavethearctic.org", "refr": "http%3A%2F%2Freferrer.com"}
        for key in expected_fields:
            self.assertEquals(from_querystring(key, querystrings[-1]), expected_fields[key])            

    def test_integration_ecommerce_transaction_item(self):
        t = tracker.Tracker(default_emitter, default_subject)
        with HTTMock(pass_response_content):
            t.track_ecommerce_transaction_item("12345", "pbz0025", 7.99, 2, "black-tarot", "tarot", currency="GBP")
        expected_fields = {"ti_ca": "tarot", "ti_id": "12345", "ti_qu": "2", "ti_sk": "pbz0025", "e": "ti", "ti_nm": "black-tarot", "ti_pr": "7.99", "ti_cu": "GBP"}
        for key in expected_fields:
            self.assertEquals(from_querystring(key, querystrings[-1]), expected_fields[key])

    def test_integration_ecommerce_transaction(self):
        t = tracker.Tracker(default_emitter, default_subject)
        with HTTMock(pass_response_content):
            t.track_ecommerce_transaction("6a8078be", 35, city="London", currency="GBP", items=
                [{  
                    "sku": "pbz0026",
                    "price": 20,
                    "quantity": 1
                },
                {
                    "sku": "pbz0038",
                    "price": 15,
                    "quantity": 1  
                }])

        expected_fields = {"e": "tr", "tr_id": "6a8078be", "tr_tt": "35", "tr_ci": "London", "tr_cu": "GBP"}
        for key in expected_fields:
            self.assertEquals(from_querystring(key, querystrings[-3]), expected_fields[key])

        expected_fields = {"e": "ti",  "ti_id": "6a8078be", "ti_sk": "pbz0026", "ti_pr": "20", "ti_cu": "GBP"}
        for key in expected_fields:
            self.assertEquals(from_querystring(key, querystrings[-2]), expected_fields[key])

        expected_fields = {"e": "ti",  "ti_id": "6a8078be", "ti_sk": "pbz0038", "ti_pr": "15", "ti_cu": "GBP"}
        for key in expected_fields:
            self.assertEquals(from_querystring(key, querystrings[-1]), expected_fields[key])

        self.assertEquals(from_querystring("dtm", querystrings[-3]), from_querystring("dtm", querystrings[-2]))

    def test_integration_screen_view(self):
        t = tracker.Tracker(default_emitter, default_subject)
        with HTTMock(pass_response_content):
            t.track_screen_view("Game HUD 2", "Hello!")
        expected_fields = {"e": "ue"}
        for key in expected_fields:          
            self.assertEquals(from_querystring(key, querystrings[-1]), expected_fields[key])

    def test_integration_struct_event(self):
        t = tracker.Tracker(default_emitter, default_subject)
        with HTTMock(pass_response_content):
            t.track_struct_event("Ecomm", "add-to-basket", "dog-skateboarding-video", "hd", 13.99)
        expected_fields = {"se_ca": "Ecomm", "se_pr": "hd", "se_la": "dog-skateboarding-video", "se_va": "13.99", "se_ac": "add-to-basket", "e": "se"}
        for key in expected_fields:
            self.assertEquals(from_querystring(key, querystrings[-1]), expected_fields[key])

    def test_integration_unstruct_event_non_base64(self):
        t = tracker.Tracker(default_emitter, default_subject, encode_base64=False)
        with HTTMock(pass_response_content):
            t.track_unstruct_event({"schema": "iglu://com.acme/viewed_product/jsonschema/2-0-2", "data": {"product_id": "ASO01043", "price$flt": 49.95, "walrus$tms": 1000}})
        expected_fields = {"e": "ue"}
        for key in expected_fields:
            self.assertEquals(from_querystring(key, querystrings[-1]), expected_fields[key])
        envelope_string = from_querystring("ue_pr", querystrings[-1])
        envelope = json.loads(unquote_plus(envelope_string))
        self.assertEquals(envelope, {
            "schema": "iglu://com.snowplowanalytics/unstruct_event/jsonschema/1-0-0",
            "data": {"schema": "iglu://com.acme/viewed_product/jsonschema/2-0-2", "data": {"product_id": "ASO01043", "price$flt": 49.95, "walrus$tms": 1000}}
        })

    def test_integration_unstruct_event_base64(self):
        t = tracker.Tracker(default_emitter, default_subject, encode_base64=True)
        with HTTMock(pass_response_content):
            t.track_unstruct_event({"schema": "iglu://com.acme/viewed_product/jsonschema/2-0-2", "data": {"product_id": "ASO01043", "price$flt": 49.95, "walrus$tms": 1000}})
        expected_fields = {"e": "ue"}
        for key in expected_fields:
            self.assertEquals(from_querystring(key, querystrings[-1]), expected_fields[key])
        envelope_string = from_querystring("ue_px", querystrings[-1])
        envelope = json.loads((base64.urlsafe_b64decode(bytearray(envelope_string, "utf-8"))).decode("utf-8"))
        self.assertEquals(envelope, {
            "schema": "iglu://com.snowplowanalytics/unstruct_event/jsonschema/1-0-0",
            "data": {"schema": "iglu://com.acme/viewed_product/jsonschema/2-0-2", "data": {"product_id": "ASO01043", "price$flt": 49.95, "walrus$tms": 1000}}
        })

    def test_integration_context_non_base64(self):
        t = tracker.Tracker(default_emitter, default_subject, encode_base64=False)
        with HTTMock(pass_response_content):
            t.track_page_view("localhost", "local host", None, [{"schema": "iglu://com.example/user/jsonschema/2-0-3", "data": {"user_type": "tester"}}])
        envelope_string = from_querystring("co", querystrings[-1])
        envelope = json.loads(unquote_plus(envelope_string))
        self.assertEquals(envelope, {
            "schema": "iglu://com.snowplowanalytics/contexts/jsonschema/1-0-0",
            "data":[{"schema": "iglu://com.example/user/jsonschema/2-0-3", "data": {"user_type": "tester"}}]
        })

    def test_integration_context_base64(self):
        t = tracker.Tracker(default_emitter, default_subject, encode_base64=True)
        with HTTMock(pass_response_content):
            t.track_page_view("localhost", "local host", None, [{"schema": "iglu://com.example/user/jsonschema/2-0-3", "data": {"user_type": "tester"}}])
        envelope_string = from_querystring("cx", querystrings[-1])
        envelope = json.loads((base64.urlsafe_b64decode(bytearray(envelope_string, "utf-8"))).decode("utf-8"))
        self.assertEquals(envelope, {
            "schema": "iglu://com.snowplowanalytics/contexts/jsonschema/1-0-0",
            "data":[{"schema": "iglu://com.example/user/jsonschema/2-0-3", "data": {"user_type": "tester"}}]
        })
<commit_msg>Fixed base-64 tests by decoding "=" signs<commit_after>"""
    test_integration.py

    Copyright (c) 2013-2014 Snowplow Analytics Ltd. All rights reserved.

    This program is licensed to you under the Apache License Version 2.0,
    and you may not use this file except in compliance with the Apache License
    Version 2.0. You may obtain a copy of the Apache License Version 2.0 at
    http://www.apache.org/licenses/LICENSE-2.0.

    Unless required by applicable law or agreed to in writing,
    software distributed under the Apache License Version 2.0 is distributed on
    an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
    express or implied. See the Apache License Version 2.0 for the specific
    language governing permissions and limitations there under.

    Authors: <NAME>, <NAME>, <NAME>
    Copyright: Copyright (c) 2013-2014 Snowplow Analytics Ltd
    License: Apache License Version 2.0
"""

import unittest
import time
import re
import redis
import json
import base64
from snowplow_tracker import tracker, _version, emitters, subject
from httmock import all_requests, HTTMock

try:
    from urllib.parse import unquote_plus  # Python 3

except ImportError:
    from urllib import unquote_plus        # Python 2

querystrings = [""]

default_emitter = emitters.Emitter("localhost", protocol="http", port=80)

default_subject = subject.Subject()

def from_querystring(field, url):
    pattern = re.compile("^[^#]*[?&]" + field + "=([^&#]*)")
    match = pattern.match(url)
    if match:
        return match.groups()[0]

@all_requests
def pass_response_content(url, request):
    querystrings.append(request.url)
    return {
        "url": request.url,
        "status_code": 200
    }

@all_requests
def fail_response_content(url, request):
    return {
        "url": request.url,
        "status_code": 501
    }


class IntegrationTest(unittest.TestCase):

    def test_integration_page_view(self):
        t = tracker.Tracker(default_emitter, default_subject)
        with HTTMock(pass_response_content):
            t.track_page_view("http://savethearctic.org", "Save The Arctic", "http://referrer.com")
        expected_fields = {"e": "pv", "page": "Save+The+Arctic", "url": "http%3A%2F%2Fsavethearctic.org", "refr": "http%3A%2F%2Freferrer.com"}
        for key in expected_fields:
            self.assertEquals(from_querystring(key, querystrings[-1]), expected_fields[key])            

    def test_integration_ecommerce_transaction_item(self):
        t = tracker.Tracker(default_emitter, default_subject)
        with HTTMock(pass_response_content):
            t.track_ecommerce_transaction_item("12345", "pbz0025", 7.99, 2, "black-tarot", "tarot", currency="GBP")
        expected_fields = {"ti_ca": "tarot", "ti_id": "12345", "ti_qu": "2", "ti_sk": "pbz0025", "e": "ti", "ti_nm": "black-tarot", "ti_pr": "7.99", "ti_cu": "GBP"}
        for key in expected_fields:
            self.assertEquals(from_querystring(key, querystrings[-1]), expected_fields[key])

    def test_integration_ecommerce_transaction(self):
        t = tracker.Tracker(default_emitter, default_subject)
        with HTTMock(pass_response_content):
            t.track_ecommerce_transaction("6a8078be", 35, city="London", currency="GBP", items=
                [{  
                    "sku": "pbz0026",
                    "price": 20,
                    "quantity": 1
                },
                {
                    "sku": "pbz0038",
                    "price": 15,
                    "quantity": 1  
                }])

        expected_fields = {"e": "tr", "tr_id": "6a8078be", "tr_tt": "35", "tr_ci": "London", "tr_cu": "GBP"}
        for key in expected_fields:
            self.assertEquals(from_querystring(key, querystrings[-3]), expected_fields[key])

        expected_fields = {"e": "ti",  "ti_id": "6a8078be", "ti_sk": "pbz0026", "ti_pr": "20", "ti_cu": "GBP"}
        for key in expected_fields:
            self.assertEquals(from_querystring(key, querystrings[-2]), expected_fields[key])

        expected_fields = {"e": "ti",  "ti_id": "6a8078be", "ti_sk": "pbz0038", "ti_pr": "15", "ti_cu": "GBP"}
        for key in expected_fields:
            self.assertEquals(from_querystring(key, querystrings[-1]), expected_fields[key])

        self.assertEquals(from_querystring("dtm", querystrings[-3]), from_querystring("dtm", querystrings[-2]))

    def test_integration_screen_view(self):
        t = tracker.Tracker(default_emitter, default_subject)
        with HTTMock(pass_response_content):
            t.track_screen_view("Game HUD 2", "Hello!")
        expected_fields = {"e": "ue"}
        for key in expected_fields:          
            self.assertEquals(from_querystring(key, querystrings[-1]), expected_fields[key])

    def test_integration_struct_event(self):
        t = tracker.Tracker(default_emitter, default_subject)
        with HTTMock(pass_response_content):
            t.track_struct_event("Ecomm", "add-to-basket", "dog-skateboarding-video", "hd", 13.99)
        expected_fields = {"se_ca": "Ecomm", "se_pr": "hd", "se_la": "dog-skateboarding-video", "se_va": "13.99", "se_ac": "add-to-basket", "e": "se"}
        for key in expected_fields:
            self.assertEquals(from_querystring(key, querystrings[-1]), expected_fields[key])

    def test_integration_unstruct_event_non_base64(self):
        t = tracker.Tracker(default_emitter, default_subject, encode_base64=False)
        with HTTMock(pass_response_content):
            t.track_unstruct_event({"schema": "iglu://com.acme/viewed_product/jsonschema/2-0-2", "data": {"product_id": "ASO01043", "price$flt": 49.95, "walrus$tms": 1000}})
        expected_fields = {"e": "ue"}
        for key in expected_fields:
            self.assertEquals(from_querystring(key, querystrings[-1]), expected_fields[key])
        envelope_string = from_querystring("ue_pr", querystrings[-1])
        envelope = json.loads(unquote_plus(envelope_string))
        self.assertEquals(envelope, {
            "schema": "iglu://com.snowplowanalytics/unstruct_event/jsonschema/1-0-0",
            "data": {"schema": "iglu://com.acme/viewed_product/jsonschema/2-0-2", "data": {"product_id": "ASO01043", "price$flt": 49.95, "walrus$tms": 1000}}
        })

    def test_integration_unstruct_event_base64(self):
        t = tracker.Tracker(default_emitter, default_subject, encode_base64=True)
        with HTTMock(pass_response_content):
            t.track_unstruct_event({"schema": "iglu://com.acme/viewed_product/jsonschema/2-0-2", "data": {"product_id": "ASO01043", "price$flt": 49.95, "walrus$tms": 1000}})
        expected_fields = {"e": "ue"}
        for key in expected_fields:
            self.assertEquals(from_querystring(key, querystrings[-1]), expected_fields[key])
        envelope_string = unquote_plus(from_querystring("ue_px", querystrings[-1]))
        envelope = json.loads((base64.urlsafe_b64decode(bytearray(envelope_string, "utf-8"))).decode("utf-8"))
        self.assertEquals(envelope, {
            "schema": "iglu://com.snowplowanalytics/unstruct_event/jsonschema/1-0-0",
            "data": {"schema": "iglu://com.acme/viewed_product/jsonschema/2-0-2", "data": {"product_id": "ASO01043", "price$flt": 49.95, "walrus$tms": 1000}}
        })

    def test_integration_context_non_base64(self):
        t = tracker.Tracker(default_emitter, default_subject, encode_base64=False)
        with HTTMock(pass_response_content):
            t.track_page_view("localhost", "local host", None, [{"schema": "iglu://com.example/user/jsonschema/2-0-3", "data": {"user_type": "tester"}}])
        envelope_string = from_querystring("co", querystrings[-1])
        envelope = json.loads(unquote_plus(envelope_string))
        self.assertEquals(envelope, {
            "schema": "iglu://com.snowplowanalytics/contexts/jsonschema/1-0-0",
            "data":[{"schema": "iglu://com.example/user/jsonschema/2-0-3", "data": {"user_type": "tester"}}]
        })
    
    def test_integration_context_base64(self):
        t = tracker.Tracker(default_emitter, default_subject, encode_base64=True)
        with HTTMock(pass_response_content):
            t.track_page_view("localhost", "local host", None, [{"schema": "iglu://com.example/user/jsonschema/2-0-3", "data": {"user_type": "tester"}}])
        envelope_string = unquote_plus(from_querystring("cx", querystrings[-1]))
        envelope = json.loads((base64.urlsafe_b64decode(bytearray(envelope_string, "utf-8"))).decode("utf-8"))
        self.assertEquals(envelope, {
            "schema": "iglu://com.snowplowanalytics/contexts/jsonschema/1-0-0",
            "data":[{"schema": "iglu://com.example/user/jsonschema/2-0-3", "data": {"user_type": "tester"}}]
        })
    
{'fraction_non_alphanumeric': 0.14493917833371586, 'fraction_numerical': 0.030525591002983705, 'mean_word_length': 3.390176322418136, 'pattern_counts': {'":': 172, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, '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': '6982699', 'n_tokens_mistral': 5972, 'n_tokens_neox': 5542, 'n_words': 1109}