[Zope3-checkins] CVS: zopeproducts/sqlauth/tests - __init__.py:1.1 test_sqlusers.py:1.1

[Albertas Agejevas]

Update of /cvs-repository/zopeproducts/sqlauth/tests
In directory cvs.zope.org:/tmp/cvs-serv24080/tests

Added Files:
	__init__.py test_sqlusers.py 
Log Message:
An SQL Principal Source for the Pluggable Authentication Service.
Initial checkin.


=== Added File zopeproducts/sqlauth/tests/__init__.py ===
# Not empty


=== Added File zopeproducts/sqlauth/tests/test_sqlusers.py ===
##############################################################################
#
# Copyright (c) 2003 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL).  A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""Tests for the SQLPrincipalSource

$Id: test_sqlusers.py,v 1.1 2003/07/18 13:31:16 alga Exp $
"""

import unittest
from zope.interface import implements
from zope.interface.verify import verifyObject
from zope.app.interfaces.rdb import IDBIConnection, IConnectionService, IZopeCursor
from zope.app.services.tests.placefulsetup import PlacefulSetup
from zope.app.services.servicenames import SQLDatabaseConnections
from zope.component.service import serviceManager
from zope.app.interfaces.services.pluggableauth import IUserSchemafied
from zope.exceptions import NotFoundError

class ConnectionServiceStub:

    implements(IConnectionService)

    def getConnection(self, name):
        return ConnectionStub()

class ConnectionStub:

    implements(IDBIConnection)

    def cursor(self):
        self._cursor = CursorStub()
        return self._cursor

class CursorStub:

    implements(IZopeCursor)

    description = (('id', 'string'), ('login', 'string'),
                   ('password', 'string'))
    count = 0

    def execute(self, operation, parameters=None):
        CursorStub.count += 1
        self._operation = operation
        self.result = (
            {'SELECT login, password, title, description'
             ' from "users" where id = 42':
             (('login', 'passwd', 'title', 'descr'),),

             'SELECT login, password, title, description'
             ' from "users" where id = 43':
             (('jim', '5en6G6MezRroT3XKqkdPOmY/BfQ=', # password 'secret'
               'The Pope', 'descr'),),

             'UPDATE "users" SET login = \'foo\' WHERE id = 42':
             (),

             'UPDATE "users" SET title = \'foo\' WHERE id = 42':
             (),

             'UPDATE "users" SET description = \'foo\' WHERE id = 42':
             (),

             'UPDATE "users" SET password =  \'hashashash\' WHERE id = 42':
             (),

             'UPDATE "users" SET password ='
             ' \'5en6G6MezRroT3XKqkdPOmY/BfQ=\' WHERE id = 42':
             (),

             'SELECT id FROM "users" where login LIKE \'%foo%\'':
             ((3,), (4,), (2,)),

             'SELECT login, password, title, description'
             ' from "users" where id = 2':
             (('foo2', 'passwd', 'title', 'descr'),),

             'SELECT login, password, title, description'
             ' from "users" where id = 3':
             (('foo3', 'passwd', 'title', 'descr'),),

             'SELECT login, password, title, description'
             ' from "users" where id = 4':
             (('foo4', 'passwd', 'title', 'descr'),),

             'SELECT login, password, title, description'
             ' from "users" where id = 22':
             (),

             'SELECT id FROM "users" WHERE login = \'steve\'':
             (),

             'SELECT id FROM "users" WHERE login = \'jim\'':
             ((43,),),

             'SELECT id FROM "users"':
             ((3,), (4,), (2,), (42,), (43,)),

             'SELECT COUNT(*) FROM "users"': ((5,),),

             'DELETE FROM "users" WHERE login = \'jim\'': (),

             'INSERT INTO "users" (login, password, title, description)'
             ' VALUES (\'login\', \'passwd\', \'\', \'\')':
             ()
             }[operation])


    def fetchall(self):
        return self.result

    def fetchone(self):
        return self.result[0]

class TestSQLPrincipalSource(PlacefulSetup, unittest.TestCase):

    def setUp(self):
        PlacefulSetup.setUp(self)
        serviceManager.defineService(SQLDatabaseConnections, IConnectionService)
        serviceManager.provideService(SQLDatabaseConnections,
                                      ConnectionServiceStub())

    def test_interface(self):
        from zopeproducts.sqlauth import SQLPrincipalSource
        from zopeproducts.sqlauth.interfaces import ISQLPrincipalSource

        verifyObject(ISQLPrincipalSource, SQLPrincipalSource())

    def test_getConnection(self):
        from zopeproducts.sqlauth import SQLPrincipalSource

        psrc = SQLPrincipalSource(connection="test")
        self.assert_(IDBIConnection.isImplementedBy(psrc._getConnection()))

    def test_getPrincipal(self):
        from zopeproducts.sqlauth import SQLPrincipalSource

        psrc = SQLPrincipalSource(connection="test")
        pr = psrc.getPrincipal(42)
        self.assert_(IUserSchemafied.isImplementedBy(pr))
        self.assertEqual(pr.login, 'login')
        self.assertEqual(pr.password, 'passwd')
        self.assertEqual(pr.title, 'title')
        self.assertEqual(pr.description, 'descr')

        self.assertRaises(NotFoundError, psrc.getPrincipal, 22)


    def test_getPrincipals(self):
        from zopeproducts.sqlauth import SQLPrincipalSource
        psrc = SQLPrincipalSource(connection="test")

        result = psrc.getPrincipals('foo')
        for principal in result:
            self.assert_(IUserSchemafied.isImplementedBy(principal))

        ids = [pr.id for pr in result]
        ids.sort()
        self.assertEqual(ids, [2,3,4])

    def test_getitem(self):
        from zopeproducts.sqlauth import SQLPrincipalSource
        psrc = SQLPrincipalSource(connection="test")

        principal = psrc['jim']
        self.assertEqual(principal.id, 43)
        self.assertEqual(principal.login, 'jim')

        self.assertEqual(psrc.get('steve'), None)

        self.assert_('steve' not in psrc)
        self.assert_('jim' in psrc)

    def test_items(self):
        from zopeproducts.sqlauth import SQLPrincipalSource
        psrc = SQLPrincipalSource(connection="test")

        for key, value in psrc.items():
            self.assert_(IUserSchemafied.isImplementedBy(value))
            self.assertEqual(key, value.login)

        self.assertEqual(len(psrc.items()), 5)
        self.assertEqual(len(psrc), 5)

        keys = tuple(psrc.__iter__())
        self.assertEqual(keys, ('foo3', 'foo4', 'foo2', 'login', 'jim'))
        self.assertEqual(psrc.keys(), ('foo3', 'foo4', 'foo2', 'login', 'jim'))

    def test_values(self):
        from zopeproducts.sqlauth import SQLPrincipalSource
        psrc = SQLPrincipalSource(connection="test")

        self.assertEqual(zip(psrc.keys(), psrc.values()), list(psrc.items()))

    def test_delitem(self):
        from zopeproducts.sqlauth import SQLPrincipalSource
        psrc = SQLPrincipalSource(connection="test")
        conn = ConnectionStub()
        def myGetConnection():
            return conn
        psrc._getConnection = myGetConnection

        del psrc['jim']

        self.assertEqual(conn._cursor._operation,
                         'DELETE FROM "users" WHERE login = \'jim\'')

    def test_setObject(self):
        from zopeproducts.sqlauth import SQLPrincipalSource
        from zope.app.services.pluggableauth import SimplePrincipal

        psrc = SQLPrincipalSource(connection="test")
        conn = ConnectionStub()
        def myGetConnection():
            return conn
        psrc._getConnection = myGetConnection
        user = SimplePrincipal('login', 'passwd', '', '')

        psrc.setObject('whatever', user)
        self.assertEqual(conn._cursor._operation,
                         'INSERT INTO "users" (login, password, title, description)'
                         ' VALUES (\'login\', \'passwd\', \'\', \'\')'
                        )

    def test_authenticate(self):
        from zopeproducts.sqlauth import SQLPrincipalSource
        psrc = SQLPrincipalSource(connection="test")

        principal = psrc.authenticate('jim', 'secret')

        self.assertEqual(principal.id, 43)
        self.assertEqual(principal.login, 'jim')


class TestSQLPrincipal(unittest.TestCase):

    def test_interface(self):
        from zopeproducts.sqlauth import SQLPrincipal

        verifyObject(IUserSchemafied, SQLPrincipal(ConnectionStub(), 42))

    def test_getters(self):
        from zopeproducts.sqlauth import SQLPrincipal

        principal = SQLPrincipal(ConnectionStub(), 42)
        self.assertEqual(principal.getId(), 42)
        self.assertEqual(principal.getTitle(), "title")
        self.assertEqual(principal.getDescription(), "descr")

        self.assertEqual(principal.id, 42)
        self.assertEqual(principal.title, "title")
        self.assertEqual(principal.description, "descr")
        self.assertEqual(principal.password, "passwd")
        self.assertEqual(principal.login, "login")

    def test_setProperties(self):
        from zopeproducts.sqlauth import SQLPrincipal

        conn = ConnectionStub()
        principal = SQLPrincipal(conn, 42)

        for attr in ('login', 'title', 'description'):
            setattr(principal, attr, 'foo')

            self.assertEqual(conn._cursor._operation,
                         'UPDATE "users" SET %s = \'foo\' WHERE id = 42' % attr)

    def test_validate(self):
        from zopeproducts.sqlauth import SQLPrincipal

        conn = ConnectionStub()
        principal = SQLPrincipal(conn, 42)

        principal.password = 'secret'
        self.assertEqual(conn._cursor._operation,
                         'UPDATE "users" SET password = '
                         '\'5en6G6MezRroT3XKqkdPOmY/BfQ=\' '
                         'WHERE id = 42')

        self.assert_(principal.validate('secret'))


def test_suite():
    suite = unittest.TestSuite()
    suite.addTest(unittest.makeSuite(TestSQLPrincipalSource))
    suite.addTest(unittest.makeSuite(TestSQLPrincipal))
    return suite


if __name__ == '__main__':
    unittest.main()