Index: ppdb/const.py ================================================================== --- ppdb/const.py +++ ppdb/const.py @@ -26,11 +26,11 @@ JINJA_ENV = None """ The Jinja2 templating environment. Initialised at app startup by ppd.py """ VERSION = '0.93.0' VERSION__doc = """ The PPDB code version """ -SCHEMA = '33.0' +SCHEMA = '34.0' SCHEMA__doc = """ The PPDB database schema version """ BREED_ID_CHAR = 'Z' Index: ppdb/lib/pglib.py ================================================================== --- ppdb/lib/pglib.py +++ ppdb/lib/pglib.py @@ -205,12 +205,11 @@ db_connect() curs = cherrypy.thread_data.dbconn.cursor() else: raise DBAPIError(inspect.stack()[0][3], "Connection retry failed") curs.execute("SET search_path TO %s,public;" % schema) - curs.execute("create temporary table if not exists curr_app_user (username text);") - curs.execute("insert into curr_app_user values ('nologin');") + curs.execute("set ppdb.username to 'nologin';") cherrypy.thread_data.dbconn.commit() except psycopg2.OperationalError as exc: raise DBAPIOperationalError(inspect.stack()[0][3], exc) except psycopg2.Error as exc: raise DBAPIError(inspect.stack()[0][3], exc) Index: ppdb/lib/sheep_schema.py ================================================================== --- ppdb/lib/sheep_schema.py +++ ppdb/lib/sheep_schema.py @@ -44,11 +44,11 @@ """ # Disable constant name case warnings # pylint: disable=C0103 # The schema version -schema_version = "33.0" +schema_version = "34.0" #: The table creation order - parent and look-up tables must be created and populated #: before their child tables. table_list = ["version", "audit_history", "mem_dates", "pmt_method", "mem_source", "nonren_reason", "phone_type", "role_type", "county", @@ -135,11 +135,11 @@ } else { set reason "New Record" } spi_exec "INSERT INTO audit_history VALUES (DEFAULT, '$TG_op', - LOCALTIMESTAMP(3), (SELECT username from curr_app_user), '$reason', + LOCALTIMESTAMP(3), (SELECT current_setting('ppdb.username')), '$reason', '$TG_table_name', '$NEW($1)', '$row_select_1', '{' || '$old_data' || '}', '{' || '$new_data' || '}');" return SKIP } else { if {$TG_op eq "UPDATE" || $TG_op eq "DELETE"} { @@ -167,11 +167,11 @@ set row_select_1 $OLD($2) } else { set row_select_1 "" } spi_exec "INSERT INTO audit_history VALUES (DEFAULT, '$TG_op', - LOCALTIMESTAMP(3), (SELECT username from curr_app_user), '$reason', + LOCALTIMESTAMP(3), (SELECT current_setting('ppdb.username')), '$reason', '$TG_table_name', '$OLD($1)', '$row_select_1', '{' || '$old_data' || '}', '{' || '$new_data' || '}');" return SKIP; } else { return SKIP; Index: ppdb/lib/tools.py ================================================================== --- ppdb/lib/tools.py +++ ppdb/lib/tools.py @@ -181,12 +181,11 @@ for role in login_roles: if role in method_roles: cherrypy.request.loginroles = login_roles with pglib.get_conn(): with pglib.cursor() as curs: - curs.execute("update curr_app_user set username = ? " - "where username = 'nologin';", (username,)) + curs.execute("set ppdb.username to ?;", (username,)) break else: raise util.APIError('403 Forbidden', 'Your login does not allow you to do that') else: Index: ppdb/test/__init__.py ================================================================== --- ppdb/test/__init__.py +++ ppdb/test/__init__.py @@ -76,14 +76,11 @@ pglib.initialise(dbconf, False) pglib.make_db_connection(dbparm) with pglib.get_conn(): with pglib.cursor() as curs: - curs.execute("create table if not exists curr_app_user (username text);") - curs.execute("delete from curr_app_user;") - curs.execute("insert into curr_app_user values ('tester');") - + curs.execute("set ppdb.username to 'tester';") print("\nTesting %s with PostgreSQL database %s" % (classname, dbname)) def start_ppd(test_pg_pool=False, test_db=True, refresh=False, auth=True, app='db'): """ Start PPD using the CherryPy web server @@ -246,17 +243,10 @@ def set_cp_domainbase(domainbase): """ Set domainbase to CherryPy's config """ cherrypy.config['domainbase'] = domainbase -def create_curr_app_user(): - """ Create the curr_app_user table """ - with pglib.get_conn(): - with pglib.cursor() as curs: - curs.execute("create table if not exists curr_app_user (username text);") - curs.execute("insert into curr_app_user values ('tester');") - class SelWeb(): """ Class to contain custom methods """ TEST_MEM_NO = "" Index: ppdb/test/appinit_test.py ================================================================== --- ppdb/test/appinit_test.py +++ ppdb/test/appinit_test.py @@ -66,10 +66,11 @@ self.uri_tree._cp_config['tools.sessions.secure'] = False # Add new test branch. These uris are used to get test data and delete test # resources self.uri_tree.test = testuri.Test(self.appconf) + self.uri_tree.test.audit = testuri.Audit(self.appconf) self.uri_tree.test.sheep = testuri.Sheep(self.appconf) self.uri_tree.test.transfer = testuri.Transfer(self.appconf) self.uri_tree.test.member = testuri.Member(self.appconf) self.uri_tree.test.person = testuri.Person(self.appconf) self.uri_tree.test.meminfo = testuri.MemInfo(self.appconf) Index: ppdb/test/selweb/test_logon.py ================================================================== --- ppdb/test/selweb/test_logon.py +++ ppdb/test/selweb/test_logon.py @@ -60,42 +60,70 @@ elem = self.wait_for_link_text("Logout User 'admin'") elem.click() self.wait_for_element_by_id("title") self.assert_equal(self.get_text("title"), "Please log in to continue") - def test_DA(self): - """ DA - Click the 'reset password' link -""" + def test_DA(self): # DA - Get login again as admin + self.set_input_field("uname", "admin", True) + self.set_input_field("passwd", "admin#", True) + self.click("loginbtn") + self.wait_for_element_by_id("sidenav-open-btn") + + def test_DB(self): # DB - Open the member data page + self.wait_for_element_by_id("sidenav-open-btn").click() + elem = self.wait_for_link_text("Membership").click() + elem = self.wait_for_link_text("Member Data").click() + + def test_DC(self): # DC - Make a change to the payments + self.wait_for_element_by_id("edit_pmts_btn").click() + self.set_input_field("pmt_ref", "test reference", True) + self.click("pmts_submit") + self.wait_for_visibility("payments_dialog", False) + + def test_DD(self): # DD Check that the audit record user is correct + resp = test.make_request(self.host, 'GET', "/test/audit/mem_pmts", 200) + self.assert_equal('admin', resp["data"]) + + def test_DE(self): # DE logout to return to the login page + self.click("sidenav-open-btn") + elem = self.wait_for_link_text("Logout User 'admin'") + elem.click() + self.wait_for_element_by_id("title") + self.assert_equal(self.get_text("title"), "Please log in to continue") + + def test_EA(self): + """ EA - Click the 'reset password' link -""" elem = self.wait_for_element_by_selector(".pseudo_link") self.assert_equal("Forgotten your password?", elem.get_attribute("innerHTML").lstrip()) elem.click() self.wait_for_visibility("reset_box", True) self.wait_for_element_by_id("reset_title") self.assert_equal(self.get_text("reset_title"), "Request Password Reset") - def test_DB(self): - """ DB - Test a user id with no login record """ + def test_EB(self): + """ EB - Test a user id with no login record """ self.set_input_field("username_field", "2098", True) self.click("resetpw_submit") self.wait_for_text("resetpw_msg", "User 2098 is not authorised to log in") - def test_DC(self): - """ DC - Test a missing email """ + def test_EC(self): + """ EC - Test a missing email """ self.set_input_field("username_field", '424', True) self.click("resetpw_submit") self.wait_for_text("resetpw_msg", "You must specify an email address") - def test_DD(self): - """ DD - Test a bad email address """ + def test_ED(self): + """ ED - Test a bad email address """ self.set_input_field("username_field", '424', True) self.set_input_field("mem_email_addr", "womble@localhost", True) self.click("resetpw_submit") self.wait_for_text("resetpw_msg", "email address womble@localhost is not" " registered to user " + '424') - def test_DE(self): - """ DE - Test a good user and email address """ + def test_EE(self): + """ EE - Test a good user and email address """ # Empty the local mail file #~ os.system('> /var/mail/" + self.logname') self.set_input_field("username_field", '424', True) self.set_input_field("mem_email_addr", self.logname + "@localhost", True) self.click("resetpw_submit") @@ -102,12 +130,12 @@ self.wait_for_visibility("login_box", True) self.wait_for_text("login_msg", "An email has been sent to {}@localhost with " "instructions
for resetting the password for" " user {}".format(self.logname, '424')) - def test_EA(self): - """ EA - Read the link from the email and go to it """ + def test_FA(self): + """ FA - Read the link from the email and go to it """ with open("/var/mail/" + self.logname, "r") as mail: for line in mail: #~ for line in reversed(list(mail)): ix = line.find("localhost:8") if ix != -1: @@ -115,28 +143,28 @@ url = "http:{}".format(line[ix:]) self.driver.get("http:{}".format(line[ix:])) self.wait_for_title("Shetland Sheep Society - Database - Change Password") self.assert_equal(self.get_text("title"), "Change Password for '424'") - def test_EB(self): - """ EB - Enter a poor password and ensure the correct message is displayed """ + def test_FB(self): + """ FB - Enter a poor password and ensure the correct message is displayed """ self.set_input_field("new_passwd", "", True) self.wait_for_text("score", "You must enter a new password!") self.assert_equal(self.get_class("score"), "err_msg") self.set_input_field("new_passwd", "Somborne", True) self.wait_for_text_starts_with("score", "Strength: 0. Add another word or two. U") self.assert_equal(self.get_class("score"), "err_msg") - def test_EC(self): - """ EC - Enter a good password and ensure the correct message is displayed """ + def test_FC(self): + """ FC - Enter a good password and ensure the correct message is displayed """ self.set_input_field("new_passwd", "sombornebellalamb", True) self.wait_for_text("score", "Strength: 3 - strong enough.") self.assert_equal(self.get_class("score"), "ok_msg") - def test_ED(self): - """ ED - test the conf_passwd field """ + def test_FD(self): + """ FD - test the conf_passwd field """ self.set_input_field("conf_passwd", "", True) self.wait_for_text("score", "The passwords do not match") self.assert_equal(self.get_class("score"), "err_msg") self.set_input_field("conf_passwd", "sombornebelllamb", True) @@ -145,12 +173,12 @@ self.set_input_field("conf_passwd", "sombornebellalamb", True) self.wait_for_text("score", "The passwords match") self.assert_equal(self.get_class("score"), "ok_msg") - def test_EE(self): - """ EE: Get a demo password and paste into the new_passwd and conf_passwd fields """ + def test_FE(self): + """ FE: Get a demo password and paste into the new_passwd and conf_passwd fields """ self.click("genpw_btn") self.pause() newpw = self.get_text("suggested_pw") print("suggested password: {}".format(newpw)) @@ -160,17 +188,18 @@ self.set_input_field("conf_passwd", newpw, True) self.wait_for_text("score", "The passwords match") self.assert_equal(self.get_class("score"), "ok_msg") - def test_EF(self): - """ EF - Change the password """ + def test_FF(self): + """ FF - Change the password """ self.click("link_change_btn") self.wait_for_visibility("info_dialog", True) self.check_info_dialog("Password has been changed") self.wait_for_visibility("login_box", True) + @attr('firefox', 'local', 'logon') class TestLoginPGFF(TCases): @classmethod Index: ppdb/test/testuri.py ================================================================== --- ppdb/test/testuri.py +++ ppdb/test/testuri.py @@ -52,11 +52,38 @@ """ The test URI """ def GET(self): """ A GET to the Test URI """ return "You requested some test function" + +@cherrypy.expose +class Audit(TestBase): + """ The test URI """ + + @cherrypy.tools.json_out() + def GET(self, tblname=None): + """ Return the 'app_user' from the latest audit record for the specified table """ + cherrypy.response.headers['content-type'] = 'application/json' + status = xhrc.XHR_NOERR + if not tblname: + cherrypy.response.status = 400 + return {"rcode": xhrc.XHR_ERROR, "data": "A table name must be specified"} + + with pglib.get_conn(True): + with pglib.cursor() as curs: + curs.execute("select app_user from audit_history where tablename = ? " + "order by action_tstamp desc limit 1;", (tblname,)) + row = curs.fetchone() + if not row: + cherrypy.response.status = 400 + return {"rcode": xhrc.XHR_ERROR, + "data": f"Table {tblname} has no audit records"} + + cherrypy.response.status = 200 + return {"rcode": xhrc.XHR_NOERR, "data": row[0]} + @cherrypy.expose class Sheep(TestBase): """ The Sheep test URI """ @cherrypy.tools.json_out() Index: ppdb/test/utests/test_audit.py ================================================================== --- ppdb/test/utests/test_audit.py +++ ppdb/test/utests/test_audit.py @@ -73,16 +73,10 @@ """ Clear the history table """ with pglib.get_conn(): with pglib.cursor() as curs: curs.execute("delete from audit_history;") -def set_app_user(user): - """ Create and populate the curr_app_user table """ - with pglib.get_conn(): - with pglib.cursor() as curs: - curs.execute("insert into curr_app_user values (?);", (user,)) - def drop_tmp_tables(): """ Drop the test tables """ with pglib.get_conn(): with pglib.cursor() as curs: for tblname in AUDIT_TEST_TABLES: @@ -173,13 +167,13 @@ """ AA1 - INSERT to a table with autoincrement key and set_ts trigger """ TCases.old_reason = 'test_aa1' TCases.gts_min = util.isots_now(True) with pglib.get_conn(): with pglib.cursor() as curs: - curs.execute("select * from curr_app_user;") - rows = curs.fetchall() - print(util.rows2dicts(rows)) + curs.execute("select current_setting('ppdb.username');") + row = curs.fetchone() + print(row[0]) with pglib.get_conn(): with pglib.cursor() as curs: curs.execute("insert into audit_test_a (field_1, field_2, " "field_3, change_reason, last_changed) " "values(3, 'value', 'extra', ?, '2012-10-09 00:01:00');", Index: ppdrole.py ================================================================== --- ppdrole.py +++ ppdrole.py @@ -51,14 +51,11 @@ print(f"No such role '{role}'") return with dbconn: with dbconn.cursor() as curs: - curs.execute("create table if not exists curr_app_user (username text);") - curs.execute("delete from curr_app_user;") - curs.execute("insert into curr_app_user values ('super');") - + curs.execute("set ppdb.username to 'super';") if action == 'add': action = 'added' curs.execute("insert into user_role values " "(%s, %s, 'superuser', %s);", (user, role, util.isots_now())) else: