171 lines
7.3 KiB
Python
171 lines
7.3 KiB
Python
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
class CatalogsyncEnvManagerTests(unittest.TestCase):
|
|
def test_load_current_strips_outer_quotes_from_command_values(self):
|
|
from musicdl.catalogsync.db import initialize_database
|
|
from musicdl.catalogsync.ops.config import CatalogsyncEnvManager
|
|
|
|
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
|
|
root = Path(tmpdir)
|
|
db_path = root / "catalogsync.db"
|
|
env_path = root / "catalogsync.env"
|
|
env_path.write_text(
|
|
"\n".join(
|
|
[
|
|
'CATALOG_EXPORT_COMMAND="sudo /volume4/Music_Cloud/Music_Server/bin/catalog-export-root.sh"',
|
|
"CATALOG_EXPORT_WORKDIR=/volume4/Music_Cloud/Music_Server",
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
initialize_database(db_path).close()
|
|
|
|
manager = CatalogsyncEnvManager(db_path=db_path, env_file_path=env_path)
|
|
current = manager.load_current()
|
|
|
|
self.assertEqual(
|
|
"sudo /volume4/Music_Cloud/Music_Server/bin/catalog-export-root.sh",
|
|
current["CATALOG_EXPORT_COMMAND"],
|
|
)
|
|
self.assertEqual(
|
|
"/volume4/Music_Cloud/Music_Server",
|
|
current["CATALOG_EXPORT_WORKDIR"],
|
|
)
|
|
|
|
def test_load_snapshot_save_revision_and_list_revisions(self):
|
|
from musicdl.catalogsync.db import initialize_database
|
|
from musicdl.catalogsync.ops.config import CatalogsyncEnvManager
|
|
|
|
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
|
|
root = Path(tmpdir)
|
|
db_path = root / "catalogsync.db"
|
|
env_path = root / "catalogsync.env"
|
|
env_path.write_text(
|
|
"\n".join(
|
|
[
|
|
"ROOT_DIR=/music",
|
|
"DOWNLOAD_SOURCES= qq, netease, , kuwo ,",
|
|
"LOG_DIR=/music/logs",
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
initialize_database(db_path).close()
|
|
|
|
manager = CatalogsyncEnvManager(db_path=db_path, env_file_path=env_path)
|
|
current = manager.load_current()
|
|
snapshot = manager.build_job_snapshot()
|
|
revision_id = manager.save_revision(note="initial import")
|
|
revisions = manager.list_revisions()
|
|
|
|
self.assertEqual("/music", current["ROOT_DIR"])
|
|
self.assertEqual(" qq, netease, , kuwo ,", current["DOWNLOAD_SOURCES"])
|
|
self.assertEqual(["qq", "netease", "kuwo"], snapshot["download_sources"])
|
|
self.assertEqual(revision_id, revisions[0]["id"])
|
|
self.assertEqual("initial import", revisions[0]["note"])
|
|
self.assertEqual("env_file", revisions[0]["source_type"])
|
|
self.assertTrue(revisions[0]["content_hash"])
|
|
self.assertIsNone(revisions[0]["applied_at"])
|
|
|
|
def test_apply_revision_overwrites_env_file_and_sets_applied_at(self):
|
|
from musicdl.catalogsync.db import initialize_database
|
|
from musicdl.catalogsync.ops.config import CatalogsyncEnvManager
|
|
|
|
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
|
|
root = Path(tmpdir)
|
|
db_path = root / "catalogsync.db"
|
|
env_path = root / "catalogsync.env"
|
|
original_env = "ROOT_DIR=/music\nDOWNLOAD_SOURCES=qq,netease\n"
|
|
env_path.write_text(original_env, encoding="utf-8")
|
|
initialize_database(db_path).close()
|
|
|
|
manager = CatalogsyncEnvManager(db_path=db_path, env_file_path=env_path)
|
|
revision_id = manager.save_revision(note="baseline")
|
|
env_path.write_text("ROOT_DIR=/other\nDOWNLOAD_SOURCES=migu\n", encoding="utf-8")
|
|
|
|
manager.apply_revision(revision_id)
|
|
|
|
restored = env_path.read_text(encoding="utf-8")
|
|
revisions = manager.list_revisions()
|
|
|
|
restored_revision = next(row for row in revisions if row["id"] == revision_id)
|
|
self.assertEqual(original_env, restored)
|
|
self.assertIsNotNone(restored_revision["applied_at"])
|
|
|
|
def test_save_revision_same_content_returns_same_id_without_new_row(self):
|
|
from musicdl.catalogsync.db import initialize_database
|
|
from musicdl.catalogsync.ops.config import CatalogsyncEnvManager
|
|
|
|
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
|
|
root = Path(tmpdir)
|
|
db_path = root / "catalogsync.db"
|
|
env_path = root / "catalogsync.env"
|
|
env_path.write_text("ROOT_DIR=/music\nDOWNLOAD_SOURCES=qq\n", encoding="utf-8")
|
|
initialize_database(db_path).close()
|
|
|
|
manager = CatalogsyncEnvManager(db_path=db_path, env_file_path=env_path)
|
|
first_id = manager.save_revision(note="first")
|
|
second_id = manager.save_revision(note="second")
|
|
revisions = manager.list_revisions()
|
|
|
|
self.assertEqual(first_id, second_id)
|
|
self.assertEqual(1, len(revisions))
|
|
|
|
def test_apply_revision_marks_only_target_and_keeps_first_applied_at(self):
|
|
from musicdl.catalogsync.db import initialize_database
|
|
from musicdl.catalogsync.ops.config import CatalogsyncEnvManager
|
|
|
|
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
|
|
root = Path(tmpdir)
|
|
db_path = root / "catalogsync.db"
|
|
env_path = root / "catalogsync.env"
|
|
initialize_database(db_path).close()
|
|
manager = CatalogsyncEnvManager(db_path=db_path, env_file_path=env_path)
|
|
|
|
env_path.write_text("ROOT_DIR=/v1\nDOWNLOAD_SOURCES=qq\n", encoding="utf-8")
|
|
revision_1 = manager.save_revision(note="v1")
|
|
env_path.write_text("ROOT_DIR=/v2\nDOWNLOAD_SOURCES=netease\n", encoding="utf-8")
|
|
revision_2 = manager.save_revision(note="v2")
|
|
|
|
manager.apply_revision(revision_1)
|
|
first_apply_state = manager.list_revisions()
|
|
first_applied_at = next(
|
|
row for row in first_apply_state if row["id"] == revision_1
|
|
)["applied_at"]
|
|
self.assertIsNotNone(first_applied_at)
|
|
untouched = next(row for row in first_apply_state if row["id"] == revision_2)
|
|
self.assertIsNone(untouched["applied_at"])
|
|
|
|
time.sleep(1.1)
|
|
manager.apply_revision(revision_1)
|
|
second_apply_state = manager.list_revisions()
|
|
second_applied_at = next(
|
|
row for row in second_apply_state if row["id"] == revision_1
|
|
)["applied_at"]
|
|
|
|
self.assertEqual(first_applied_at, second_applied_at)
|
|
|
|
def test_apply_revision_raises_for_missing_revision(self):
|
|
from musicdl.catalogsync.db import initialize_database
|
|
from musicdl.catalogsync.ops.config import CatalogsyncEnvManager
|
|
|
|
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdir:
|
|
root = Path(tmpdir)
|
|
db_path = root / "catalogsync.db"
|
|
env_path = root / "catalogsync.env"
|
|
initialize_database(db_path).close()
|
|
manager = CatalogsyncEnvManager(db_path=db_path, env_file_path=env_path)
|
|
|
|
with self.assertRaises(ValueError):
|
|
manager.apply_revision(999999)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|