~abentley/bzrtools/bzrtools.dev

« back to all changes in this revision

Viewing changes to tests/upstream_import.py

• browse files at revision 486
• compare with another revision
• download diff
• download tarball
• view history from revision 486

• reverse the comparison (486 to 531.2.2)
• stop comparing with revision 531.2.2

Show diffs side-by-side

added

removed

tests/upstream_import.py

1

1

import os

2

2

from StringIO import StringIO

3

 

from shutil import rmtree, copy2, copytree

 

3

from shutil import rmtree, copy2

4

4

import tarfile

5

 

import tempfile

6

5

from unittest import makeSuite

7

6

 

8

7

from bzrlib import osutils

9

8

from bzrlib.bzrdir import BzrDir

10

 

from bzrlib.plugins.bzrtools.upstream_import import (

11

 

    common_directory,

12

 

    import_archive,

13

 

    import_tar,

14

 

    import_zip,

15

 

    import_dir,

16

 

    top_directory,

17

 

    ZipFileWrapper,

18

 

)

 

9

try:

 

10

    from bzrlib.plugins.bzrtools.upstream_import import (

 

11

        common_directory,

 

12

        import_tar,

 

13

        import_zip,

 

14

        import_dir,

 

15

        top_directory,

 

16

        ZipFileWrapper, 

 

17

    )

 

18

except ImportError:

 

19

    from bzrtools.upstream_import import (

 

20

        common_directory,

 

21

        import_tar,

 

22

        import_zip,

 

23

        top_directory,

 

24

        ZipFileWrapper, 

 

25

    )

19

26

from bzrlib.tests import TestCaseInTempDir

20

27

 

21

28

 

22

 

def import_tar_broken(tree, tar_input):

23

 

    """

24

 

    Import a tarfile with names that that end in //, e.g. Feisty Python 2.5

25

 

    """

26

 

    tar_file = tarfile.open('lala', 'r', tar_input)

27

 

    for member in tar_file.members:

28

 

        if member.name.endswith('/'):

29

 

            member.name += '/'

30

 

    import_archive(tree, tar_file)

31

 

 

32

 

 

33

29

class DirFileWriter(object):

34

30

 

35

31

    def __init__(self, fileobj, mode):

36

 

        # We may be asked to 'append'.  If so, fileobj already has a path.

37

 

        # So we copy the existing tree, and overwrite afterward.

38

 

        fileobj.seek(0)

39

 

        existing = fileobj.read()

40

 

        fileobj.seek(0)

41

 

        path = tempfile.mkdtemp(dir=os.getcwd())

42

 

        if existing != '':

43

 

            # copytree requires the directory not to exist

44

 

            os.rmdir(path)

45

 

            copytree(existing, path)

46

 

        fileobj.write(path)

47

 

        self.root = path

 

32

        os.mkdir('tmp')

 

33

        fileobj.write('tmp')

 

34

        self.root = os.path.realpath('tmp')

48

35

 

49

36

    def add(self, path):

50

 

        target_path = os.path.join(self.root, path)

51

 

        parent = osutils.dirname(target_path)

52

 

        if not os.path.exists(parent):

53

 

            os.makedirs(parent)

54

37

        kind = osutils.file_kind(path)

55

38

        if kind == 'file':

56

 

            copy2(path, target_path)

 

39

            copy2(path, os.path.join(self.root, path))

57

40

        if kind == 'directory':

58

 

            os.mkdir(target_path)

 

41

            os.mkdir(os.path.join(self.root, path))

59

42

 

60

43

    def close(self):

61

44

        pass

81

64

                os.chdir('project-0.1')

82

65

            os.mkdir(prefix + 'junk')

83

66

            archive_file.add(prefix + 'junk')

84

 

 

 

67

            

85

68

            f = file(prefix + 'README', 'wb')

86

69

            f.write('What?')

87

70

            f.close()

100

83

        result.seek(0)

101

84

        return result

102

85

 

103

 

    def make_archive2(self, builder, subdir):

 

86

    def make_archive2(self, builder):

104

87

        result = StringIO()

105

 

        archive_file = builder(result)

 

88

        tar_file = builder(result)

106

89

        os.mkdir('project-0.2')

107

 

        try:

108

 

            if subdir:

109

 

                prefix='project-0.2/'

110

 

                archive_file.add('project-0.2')

111

 

            else:

112

 

                prefix=''

113

 

                os.chdir('project-0.2')

114

 

 

115

 

            os.mkdir(prefix + 'junk')

116

 

            archive_file.add(prefix + 'junk')

117

 

 

118

 

            f = file(prefix + 'README', 'wb')

119

 

            f.write('Now?')

120

 

            f.close()

121

 

            archive_file.add(prefix + 'README')

122

 

            archive_file.close()

123

 

 

124

 

            archive_file = builder(result, 'a')

125

 

            archive_file.add(prefix + 'README')

126

 

            archive_file.close()

127

 

 

128

 

        finally:

129

 

            if not subdir:

130

 

                os.chdir('..')

131

 

        result.seek(0)

 

90

        tar_file.add('project-0.2')

 

91

        

 

92

        os.mkdir('project-0.2/junk')

 

93

        tar_file.add('project-0.2/junk')

 

94

 

 

95

        f = file('project-0.2/README', 'wb')

 

96

        f.write('Now?')

 

97

        f.close()

 

98

        tar_file.add('project-0.2/README')

 

99

        tar_file.close()

 

100

 

 

101

        tar_file = builder(result, 'a')

 

102

        tar_file.add('project-0.2/README')

 

103

 

 

104

        rmtree('project-0.2')

132

105

        return result

133

106

 

134

107

    def make_messed_tar(self):

139

112

 

140

113

        os.mkdir('project-0.2')

141

114

        tar_file.add('project-0.2')

142

 

 

 

115

        

143

116

        f = file('project-0.1/README', 'wb')

144

117

        f.write('What?')

145

118

        f.close()

168

141

            return tarfile.open('project-0.1.tar', mode, fileobj)

169

142

        self.archive_test(builder, import_tar)

170

143

 

171

 

    def test_broken_tar(self):

172

 

        def builder(fileobj, mode='w'):

173

 

            return tarfile.open('project-0.1.tar', mode, fileobj)

174

 

        self.archive_test(builder, import_tar_broken, subdir=True)

175

 

 

176

144

    def test_unzip(self):

177

145

        def builder(fileobj, mode='w'):

178

146

            return ZipFileWrapper(fileobj, mode)

181

149

    def test_copydir_nosub(self):

182

150

        def builder(fileobj, mode='w'):

183

151

            return DirFileWriter(fileobj, mode)

184

 

        # It would be bogus to test with the result in a subdirectory,

185

 

        # because for directories, the input root is always the output root.

186

152

        self.archive_test(builder, import_dir)

187

153

 

 

154

    def test_copydir_sub(self):

 

155

        def builder(fileobj, mode='w'):

 

156

            return DirFileWriter(fileobj, mode)

 

157

        self.archive_test(builder, import_dir, subdir=True)

 

158

 

 

159

    def test_copydir_nosub(self):

 

160

        def builder(fileobj, mode='w'):

 

161

            return DirFileWriter(fileobj, mode)

 

162

        self.archive_test(builder, import_dir, subdir=True)

 

163

 

188

164

    def archive_test(self, builder, importer, subdir=False):

189

165

        archive_file = self.make_archive(builder, subdir)

190

166

        tree = BzrDir.create_standalone_workingtree('tree')

191

 

        tree.lock_write()

192

 

        try:

193

 

            importer(tree, archive_file)

194

 

            self.assertTrue(tree.path2id('README') is not None)

195

 

            self.assertTrue(tree.path2id('FEEDME') is not None)

196

 

            self.assertTrue(os.path.isfile(tree.abspath('README')))

197

 

            self.assertEqual(tree.inventory[tree.path2id('README')].kind,

198

 

                'file')

199

 

            self.assertEqual(tree.inventory[tree.path2id('FEEDME')].kind,

200

 

                'file')

201

 

            f = file(tree.abspath('junk/food'), 'wb')

202

 

            f.write('I like food\n')

203

 

            f.close()

 

167

        importer(tree, archive_file)

 

168

        self.assertTrue(tree.path2id('README') is not None) 

 

169

        self.assertTrue(tree.path2id('FEEDME') is not None)

 

170

        self.assertTrue(os.path.isfile(tree.abspath('README')))

 

171

        self.assertEqual(tree.inventory[tree.path2id('README')].kind, 'file')

 

172

        self.assertEqual(tree.inventory[tree.path2id('FEEDME')].kind, 'file')

 

173

        

 

174

        f = file(tree.abspath('junk/food'), 'wb')

 

175

        f.write('I like food\n')

 

176

        f.close()

204

177

 

205

 

            archive_file = self.make_archive2(builder, subdir)

206

 

            importer(tree, archive_file)

207

 

            self.assertTrue(tree.path2id('README') is not None)

208

 

            self.assertTrue(not os.path.exists(tree.abspath('FEEDME')))

209

 

        finally:

210

 

            tree.unlock()

 

178

        archive_file = self.make_archive2(builder)

 

179

        importer(tree, archive_file)

 

180

        self.assertTrue(tree.path2id('README') is not None) 

 

181

        self.assertTrue(not os.path.exists(tree.abspath('FEEDME')))

211

182

 

212

183

 

213

184

    def test_untar2(self):

214

185

        tar_file = self.make_messed_tar()

215

186

        tree = BzrDir.create_standalone_workingtree('tree')

216

187

        import_tar(tree, tar_file)

217

 

        self.assertTrue(tree.path2id('project-0.1/README') is not None)

 

188

        self.assertTrue(tree.path2id('project-0.1/README') is not None) 

218

189

 

219

190

    def test_untar_gzip(self):

220

191

        tar_file = self.make_tar(mode='w:gz')

221

192

        tree = BzrDir.create_standalone_workingtree('tree')

222

193

        import_tar(tree, tar_file)

223

 

        self.assertTrue(tree.path2id('README') is not None)

 

194

        self.assertTrue(tree.path2id('README') is not None) 

224

195

 

225

196

def test_suite():

226

197

    return makeSuite(TestImport)

• Older »

Loggerhead is a web-based interface for Breezy
Version: 2.0.1