Mods to tests; replaced diff_status = null to 'i'

test.py

@@ -1,10 +1,23 @@
 #!/usr/bin/python
 # coding=UTF-8
+'''
+To run this script from the current directory where this file resides :
+  - set the PYTHONPATH environment variable : 'export PYTHONPATH=$PWD'
+
+All files must be in the 'test' directory and end with the string '_test.py'
+'''
 
 import os
 import sys
 from subprocess import Popen, PIPE
 
+try:
+    print os.environ['PYTHONPATH']
+except(KeyError):
+    print "Missing PYTHONPATH environment variable"
+    print __doc__
+    exit()
+
 __currendir = os.path.dirname(__file__)
 tests = []
 test_directories = ["test"]
@@ -32,8 +45,8 @@
     exit(1)
 else:
     sys.stdout.write("%d/%d test passed (%d%%)\n"%(
-        len(tests)-failed, 
-        len(tests), 
+        len(tests)-failed,
+        len(tests),
         int((100.*(len(tests)-failed))/len(tests))))
 
 exit(0)

test/history_creation_test.py

@@ -21,9 +21,9 @@
     CREATE TABLE epanet.junctions (
         hid serial PRIMARY KEY,
         id varchar,
-        elevation float, 
-        base_demand_flow float, 
-        demand_pattern_id varchar, 
+        elevation float,
+        base_demand_flow float,
+        demand_pattern_id varchar,
         geom geometry('POINT',2154)
     )""")
 
@@ -55,7 +55,7 @@
 
 pcur.execute("""
     INSERT INTO epanet.pipes
-        (id, start_node, end_node, length, diameter, geom) 
+        (id, start_node, end_node, length, diameter, geom)
         VALUES
         ('0','0','1',1,2,ST_GeometryFromText('LINESTRING(1 0,0 1)',2154))""")
 
@@ -67,7 +67,7 @@
 failed = False
 try:
     versioning_base.add_branch( pg_conn_info, 'epanet', 'trunk' )
-except: 
+except:
     failed = True
 assert( failed )
 
@@ -87,13 +87,19 @@
 pcur.execute("SELECT * FROM epanet_mybranch_rev_head.pipes")
 assert( len(pcur.fetchall()) == 1 )
 
-versioning_base.add_revision_view( pg_conn_info, 'epanet', 'mybranch', 2)
-pcur.execute("SELECT * FROM epanet_mybranch_rev_2.junctions")
+##versioning_base.add_revision_view( pg_conn_info, 'epanet', 'mybranch', 2)
+##pcur.execute("SELECT * FROM epanet_mybranch_rev_2.junctions")
+##assert( len(pcur.fetchall()) == 2 )
+##pcur.execute("SELECT * FROM epanet_mybranch_rev_2.pipes")
+##assert( len(pcur.fetchall()) == 1 )
+
+select_and_where_str =  versioning_base.rev_view_str( pg_conn_info, 'epanet', 'junctions','mybranch', 2)
+#print select_and_where_str[0] + " WHERE " + select_and_where_str[1]
+pcur.execute(select_and_where_str[0] + " WHERE " + select_and_where_str[1])
 assert( len(pcur.fetchall()) == 2 )
-pcur.execute("SELECT * FROM epanet_mybranch_rev_2.pipes")
+select_and_where_str =  versioning_base.rev_view_str( pg_conn_info, 'epanet', 'pipes','mybranch', 2)
+#print select_and_where_str[0] + " WHERE " + select_and_where_str[1]
+pcur.execute(select_and_where_str[0] + " WHERE " + select_and_where_str[1])
 assert( len(pcur.fetchall()) == 1 )
 
 pcur.close()
-
-
-

test/issue485_test.py

@@ -21,9 +21,9 @@
     CREATE TABLE epanet.junctions (
         hid serial PRIMARY KEY,
         id varchar,
-        elevation float, 
-        base_demand_flow float, 
-        demand_pattern_id varchar, 
+        elevation float,
+        base_demand_flow float,
+        demand_pattern_id varchar,
         geometry geometry('POINT',2154),
         geometry_schematic geometry('POLYGON',2154)
     )""")
@@ -58,7 +58,7 @@
 
 pcur.execute("""
     INSERT INTO epanet.pipes
-        (id, start_node, end_node, length, diameter, geometry) 
+        (id, start_node, end_node, length, diameter, geometry)
         VALUES
         ('0','0','1',1,2,ST_GeometryFromText('LINESTRING(1 0,0 1)',2154))""")
 
@@ -70,7 +70,7 @@
 failed = False
 try:
     versioning_base.add_branch( pg_conn_info, 'epanet', 'trunk' )
-except: 
+except:
     failed = True
 assert( failed )
 
@@ -90,20 +90,30 @@
 pcur.execute("SELECT * FROM epanet_mybranch_rev_head.pipes")
 assert( len(pcur.fetchall()) == 1 )
 
-versioning_base.add_revision_view( pg_conn_info, 'epanet', 'mybranch', 2)
-pcur.execute("SELECT * FROM epanet_mybranch_rev_2.junctions")
+##versioning_base.add_revision_view( pg_conn_info, 'epanet', 'mybranch', 2)
+##pcur.execute("SELECT * FROM epanet_mybranch_rev_2.junctions")
+##assert( len(pcur.fetchall()) == 2 )
+##pcur.execute("SELECT * FROM epanet_mybranch_rev_2.pipes")
+##assert( len(pcur.fetchall()) == 1 )
+
+select_and_where_str =  versioning_base.rev_view_str( pg_conn_info, 'epanet', 'junctions','mybranch', 2)
+print select_and_where_str[0] + " WHERE " + select_and_where_str[1]
+pcur.execute(select_and_where_str[0] + " WHERE " + select_and_where_str[1])
 assert( len(pcur.fetchall()) == 2 )
-pcur.execute("SELECT * FROM epanet_mybranch_rev_2.pipes")
+select_and_where_str =  versioning_base.rev_view_str( pg_conn_info, 'epanet', 'pipes','mybranch', 2)
+print select_and_where_str[0] + " WHERE " + select_and_where_str[1]
+pcur.execute(select_and_where_str[0] + " WHERE " + select_and_where_str[1])
 assert( len(pcur.fetchall()) == 1 )
 
-pcur.execute("SELECT ST_AsText(geometry), ST_AsText(geometry_schematic) FROM epanet_mybranch_rev_2.junctions")
+##pcur.execute("SELECT ST_AsText(geometry), ST_AsText(geometry_schematic) FROM epanet_mybranch_rev_2.junctions")
+pcur.execute("SELECT ST_AsText(geometry), ST_AsText(geometry_schematic) FROM epanet.junctions")
 res = pcur.fetchall()
 assert( res[0][0] == 'POINT(0 0)' )
 assert( res[1][1] == 'POLYGON((0 0,2 0,2 2,0 2,0 0))' )
 
 
 wc = tmp_dir+'/wc_multiple_geometry_test.sqlite'
-if os.path.isfile(wc): os.remove(wc) 
+if os.path.isfile(wc): os.remove(wc)
 versioning_base.checkout( pg_conn_info, ['epanet_trunk_rev_head.pipes','epanet_trunk_rev_head.junctions'], wc )
 
 
@@ -130,7 +140,7 @@
 pcur.close()
 
 # edit a little with a new wc
-os.remove(wc) 
+os.remove(wc)
 versioning_base.checkout( pg_conn_info, ['epanet_b1_rev_head.junctions'], wc )
 
 scur = versioning_base.Db( dbapi2.connect(wc) )

test/multiple_geometry_test.py

@@ -21,9 +21,9 @@
     CREATE TABLE epanet.junctions (
         hid serial PRIMARY KEY,
         id varchar,
-        elevation float, 
-        base_demand_flow float, 
-        demand_pattern_id varchar, 
+        elevation float,
+        base_demand_flow float,
+        demand_pattern_id varchar,
         geometry geometry('POINT',2154),
         geometry_schematic geometry('POLYGON',2154)
     )""")
@@ -58,7 +58,7 @@
 
 pcur.execute("""
     INSERT INTO epanet.pipes
-        (id, start_node, end_node, length, diameter, geometry) 
+        (id, start_node, end_node, length, diameter, geometry)
         VALUES
         ('0','0','1',1,2,ST_GeometryFromText('LINESTRING(1 0,0 1)',2154))""")
 
@@ -70,7 +70,7 @@
 failed = False
 try:
     versioning_base.add_branch( pg_conn_info, 'epanet', 'trunk' )
-except: 
+except:
     failed = True
 assert( failed )
 
@@ -90,20 +90,30 @@
 pcur.execute("SELECT * FROM epanet_mybranch_rev_head.pipes")
 assert( len(pcur.fetchall()) == 1 )
 
-versioning_base.add_revision_view( pg_conn_info, 'epanet', 'mybranch', 2)
-pcur.execute("SELECT * FROM epanet_mybranch_rev_2.junctions")
+##versioning_base.add_revision_view( pg_conn_info, 'epanet', 'mybranch', 2)
+##pcur.execute("SELECT * FROM epanet_mybranch_rev_2.junctions")
+##assert( len(pcur.fetchall()) == 2 )
+##pcur.execute("SELECT * FROM epanet_mybranch_rev_2.pipes")
+##assert( len(pcur.fetchall()) == 1 )
+
+select_and_where_str =  versioning_base.rev_view_str( pg_conn_info, 'epanet', 'junctions','mybranch', 2)
+#print select_and_where_str[0] + " WHERE " + select_and_where_str[1]
+pcur.execute(select_and_where_str[0] + " WHERE " + select_and_where_str[1])
 assert( len(pcur.fetchall()) == 2 )
-pcur.execute("SELECT * FROM epanet_mybranch_rev_2.pipes")
+select_and_where_str =  versioning_base.rev_view_str( pg_conn_info, 'epanet', 'pipes','mybranch', 2)
+#print select_and_where_str[0] + " WHERE " + select_and_where_str[1]
+pcur.execute(select_and_where_str[0] + " WHERE " + select_and_where_str[1])
 assert( len(pcur.fetchall()) == 1 )
 
-pcur.execute("SELECT ST_AsText(geometry), ST_AsText(geometry_schematic) FROM epanet_mybranch_rev_2.junctions")
+##pcur.execute("SELECT ST_AsText(geometry), ST_AsText(geometry_schematic) FROM epanet_mybranch_rev_2.junctions")
+pcur.execute("SELECT ST_AsText(geometry), ST_AsText(geometry_schematic) FROM epanet.junctions")
 res = pcur.fetchall()
 assert( res[0][0] == 'POINT(0 0)' )
 assert( res[1][1] == 'POLYGON((0 0,2 0,2 2,0 2,0 0))' )
 
 
 wc = tmp_dir+'/wc_multiple_geometry_test.sqlite'
-if os.path.isfile(wc): os.remove(wc) 
+if os.path.isfile(wc): os.remove(wc)
 versioning_base.checkout( pg_conn_info, ['epanet_trunk_rev_head.pipes','epanet_trunk_rev_head.junctions'], wc )
 
 
@@ -119,5 +129,3 @@
 assert( res[0][0] == 'POINT(3 3)' )
 assert( res[1][1] == 'POLYGON((0 0,2 0,2 2,0 2,0 0))' )
 pcur.close()
-
-

test/versioning_base_test.py

@@ -9,8 +9,12 @@
 
 sqlite_test_filename1 = tmp_dir+"/versioning_base_test1.sqlite"
 sqlite_test_filename2 = tmp_dir+"/versioning_base_test2.sqlite"
+sqlite_test_filename3 = tmp_dir+"/versioning_base_test3.sqlite"
+sqlite_test_filename4 = tmp_dir+"/versioning_base_test4.sqlite"
 if os.path.isfile(sqlite_test_filename1): os.remove(sqlite_test_filename1)
 if os.path.isfile(sqlite_test_filename2): os.remove(sqlite_test_filename2)
+if os.path.isfile(sqlite_test_filename3): os.remove(sqlite_test_filename3)
+if os.path.isfile(sqlite_test_filename4): os.remove(sqlite_test_filename4)
 
 # create the test database
 
@@ -22,22 +26,22 @@
 # chechout two tables
 
 try:
-    versioning_base.checkout("dbname=epanet_test_db",["epanet_trunk_rev_head.junctions","epanet.pipes"], sqlite_test_filename1) 
+    versioning_base.checkout("dbname=epanet_test_db",["epanet_trunk_rev_head.junctions","epanet.pipes"], sqlite_test_filename1)
     assert(False and "checkout from schema withouti suffix _branch_rev_head should not be successfull")
 except RuntimeError:
     pass
 
 assert( not os.path.isfile(sqlite_test_filename1) and "sqlite file must not exist at this point" )
-versioning_base.checkout("dbname=epanet_test_db",["epanet_trunk_rev_head.junctions","epanet_trunk_rev_head.pipes"], sqlite_test_filename1) 
+versioning_base.checkout("dbname=epanet_test_db",["epanet_trunk_rev_head.junctions","epanet_trunk_rev_head.pipes"], sqlite_test_filename1)
 assert( os.path.isfile(sqlite_test_filename1) and "sqlite file must exist at this point" )
 
 try:
-    versioning_base.checkout("dbname=epanet_test_db",["epanet_trunk_rev_head.junctions","epanet_trunk_rev_head.pipes"], sqlite_test_filename1) 
+    versioning_base.checkout("dbname=epanet_test_db",["epanet_trunk_rev_head.junctions","epanet_trunk_rev_head.pipes"], sqlite_test_filename1)
     assert(False and "trying to checkout on an existing file must fail")
 except RuntimeError:
     pass
 
-# edit one table and commit changes
+# edit one table and commit changes; rev = 2
 
 scon = dbapi2.connect(sqlite_test_filename1)
 scur = scon.cursor()
@@ -46,13 +50,78 @@
 scur.execute("SELECT COUNT(*) FROM junctions")
 assert( scur.fetchone()[0] == 3 )
 scon.close()
-versioning_base.commit(sqlite_test_filename1, 'first test commit', "dbname=epanet_test_db")
+versioning_base.commit(sqlite_test_filename1, 'first edit commit', "dbname=epanet_test_db")
 pcon = psycopg2.connect("dbname=epanet_test_db")
 pcur = pcon.cursor()
 pcur.execute("SELECT COUNT(*) FROM epanet.junctions")
 assert( pcur.fetchone()[0] == 3 )
 pcur.execute("SELECT COUNT(*) FROM epanet.revisions")
 assert( pcur.fetchone()[0] == 2 )
 
+# add revision : edit one table and commit changes; rev = 3
 
+versioning_base.checkout("dbname=epanet_test_db",["epanet_trunk_rev_head.junctions"], sqlite_test_filename2)
 
+scon = dbapi2.connect(sqlite_test_filename2)
+scur = scon.cursor()
+scur.execute("UPDATE junctions_view SET elevation = '22' WHERE id = '1'")
+scon.commit()
+#scur.execute("SELECT COUNT(*) FROM junctions")
+#assert( scur.fetchone()[0] == 3 )
+scon.close()
+versioning_base.commit(sqlite_test_filename2, 'second edit commit', "dbname=epanet_test_db")
+
+# add revision : insert one junction and commit changes; rev = 4
+
+versioning_base.checkout("dbname=epanet_test_db",["epanet_trunk_rev_head.junctions"], sqlite_test_filename3)
+
+scon = dbapi2.connect(sqlite_test_filename3)
+scur = scon.cursor()
+scur.execute("INSERT INTO junctions_view(id, elevation, GEOMETRY) VALUES ('10','100',GeomFromText('POINT(2 0)',2154))")
+scon.commit()
+#scur.execute("SELECT COUNT(*) FROM junctions")
+#assert( scur.fetchone()[0] == 3 )
+scon.close()
+versioning_base.commit(sqlite_test_filename3, 'insert commit', "dbname=epanet_test_db")
+
+# add revision : delete one junction and commit changes; rev = 5
+
+versioning_base.checkout("dbname=epanet_test_db",["epanet_trunk_rev_head.junctions"], sqlite_test_filename4)
+
+scon = dbapi2.connect(sqlite_test_filename4)
+scur = scon.cursor()
+scur.execute("DELETE FROM junctions_view  WHERE id = 0")
+scon.commit()
+#scur.execute("SELECT COUNT(*) FROM junctions")
+#assert( scur.fetchone()[0] == 3 )
+scon.close()
+versioning_base.commit(sqlite_test_filename4, 'delete id=0 commit', "dbname=epanet_test_db")
+
+pg_conn_info = "dbname=epanet_test_db"
+select_str = versioning_base.diff_rev_view_str(pg_conn_info, 'epanet', 'junctions','trunk', 1,2)
+#print "select_str = " + select_str
+pcur.execute(select_str)
+res = pcur.fetchall()
+print "fetchall 1 vs 2 = " + str(res)
+
+select_and_where_str =  versioning_base.rev_view_str(pg_conn_info, 'epanet', 'junctions','trunk', 1)
+select_str = select_and_where_str[0]
+where_str = select_and_where_str[1]
+
+select_str = versioning_base.diff_rev_view_str(pg_conn_info, 'epanet', 'junctions','trunk', 1,3)
+#print "select_str = " + select_str
+pcur.execute(select_str)
+res = pcur.fetchall()
+print "fetchall 1 vs 3 = " + str(res)
+
+select_str = versioning_base.diff_rev_view_str(pg_conn_info, 'epanet', 'junctions','trunk', 1,4)
+print "select_str 1 vs 4 = " + select_str
+pcur.execute(select_str)
+res = pcur.fetchall()
+print "fetchall 1 vs 4 = " + str(res)
+
+select_str = versioning_base.diff_rev_view_str(pg_conn_info, 'epanet', 'junctions','trunk', 1,5)
+print "select_str 1 vs 5 = " + select_str
+pcur.execute(select_str)
+res = pcur.fetchall()
+print "fetchall 1 vs 5 = " + str(res)