~bzr-pqm/bzr/bzr.dev : revision 1543

1

2

3

# This program is free software; you can redistribute it and/or modify

4

# it under the terms of the GNU General Public License as published by

5

# the Free Software Foundation; either version 2 of the License, or

6

# (at your option) any later version.

7

8

# This program is distributed in the hope that it will be useful,

9

# but WITHOUT ANY WARRANTY; without even the implied warranty of

10

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

11

# GNU General Public License for more details.

12

13

# You should have received a copy of the GNU General Public License

14

# along with this program; if not, write to the Free Software

15

# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA

16

17

"""Tests for Transport implementations.

18

19

Transport implementations tested here are supplied by

20

TransportTestProviderAdapter.

21

"""

22

23

import os

24

from cStringIO import StringIO

25

import stat

26

import sys

27

28

from bzrlib.errors import (NoSuchFile, FileExists,

29

TransportNotPossible, ConnectionError)

30

from bzrlib.tests import TestCaseInTempDir, TestSkipped

31

from bzrlib.transport import memory, urlescape

32

import bzrlib.transport

33

34

35

def _append(fn, txt):

36

"""Append the given text (file-like object) to the supplied filename."""

37

f = open(fn, 'ab')

38

try:

39

f.write(txt.read())

40

finally:

41

f.close()

42

43

44

class TestTransportImplementation(TestCaseInTempDir):

45

"""Implementation verification for transports.

46

47

To verify a transport we need a server factory, which is a callable

48

that accepts no parameters and returns an implementation of

49

bzrlib.transport.Server.

50

51

That Server is then used to construct transport instances and test

52

the transport via loopback activity.

53

54

Currently this assumes that the Transport object is connected to the

55

current working directory. So that whatever is done

56

through the transport, should show up in the working

57

directory, and vice-versa. This is a bug, because its possible to have

58

URL schemes which provide access to something that may not be

59

result in storage on the local disk, i.e. due to file system limits, or

60

due to it being a database or some other non-filesystem tool.

61

62

This also tests to make sure that the functions work with both

63

generators and lists (assuming iter(list) is effectively a generator)

64

"""

65

66

def setUp(self):

67

super(TestTransportImplementation, self).setUp()

68

self._server = self.transport_server()

69

self._server.setUp()

70

71

def tearDown(self):

72

super(TestTransportImplementation, self).tearDown()

73

self._server.tearDown()

74

75

def check_transport_contents(self, content, transport, relpath):

76

"""Check that transport.get(relpath).read() == content."""

77

self.assertEqualDiff(content, transport.get(relpath).read())

78

79

def get_transport(self):

80

"""Return a connected transport to the local directory."""

81

t = bzrlib.transport.get_transport(self._server.get_url())

82

self.failUnless(isinstance(t, self.transport_class),

83

"Got the wrong class from get_transport"

84

"(%r, expected %r)" % (t.__class__,

85

self.transport_class))

86

return t

87

88

def assertListRaises(self, excClass, func, *args, **kwargs):

89

"""Fail unless excClass is raised when the iterator from func is used.

90

91

Many transport functions can return generators this makes sure

92

to wrap them in a list() call to make sure the whole generator

93

is run, and that the proper exception is raised.

94

"""

95

try:

96

list(func(*args, **kwargs))

97

except excClass:

98

return

99

else:

100

if hasattr(excClass,'__name__'): excName = excClass.__name__

101

else: excName = str(excClass)

102

raise self.failureException, "%s not raised" % excName

103

104

def test_has(self):

105

t = self.get_transport()

106

107

files = ['a', 'b', 'e', 'g', '%']

108

self.build_tree(files, transport=t)

109

self.assertEqual(True, t.has('a'))

110

self.assertEqual(False, t.has('c'))

111

self.assertEqual(True, t.has(urlescape('%')))

112

self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),

113

[True, True, False, False, True, False, True, False])

114

self.assertEqual(True, t.has_any(['a', 'b', 'c']))

115

self.assertEqual(False, t.has_any(['c', 'd', 'f', urlescape('%%')]))

116

self.assertEqual(list(t.has_multi(iter(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']))),

117

[True, True, False, False, True, False, True, False])

118

self.assertEqual(False, t.has_any(['c', 'c', 'c']))

119

self.assertEqual(True, t.has_any(['b', 'b', 'b']))

120

121

def test_get(self):

122

t = self.get_transport()

123

124

files = ['a', 'b', 'e', 'g']

125

contents = ['contents of a\n',

126

'contents of b\n',

127

'contents of e\n',

128

'contents of g\n',

129

]

130

self.build_tree(files, transport=t)

131

self.check_transport_contents('contents of a\n', t, 'a')

132

content_f = t.get_multi(files)

133

for content, f in zip(contents, content_f):

134

self.assertEqual(content, f.read())

135

136

content_f = t.get_multi(iter(files))

137

for content, f in zip(contents, content_f):

138

self.assertEqual(content, f.read())

139

140

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

141

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

142

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

143

144

def test_put(self):

145

t = self.get_transport()

146

147

if t.is_readonly():

148

self.assertRaises(TransportNotPossible,

149

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

150

return

151

152

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

153

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

154

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

155

# Make sure 'has' is updated

156

self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),

157

[True, False, False, False, False])

158

# Put also replaces contents

159

self.assertEqual(t.put_multi([('a', StringIO('new\ncontents for\na\n')),

160

('d', StringIO('contents\nfor d\n'))]),

161

2)

162

self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),

163

[True, False, False, True, False])

164

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

165

self.check_transport_contents('contents\nfor d\n', t, 'd')

166

167

self.assertEqual(

168

t.put_multi(iter([('a', StringIO('diff\ncontents for\na\n')),

169

('d', StringIO('another contents\nfor d\n'))])),

170

2)

171

self.check_transport_contents('diff\ncontents for\na\n', t, 'a')

172

self.check_transport_contents('another contents\nfor d\n', t, 'd')

173

174

self.assertRaises(NoSuchFile,

175

t.put, 'path/doesnt/exist/c', 'contents')

176

177

def test_put_permissions(self):

178

t = self.get_transport()

179

180

if t.is_readonly():

181

return

182

t.put('mode644', StringIO('test text\n'), mode=0644)

183

self.assertTransportMode(t, 'mode644', 0644)

184

t.put('mode666', StringIO('test text\n'), mode=0666)

185

self.assertTransportMode(t, 'mode666', 0666)

186

t.put('mode600', StringIO('test text\n'), mode=0600)

187

self.assertTransportMode(t, 'mode600', 0600)

188

# Yes, you can put a file such that it becomes readonly

189

t.put('mode400', StringIO('test text\n'), mode=0400)

190

self.assertTransportMode(t, 'mode400', 0400)

191

t.put_multi([('mmode644', StringIO('text\n'))], mode=0644)

192

self.assertTransportMode(t, 'mmode644', 0644)

193

194

def test_mkdir(self):

195

t = self.get_transport()

196

197

if t.is_readonly():

198

# cannot mkdir on readonly transports. We're not testing for

199

# cache coherency because cache behaviour is not currently

200

# defined for the transport interface.

201

self.assertRaises(TransportNotPossible, t.mkdir, '.')

202

self.assertRaises(TransportNotPossible, t.mkdir, 'new_dir')

203

self.assertRaises(TransportNotPossible, t.mkdir_multi, ['new_dir'])

204

self.assertRaises(TransportNotPossible, t.mkdir, 'path/doesnt/exist')

205

return

206

# Test mkdir

207

t.mkdir('dir_a')

208

self.assertEqual(t.has('dir_a'), True)

209

self.assertEqual(t.has('dir_b'), False)

210

211

t.mkdir('dir_b')

212

self.assertEqual(t.has('dir_b'), True)

213

214

t.mkdir_multi(['dir_c', 'dir_d'])

215

216

t.mkdir_multi(iter(['dir_e', 'dir_f']))

217

self.assertEqual(list(t.has_multi(

218

['dir_a', 'dir_b', 'dir_c', 'dir_q',

219

'dir_d', 'dir_e', 'dir_f', 'dir_b'])),

220

[True, True, True, False,

221

True, True, True, True])

222

223

# we were testing that a local mkdir followed by a transport

224

# mkdir failed thusly, but given that we * in one process * do not

225

# concurrently fiddle with disk dirs and then use transport to do

226

# things, the win here seems marginal compared to the constraint on

227

# the interface. RBC 20051227

228

t.mkdir('dir_g')

229

self.assertRaises(FileExists, t.mkdir, 'dir_g')

230

231

# Test get/put in sub-directories

232

self.assertEqual(

233

t.put_multi([('dir_a/a', StringIO('contents of dir_a/a')),

234

('dir_b/b', StringIO('contents of dir_b/b'))])

235

, 2)

236

self.check_transport_contents('contents of dir_a/a', t, 'dir_a/a')

237

self.check_transport_contents('contents of dir_b/b', t, 'dir_b/b')

238

239

# mkdir of a dir with an absent parent

240

self.assertRaises(NoSuchFile, t.mkdir, 'missing/dir')

241

242

def test_mkdir_permissions(self):

243

t = self.get_transport()

244

if t.is_readonly():

245

return

246

# Test mkdir with a mode

247

t.mkdir('dmode755', mode=0755)

248

self.assertTransportMode(t, 'dmode755', 0755)

249

t.mkdir('dmode555', mode=0555)

250

self.assertTransportMode(t, 'dmode555', 0555)

251

t.mkdir('dmode777', mode=0777)

252

self.assertTransportMode(t, 'dmode777', 0777)

253

t.mkdir('dmode700', mode=0700)

254

self.assertTransportMode(t, 'dmode700', 0700)

255

# TODO: jam 20051215 test mkdir_multi with a mode

256

t.mkdir_multi(['mdmode755'], mode=0755)

257

self.assertTransportMode(t, 'mdmode755', 0755)

258

259

def test_copy_to(self):

260

from bzrlib.transport.memory import MemoryTransport

261

t = self.get_transport()

262

263

files = ['a', 'b', 'c', 'd']

264

self.build_tree(files, transport=t)

265

266

temp_transport = MemoryTransport('memory:/')

267

268

t.copy_to(files, temp_transport)

269

for f in files:

270

self.check_transport_contents(temp_transport.get(f).read(),

271

t, f)

272

273

# Test that copying into a missing directory raises

274

# NoSuchFile

275

if t.is_readonly():

276

self.build_tree(['e/', 'e/f'])

277

else:

278

t.mkdir('e')

279

t.put('e/f', StringIO('contents of e'))

280

self.assertRaises(NoSuchFile, t.copy_to, ['e/f'], temp_transport)

281

temp_transport.mkdir('e')

282

t.copy_to(['e/f'], temp_transport)

283

284

del temp_transport

285

temp_transport = MemoryTransport('memory:/')

286

287

files = ['a', 'b', 'c', 'd']

288

t.copy_to(iter(files), temp_transport)

289

for f in files:

290

self.check_transport_contents(temp_transport.get(f).read(),

291

t, f)

292

del temp_transport

293

294

for mode in (0666, 0644, 0600, 0400):

295

temp_transport = MemoryTransport("memory:/")

296

t.copy_to(files, temp_transport, mode=mode)

297

for f in files:

298

self.assertTransportMode(temp_transport, f, mode)

299

300

def test_append(self):

301

t = self.get_transport()

302

303

if t.is_readonly():

304

open('a', 'wb').write('diff\ncontents for\na\n')

305

open('b', 'wb').write('contents\nfor b\n')

306

else:

307

t.put_multi([

308

('a', StringIO('diff\ncontents for\na\n')),

309

('b', StringIO('contents\nfor b\n'))

310

])

311

312

if t.is_readonly():

313

self.assertRaises(TransportNotPossible,

314

t.append, 'a', 'add\nsome\nmore\ncontents\n')

315

_append('a', StringIO('add\nsome\nmore\ncontents\n'))

316

else:

317

t.append('a', StringIO('add\nsome\nmore\ncontents\n'))

318

319

self.check_transport_contents(

320

'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n',

321

t, 'a')

322

323

if t.is_readonly():

324

self.assertRaises(TransportNotPossible,

325

t.append_multi,

326

[('a', 'and\nthen\nsome\nmore\n'),

327

('b', 'some\nmore\nfor\nb\n')])

328

_append('a', StringIO('and\nthen\nsome\nmore\n'))

329

_append('b', StringIO('some\nmore\nfor\nb\n'))

330

else:

331

t.append_multi([('a', StringIO('and\nthen\nsome\nmore\n')),

332

('b', StringIO('some\nmore\nfor\nb\n'))])

333

self.check_transport_contents(

334

'diff\ncontents for\na\n'

335

'add\nsome\nmore\ncontents\n'

336

'and\nthen\nsome\nmore\n',

337

t, 'a')

338

self.check_transport_contents(

339

'contents\nfor b\n'

340

'some\nmore\nfor\nb\n',

341

t, 'b')

342

343

if t.is_readonly():

344

_append('a', StringIO('a little bit more\n'))

345

_append('b', StringIO('from an iterator\n'))

346

else:

347

t.append_multi(iter([('a', StringIO('a little bit more\n')),

348

('b', StringIO('from an iterator\n'))]))

349

self.check_transport_contents(

350

'diff\ncontents for\na\n'

351

'add\nsome\nmore\ncontents\n'

352

'and\nthen\nsome\nmore\n'

353

'a little bit more\n',

354

t, 'a')

355

self.check_transport_contents(

356

'contents\nfor b\n'

357

'some\nmore\nfor\nb\n'

358

'from an iterator\n',

359

t, 'b')

360

361

if t.is_readonly():

362

_append('c', StringIO('some text\nfor a missing file\n'))

363

_append('a', StringIO('some text in a\n'))

364

_append('d', StringIO('missing file r\n'))

365

else:

366

t.append('c', StringIO('some text\nfor a missing file\n'))

367

t.append_multi([('a', StringIO('some text in a\n')),

368

('d', StringIO('missing file r\n'))])

369

self.check_transport_contents(

370

'diff\ncontents for\na\n'

371

'add\nsome\nmore\ncontents\n'

372

'and\nthen\nsome\nmore\n'

373

'a little bit more\n'

374

'some text in a\n',

375

t, 'a')

376

self.check_transport_contents('some text\nfor a missing file\n',

377

t, 'c')

378

self.check_transport_contents('missing file r\n', t, 'd')

379

380

# a file with no parent should fail..

381

if not t.is_readonly():

382

self.assertRaises(NoSuchFile,

383

t.append, 'missing/path',

384

StringIO('content'))

385

386

def test_append_file(self):

387

t = self.get_transport()

388

389

contents = [

390

('f1', StringIO('this is a string\nand some more stuff\n')),

391

('f2', StringIO('here is some text\nand a bit more\n')),

392

('f3', StringIO('some text for the\nthird file created\n')),

393

('f4', StringIO('this is a string\nand some more stuff\n')),

394

('f5', StringIO('here is some text\nand a bit more\n')),

395

('f6', StringIO('some text for the\nthird file created\n'))

396

]

397

398

if t.is_readonly():

399

for f, val in contents:

400

open(f, 'wb').write(val.read())

401

else:

402

t.put_multi(contents)

403

404

a1 = StringIO('appending to\none\n')

405

if t.is_readonly():

406

_append('f1', a1)

407

else:

408

t.append('f1', a1)

409

410

del a1

411

412

self.check_transport_contents(

413

'this is a string\nand some more stuff\n'

414

'appending to\none\n',

415

t, 'f1')

416

417

a2 = StringIO('adding more\ntext to two\n')

418

a3 = StringIO('some garbage\nto put in three\n')

419

420

if t.is_readonly():

421

_append('f2', a2)

422

_append('f3', a3)

423

else:

424

t.append_multi([('f2', a2), ('f3', a3)])

425

426

del a2, a3

427

428

self.check_transport_contents(

429

'here is some text\nand a bit more\n'

430

'adding more\ntext to two\n',

431

t, 'f2')

432

self.check_transport_contents(

433

'some text for the\nthird file created\n'

434

'some garbage\nto put in three\n',

435

t, 'f3')

436

437

# Test that an actual file object can be used with put

438

a4 = t.get('f1')

439

if t.is_readonly():

440

_append('f4', a4)

441

else:

442

t.append('f4', a4)

443

444

del a4

445

446

self.check_transport_contents(

447

'this is a string\nand some more stuff\n'

448

'this is a string\nand some more stuff\n'

449

'appending to\none\n',

450

t, 'f4')

451

452

a5 = t.get('f2')

453

a6 = t.get('f3')

454

if t.is_readonly():

455

_append('f5', a5)

456

_append('f6', a6)

457

else:

458

t.append_multi([('f5', a5), ('f6', a6)])

459

460

del a5, a6

461

462

self.check_transport_contents(

463

'here is some text\nand a bit more\n'

464

'here is some text\nand a bit more\n'

465

'adding more\ntext to two\n',

466

t, 'f5')

467

self.check_transport_contents(

468

'some text for the\nthird file created\n'

469

'some text for the\nthird file created\n'

470

'some garbage\nto put in three\n',

471

t, 'f6')

472

473

a5 = t.get('f2')

474

a6 = t.get('f2')

475

a7 = t.get('f3')

476

if t.is_readonly():

477

_append('c', a5)

478

_append('a', a6)

479

_append('d', a7)

480

else:

481

t.append('c', a5)

482

t.append_multi([('a', a6), ('d', a7)])

483

del a5, a6, a7

484

self.check_transport_contents(t.get('f2').read(), t, 'c')

485

self.check_transport_contents(t.get('f3').read(), t, 'd')

486

487

488

def test_delete(self):

489

# TODO: Test Transport.delete

490

t = self.get_transport()

491

492

# Not much to do with a readonly transport

493

if t.is_readonly():

494

return

495

496

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

497

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

498

t.delete('a')

499

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

500

501

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

502

503

t.put('a', StringIO('a text\n'))

504

t.put('b', StringIO('b text\n'))

505

t.put('c', StringIO('c text\n'))

506

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

507

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

508

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

509

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

510

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

511

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

512

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

513

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

514

515

self.assertRaises(NoSuchFile,

516

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

517

518

self.assertRaises(NoSuchFile,

519

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

520

521

t.put('a', StringIO('another a text\n'))

522

t.put('c', StringIO('another c text\n'))

523

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

524

525

# We should have deleted everything

526

# SftpServer creates control files in the

527

# working directory, so we can just do a

528

# plain "listdir".

529

# self.assertEqual([], os.listdir('.'))

530

531

def test_move(self):

532

t = self.get_transport()

533

534

if t.is_readonly():

535

return

536

537

# TODO: I would like to use os.listdir() to

538

# make sure there are no extra files, but SftpServer

539

# creates control files in the working directory

540

# perhaps all of this could be done in a subdirectory

541

542

t.put('a', StringIO('a first file\n'))

543

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

544

545

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

546

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

547

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

548

549

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

550

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

551

552

# Overwrite a file

553

t.put('c', StringIO('c this file\n'))

554

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

555

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

556

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

557

558

# TODO: Try to write a test for atomicity

559

# TODO: Test moving into a non-existant subdirectory

560

# TODO: Test Transport.move_multi

561

562

def test_copy(self):

563

t = self.get_transport()

564

565

if t.is_readonly():

566

return

567

568

t.put('a', StringIO('a file\n'))

569

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

570

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

571

572

self.assertRaises(NoSuchFile, t.copy, 'c', 'd')

573

os.mkdir('c')

574

# What should the assert be if you try to copy a

575

# file over a directory?

576

#self.assertRaises(Something, t.copy, 'a', 'c')

577

t.put('d', StringIO('text in d\n'))

578

t.copy('d', 'b')

579

self.check_transport_contents('text in d\n', t, 'b')

580

581

# TODO: test copy_multi

582

583

def test_connection_error(self):

584

"""ConnectionError is raised when connection is impossible"""

585

try:

586

url = self._server.get_bogus_url()

587

except NotImplementedError:

588

raise TestSkipped("Transport %s has no bogus URL support." %

589

self._server.__class__)

590

t = bzrlib.transport.get_transport(url)

591

try:

592

t.get('.bzr/branch')

593

except (ConnectionError, NoSuchFile), e:

594

pass

595

except (Exception), e:

596

self.failIf(True, 'Wrong exception thrown: %s' % e)

597

else:

598

self.failIf(True, 'Did not get the expected exception.')

599

600

def test_stat(self):

601

# TODO: Test stat, just try once, and if it throws, stop testing

602

from stat import S_ISDIR, S_ISREG

603

604

t = self.get_transport()

605

606

try:

607

st = t.stat('.')

608

except TransportNotPossible, e:

609

# This transport cannot stat

610

return

611

612

paths = ['a', 'b/', 'b/c', 'b/d/', 'b/d/e']

613

sizes = [14, 0, 16, 0, 18]

614

self.build_tree(paths, transport=t)

615

616

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

617

st = t.stat(path)

618

if path.endswith('/'):

619

self.failUnless(S_ISDIR(st.st_mode))

620

# directory sizes are meaningless

621

else:

622

self.failUnless(S_ISREG(st.st_mode))

623

self.assertEqual(size, st.st_size)

624

625

remote_stats = list(t.stat_multi(paths))

626

remote_iter_stats = list(t.stat_multi(iter(paths)))

627

628

self.assertRaises(NoSuchFile, t.stat, 'q')

629

self.assertRaises(NoSuchFile, t.stat, 'b/a')

630

631

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

632

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

633

634

def test_list_dir(self):

635

# TODO: Test list_dir, just try once, and if it throws, stop testing

636

t = self.get_transport()

637

638

if not t.listable():

639

self.assertRaises(TransportNotPossible, t.list_dir, '.')

640

return

641

642

def sorted_list(d):

643

l = list(t.list_dir(d))

644

l.sort()

645

return l

646

647

# SftpServer creates control files in the working directory

648

# so lets move down a directory to avoid those.

649

t.mkdir('wd')

650

t = t.clone('wd')

651

652

self.assertEqual([], sorted_list(u'.'))

653

self.build_tree(['a', 'b', 'c/', 'c/d', 'c/e'], transport=t)

654

655

self.assertEqual([u'a', u'b', u'c'], sorted_list(u'.'))

656

self.assertEqual([u'd', u'e'], sorted_list(u'c'))

657

658

t.delete('c/d')

659

t.delete('b')

660

self.assertEqual([u'a', u'c'], sorted_list('.'))

661

self.assertEqual([u'e'], sorted_list(u'c'))

662

663

self.assertListRaises(NoSuchFile, t.list_dir, 'q')

664

self.assertListRaises(NoSuchFile, t.list_dir, 'c/f')

665

self.assertListRaises(NoSuchFile, t.list_dir, 'a')

666

667

def test_clone(self):

668

# TODO: Test that clone moves up and down the filesystem

669

t1 = self.get_transport()

670

671

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

672

673

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

674

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

675

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

676

677

t2 = t1.clone('b')

678

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

679

680

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

681

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

682

683

t3 = t2.clone('..')

684

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

685

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

686

687

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

688

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

689

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

690

691

if t1.is_readonly():

692

open('b/d', 'wb').write('newfile\n')

693

else:

694

t2.put('d', StringIO('newfile\n'))

695

696

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

697

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

698

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

699

700

def test_relpath(self):

701

t = self.get_transport()

702

self.assertEqual('', t.relpath(t.base))

703

# base ends with /

704

self.assertEqual('', t.relpath(t.base[:-1]))

705

# subdirs which dont exist should still give relpaths.

706

self.assertEqual('foo', t.relpath(t.base + 'foo'))

707

# trailing slash should be the same.

708

self.assertEqual('foo', t.relpath(t.base + 'foo/'))

709

710

def test_abspath(self):

711

# smoke test for abspath. Corner cases for backends like unix fs's

712

# that have aliasing problems like symlinks should go in backend

713

# specific test cases.

714

transport = self.get_transport()

715

self.assertEqual(transport.base + 'relpath',

716

transport.abspath('relpath'))

717

718

def test_iter_files_recursive(self):

719

transport = self.get_transport()

720

if not transport.listable():

721

self.assertRaises(TransportNotPossible,

722

transport.iter_files_recursive)

723

return

724

self.build_tree(['isolated/',

725

'isolated/dir/',

726

'isolated/dir/foo',

727

'isolated/dir/bar',

728

'isolated/bar'],

729

transport=transport)

730

transport = transport.clone('isolated')

731

paths = set(transport.iter_files_recursive())

732

self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)