/search.css" rel="stylesheet" type="text/css"/> /search.js">
Classes | |
class | AsciiRow |
class | AsciiCSV |
class | AsciiCat |
class | DD |
Functions | |
def | validate_csv_update |
def | test_diddle |
def | test_merge |
def | test_fake_write |
def | test_compare_csv |
def | test_update_relative_to_target |
def | test_copy_cat |
def | test_fake_updates |
def | test_dump_cat |
def | test_partial_rloadcat |
Variables | |
tuple | log = logging.getLogger(__name__) |
tuple | pathx = lambda_:os.path.abspath(os.path.expanduser(os.path.expandvars(_))) |
string | catd = "~/dbicopy/tmp_offline_db" |
string | tmpd = "/tmp/tdbicopy" |
def DybPython::asciicat::validate_csv_update | ( | tn, | |
old_, | |||
new_ | |||
) |
Standalone comparison of parsed .csv :param tn: table name :param old_: path to starting csv :param new_: path to changed csv Hmm when operating decoupled the new LOCALSEQNO will
Definition at line 851 of file asciicat.py.
00852 : 00853 """ 00854 Standalone comparison of parsed .csv 00855 00856 :param tn: table name 00857 :param old_: path to starting csv 00858 :param new_: path to changed csv 00859 00860 Hmm when operating decoupled the new LOCALSEQNO will 00861 00862 """ 00863 if tn in ('LOCALSEQNO','GLOBALSEQNO',): 00864 old = AsciiCSV(None, old_) 00865 old.read() 00866 log.info( "oldcsv %r %r " % ( old, dict(old) ) ) 00867 new = AsciiCSV(None, new_) 00868 new.read() 00869 log.info( "newcsv %r %r " % ( new, dict(new) ) ) 00870 pass 00871 assert len(new) >= len(old) , "LOCALSEQNO cannot shrink " 00872 00873
def DybPython::asciicat::test_diddle | ( | ) |
Definition at line 874 of file asciicat.py.
00875 : 00876 cat = AsciiCat("~/dybaux/catalog/tmp_offline_db") 00877 diddle = False 00878 if diddle: 00879 now = datetime.utcnow() 00880 for pk,row in cat['CableMapVld'].items(): 00881 row['INSERTDATE'] = now.strftime("%Y-%m-%d %H:%M:%S") 00882 #print cat.allseqno 00883 print cat.fabseqno 00884 #cat.write( "/tmp/demo" ) 00885 #cat.diff( "/tmp/demo" ) 00886
def DybPython::asciicat::test_merge | ( | ) |
Definition at line 887 of file asciicat.py.
00888 : 00889 00890 old_ = """TABLENAME char(64),LASTUSEDSEQNO int(11),PRIMARY KEY (TABLENAME) 00891 "*",0 00892 "CableMap",440 00893 "CalibFeeSpec",113 00894 "CalibPmtSpec",29 00895 "FeeCableMap",3 00896 "HardwareID",358 00897 """ 00898 00899 new_ = """TABLENAME char(64),LASTUSEDSEQNO int(11),PRIMARY KEY (TABLENAME) 00900 "*",0 00901 "CableMap",1440 00902 "HardwareID",1358 00903 "Lettuce",100 00904 "Tomato",200 00905 "CalibFeeZiggy",300 00906 """ 00907 00908 old = AsciiCSV( None, None )( old_ ) 00909 print "old", old.lastusedseqno 00910 print "old content " + "*" * 100 00911 print old.content() 00912 00913 new = AsciiCSV( None, None )( new_ ) 00914 print "new", new.lastusedseqno 00915 print "new content " + "*" * 100 00916 print new.content() 00917 00918 mrga = old.merged( new ) 00919 print "mrga ", mrga.lastusedseqno 00920 print "mrga content " + "*" * 100 00921 print mrga.content() 00922 00923 mrgb = new.merged( old ) 00924 print "mrgb ", mrgb.lastusedseqno 00925 print "mrgb content " + "*" * 100 00926 print mrgb.content() 00927 00928 ## merge direction matters ... in the CableMap and HardwareID values 00929 00930 om = DD( old.lastusedseqno, mrga.lastusedseqno , name="om", increments=True ) 00931 print om 00932 00933
def DybPython::asciicat::test_fake_write | ( | ) |
Requires test_copy_cat run before to create sacrificial AsciiCat to diddle with #. fake write in memory #. inplace write in filesystem tmpd
Definition at line 934 of file asciicat.py.
def DybPython::asciicat::test_compare_csv | ( | ) |
Definition at line 940 of file asciicat.py.
def DybPython::asciicat::test_update_relative_to_target | ( | ) |
Definition at line 947 of file asciicat.py.
00948 : 00949 from DybPython import DB 00950 target = DB("tmp_offline_db") 00951 cat = AsciiCat("~/dbicopy/tmp_offline_db", skip_pay_check=True, allow_partial=True) 00952 tselect = [tn] 00953 00954 log.info(" cat.seqno=%r" % cat.seqno) 00955 log.info("target.seqno=%r" % target.seqno) 00956 log.info(" cat.allseqno=%r" % cat.allseqno) 00957 #log.info("target.allseqno=%r" % target.allseqno) 00958 00959 updates = cat.updates( target , tselect=tselect , fastforward=False ) 00960 print "asciicat updates relative to target %s " % updates 00961 00962 00963
def DybPython::asciicat::test_copy_cat | ( | ) |
#. scrubs any pre-existing tmpd #. copies asciicat into tmpd #. checks duplication
Definition at line 967 of file asciicat.py.
00968 : 00969 """ 00970 #. scrubs any pre-existing tmpd 00971 #. copies asciicat into tmpd 00972 #. checks duplication 00973 """ 00974 import shutil 00975 cat = AsciiCat(catd, skip_pay_check=True, allow_partial=True) 00976 if os.path.isdir(tmpd): 00977 log.info("removing %s " % tmpd ) 00978 shutil.rmtree(tmpd) 00979 pass 00980 tn = 'PhysAd' # conveniently small table 00981 cat[tn]() # reads in payload CSV 00982 cat.write(tmpd) # info file system Ascii Catalog 00983 00984 tmp = AsciiCat(tmpd, skip_pay_check=True, allow_partial=True) 00985 tmp[tn]() # reads in payload CSV 00986 assert cat == tmp
def DybPython::asciicat::test_fake_updates | ( | ) |
#. for updates call the target slot (argument 1) `cat` is required to be at/behind base `tmp` slot (argument 0)
Definition at line 1005 of file asciicat.py.
01006 : 01007 """ 01008 #. for updates call the target slot (argument 1) `cat` is required to be at/behind base `tmp` slot (argument 0) 01009 """ 01010 cat = AsciiCat(catd, skip_pay_check=True, allow_partial=True) 01011 tmp = AsciiCat(tmpd, skip_pay_check=True, allow_partial=True) 01012 tn = 'PhysAd' 01013 updates = tmp.updates( cat , tselect=[tn] , fastforward=False ) 01014 print updates
def DybPython::asciicat::test_dump_cat | ( | ) |
Definition at line 1015 of file asciicat.py.
def DybPython::asciicat::test_partial_rloadcat | ( | ) |
TODO: #. make re-testable : by controlling DB status (use fixture control?), #. currently does nothing after 1st usage as detects nothing to do
Definition at line 1024 of file asciicat.py.
01025 : 01026 """ 01027 TODO: 01028 01029 #. make re-testable : by controlling DB status (use fixture control?), 01030 #. currently does nothing after 1st usage as detects nothing to do 01031 """ 01032 tn = 'PhysAd' 01033 _tmpd = tmpd 01034 cmd = "db.py --noconfirm --ALLOW_PARTIAL -t %(tn)s tmp_offline_db rloadcat %(_tmpd)s " % locals() 01035 print cmd 01036
tuple DybPython::asciicat::log = logging.getLogger(__name__) |
Definition at line 29 of file asciicat.py.
tuple DybPython::asciicat::pathx = lambda_:os.path.abspath(os.path.expanduser(os.path.expandvars(_))) |
Definition at line 30 of file asciicat.py.
string DybPython::asciicat::catd = "~/dbicopy/tmp_offline_db" |
Definition at line 964 of file asciicat.py.
string DybPython::asciicat::tmpd = "/tmp/tdbicopy" |
Definition at line 965 of file asciicat.py.