~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/per_transport.py

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

Show diffs side-by-side

added

removed

bzrlib/tests/per_transport.py

207

207

                    ]

208

208

        self.build_tree(files, transport=t, line_endings='binary')

209

209

        self.assertRaises(NoSuchFile, t.get, 'c')

210

 

        self.assertListRaises(NoSuchFile, t.get_multi, ['a', 'b', 'c'])

211

 

        self.assertListRaises(NoSuchFile, t.get_multi, iter(['a', 'b', 'c']))

 

210

        def iterate_and_close(func, *args):

 

211

            for f in func(*args):

 

212

                # We call f.read() here because things like paramiko actually

 

213

                # spawn a thread to prefetch the content, which we want to

 

214

                # consume before we close the handle.

 

215

                content = f.read()

 

216

                f.close()

 

217

        self.assertRaises(NoSuchFile, iterate_and_close,

 

218

                          t.get_multi, ['a', 'b', 'c'])

 

219

        self.assertRaises(NoSuchFile, iterate_and_close,

 

220

                          t.get_multi, iter(['a', 'b', 'c']))

212

221

 

213

222

    def test_get_directory_read_gives_ReadError(self):

214

223

        """consistent errors for read() on a file returned by get()."""

285

294

            return

286

295

 

287

296

        t.put_bytes('a', 'some text for a\n')

288

 

        self.failUnless(t.has('a'))

 

297

        self.assertTrue(t.has('a'))

289

298

        self.check_transport_contents('some text for a\n', t, 'a')

290

299

 

291

300

        # The contents should be overwritten

303

312

                    t.put_bytes_non_atomic, 'a', 'some text for a\n')

304

313

            return

305

314

 

306

 

        self.failIf(t.has('a'))

 

315

        self.assertFalse(t.has('a'))

307

316

        t.put_bytes_non_atomic('a', 'some text for a\n')

308

 

        self.failUnless(t.has('a'))

 

317

        self.assertTrue(t.has('a'))

309

318

        self.check_transport_contents('some text for a\n', t, 'a')

310

319

        # Put also replaces contents

311

320

        t.put_bytes_non_atomic('a', 'new\ncontents for\na\n')

323

332

        # Now test the create_parent flag

324

333

        self.assertRaises(NoSuchFile, t.put_bytes_non_atomic, 'dir/a',

325

334

                                       'contents\n')

326

 

        self.failIf(t.has('dir/a'))

 

335

        self.assertFalse(t.has('dir/a'))

327

336

        t.put_bytes_non_atomic('dir/a', 'contents for dir/a\n',

328

337

                               create_parent_dir=True)

329

338

        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')

401

410

        result = t.put_file('a', StringIO('some text for a\n'))

402

411

        # put_file returns the length of the data written

403

412

        self.assertEqual(16, result)

404

 

        self.failUnless(t.has('a'))

 

413

        self.assertTrue(t.has('a'))

405

414

        self.check_transport_contents('some text for a\n', t, 'a')

406

415

        # Put also replaces contents

407

416

        result = t.put_file('a', StringIO('new\ncontents for\na\n'))

419

428

                    t.put_file_non_atomic, 'a', StringIO('some text for a\n'))

420

429

            return

421

430

 

422

 

        self.failIf(t.has('a'))

 

431

        self.assertFalse(t.has('a'))

423

432

        t.put_file_non_atomic('a', StringIO('some text for a\n'))

424

 

        self.failUnless(t.has('a'))

 

433

        self.assertTrue(t.has('a'))

425

434

        self.check_transport_contents('some text for a\n', t, 'a')

426

435

        # Put also replaces contents

427

436

        t.put_file_non_atomic('a', StringIO('new\ncontents for\na\n'))

439

448

        # Now test the create_parent flag

440

449

        self.assertRaises(NoSuchFile, t.put_file_non_atomic, 'dir/a',

441

450

                                       StringIO('contents\n'))

442

 

        self.failIf(t.has('dir/a'))

 

451

        self.assertFalse(t.has('dir/a'))

443

452

        t.put_file_non_atomic('dir/a', StringIO('contents for dir/a\n'),

444

453

                              create_parent_dir=True)

445

454

        self.check_transport_contents('contents for dir/a\n', t, 'dir/a')

822

831

            return

823

832

 

824

833

        t.put_bytes('a', 'a little bit of text\n')

825

 

        self.failUnless(t.has('a'))

 

834

        self.assertTrue(t.has('a'))

826

835

        t.delete('a')

827

 

        self.failIf(t.has('a'))

 

836

        self.assertFalse(t.has('a'))

828

837

 

829

838

        self.assertRaises(NoSuchFile, t.delete, 'a')

830

839

 

836

845

        t.delete_multi(['a', 'c'])

837

846

        self.assertEqual([False, True, False],

838

847

                list(t.has_multi(['a', 'b', 'c'])))

839

 

        self.failIf(t.has('a'))

840

 

        self.failUnless(t.has('b'))

841

 

        self.failIf(t.has('c'))

 

848

        self.assertFalse(t.has('a'))

 

849

        self.assertTrue(t.has('b'))

 

850

        self.assertFalse(t.has('c'))

842

851

 

843

852

        self.assertRaises(NoSuchFile,

844

853

                t.delete_multi, ['a', 'b', 'c'])

905

914

        t.mkdir('foo-baz')

906

915

        t.rmdir('foo')

907

916

        self.assertRaises((NoSuchFile, PathError), t.rmdir, 'foo')

908

 

        self.failUnless(t.has('foo-bar'))

 

917

        self.assertTrue(t.has('foo-bar'))

909

918

 

910

919

    def test_rename_dir_succeeds(self):

911

920

        t = self.get_transport()

994

1003

        self.assertEquals([True, False], list(t.has_multi(['a', 'b'])))

995

1004

 

996

1005

        t.move('a', 'b')

997

 

        self.failUnless(t.has('b'))

998

 

        self.failIf(t.has('a'))

 

1006

        self.assertTrue(t.has('b'))

 

1007

        self.assertFalse(t.has('a'))

999

1008

 

1000

1009

        self.check_transport_contents('a first file\n', t, 'b')

1001

1010

        self.assertEquals([False, True], list(t.has_multi(['a', 'b'])))

1003

1012

        # Overwrite a file

1004

1013

        t.put_bytes('c', 'c this file\n')

1005

1014

        t.move('c', 'b')

1006

 

        self.failIf(t.has('c'))

 

1015

        self.assertFalse(t.has('c'))

1007

1016

        self.check_transport_contents('c this file\n', t, 'b')

1008

1017

 

1009

1018

        # TODO: Try to write a test for atomicity

1063

1072

        for path, size in zip(paths, sizes):

1064

1073

            st = t.stat(path)

1065

1074

            if path.endswith('/'):

1066

 

                self.failUnless(S_ISDIR(st.st_mode))

 

1075

                self.assertTrue(S_ISDIR(st.st_mode))

1067

1076

                # directory sizes are meaningless

1068

1077

            else:

1069

 

                self.failUnless(S_ISREG(st.st_mode))

 

1078

                self.assertTrue(S_ISREG(st.st_mode))

1070

1079

                self.assertEqual(size, st.st_size)

1071

1080

 

1072

1081

        remote_stats = list(t.stat_multi(paths))

1079

1088

        self.assertListRaises(NoSuchFile, t.stat_multi, iter(['a', 'c', 'd']))

1080

1089

        self.build_tree(['subdir/', 'subdir/file'], transport=t)

1081

1090

        subdir = t.clone('subdir')

1082

 

        subdir.stat('./file')

1083

 

        subdir.stat('.')

 

1091

        st = subdir.stat('./file')

 

1092

        st = subdir.stat('.')

1084

1093

 

1085

1094

    def test_hardlink(self):

1086

1095

        from stat import ST_NLINK

1095

1104

        try:

1096

1105

            t.hardlink(source_name, link_name)

1097

1106

 

1098

 

            self.failUnless(t.has(source_name))

1099

 

            self.failUnless(t.has(link_name))

 

1107

            self.assertTrue(t.has(source_name))

 

1108

            self.assertTrue(t.has(link_name))

1100

1109

 

1101

1110

            st = t.stat(link_name)

1102

 

            self.failUnlessEqual(st[ST_NLINK], 2)

 

1111

            self.assertEqual(st[ST_NLINK], 2)

1103

1112

        except TransportNotPossible:

1104

1113

            raise TestSkipped("Transport %s does not support hardlinks." %

1105

1114

                              self._server.__class__)

1117

1126

        try:

1118

1127

            t.symlink(source_name, link_name)

1119

1128

 

1120

 

            self.failUnless(t.has(source_name))

1121

 

            self.failUnless(t.has(link_name))

 

1129

            self.assertTrue(t.has(source_name))

 

1130

            self.assertTrue(t.has(link_name))

1122

1131

 

1123

1132

            st = t.stat(link_name)

1124

 

            self.failUnless(S_ISLNK(st.st_mode),

 

1133

            self.assertTrue(S_ISLNK(st.st_mode),

1125

1134

                "expected symlink, got mode %o" % st.st_mode)

1126

1135

        except TransportNotPossible:

1127

1136

            raise TestSkipped("Transport %s does not support symlinks." %

1293

1302

 

1294

1303

        self.build_tree(['a', 'b/', 'b/c'], transport=t1)

1295

1304

 

1296

 

        self.failUnless(t1.has('a'))

1297

 

        self.failUnless(t1.has('b/c'))

1298

 

        self.failIf(t1.has('c'))

 

1305

        self.assertTrue(t1.has('a'))

 

1306

        self.assertTrue(t1.has('b/c'))

 

1307

        self.assertFalse(t1.has('c'))

1299

1308

 

1300

1309

        t2 = t1.clone('b')

1301

1310

        self.assertEqual(t1.base + 'b/', t2.base)

1302

1311

 

1303

 

        self.failUnless(t2.has('c'))

1304

 

        self.failIf(t2.has('a'))

 

1312

        self.assertTrue(t2.has('c'))

 

1313

        self.assertFalse(t2.has('a'))

1305

1314

 

1306

1315

        t3 = t2.clone('..')

1307

 

        self.failUnless(t3.has('a'))

1308

 

        self.failIf(t3.has('c'))

 

1316

        self.assertTrue(t3.has('a'))

 

1317

        self.assertFalse(t3.has('c'))

1309

1318

 

1310

 

        self.failIf(t1.has('b/d'))

1311

 

        self.failIf(t2.has('d'))

1312

 

        self.failIf(t3.has('b/d'))

 

1319

        self.assertFalse(t1.has('b/d'))

 

1320

        self.assertFalse(t2.has('d'))

 

1321

        self.assertFalse(t3.has('b/d'))

1313

1322

 

1314

1323

        if t1.is_readonly():

1315

1324

            self.build_tree_contents([('b/d', 'newfile\n')])

1316

1325

        else:

1317

1326

            t2.put_bytes('d', 'newfile\n')

1318

1327

 

1319

 

        self.failUnless(t1.has('b/d'))

1320

 

        self.failUnless(t2.has('d'))

1321

 

        self.failUnless(t3.has('b/d'))

 

1328

        self.assertTrue(t1.has('b/d'))

 

1329

        self.assertTrue(t2.has('d'))

 

1330

        self.assertTrue(t3.has('b/d'))

1322

1331

 

1323

1332

    def test_clone_to_root(self):

1324

1333

        orig_transport = self.get_transport()

1410

1419

        # smoke test for abspath on win32.

1411

1420

        # a transport based on 'file:///' never fully qualifies the drive.

1412

1421

        transport = _mod_transport.get_transport("file:///")

1413

 

        self.failUnlessEqual(transport.abspath("/"), "file:///")

 

1422

        self.assertEqual(transport.abspath("/"), "file:///")

1414

1423

 

1415

1424

        # but a transport that starts with a drive spec must keep it.

1416

1425

        transport = _mod_transport.get_transport("file:///C:/")

1417

 

        self.failUnlessEqual(transport.abspath("/"), "file:///C:/")

 

1426

        self.assertEqual(transport.abspath("/"), "file:///C:/")

1418

1427

 

1419

1428

    def test_local_abspath(self):

1420

1429

        transport = self.get_transport()

• Older »

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