tests/teststore.py

   1 #!/usr/bin/env python2.5
   2 from basetest import BaseTest
   3 import sys, tempfile, os
   4 from StringIO import StringIO
   5 import unittest
   6
   7 sys.path.insert(0, '..')
   8
   9 from zeroinstall.zerostore import Store, manifest, BadDigest, cli
  10 from zeroinstall import SafeException, support
  11
  12 class TestStore(BaseTest):
  13         def setUp(self):
  14                 BaseTest.setUp(self)
  15
  16                 self.store_parent = tempfile.mktemp()
  17                 os.mkdir(self.store_parent, 0700)
  18                 self.store = Store(self.store_parent + '/implementations')
  19                 os.mkdir(self.store.dir, 0700)
  20
  21                 self.tmp = tempfile.mktemp()
  22                 os.mkdir(self.tmp)
  23
  24         def tearDown(self):
  25                 BaseTest.tearDown(self)
  26
  27                 support.ro_rmtree(self.store_parent)
  28                 support.ro_rmtree(self.tmp)
  29
  30         def testInit(self):
  31                 assert os.path.isdir(self.store.dir)
  32                 self.assertEquals([], os.listdir(self.store.dir))
  33
  34         def testEmptyManifest(self):
  35                 lines = list(manifest.generate_manifest(self.tmp))
  36                 self.assertEquals([], lines)
  37
  38         def testSimpleManifest(self):
  39                 path = os.path.join(self.tmp, 'MyFile')
  40                 f = file(path, 'w')
  41                 f.write('Hello')
  42                 f.close()
  43                 os.utime(path, (1, 2))
  44                 lines = list(manifest.generate_manifest(self.tmp))
  45                 self.assertEquals(['F f7ff9e8b7bb2e09b70935a5d785e0cc5d9d0abf0 2 5 MyFile'],
  46                                 lines)
  47
  48         def testLinkManifest(self):
  49                 path = os.path.join(self.tmp, 'MyLink')
  50                 os.symlink('Hello', path)
  51                 lines = list(manifest.generate_manifest(self.tmp))
  52                 self.assertEquals(['S f7ff9e8b7bb2e09b70935a5d785e0cc5d9d0abf0 5 MyLink'],
  53                                 lines)
  54
  55         def testVerify(self):
  56                 path = os.path.join(self.tmp, 'MyLink')
  57                 os.symlink('Hello', path)
  58                 mfile = os.path.join(self.tmp, '.manifest')
  59                 for alg_name in ['sha1', 'sha256', 'sha1new']:
  60                         try:
  61                                 alg = manifest.get_algorithm(alg_name)
  62                                 added_digest = alg.getID(manifest.add_manifest_file(self.tmp, alg))
  63                                 digest = alg.new_digest()
  64                                 digest.update('Hello')
  65                                 self.assertEquals("S %s 5 MyLink\n" % digest.hexdigest(),
  66                                                 file(mfile).read())
  67                                 manifest.verify(self.tmp, added_digest)
  68                                 os.chmod(self.tmp, 0700)
  69                                 os.unlink(mfile)
  70                         except BadDigest, ex:
  71                                 raise Exception("%s: %s\n%s" % (alg_name, ex, ex.detail))
  72
  73         def populate_sample(self, target):
  74                 """Create a set of files, links and directories in target for testing."""
  75                 path = os.path.join(target, 'MyFile')
  76                 f = file(path, 'w')
  77                 f.write('Hello')
  78                 f.close()
  79                 os.utime(path, (1, 2))
  80
  81                 subdir = os.path.join(target, 'My Dir')
  82                 os.mkdir(subdir)
  83
  84                 subfile = os.path.join(subdir, '!a file!')
  85                 f = file(subfile, 'w')
  86                 f.write('Some data.')
  87                 f.close()
  88                 os.utime(subfile, (1, 2))
  89
  90                 subfile += '.exe'
  91                 f = file(subfile, 'w')
  92                 f.write('Some code.')
  93                 f.close()
  94                 os.chmod(subfile, 0500)
  95                 os.utime(subfile, (1, 2))
  96
  97                 os.symlink('/the/symlink/target',
  98                            os.path.join(target, 'a symlink'))
  99
 100         def testOptimise(self):
 101                 sample = os.path.join(self.tmp, 'sample')
 102                 os.mkdir(sample)
 103                 self.populate_sample(sample)
 104                 self.store.add_dir_to_cache('sha1new=7e3eb25a072988f164bae24d33af69c1814eb99a',
 105                                            sample,
 106                                            try_helper = False)
 107                 subfile = os.path.join(sample, 'My Dir', '!a file!.exe')
 108                 mtime = os.stat(subfile).st_mtime
 109                 os.chmod(subfile, 0755)
 110                 stream = file(subfile, 'w')
 111                 stream.write('Extra!\n')
 112                 stream.close()
 113                 os.utime(subfile, (mtime, mtime))
 114                 self.store.add_dir_to_cache('sha1new=40861a33dba4e7c26d37505bd9693511808c0c35',
 115                                            sample,
 116                                            try_helper = False)
 117
 118                 impl_a = self.store.lookup('sha1new=7e3eb25a072988f164bae24d33af69c1814eb99a')
 119                 impl_b = self.store.lookup('sha1new=40861a33dba4e7c26d37505bd9693511808c0c35')
 120
 121                 def same_inode(name):
 122                         info_a = os.lstat(os.path.join(impl_a, name))
 123                         info_b = os.lstat(os.path.join(impl_b, name))
 124                         return info_a.st_ino == info_b.st_ino
 125
 126                 assert not same_inode('My Dir/!a file!')
 127                 assert not same_inode('My Dir/!a file!.exe')
 128
 129                 old_stdout = sys.stdout
 130                 sys.stdout = StringIO()
 131                 try:
 132                         cli.do_optimise([self.store.dir])
 133                         got = sys.stdout.getvalue()
 134                 finally:
 135                         sys.stdout = old_stdout
 136                 assert 'Space freed up : 15 bytes' in got
 137
 138                 old_stdout = sys.stdout
 139                 sys.stdout = StringIO()
 140                 try:
 141                         cli.do_optimise([self.store.dir])
 142                         got = sys.stdout.getvalue()
 143                 finally:
 144                         sys.stdout = old_stdout
 145                 assert 'No duplicates found; no changes made.' in got
 146
 147                 assert same_inode('My Dir/!a file!')
 148                 assert not same_inode('My Dir/!a file!.exe')
 149
 150         def testCopy(self):
 151                 sha1 = manifest.get_algorithm('sha1')
 152                 sha1new = manifest.get_algorithm('sha1new')
 153                 source = os.path.join(self.tmp, 'badname')
 154                 os.mkdir(source)
 155
 156                 self.populate_sample(source)
 157
 158                 lines = list(sha1new.generate_manifest(source))
 159                 self.assertEquals(['F f7ff9e8b7bb2e09b70935a5d785e0cc5d9d0abf0 2 5 MyFile',
 160                                    'S 570b0ce957ab43e774c82fca0ea3873fc452278b 19 a symlink',
 161                                    'D /My Dir',
 162                                    'F 0236ef92e1e37c57f0eb161e7e2f8b6a8face705 2 10 !a file!',
 163                                    'X b4ab02f2c791596a980fd35f51f5d92ee0b4705c 2 10 !a file!.exe'],
 164                                 lines)
 165                 digest = sha1.getID(manifest.add_manifest_file(source, sha1))
 166
 167                 copy = tempfile.mktemp()
 168                 os.mkdir(copy)
 169                 try:
 170                         # Source must be in the form alg=value
 171                         try:
 172                                 cli.do_copy([source, copy])
 173                                 assert 0
 174                         except BadDigest, ex:
 175                                 assert 'badname' in str(ex)
 176                         source, badname = os.path.join(self.tmp, digest), source
 177                         os.rename(badname, source)
 178
 179                         # Can't copy sha1 implementations (unsafe)
 180                         try:
 181                                 cli.do_copy([source, copy])
 182                         except SafeException, ex:
 183                                 assert 'sha1' in str(ex)
 184
 185                         # Already have a .manifest
 186                         try:
 187                                 manifest.add_manifest_file(source, sha1new)
 188                                 assert 0
 189                         except SafeException, ex:
 190                                 assert '.manifest' in str(ex)
 191
 192                         os.chmod(source, 0700)
 193                         os.unlink(os.path.join(source, '.manifest'))
 194
 195                         # Switch to sha1new
 196                         digest = sha1new.getID(manifest.add_manifest_file(source, sha1new))
 197                         source, badname = os.path.join(self.tmp, digest), source
 198                         os.rename(badname, source)
 199
 200                         cli.do_copy([source, copy])
 201
 202                         self.assertEquals('Hello', file(os.path.join(copy, digest, 'MyFile')).read())
 203                 finally:
 204                         support.ro_rmtree(copy)
 205
 206 suite = unittest.makeSuite(TestStore)
 207 if __name__ == '__main__':
 208         sys.argv.append('-v')
 209         unittest.main()