Coverage for tests/test_library.py: 99%

1# -*- encoding: utf-8 -*- 

2"""Tests for the Nendo Core default library implementation.""" 

3 

4import os 

5import unittest 

6from types import GeneratorType 

7 

8from nendo import Nendo, NendoCollection, NendoConfig, NendoTrack 

9 

10nd = Nendo( 

11 config=NendoConfig( 

12 log_level="DEBUG", 

13 library_plugin="default", 

14 library_path="tests/library", 

15 copy_to_library=False, 

16 max_threads=1, 

17 plugins=[], 

18 stream_mode=False, 

19 stream_chunk_size=3, 

20 ), 

21) 

22 

23 

24class DefaultLibraryTests(unittest.TestCase): 

25 """Unit test class for testing the default (DuckDB) library.""" 

26 

27 def test_len_library(self): 

28 """Test `len(nd.library)`.""" 

29 nd.library.reset(force=True) 

30 nd.library.add_track(file_path="tests/assets/test.mp3") 

31 nd.library.add_track(file_path="tests/assets/test.wav") 

32 self.assertEqual(len(nd.library), 2) 

33 

34 def test_iter_library(self): 

35 """Test iterating over the library items.""" 

36 nd.library.reset(force=True) 

37 nd.library.add_track(file_path="tests/assets/test.mp3") 

38 nd.library.add_track(file_path="tests/assets/test.wav") 

39 for track in nd.library: 

40 self.assertTrue(isinstance(track, NendoTrack)) 

41 

42 def test_add_file_to_library(self): 

43 """Test adding a file to the library using the `add_track()` method.""" 

44 nd.library.reset(force=True) 

45 inserted_track = nd.library.add_track(file_path="tests/assets/test.wav") 

46 queried_track = nd.library.get_track(inserted_track.id) 

47 self.assertIsNotNone(queried_track) 

48 

49 queried_tracks = nd.library.get_tracks() 

50 self.assertEqual(len(queried_tracks), 1) 

51 

52 def test_add_related_to_library(self): 

53 """Test adding a related track to the library.""" 

54 nd.library.reset(force=True) 

55 inserted_track1 = nd.library.add_track(file_path="tests/assets/test.mp3") 

56 inserted_track2 = nd.library.add_related_track( 

57 file_path="tests/assets/test.wav", 

58 related_track_id=inserted_track1.id, 

59 relationship_type="stem", 

60 meta={"test": "value"}, 

61 ) 

62 self.assertTrue(inserted_track2.has_relationship("stem")) 

63 self.assertEqual(len(inserted_track2.related_tracks), 1) 

64 self.assertEqual(inserted_track2.related_tracks[0].meta, {"test": "value"}) 

65 

66 def test_add_track_relationship_with_track_ids_library(self): 

67 """Test the `add_track_relationship()` method.""" 

68 nd.library.reset(force=True) 

69 inserted_track1 = nd.library.add_track(file_path="tests/assets/test.mp3") 

70 inserted_track2 = nd.library.add_track(file_path="tests/assets/test.wav") 

71 

72 nd.library.add_track_relationship( 

73 track_one_id=inserted_track1.id, 

74 track_two_id=inserted_track2.id, 

75 relationship_type="stem", 

76 meta={"test": "value"}, 

77 ) 

78 

79 related_tracks = nd.library.get_related_tracks(inserted_track1.id) 

80 self.assertEqual(len(related_tracks), 2) 

81 

82 def test_get_tracks_returns_tracks(self): 

83 """Test the `nd.library.get_tracks()` method.""" 

84 nd.library.reset(force=True) 

85 nd.library.add_track(file_path="tests/assets/test.mp3") 

86 nd.library.add_track(file_path="tests/assets/test.wav") 

87 all_tracks = nd.library.get_tracks() 

88 self.assertEqual(len(all_tracks), 2) 

89 

90 def test_get_tracks_order_by_returns_asc_ordered_tracks(self): 

91 """Test the ordering of tracks returned by the `get_tracks()` method.""" 

92 nd.library.reset(force=True) 

93 nd.library.add_track(file_path="tests/assets/test.mp3") 

94 nd.library.add_track(file_path="tests/assets/test.wav") 

95 all_tracks = nd.library.get_tracks(order_by="created_at", order="asc") 

96 self.assertEqual(len(all_tracks), 2) 

97 self.assertTrue(all_tracks[1].created_at > all_tracks[0].created_at) 

98 

99 def test_get_tracks_order_by_returns_desc_ordered_tracks(self): 

100 """Test the ordering of tracks returned by the `get_tracks()` method.""" 

101 nd.library.reset(force=True) 

102 nd.library.add_track(file_path="tests/assets/test.mp3") 

103 nd.library.add_track(file_path="tests/assets/test.wav") 

104 all_tracks = nd.library.get_tracks(order_by="created_at", order="desc") 

105 self.assertEqual(len(all_tracks), 2) 

106 self.assertTrue(all_tracks[1].created_at < all_tracks[0].created_at) 

107 

108 def test_get_tracks_returns_limited_offset_tracks(self): 

109 """Test the limit/offset functionality of `nd.library.get_tracks()`.""" 

110 nd.library.reset(force=True) 

111 nd.library.add_track(file_path="tests/assets/test.mp3") 

112 nd.library.add_track(file_path="tests/assets/test.wav") 

113 limit_tracks = nd.library.get_tracks(limit=1) 

114 offset_tracks = nd.library.get_tracks(limit=1, offset=1) 

115 

116 self.assertEqual(len(limit_tracks), 1) 

117 self.assertEqual(len(offset_tracks), 1) 

118 self.assertNotEqual(limit_tracks, offset_tracks) 

119 

120 def test_get_tracks_returns_limit_tracks(self): 

121 """Test the limit argument of `nd.library.get_tracks()`.""" 

122 nd.library.reset(force=True) 

123 nd.library.add_track(file_path="tests/assets/test.mp3") 

124 nd.library.add_track(file_path="tests/assets/test.wav") 

125 all_tracks = nd.library.get_tracks(limit=2) 

126 self.assertEqual(len(all_tracks), 2) 

127 

128 def test_filter_tracks_returns_filtered_tracks(self): 

129 """Test filtering of tracks.""" 

130 nd.library.reset(force=True) 

131 inserted_track1 = nd.library.add_track(file_path="tests/assets/test.mp3") 

132 nd.library.add_track(file_path="tests/assets/test.wav") 

133 nd.library.add_plugin_data( 

134 track_id=inserted_track1.id, 

135 plugin_name="test_plugin", 

136 plugin_version="1.0", 

137 key="test", 

138 value="value", 

139 ) 

140 # with explicit plugin_names 

141 retrieved_tracks = nd.library.filter_tracks( 

142 filters={"test": "value"}, 

143 plugin_names=["test_plugin"], 

144 ) 

145 self.assertEqual(len(retrieved_tracks), 1) 

146 self.assertEqual(retrieved_tracks[0].id, inserted_track1.id) 

147 # without plugin_names should just return all 

148 retrieved_tracks = nd.library.filter_tracks( 

149 filters={"test": "value"}, 

150 ) 

151 self.assertEqual(len(retrieved_tracks), 1) 

152 self.assertEqual(retrieved_tracks[0].id, inserted_track1.id) 

153 # with empty plugin_names list should also return all 

154 retrieved_tracks = nd.library.filter_tracks( 

155 filters={"test": "value"}, 

156 plugin_names=[], 

157 ) 

158 self.assertEqual(len(retrieved_tracks), 1) 

159 self.assertEqual(retrieved_tracks[0].id, inserted_track1.id) 

160 

161 def test_filter_by_track_type(self): 

162 """Test filtering by track type.""" 

163 nd.library.reset(force=True) 

164 test_track_1 = nd.library.add_track( 

165 file_path="tests/assets/test.mp3", 

166 track_type="stem", 

167 ) 

168 nd.library.add_track(file_path="tests/assets/test.wav", track_type="track") 

169 result = nd.library.filter_tracks(track_type="stem") 

170 self.assertEqual(len(result), 1) 

171 self.assertEqual(result[0].id, test_track_1.id) 

172 result = nd.library.filter_tracks(track_type=["stem", "track"]) 

173 self.assertEqual(len(result), 2) 

174 

175 def test_get_tracks_filtered_by_collection(self): 

176 """Test filtering of tracks by collection.""" 

177 nd.library.reset(force=True) 

178 inserted_track1 = nd.library.add_track(file_path="tests/assets/test.mp3") 

179 nd.library.add_track(file_path="tests/assets/test.wav") 

180 

181 collection = nd.library.add_collection( 

182 name="test_collection_filter", 

183 track_ids=[inserted_track1.id], 

184 ) 

185 

186 all_tracks = nd.library.filter_tracks(collection_id=collection.id) 

187 self.assertEqual(len(all_tracks), 1) 

188 

189 def test_find_in_library(self): 

190 """Test the function for finding tracks in the library.""" 

191 nd.library.reset(force=True) 

192 inserted_track1 = nd.library.add_track(file_path="tests/assets/test.mp3") 

193 inserted_track2 = nd.library.add_track(file_path="tests/assets/test.wav") 

194 result = nd.library.find_tracks(value="Test Artist") 

195 self.assertEqual(len(result), 2) 

196 self.assertEqual(result[0].id, inserted_track1.id) 

197 result = nd.library.find_tracks(value="test.wav") 

198 self.assertEqual(len(result), 1) 

199 self.assertEqual(result[0].id, inserted_track2.id) 

200 

201 def test_find_related_tracks_in_library(self): 

202 """Test the finding of related tracks in the library.""" 

203 nd.library.reset(force=True) 

204 inserted_track1 = nd.library.add_track(file_path="tests/assets/test.mp3") 

205 nd.library.add_related_track( 

206 file_path="tests/assets/test.wav", 

207 related_track_id=inserted_track1.id, 

208 relationship_type="stem", 

209 meta={"test": "value"}, 

210 ) 

211 nd.library.add_related_track( 

212 file_path="tests/assets/test.mp3", 

213 related_track_id=inserted_track1.id, 

214 relationship_type="stem", 

215 meta={"test": "value"}, 

216 ) 

217 

218 related_tracks = nd.library.get_related_tracks(inserted_track1.id) 

219 self.assertEqual(len(related_tracks), 2) 

220 

221 def test_add_file_without_conversion(self): 

222 """Test adding a file to the library without conversion.""" 

223 nd.config.copy_to_library = True 

224 nd.config.auto_convert = False 

225 nd.library.reset(force=True) 

226 inserted_track = nd.library.add_track(file_path="tests/assets/test.mp3") 

227 self.assertEqual(os.path.splitext(inserted_track.resource.src)[1], ".mp3") 

228 nd.config.auto_convert = True 

229 nd.config.copy_to_library = False 

230 

231 def test_add_file_skip_duplicate(self): 

232 """Test the `skip_duplicate` config variable.""" 

233 nd.config.skip_duplicate = False 

234 nd.library.reset(force=True) 

235 inserted_track_1 = nd.library.add_track(file_path="tests/assets/test.mp3") 

236 inserted_track_2 = nd.library.add_track( 

237 file_path="tests/assets/test.mp3", 

238 skip_duplicate=True, 

239 ) 

240 self.assertEqual(len(nd.library), 1) 

241 self.assertEqual(inserted_track_1.id, inserted_track_2.id) 

242 _ = nd.library.add_track(file_path="tests/assets/test.mp3") 

243 self.assertEqual(len(nd.library), 2) 

244 nd.config.skip_duplicate = True 

245 

246 def test_add_file_stores_file_namename(self): 

247 """Test the `copy_to_library` config variable.""" 

248 nd.config.copy_to_library = True 

249 nd.library.reset(force=True) 

250 inserted_track = nd.library.add_track(file_path="tests/assets/test.wav") 

251 queried_track = nd.library.get_track(inserted_track.id) 

252 self.assertEqual(queried_track.resource.meta["original_filename"], "test.wav") 

253 nd.config.copy_to_library = False 

254 

255 def test_add_tracks_adds_all_files_in_folder(self): 

256 """Test the `nd.library.add_tracks()` function.""" 

257 nd.library.reset(force=True) 

258 nd.library.add_tracks(path="tests/assets") 

259 results = nd.library.get_tracks() 

260 self.assertEqual(len(results), 3) 

261 results = nd.library.get_tracks(limit=1) 

262 self.assertEqual(len(results), 1) 

263 

264 def test_remove_file_from_library(self): 

265 """Test the `nd.library.remove_track()` function.""" 

266 nd.library.reset(force=True) 

267 inserted_track = nd.library.add_track(file_path="tests/assets/test.mp3") 

268 results_before_remove = nd.library.find_tracks(value="Test Artist") 

269 nd.library.remove_track(track_id=inserted_track.id) 

270 results_after_remove = nd.library.find_tracks(value="Test Artist") 

271 self.assertTrue(len(results_before_remove) > len(results_after_remove)) 

272 self.assertFalse(os.path.exists(inserted_track.resource.src)) 

273 

274 def test_remove_file_with_relations_returns_error(self): 

275 """Test removal of tracks with existing relations (without forcing).""" 

276 nd.library.reset(force=True) 

277 inserted_track1 = nd.library.add_track(file_path="tests/assets/test.mp3") 

278 inserted_track2 = nd.library.add_related_track( 

279 file_path="tests/assets/test.wav", 

280 related_track_id=inserted_track1.id, 

281 relationship_type="stem", 

282 meta={"test": "value"}, 

283 ) 

284 inserted_track3 = nd.library.add_related_track( 

285 file_path="tests/assets/test.wav", 

286 related_track_id=inserted_track1.id, 

287 relationship_type="stem", 

288 meta={"test": "value"}, 

289 ) 

290 

291 result = nd.library.remove_track(inserted_track1.id, remove_relationships=False) 

292 track_1_after_remove = nd.library.get_track(inserted_track1.id) 

293 self.assertTrue(track_1_after_remove.id == inserted_track1.id) 

294 

295 track_2_after_remove = nd.library.get_track(inserted_track2.id) 

296 self.assertTrue(track_2_after_remove.id == inserted_track2.id) 

297 

298 track_3_after_remove = nd.library.get_track(inserted_track3.id) 

299 self.assertTrue(track_3_after_remove.id == inserted_track3.id) 

300 

301 self.assertFalse(result) 

302 

303 def test_remove_file_with_relations_removes_relations(self): 

304 """Test the removal of tracks with relations (with forcing).""" 

305 nd.config.skip_duplicate = False 

306 nd.library.reset(force=True) 

307 inserted_track1 = nd.library.add_track(file_path="tests/assets/test.wav") 

308 inserted_track2 = nd.library.add_related_track( 

309 file_path="tests/assets/test.wav", 

310 related_track_id=inserted_track1.id, 

311 relationship_type="stem", 

312 meta={"test": "value"}, 

313 ) 

314 inserted_track3 = nd.library.add_related_track( 

315 file_path="tests/assets/test.wav", 

316 related_track_id=inserted_track1.id, 

317 relationship_type="stem", 

318 meta={"test": "value"}, 

319 ) 

320 

321 # confirm that the related_tracks exist 

322 inserted_track2 = nd.library.get_track(inserted_track2.id) 

323 self.assertTrue(inserted_track2.has_relationship("stem")) 

324 

325 inserted_track3 = nd.library.get_track(inserted_track3.id) 

326 self.assertTrue(inserted_track3.has_relationship("stem")) 

327 

328 result = nd.library.remove_track(inserted_track1.id, remove_relationships=True) 

329 inserted_track1 = nd.library.get_track(inserted_track1.id) 

330 

331 self.assertIsNone(inserted_track1) 

332 self.assertTrue(result) 

333 

334 # confirm that the related_tracks are removed 

335 inserted_track2 = nd.library.get_track(inserted_track2.id) 

336 self.assertFalse(inserted_track2.has_relationship("stem")) 

337 

338 inserted_track3 = nd.library.get_track(inserted_track3.id) 

339 self.assertFalse(inserted_track3.has_relationship("stem")) 

340 

341 nd.config.skip_duplicate = True 

342 

343 def test_add_collection_adds_collection(self): 

344 """Test the `nd.library.add_collection()` method.""" 

345 nd.library.reset(force=True) 

346 

347 test_track_1 = nd.library.add_track(file_path="tests/assets/test.mp3") 

348 test_track_2 = nd.library.add_track(file_path="tests/assets/test.wav") 

349 test_collection = nd.library.add_collection( 

350 track_ids=[test_track_1.id, test_track_2.id], 

351 name="Testcollection", 

352 ) 

353 self.assertEqual(test_collection.name, "Testcollection") 

354 retrieved_test_collection = nd.library.get_collection( 

355 collection_id=test_collection.id, 

356 ) 

357 self.assertEqual(test_collection.name, retrieved_test_collection.name) 

358 self.assertEqual( 

359 len(test_collection.related_tracks), 

360 len(retrieved_test_collection.related_tracks), 

361 ) 

362 rts = [rt.source.id for rt in retrieved_test_collection.related_tracks] 

363 self.assertTrue(test_track_1.id in rts) 

364 

365 def test_add_track_to_collection(self): 

366 """Test the adding of tracks to collections.""" 

367 nd.library.reset(force=True) 

368 test_track_1 = nd.library.add_track(file_path="tests/assets/test.mp3") 

369 test_track_2 = nd.library.add_track(file_path="tests/assets/test.wav") 

370 test_collection = nd.library.add_collection( 

371 track_ids=[test_track_1.id], 

372 name="Testcollection", 

373 ) 

374 self.assertEqual(len(test_collection), 1) 

375 nd.library.add_track_to_collection( 

376 track_id=test_track_2.id, 

377 collection_id=test_collection.id, 

378 ) 

379 retrieved_test_collection = nd.library.get_collection( 

380 collection_id=test_collection.id, 

381 ) 

382 self.assertEqual(len(retrieved_test_collection), 2) 

383 

384 def test_find_collections_by_empty_value(self): 

385 """Test finding collections by empty search value.""" 

386 nd.library.reset(force=True) 

387 test_track_1 = nd.library.add_track(file_path="tests/assets/test.mp3") 

388 test_track_2 = nd.library.add_track(file_path="tests/assets/test.wav") 

389 test_collection = nd.library.add_collection( 

390 track_ids=[test_track_1.id, test_track_2.id], 

391 name="Testcollection", 

392 ) 

393 test_collection_2 = nd.library.add_collection( 

394 track_ids=[test_track_1.id, test_track_2.id], 

395 name="Testcollection2", 

396 ) 

397 retrieved_test_collection = nd.library.find_collections() 

398 self.assertEqual(len(retrieved_test_collection), 2) 

399 self.assertEqual(retrieved_test_collection[0].id, test_collection.id) 

400 self.assertEqual(retrieved_test_collection[1].id, test_collection_2.id) 

401 

402 def test_find_collections_by_collection_name(self): 

403 """Test finding collections by name as search value.""" 

404 nd.library.reset(force=True) 

405 test_track_1 = nd.library.add_track(file_path="tests/assets/test.mp3") 

406 test_track_2 = nd.library.add_track(file_path="tests/assets/test.wav") 

407 test_collection = nd.library.add_collection( 

408 track_ids=[test_track_1.id, test_track_2.id], 

409 name="Testcollection", 

410 ) 

411 nd.library.add_collection( 

412 track_ids=[test_track_1.id, test_track_2.id], 

413 name="iwontbefoundcollection", 

414 ) 

415 retrieved_test_collection = nd.library.find_collections("Testc") 

416 self.assertEqual(len(retrieved_test_collection), 1) 

417 self.assertEqual(retrieved_test_collection[0].id, test_collection.id) 

418 

419 def test_get_all_collections(self): 

420 """Test getting all collections.""" 

421 nd.library.reset(force=True) 

422 test_track_1 = nd.library.add_track(file_path="tests/assets/test.mp3") 

423 test_track_2 = nd.library.add_track(file_path="tests/assets/test.wav") 

424 test_collection = nd.library.add_collection( 

425 track_ids=[test_track_1.id, test_track_2.id], 

426 name="Testcollection", 

427 ) 

428 test_collection_2 = nd.library.add_collection( 

429 track_ids=[test_track_1.id, test_track_2.id], 

430 name="Testcollection2", 

431 ) 

432 all_collections = nd.library.get_collections() 

433 self.assertEqual(len(all_collections), 2) 

434 self.assertEqual(all_collections[0].id, test_collection.id) 

435 self.assertEqual(all_collections[1].id, test_collection_2.id) 

436 

437 def test_remove_collection_without_relationships(self): 

438 """Test removing of collections that have no relationships.""" 

439 nd.library.reset(force=True) 

440 test_track_1 = nd.library.add_track(file_path="tests/assets/test.mp3") 

441 test_track_2 = nd.library.add_track(file_path="tests/assets/test.wav") 

442 test_collection = nd.library.add_collection( 

443 track_ids=[test_track_1.id, test_track_2.id], 

444 name="Testcollection", 

445 ) 

446 self.assertTrue( 

447 nd.library.remove_collection( 

448 collection_id=test_collection.id, 

449 remove_relationships=False, 

450 ), 

451 ) 

452 

453 def test_remove_collection_with_relationships_failing(self): 

454 """Test the removing of collections with relationships (without forcing).""" 

455 nd.library.reset(force=True) 

456 test_track_1 = nd.library.add_track(file_path="tests/assets/test.mp3") 

457 test_track_2 = nd.library.add_track(file_path="tests/assets/test.wav") 

458 test_collection_1 = nd.library.add_collection( 

459 track_ids=[test_track_1.id, test_track_2.id], 

460 name="Testcollection1", 

461 ) 

462 nd.library.add_related_collection( 

463 track_ids=[test_track_1.id, test_track_2.id], 

464 collection_id=test_collection_1.id, 

465 name="Testcollection2", 

466 ) 

467 self.assertFalse( 

468 nd.library.remove_collection( 

469 collection_id=test_collection_1.id, 

470 remove_relationships=False, 

471 ), 

472 ) 

473 

474 def test_remove_collection_with_relationships_succeeding(self): 

475 """Test the removing of collections with relationships (with forcing).""" 

476 nd.library.reset(force=True) 

477 test_track_1 = nd.library.add_track(file_path="tests/assets/test.mp3") 

478 test_track_2 = nd.library.add_track(file_path="tests/assets/test.wav") 

479 test_collection_1 = nd.library.add_collection( 

480 track_ids=[test_track_1.id, test_track_2.id], 

481 name="Testcollection", 

482 ) 

483 nd.library.add_related_collection( 

484 track_ids=[test_track_1.id, test_track_2.id], 

485 collection_id=test_collection_1.id, 

486 name="Testcollection2", 

487 ) 

488 self.assertTrue( 

489 nd.library.remove_collection( 

490 collection_id=test_collection_1.id, 

491 remove_relationships=True, 

492 ), 

493 ) 

494 

495 def test_remove_track_updates_collections(self): 

496 """Test that removing tracks also removes them from relevant collections.""" 

497 nd.library.reset(force=True) 

498 test_track_1 = nd.library.add_track(file_path="tests/assets/test.mp3") 

499 test_track_2 = nd.library.add_track(file_path="tests/assets/test.wav") 

500 test_collection = nd.library.add_collection(name="Testcollection") 

501 nd.library.add_track_to_collection( 

502 collection_id=test_collection.id, 

503 track_id=test_track_1.id, 

504 ) 

505 nd.library.add_track_to_collection( 

506 collection_id=test_collection.id, 

507 track_id=test_track_2.id, 

508 ) 

509 

510 # Remove the first test track 

511 nd.library.remove_track(track_id=test_track_1.id, remove_relationships=True) 

512 

513 # Get the tracks from the collection 

514 test_collection = nd.library.get_collection(collection_id=test_collection.id) 

515 

516 # Assert that the second track was removed and 

517 # positions of other tracks were updated 

518 self.assertIsNotNone(test_collection.related_tracks[0]) 

519 self.assertIsNotNone(test_collection.related_tracks[0].relationship_position) 

520 self.assertEqual(test_collection.related_tracks[0].relationship_position, 0) 

521 

522 def test_remove_track_from_collection_updates_positions(self): 

523 """Test that removing tracks updates the positions in affected collections.""" 

524 nd.library.reset(force=True) 

525 test_track_1 = nd.library.add_track(file_path="tests/assets/test.mp3") 

526 test_track_2 = nd.library.add_track(file_path="tests/assets/test.wav") 

527 test_collection = nd.library.add_collection(name="Testcollection") 

528 nd.library.add_track_to_collection( 

529 collection_id=test_collection.id, 

530 track_id=test_track_1.id, 

531 position=0, 

532 ) 

533 nd.library.add_track_to_collection( 

534 collection_id=test_collection.id, 

535 track_id=test_track_2.id, 

536 position=1, 

537 ) 

538 

539 # Remove the first track from the collection 

540 nd.library.remove_track_from_collection( 

541 track_id=test_track_1.id, 

542 collection_id=test_collection.id, 

543 ) 

544 

545 # Get the tracks from the collection 

546 test_collection = nd.library.get_collection(collection_id=test_collection.id) 

547 

548 # Assert that the second track was removed and 

549 # positions of other tracks were updated 

550 self.assertIsNotNone(test_collection.related_tracks[0]) 

551 self.assertIsNotNone(test_collection.related_tracks[0].relationship_position) 

552 self.assertEqual(test_collection.related_tracks[0].relationship_position, 0) 

553 

554 def test_library_get_tracks_stream(self): 

555 """Test the `stream_mode` functionality of Nendo.""" 

556 # with stream_chunk_size > 1, the generator should return 

557 # chunks (lists) of tracks 

558 nd.config.skip_duplicate = False 

559 nd.config.stream_mode = True 

560 nd.config.stream_chunk_size = 4 

561 nd.library.reset(force=True) 

562 nd.library.add_track(file_path="tests/assets/test.wav") 

563 nd.library.add_track(file_path="tests/assets/test.wav") 

564 nd.library.add_track(file_path="tests/assets/test.wav") 

565 nd.library.add_track(file_path="tests/assets/test.wav") 

566 nd.library.add_track(file_path="tests/assets/test.wav") 

567 tracks_iterator = nd.library.get_tracks() 

568 self.assertEqual(type(tracks_iterator), GeneratorType) 

569 i = 0 

570 for chunk in tracks_iterator: 

571 self.assertEqual(type(chunk), list) 

572 self.assertEqual(type(chunk[0]), NendoTrack) 

573 if i == 0: 

574 self.assertEqual(len(chunk), 4) 

575 else: 

576 self.assertEqual(len(chunk), 1) 

577 i += 1 

578 # with stream_chunk_size == 1, the generator should return 

579 # individual tracks 

580 nd.config.stream_chunk_size = 1 

581 tracks_iterator = nd.library.get_tracks() 

582 for chunk in tracks_iterator: 

583 self.assertEqual(type(chunk), NendoTrack) 

584 nd.config.skip_duplicate = True 

585 nd.config.stream_mode = False 

586 nd.config.stream_chunk_size = 1 

587 

588 def test_get_track_or_collection(self): 

589 """Test the `nd.library.get_track_or_collection()` method.""" 

590 nd.library.reset(force=True) 

591 test_track = nd.library.add_track(file_path="tests/assets/test.mp3") 

592 test_collection = nd.library.add_collection( 

593 track_ids=[test_track.id], 

594 name="Testcollection", 

595 ) 

596 result = nd.library.get_track_or_collection(target_id=test_track.id) 

597 self.assertTrue(isinstance(result, NendoTrack)) 

598 self.assertEqual(result.id, test_track.id) 

599 result = nd.library.get_track_or_collection(target_id=test_collection.id) 

600 self.assertTrue(isinstance(result, NendoCollection)) 

601 self.assertEqual(result.id, test_collection.id) 

602 

603 def test_store_blob(self): 

604 """Test the `nd.library.store_blob()` function.""" 

605 nd.library.reset(force=True) 

606 

607 test_blob = nd.library.store_blob(file_path="tests/assets/test.wav") 

608 test_blob_2 = nd.library.store_blob(file_path="tests/assets/test.mp3") 

609 self.assertIsNone(test_blob.data) 

610 self.assertIsNone(test_blob_2.data) 

611 self.assertTrue(os.path.isfile(test_blob.resource.src)) 

612 self.assertTrue(os.path.isfile(test_blob_2.resource.src)) 

613 

614 def test_store_blob_from_bytes(self): 

615 """Test the storing of blobs from `bytes` objects.""" 

616 nd.library.reset(force=True) 

617 

618 data_1 = b"test_blob" 

619 data_2 = b"test_blob_2" 

620 

621 test_blob = nd.library.store_blob_from_bytes(data=data_1) 

622 test_blob_2 = nd.library.store_blob_from_bytes(data=data_2) 

623 self.assertIsNone(test_blob.data) 

624 self.assertIsNone(test_blob_2.data) 

625 self.assertTrue(os.path.isfile(test_blob.resource.src)) 

626 self.assertTrue(os.path.isfile(test_blob_2.resource.src)) 

627 

628 def test_get_blob(self): 

629 """Test the retrieval of blobs using `nd.library.get_blob()`.""" 

630 nd.library.reset(force=True) 

631 

632 data_1 = b"test_blob" 

633 data_2 = b"test_blob_2" 

634 

635 test_blob = nd.library.store_blob_from_bytes(data=data_1) 

636 test_blob_2 = nd.library.store_blob_from_bytes(data=data_2) 

637 

638 test_blob = nd.library.load_blob(blob_id=test_blob.id) 

639 self.assertEqual(test_blob.data, data_1) 

640 

641 test_blob_2 = nd.library.load_blob(blob_id=test_blob_2.id) 

642 self.assertEqual(test_blob_2.data, data_2) 

643 self.assertTrue(os.path.isfile(test_blob.resource.src)) 

644 self.assertTrue(os.path.isfile(test_blob_2.resource.src)) 

645 

646 def test_remove_blob(self): 

647 """Test the removal of blobs using `nd.library.remove_blob()`.""" 

648 nd.library.reset(force=True) 

649 

650 data_1 = b"test_blob" 

651 data_2 = b"test_blob_2" 

652 

653 test_blob = nd.library.store_blob_from_bytes(data=data_1) 

654 test_blob_2 = nd.library.store_blob_from_bytes(data=data_2) 

655 

656 test_blob = nd.library.load_blob(blob_id=test_blob.id) 

657 self.assertEqual(test_blob.data, data_1) 

658 

659 test_blob_2 = nd.library.load_blob(blob_id=test_blob_2.id) 

660 self.assertEqual(test_blob_2.data, data_2) 

661 self.assertTrue(os.path.isfile(test_blob.resource.src)) 

662 self.assertTrue(os.path.isfile(test_blob_2.resource.src)) 

663 

664 nd.library.remove_blob(blob_id=test_blob.id) 

665 nd.library.remove_blob(blob_id=test_blob_2.id) 

666 self.assertFalse(os.path.isfile(test_blob.resource.src)) 

667 self.assertFalse(os.path.isfile(test_blob_2.resource.src)) 

668 

669 def test_verify_delete_and_ignore(self): 

670 """Test the `nd.library.verify()` method.""" 

671 nd.library.reset(force=True) 

672 # test verification of orphaned DB entries 

673 test_track_1 = nd.library.add_track(file_path="tests/assets/test.mp3") 

674 nd.library.storage_driver.remove_file( 

675 file_name=test_track_1.resource.file_name, 

676 user_id=nd.config.user_id, 

677 ) 

678 # first, ignore inconsistency 

679 nd.library.verify(action="i", user_id=nd.config.user_id) 

680 self.assertEqual( 

681 nd.library.get_track(track_id=test_track_1.id).id, 

682 test_track_1.id,