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

Timestamp:

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

Author:

uli

Message:

Store work from last days.

File:

• main/waeup.cas/trunk/waeup/cas/tests/test_db.py (modified) (2 diffs)

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)