<commit_before># -*- coding: utf-8 -*-
# ###
# Copyright (c) 2013, Rice University
# This software is subject to the provisions of the GNU Affero General
# Public License version 3 (AGPLv3).
# See LICENCE.txt for details.
# ###
import os
import unittest
import json
import socket
import tempfile
import uuid
from urlparse import parse_qs, urlparse

import transaction
from webob.multidict import MultiDict
from pyramid import testing

from .models import DBSession


here = os.path.abspath(os.path.dirname(__file__))
test_data_directory = os.path.join(here, 'test-data')
with open(os.path.join(test_data_directory, 'idents.json'), 'r') as fp:
    _TEST_IDENTS = json.load(fp)
TEST_OPENID_IDENTS = _TEST_IDENTS['openid']
TEST_GOOGLE_IDENTS = _TEST_IDENTS['google']


def _resolve_test_domain(domain, port=80):
    """Given a domain, resolve it down to it's FQDN and ip address."""
    addrs = socket.getaddrinfo(domain, port)
    addr = addrs[0][4][0]
    fqdn = socket.gethostbyaddr(addr)[0]
    return fqdn, addr


class UidDiscoverTests(unittest.TestCase):
    # Tests the utils.discover_uid function.
    # The data here is a mock of the profile data coming out of a
    #   velruse.AuthenticationComplete object.

    def make_one(self, profile):
        from velruse import AuthenticationComplete
        obj = AuthenticationComplete()
        setattr(obj, 'profile', profile)
        return obj

    def test_w_no_info(self):
        # Case where the function is supplied with little or no data.
        # I can't say this would ever happen, but if it does it will
        #   be ready for it.
        from .utils import discover_uid
        data = self.make_one({"accounts": []})
        with self.assertRaises(ValueError):
            uid = discover_uid(data)

    def test_w_general_info(self):
        from .utils import discover_uid
        expected_uid = 'http://junk.myopenid.com'
        data = self.make_one({"accounts": [{"username": expected_uid,
                                            "domain": "openid.net"}]})
        uid = discover_uid(data)
        self.assertEqual(uid, expected_uid)

    def test_w_double_username(self):
        # Case where more than one username is supplied. We should take
        #   the first when a prefered username option hasn't been specified.
        from .utils import discover_uid
        expected_uid = 'http://junk.myopenid.com'
        data = self.make_one({"accounts": [{"username": expected_uid,
                                          "domain": "openid.net"},
                                         {"username": "not-this-one",
                                          "domain": "example.com"},
                                         ]})
        uid = discover_uid(data)
        self.assertEqual(uid, expected_uid)

    def test_w_preferred_username(self):
        # Case where a preferred username has been supplied.
        from .utils import discover_uid
        expected_uid = '<EMAIL>'
        data = self.make_one({"accounts": [{"username": "not-this-one",
                                            "domain": "openid.net"},
                                           {"username": expected_uid,
                                            "domain": "example.com"},
                                           ],
                              "preferredUsername": expected_uid})
        uid = discover_uid(data)
        self.assertEqual(uid, expected_uid)


class ModelRelationshipTests(unittest.TestCase):

    def setUp(self):
        self.config = testing.setUp()
        from sqlalchemy import create_engine
        engine = create_engine('sqlite://')
        from .models import Base
        DBSession.configure(bind=engine)
        Base.metadata.create_all(engine)

    def tearDown(self):
        DBSession.remove()
        testing.tearDown()

    def test_user_to_ident_rel(self):
        firstname = "Jone"
        lastname = "Hao"
        jone_identifier = "https://jone.openid.example.com"
        hao_identifier = "https://hao.openid.example.com"
        from .models import User, Identity
        with transaction.manager:
            user = User()
            user.firstname = firstname
            user.lastname = lastname
            DBSession.add(user)
            DBSession.flush()
            jone_ident = Identity(jone_identifier, '', '', user=user)
            hao_ident = Identity(hao_identifier, '', '', user=user)
            DBSession.add_all([jone_ident, hao_ident])
        with transaction.manager:
            user = DBSession.query(User) \
                .filter(User.firstname==firstname) \
                .one()
            ident = DBSession.query(Identity) \
                .filter(Identity.identifier==jone_identifier) \
                .one()
            self.assertIn(ident, user.identities)
            ident2 = DBSession.query(Identity) \
                .filter(Identity.identifier==hao_identifier) \
                .one()
            self.assertIn(ident2, user.identities)
            self.assertEqual(ident2.user, user)


class PutUserTests(unittest.TestCase):

    def setUp(self):
        self.config = testing.setUp(settings={})
        from sqlalchemy import create_engine
        sql_connect_str = 'sqlite://'
        engine = create_engine(sql_connect_str)
        from .models import Base
        DBSession.configure(bind=engine)
        Base.metadata.create_all(engine)

    def tearDown(self):
        DBSession.remove()
        testing.tearDown()

    def test_success(self):
        # Given a good set of data in the request, verify the request
        #   goes through and the model is updated.
        # Create a blank user.
        from .models import User
        with transaction.manager:
            user = User()
            DBSession.add(user)
            DBSession.flush()
            user_id = user.id
        request = testing.DummyRequest()
        request.matchdict = {'user_id': user_id}
        data = request.json = request.json_body = {
            'email': '<EMAIL>',
            'firstname': 'Smoo',
            'lastname': 'Smoo',
            }

        from .views import put_user
        with transaction.manager:
            user = put_user(request)

        with transaction.manager:
            user = DBSession.query(User).filter(User.id==user_id).one()
            self.assertEqual(user.email, data['email'])
            fullname = "{} {}".format(data['firstname'], data['lastname'])
            self.assertEqual(user.fullname, fullname)

    def test_invalid(self):
        # Given a set of data that includes bogus data as well as fields
        #   that are immuntable by the user, ignore them.
        # We don't have anything in the schema fields that require
        #   strict validation. So we'll ignore most problems.
        from .models import User
        with transaction.manager:
            user = User()
            DBSession.add(user)
            DBSession.flush()
            user_id = user.id
        request = testing.DummyRequest()
        request.matchdict = {'user_id': user_id}
        data = request.json = request.json_body = {
            'id': "select * from users where id='1234';",
            'email': '<EMAIL>',  # valid
            'fullname': 'Smoo Smoo',
            }

        from .views import put_user
        with transaction.manager:
            user = put_user(request)
            self.assertEqual(user.id, user_id)
            self.assertEqual(user.fullname, '')

        with transaction.manager:
            user = DBSession.query(User).filter(User.id==user_id).one()
            self.assertEqual(user.email, data['email'])
            self.assertEqual(user.fullname, '')


class IdentityDeletionTests(unittest.TestCase):

    def setUp(self):
        self.config = testing.setUp(settings={})
        from sqlalchemy import create_engine
        sql_connect_str = 'sqlite://'
        engine = create_engine(sql_connect_str)
        from .models import Base
        DBSession.configure(bind=engine)
        Base.metadata.create_all(engine)

    def tearDown(self):
        DBSession.remove()
        testing.tearDown()

    def test_success(self):
        # Given a user with two or more connected identies remove one
        #   of the identies at the user's request.
        self.fail()

    def test_error_on_no_remaining_identities(self):
        # Given a user with only one connected identity, produce an error
        #   disabling the user from removing the only remaining identity.
        # Note, that the we prevent this from happening server side, but
        #   on the client side the user should be given the option to
        #   completely remove the user account.
        self.fail()


# TODO The parse_service_url function needs testing for:<commit_msg>Test for successful deletion.<commit_after># -*- coding: utf-8 -*-
# ###
# Copyright (c) 2013, Rice University
# This software is subject to the provisions of the GNU Affero General
# Public License version 3 (AGPLv3).
# See LICENCE.txt for details.
# ###
import os
import unittest
import json
import socket
import tempfile
import uuid
from urlparse import parse_qs, urlparse

import transaction
from webob.multidict import MultiDict
from pyramid import testing

from .models import DBSession


here = os.path.abspath(os.path.dirname(__file__))
test_data_directory = os.path.join(here, 'test-data')
with open(os.path.join(test_data_directory, 'idents.json'), 'r') as fp:
    _TEST_IDENTS = json.load(fp)
TEST_OPENID_IDENTS = _TEST_IDENTS['openid']
TEST_GOOGLE_IDENTS = _TEST_IDENTS['google']


def _resolve_test_domain(domain, port=80):
    """Given a domain, resolve it down to it's FQDN and ip address."""
    addrs = socket.getaddrinfo(domain, port)
    addr = addrs[0][4][0]
    fqdn = socket.gethostbyaddr(addr)[0]
    return fqdn, addr


class UidDiscoverTests(unittest.TestCase):
    # Tests the utils.discover_uid function.
    # The data here is a mock of the profile data coming out of a
    #   velruse.AuthenticationComplete object.

    def make_one(self, profile):
        from velruse import AuthenticationComplete
        obj = AuthenticationComplete()
        setattr(obj, 'profile', profile)
        return obj

    def test_w_no_info(self):
        # Case where the function is supplied with little or no data.
        # I can't say this would ever happen, but if it does it will
        #   be ready for it.
        from .utils import discover_uid
        data = self.make_one({"accounts": []})
        with self.assertRaises(ValueError):
            uid = discover_uid(data)

    def test_w_general_info(self):
        from .utils import discover_uid
        expected_uid = 'http://junk.myopenid.com'
        data = self.make_one({"accounts": [{"username": expected_uid,
                                            "domain": "openid.net"}]})
        uid = discover_uid(data)
        self.assertEqual(uid, expected_uid)

    def test_w_double_username(self):
        # Case where more than one username is supplied. We should take
        #   the first when a prefered username option hasn't been specified.
        from .utils import discover_uid
        expected_uid = 'http://junk.myopenid.com'
        data = self.make_one({"accounts": [{"username": expected_uid,
                                          "domain": "openid.net"},
                                         {"username": "not-this-one",
                                          "domain": "example.com"},
                                         ]})
        uid = discover_uid(data)
        self.assertEqual(uid, expected_uid)

    def test_w_preferred_username(self):
        # Case where a preferred username has been supplied.
        from .utils import discover_uid
        expected_uid = '<EMAIL>'
        data = self.make_one({"accounts": [{"username": "not-this-one",
                                            "domain": "openid.net"},
                                           {"username": expected_uid,
                                            "domain": "example.com"},
                                           ],
                              "preferredUsername": expected_uid})
        uid = discover_uid(data)
        self.assertEqual(uid, expected_uid)


class ModelRelationshipTests(unittest.TestCase):

    def setUp(self):
        self.config = testing.setUp()
        from sqlalchemy import create_engine
        engine = create_engine('sqlite://')
        from .models import Base
        DBSession.configure(bind=engine)
        Base.metadata.create_all(engine)

    def tearDown(self):
        DBSession.remove()
        testing.tearDown()

    def test_user_to_ident_rel(self):
        firstname = "Jone"
        lastname = "Hao"
        jone_identifier = "https://jone.openid.example.com"
        hao_identifier = "https://hao.openid.example.com"
        from .models import User, Identity
        with transaction.manager:
            user = User()
            user.firstname = firstname
            user.lastname = lastname
            DBSession.add(user)
            DBSession.flush()
            jone_ident = Identity(jone_identifier, '', '', user=user)
            hao_ident = Identity(hao_identifier, '', '', user=user)
            DBSession.add_all([jone_ident, hao_ident])
        with transaction.manager:
            user = DBSession.query(User) \
                .filter(User.firstname==firstname) \
                .one()
            ident = DBSession.query(Identity) \
                .filter(Identity.identifier==jone_identifier) \
                .one()
            self.assertIn(ident, user.identities)
            ident2 = DBSession.query(Identity) \
                .filter(Identity.identifier==hao_identifier) \
                .one()
            self.assertIn(ident2, user.identities)
            self.assertEqual(ident2.user, user)


class PutUserTests(unittest.TestCase):

    def setUp(self):
        self.config = testing.setUp(settings={})
        from sqlalchemy import create_engine
        sql_connect_str = 'sqlite://'
        engine = create_engine(sql_connect_str)
        from .models import Base
        DBSession.configure(bind=engine)
        Base.metadata.create_all(engine)

    def tearDown(self):
        DBSession.remove()
        testing.tearDown()

    def test_success(self):
        # Given a good set of data in the request, verify the request
        #   goes through and the model is updated.
        # Create a blank user.
        from .models import User
        with transaction.manager:
            user = User()
            DBSession.add(user)
            DBSession.flush()
            user_id = user.id
        request = testing.DummyRequest()
        request.matchdict = {'user_id': user_id}
        data = request.json = request.json_body = {
            'email': '<EMAIL>',
            'firstname': 'Smoo',
            'lastname': 'Smoo',
            }

        from .views import put_user
        with transaction.manager:
            user = put_user(request)

        with transaction.manager:
            user = DBSession.query(User).filter(User.id==user_id).one()
            self.assertEqual(user.email, data['email'])
            fullname = "{} {}".format(data['firstname'], data['lastname'])
            self.assertEqual(user.fullname, fullname)

    def test_invalid(self):
        # Given a set of data that includes bogus data as well as fields
        #   that are immuntable by the user, ignore them.
        # We don't have anything in the schema fields that require
        #   strict validation. So we'll ignore most problems.
        from .models import User
        with transaction.manager:
            user = User()
            DBSession.add(user)
            DBSession.flush()
            user_id = user.id
        request = testing.DummyRequest()
        request.matchdict = {'user_id': user_id}
        data = request.json = request.json_body = {
            'id': "select * from users where id='1234';",
            'email': '<EMAIL>',  # valid
            'fullname': '<NAME>',
            }

        from .views import put_user
        with transaction.manager:
            user = put_user(request)
            self.assertEqual(user.id, user_id)
            self.assertEqual(user.fullname, '')

        with transaction.manager:
            user = DBSession.query(User).filter(User.id==user_id).one()
            self.assertEqual(user.email, data['email'])
            self.assertEqual(user.fullname, '')


class IdentityDeletionTests(unittest.TestCase):

    def setUp(self):
        self.config = testing.setUp(settings={})
        from sqlalchemy import create_engine
        sql_connect_str = 'sqlite://'
        engine = create_engine(sql_connect_str)
        from .models import Base
        DBSession.configure(bind=engine)
        Base.metadata.create_all(engine)

    def tearDown(self):
        DBSession.remove()
        testing.tearDown()

    def test_success(self):
        # Given a user with two or more connected identies remove one
        #   of the identies at the user's request.
        # Create the user and connected identities.
        from .models import Identity, User
        with transaction.manager:
            # Create the user first...
            user = User()
            DBSession.add(user)
            DBSession.flush()
            user_id = user.id
        # It's easier to do this in two transactions to avoid unrealistic
        #   flushing issues.
        with transaction.manager:
            # ...now create the connected identities.
            identities = []
            for name in ('Hao', 'Jone', 'Suse',):
                identity = Identity(name, name, 'openid')
                identity.user_id = str(user_id)
                identities.append(identity)
            DBSession.add_all(identities)
            DBSession.flush()
            identity_ids = [ident.id for ident in identities]

        request = testing.DummyRequest()
        removed_identity_id = identity_ids[0]
        request.matchdict = {'user_id': user_id,
                             'identity_id': removed_identity_id,
                             }

        from .views import delete_identity
        delete_identity(request)  # No response, just a status code.

        # Does the view tell us all went well?
        self.assertEqual(request.response.status_code, 200)
        # Did it really remove the identity?
        with transaction.manager:
            identity = DBSession.query(Identity) \
                .filter(Identity.id==removed_identity_id) \
                .first()
            self.assertEqual(identity, None)

    def test_error_on_no_remaining_identities(self):
        # Given a user with only one connected identity, produce an error
        #   disabling the user from removing the only remaining identity.
        # Note, that the we prevent this from happening server side, but
        #   on the client side the user should be given the option to
        #   completely remove the user account.
        self.fail()


# TODO The parse_service_url function needs testing for:
{'fraction_non_alphanumeric': 0.096282722513089, 'fraction_numerical': 0.0023036649214659686, 'mean_word_length': 2.036724960254372, 'pattern_counts': {'":': 30, '<': 10, '<?xml version=': 0, '>': 10, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 34, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5168769', 'n_tokens_mistral': 5170, 'n_tokens_neox': 5089, 'n_words': 1375}