error = False
if file or url:
+
+ # Clear pending dreams import
+ if song_data["queueid"] is not None:
+ dreams_importer.delete_from_queue(song_data["queueid"])
+
with tempfile.NamedTemporaryFile(delete=False) as tmp_file:
passed = convert_song(tmp_file, file, url, upload_type)
[collab, songid])
if upload_type == "dreams" and url:
- if song_data["queueid"] is not None:
- dreams_importer.delete_from_queue(song_data["queueid"])
duration = duration_in_seconds(song_duration)
dreams_importer.add_to_queue(songid, url, duration)
assert response.json["next"]["indreamsurl"] == TEST_URL + "_new"
assert response.json["next"]["songid"] == 1
+def test_import_canceled_by_file(client, user):
+ upload_song(
+ client, b"Queued for import from Dreams",
+ upload_type="dreams",
+ song_url=TEST_URL)
+
+ # Initial queue position
+ response = client.get(f"/users/user")
+ assert b"[Hidden]" in response.data
+ assert b"[Queue Pos: 1]" in response.data
+
+ upload_song(client, b"Success", songid=1)
+
+ # Queue gone now that file was used
+ response = client.get(f"/users/user")
+ assert b"[Hidden]" not in response.data
+ assert b"[Queue Pos: 1]" not in response.data
+ response = client.get("/dreams-importer/next-in-queue")
+ assert response.json["next"] is None
+