Changeset 10394 for main/waeup.cas/trunk/waeup/cas/tests

Timestamp:

4 Jul 2013, 09:04:41 (12 years ago)

Author:

uli

Message:

Store work from last days.

Location:

main/waeup.cas/trunk/waeup/cas/tests

Files:

• sample2.ini (modified) (1 diff)
• test_authenticators.py (added)
• test_db.py (modified) (2 diffs)
• test_server.py (modified) (5 diffs)

main/waeup.cas/trunk/waeup/cas/tests/sample2.ini

@@ -3 +3 @@
 [app:main]
 use = egg:waeup.cas#server
-# what file to use to store tickets, credentials, etc.
-db_path = /a/path/to/cas.db
+# SQLAlchemy connection string
+# By default we get an in-memory SQLite database
+# Samples:
+#   sqlite in-memory db:  sqlite://:memory:
+#   sqlite file-based db: sqlite:////path/to/db
+db = sqlite:///:memory:
+auth = dummy

main/waeup.cas/trunk/waeup/cas/tests/test_db.py

@@ -5 +5 @@
 import tempfile
 import unittest
-from waeup.cas.db import create_db
+from sqlalchemy.engine import Engine
+from waeup.cas.db import (
+    DB, LoginTicket, ServiceTicket, TicketGrantingCookie)
 
 
@@ -17 +19 @@
     def test_create_db(self):
         # we can create a database
-        db_path = os.path.join(self.workdir, 'sample-cas.db')
-        create_db(db_path)
-        assert os.path.isfile(db_path)
-        conn = sqlite3.connect(db_path)
-        assert conn is not None
-        conn.close()
+        conn_string = 'sqlite:///'
+        db = DB(conn_string)
+        assert hasattr(db, 'engine')
+        assert isinstance(db.engine, Engine)
 
     def test_create_db_exists(self):
         # an already existing db will be left untouched
         db_path = os.path.join(self.workdir, 'sample-cas.db')
-        create_db(db_path)
+        conn_string = 'sqlite:///%s' % db_path
+        assert os.path.isdir(self.workdir)
+        DB(conn_string)  # create database
         conn = sqlite3.connect(db_path)
-        cursor = conn.cursor()
-        cursor.execute('''CREATE TABLE mytest (name text, forname text)''')
-        cursor.execute('''INSERT INTO mytest VALUES ("Foo", "Bar")''')
-        conn.commit()
+        with conn:
+            conn.execute('''CREATE TABLE mytest (name text, forname text)''')
+            conn.execute('''INSERT INTO mytest VALUES ("Foo", "Bar")''')
         conn.close()
 
-        create_db(db_path)
         conn = sqlite3.connect(db_path)
-        cursor = conn.cursor()
-        cursor.execute('''SELECT * FROM mytest''')
-        assert cursor.fetchall() == [('Foo', 'Bar')]
+        with conn:
+            result = [x for x in conn.execute('''SELECT * FROM mytest''')]
+        conn.close()
+        assert result == [('Foo', 'Bar')]
+
+    def test_create_db_populates_db(self):
+        # created DBs will be populated with the required tables
+        db_path = os.path.join(self.workdir, 'sample-cas.db')
+        conn_string = 'sqlite:///%s' % db_path
+        DB(conn_string)  # creates database
+        conn = sqlite3.connect(db_path)
+        with conn:
+            result = [x[:2] for x in conn.execute(
+                '''SELECT * FROM sqlite_master ORDER BY type''')]
+        conn.close()
+        assert ('table', 'service_tickets') in result
+        assert ('table', 'login_tickets') in result
+
+
+class TicketTests(unittest.TestCase):
+
+    def setUp(self):
+        self.db = DB('sqlite:///')
+        self.db.session()
+
+    def tearDown(self):
+        self.db.session.remove()
+
+    def test_login_ticket_add(self):
+        # we can add login tickets
+        assert self.db.engine.has_table('login_tickets')
+
+        self.db.add(LoginTicket('foo'))
+        assert [x.ticket for x in self.db.query(LoginTicket)] == ['foo']
+
+    def test_login_ticket_delete(self):
+        # we can delete single login tickets
+        assert self.db.engine.has_table('login_tickets')
+
+        self.db.add(LoginTicket('foo'))
+        contents = [x for x in self.db.query(LoginTicket)]
+        assert len(contents) == 1
+        lt = contents[0]
+        self.db.delete(lt)
+        assert [x.ticket for x in self.db.query(LoginTicket)] == []
+
+    def test_login_ticket_repr(self):
+        # we can get a proper LoginTicket representation
+        # (i.e. one that can be fed to `eval`)
+        ticket = LoginTicket('foo', 12.1)
+        assert ticket.__repr__() == "LoginTicket('foo', 12.1)"
+
+    def test_login_ticket_timestamp(self):
+        # we get a timestamp stored, if none is passed to init
+        lticket = LoginTicket('foo')
+        assert isinstance(lticket.ts, float)
+
+    def test_add_service_ticket(self):
+        # we can add service tickets
+        self.db.add(ServiceTicket('foo', 'bar', 'baz', 12.1))
+        result = [(x.ticket, x.user, x.service, x.ts)
+                  for x in self.db.query(ServiceTicket)]
+        assert result == [('foo', 'bar', 'baz', 12.1)]
+
+    def test_service_ticket_repr(self):
+        # we can get a proper ServiceTicket representation
+        # (i.e. one that can be fed to `eval`)
+        sticket = ServiceTicket('foo', 'bar', 'baz', 12.1)
+        st_repr = sticket.__repr__()
+        assert st_repr == "ServiceTicket('foo', 'bar', 'baz', 12.1)"
+
+    def test_service_ticket_timestamp(self):
+        # we get a timestamp stored, if none is passed to init
+        sticket = ServiceTicket('foo', 'bar', 'baz')
+        assert isinstance(sticket.ts, float)
+
+    def test_ticket_granting_cookie_add(self):
+        # we can add ticket granting cookies
+        assert self.db.engine.has_table('ticket_granting_cookies')
+
+        self.db.add(TicketGrantingCookie('foo'))
+        assert [x.value for x in self.db.query(
+            TicketGrantingCookie)] == ['foo']
+
+    def test_ticket_granting_cookie_repr(self):
+        # we can get a proper ticket-granting cookie representation
+        # (i.e. one that can be fed to `eval`)
+        tgc = TicketGrantingCookie('foo', 12.1)
+        assert tgc.__repr__() == "TicketGrantingCookie('foo', 12.1)"
+
+    def test_ticket_granting_cookie_timestamp(self):
+        # we get a timestamp stored, if none is passed to init
+        tgc = TicketGrantingCookie('foo')
+        assert isinstance(tgc.ts, float)

main/waeup.cas/trunk/waeup/cas/tests/test_server.py

@@ -1 +1 @@
 import os
+import re
 import shutil
 import tempfile
 import unittest
 from paste.deploy import loadapp
-from webob import Request
-from waeup.cas.server import CASServer
+from webob import Request, Response
+from webtest import TestApp as WebTestApp  # avoid py.test skip message
+from waeup.cas.authenticators import DummyAuthenticator
+from waeup.cas.db import DB, LoginTicket, ServiceTicket, TicketGrantingCookie
+from waeup.cas.server import (
+    CASServer, create_service_ticket, create_login_ticket,
+    create_tgc_value, check_login_ticket, set_session_cookie,
+    )
+
+
+RE_ALPHABET = re.compile('^[a-zA-Z0-9\-]*$')
+RE_COOKIE = re.compile('^cas-tgc=[A-Za-z0-9\-]+; Path=/; secure; HttpOnly$')
 
 
@@ -19 +30 @@
         self.workdir = os.path.join(self._new_tmpdir, 'home')
         self.db_path = os.path.join(self.workdir, 'mycas.db')
+        os.mkdir(self.workdir)
         self.paste_conf1 = os.path.join(
             os.path.dirname(__file__), 'sample1.ini')
@@ -34 +46 @@
         app = loadapp('config:%s' % self.paste_conf1)
         assert isinstance(app, CASServer)
-        assert hasattr(app, 'db_path')
-        assert app.db_path is not None
+        assert hasattr(app, 'db')
+        assert isinstance(app.db, DB)
+        assert hasattr(app, 'auth')
 
     def test_paste_deploy_options(self):
@@ -41 +54 @@
         app = loadapp('config:%s' % self.paste_conf2)
         assert isinstance(app, CASServer)
-        assert app.db_path == '/a/path/to/cas.db'
+        assert app.db_connection_string == 'sqlite:///:memory:'
+        assert isinstance(app.auth, DummyAuthenticator)
 
     def test_init(self):
-        # we get a db dir set automatically
-        app = CASServer()
-        assert hasattr(app, 'db_path')
-        assert app.db_path is not None
-        assert os.path.exists(os.path.dirname(app.db_path))
+        # we get a `DB` instance created automatically
+        app = CASServer()
+        assert hasattr(app, 'db')
+        assert app.db is not None
 
     def test_init_explicit_db_path(self):
         # we can set a db_path explicitly
-        app = CASServer(db_path=self.db_path)
-        assert hasattr(app, 'db_path')
-        assert app.db_path == self.db_path
+        app = CASServer(db='sqlite:///%s' % self.db_path)
+        assert hasattr(app, 'db')
+        assert isinstance(app.db, DB)
+        assert os.path.isfile(self.db_path)
+
+    def test_get_template(self):
+        app = CASServer()
+        assert app._get_template('login.html') is not None
+        assert app._get_template('not-existent.html') is None
 
     def test_call_root(self):
@@ -93 +112 @@
         assert resp.content_type == 'text/html'
         assert b'<form ' in resp.body
+
+
+class BrowserTests(unittest.TestCase):
+
+    def setUp(self):
+        self.raw_app = CASServer(auth=DummyAuthenticator())
+        self.app = WebTestApp(self.raw_app)
+
+    def test_login(self):
+        resp = self.app.get('/login')
+        assert resp.status == '200 OK'
+        form = resp.forms[0]
+        # 2.1.3: form must be submitted by POST
+        assert form.method == 'post'
+        fieldnames = form.fields.keys()
+        # 2.1.3: form must contain: username, password, lt
+        assert 'username' in fieldnames
+        assert 'password' in fieldnames
+        assert 'lt' in fieldnames
+        assert RE_ALPHABET.match(form['lt'].value)
+
+    def test_login_no_service(self):
+        # w/o a service passed in, the form should not contain service
+        # (not a strict protocol requirement, but handy)
+        resp = self.app.get('/login')
+        assert 'service' not in resp.forms[0].fields.keys()
+
+    def test_login_service_replayed(self):
+        # 2.1.3: the login form must contain the service param sent
+        resp = self.app.get('/login?service=http%3A%2F%2Fwww.service.com')
+        form = resp.forms[0]
+        assert resp.status == '200 OK'
+        assert 'service' in form.fields.keys()
+        assert form['service'].value == 'http://www.service.com'
+
+    def test_login_cred_acceptor_valid_no_service(self):
+        # 2.2.4: successful login w/o service yields a message
+        lt = create_login_ticket()
+        self.raw_app.db.add(lt)
+        lt_string = lt.ticket
+        resp = self.app.post('/login', dict(
+            username='bird', password='bebop', lt=lt_string))
+        assert resp.status == '200 OK'
+        assert b'successful' in resp.body
+        # single-sign-on session initiated
+        assert 'Set-Cookie' in resp.headers
+        cookie = resp.headers['Set-Cookie']
+        assert cookie.startswith('cas-tgc=')
+
+    def test_login_cred_acceptor_valid_w_service(self):
+        # 2.2.4: successful login with service makes a redirect
+        # Appendix B: safe redirect
+        lt = create_login_ticket()
+        self.raw_app.db.add(lt)
+        lt_string = lt.ticket
+        resp = self.app.post('/login', dict(
+            username='bird', password='bebop', lt=lt_string,
+            service='http://example.com/Login'))
+        assert resp.status == '303 See Other'
+        assert 'Location' in resp.headers
+        assert resp.headers['Location'].startswith(
+            'http://example.com/Login?ticket=ST-')
+        assert 'Pragma' in resp.headers
+        assert resp.headers['Pragma'] == 'no-cache'
+        assert 'Cache-Control' in resp.headers
+        assert resp.headers['Cache-Control'] == 'no-store'
+        assert 'Expires' in resp.headers
+        assert resp.headers['Expires'] == 'Thu, 01 Dec 1994 16:00:00 GMT'
+        assert b'window.location.href' in resp.body
+        assert b'noscript' in resp.body
+        assert b'ticket=ST-' in resp.body
+
+
+class CASServerHelperTests(unittest.TestCase):
+
+    def setUp(self):
+        self.workdir = tempfile.mkdtemp()
+        self.db_file = os.path.join(self.workdir, 'mycas.db')
+        self.conn_string = 'sqlite:///%s' % self.db_file
+        self.db = DB(self.conn_string)
+
+    def tearDown(self):
+        shutil.rmtree(self.workdir)
+
+    def test_create_service_ticket(self):
+        # we can create service tickets
+        st = create_service_ticket(
+            user='bob', service='http://www.example.com')
+        assert isinstance(st, ServiceTicket)
+        # 3.1.1: service not part of ticket
+        assert 'example.com' not in st.ticket
+        # 3.1.1: ticket must start with 'ST-'
+        assert st.ticket.startswith('ST-')
+        # 3.1.1: min. ticket length clients must be able to process is 32
+        assert len(st.ticket) < 33
+        # 3.7: allowed character set == [a-zA-Z0-9\-]
+        assert RE_ALPHABET.match(st.ticket), (
+            'Ticket contains forbidden chars: %s' % st)
+
+    def test_create_login_ticket(self):
+        # we can create login tickets
+        lt = create_login_ticket()
+        # 3.5.1: ticket should start with 'LT-'
+        assert lt.ticket.startswith('LT-')
+        # 3.7: allowed character set == [a-zA-Z0-9\-]
+        assert RE_ALPHABET.match(lt.ticket), (
+            'Ticket contains forbidden chars: %s' % lt)
+
+    def test_create_login_ticket_unique(self):
+        # 3.5.1: login tickets are unique (although not hard to guess)
+        ticket_num = 1000  # increase to test more thoroughly
+        lt_list = [create_login_ticket() for x in range(ticket_num)]
+        assert len(set(lt_list)) == ticket_num
+
+    def test_create_tgc_value(self):
+        # we can create ticket granting cookies
+        tgc = create_tgc_value()
+        assert isinstance(tgc, TicketGrantingCookie)
+        # 3.6.1: cookie value should start with 'TGC-'
+        assert tgc.value.startswith('TGC-')
+        # 3.7: allowed character set == [a-zA-Z0-9\-]
+        assert RE_ALPHABET.match(tgc.value), (
+            'Cookie value contains forbidden chars: %s' % tgc)
+
+    def test_check_login_ticket(self):
+        db = DB('sqlite:///')
+        lt = LoginTicket('LT-123456')
+        db.add(lt)
+        assert check_login_ticket(db, None) is False
+        assert check_login_ticket(db, 'LT-123456') is True
+        # the ticket will be removed after check
+        assert check_login_ticket(db, 'LT-123456') is False
+        assert check_login_ticket(db, 'LT-654321') is False
+
+    def test_set_session_cookie(self):
+        # make sure we can add session cookies to responses
+        db = DB('sqlite:///')
+        resp = set_session_cookie(Response(), db)
+        assert 'Set-Cookie' in resp.headers
+        cookie = resp.headers['Set-Cookie']
+        assert RE_COOKIE.match(cookie), (
+            'Cookie in unexpected format: %s' % cookie)
+        # the cookie is stored in database
+        value = cookie.split('=')[1].split(';')[0]
+        q = db.query(TicketGrantingCookie).filter(
+            TicketGrantingCookie.value == value)
+        assert len(list(q)) == 1