132 lines
6.4 KiB
Python
132 lines
6.4 KiB
Python
# dabdatasync (c) by chacha
|
|
#
|
|
# dabdatasync is licensed under a
|
|
# Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International Unported License.
|
|
#
|
|
# You should have received a copy of the license along with this
|
|
# work. If not, see <https://creativecommons.org/licenses/by-nc-sa/4.0/>.
|
|
|
|
import unittest
|
|
from os import chdir
|
|
from pathlib import Path
|
|
import pprint
|
|
import shutil
|
|
|
|
print(__name__)
|
|
print(__package__)
|
|
|
|
from src import dabdatasync
|
|
|
|
testdir_path = Path(__file__).parent.resolve()
|
|
chdir(testdir_path.parent.resolve())
|
|
|
|
|
|
class TestDabDataSync(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
chdir(testdir_path.parent.resolve())
|
|
shutil.rmtree(testdir_path / "test_data", ignore_errors=True)
|
|
shutil.rmtree(testdir_path / "test_data2", ignore_errors=True)
|
|
shutil.copytree(testdir_path / "test_data_origin", testdir_path / "test_data")
|
|
shutil.copytree(testdir_path / "test_data_origin", testdir_path / "test_data2")
|
|
|
|
def tearDown(self) -> None:
|
|
shutil.rmtree(testdir_path / "test_data", ignore_errors=True)
|
|
shutil.rmtree(testdir_path / "test_data2", ignore_errors=True)
|
|
|
|
def test_version(self):
|
|
self.assertNotEqual(dabdatasync.__version__, "?.?.?")
|
|
|
|
def test_load_nextcloud(self):
|
|
dabdatasync.I_DataSync.manifest_path = testdir_path / "test_manifest_nextcloud.json"
|
|
datasync = dabdatasync.DataSync_Factory.get_DataSync()
|
|
self.assertIsInstance(datasync, dabdatasync.I_DataSync)
|
|
self.assertIsInstance(datasync, dabdatasync.C_DataSync_NextCloud)
|
|
|
|
self.assertEqual(len(datasync.ar_datasync_record), 2)
|
|
self.assertEqual(datasync.ar_datasync_record[0].name, "SOTF_map")
|
|
self.assertEqual(datasync.ar_datasync_record[1].name, "SOTF_map2")
|
|
self.assertEqual(datasync.ar_datasync_record[0].rec_type, "fs")
|
|
self.assertEqual(datasync.ar_datasync_record[1].rec_type, "fs")
|
|
self.assertEqual(datasync.ar_datasync_record[0].value, "test/test_data")
|
|
self.assertEqual(datasync.ar_datasync_record[1].value, "test/test_data2/SAVE_FILE.txt")
|
|
self.assertEqual(datasync.ar_datasync_record[0].suffix, ".tar.gz")
|
|
self.assertEqual(datasync.ar_datasync_record[1].suffix, ".tar.gz")
|
|
|
|
with open(testdir_path / "test_data" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
|
self.assertEqual(testfile.read(), "SAVED_VALUE")
|
|
with open(testdir_path / "test_data2" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
|
self.assertEqual(testfile.read(), "SAVED_VALUE")
|
|
|
|
datasync.write_data()
|
|
|
|
with open(testdir_path / "test_data" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
|
self.assertEqual(testfile.read(), "SAVED_VALUE")
|
|
with open(testdir_path / "test_data2" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
|
self.assertEqual(testfile.read(), "SAVED_VALUE")
|
|
|
|
with open(testdir_path / "test_data" / "SAVE_FILE.txt", "w", encoding="utf-8") as testfile:
|
|
testfile.write("MODIFIED_VALUE")
|
|
with open(testdir_path / "test_data2" / "SAVE_FILE.txt", "w", encoding="utf-8") as testfile:
|
|
testfile.write("MODIFIED_VALUE2")
|
|
|
|
with open(testdir_path / "test_data" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
|
self.assertEqual(testfile.read(), "MODIFIED_VALUE")
|
|
with open(testdir_path / "test_data2" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
|
self.assertEqual(testfile.read(), "MODIFIED_VALUE2")
|
|
|
|
datasync.read_data()
|
|
|
|
with open(testdir_path / "test_data" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
|
self.assertEqual(testfile.read(), "SAVED_VALUE")
|
|
with open(testdir_path / "test_data2" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
|
self.assertEqual(testfile.read(), "SAVED_VALUE")
|
|
|
|
with open(testdir_path / "test_data" / "SAVE_FILE.txt", "w", encoding="utf-8") as testfile:
|
|
testfile.write("MODIFIED_VALUE3")
|
|
with open(testdir_path / "test_data2" / "SAVE_FILE.txt", "w", encoding="utf-8") as testfile:
|
|
testfile.write("MODIFIED_VALUE32")
|
|
|
|
datasync.write_data()
|
|
|
|
with open(testdir_path / "test_data" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
|
self.assertEqual(testfile.read(), "MODIFIED_VALUE3")
|
|
with open(testdir_path / "test_data2" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
|
self.assertEqual(testfile.read(), "MODIFIED_VALUE32")
|
|
|
|
with open(testdir_path / "test_data" / "SAVE_FILE.txt", "w", encoding="utf-8") as testfile:
|
|
testfile.write("MODIFIED_VALUE")
|
|
with open(testdir_path / "test_data2" / "SAVE_FILE.txt", "w", encoding="utf-8") as testfile:
|
|
testfile.write("MODIFIED_VALUE")
|
|
|
|
with open(testdir_path / "test_data" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
|
self.assertEqual(testfile.read(), "MODIFIED_VALUE")
|
|
with open(testdir_path / "test_data2" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
|
self.assertEqual(testfile.read(), "MODIFIED_VALUE")
|
|
|
|
datasync.read_data()
|
|
|
|
with open(testdir_path / "test_data" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
|
self.assertEqual(testfile.read(), "MODIFIED_VALUE3")
|
|
with open(testdir_path / "test_data2" / "SAVE_FILE.txt", "rt", encoding="utf-8") as testfile:
|
|
self.assertEqual(testfile.read(), "MODIFIED_VALUE32")
|
|
|
|
def test_load_empty(self):
|
|
dabdatasync.I_DataSync.manifest_path = testdir_path / "test_manifest_empty.json"
|
|
datasync = dabdatasync.DataSync_Factory.get_DataSync()
|
|
self.assertIsNone(datasync)
|
|
|
|
def test_load_nextcloud_disabled(self):
|
|
dabdatasync.I_DataSync.manifest_path = testdir_path / "test_manifest_nextcloud_disabled.json"
|
|
datasync = dabdatasync.DataSync_Factory.get_DataSync()
|
|
self.assertIsNone(datasync)
|
|
|
|
def test_load_invalid(self):
|
|
dabdatasync.I_DataSync.manifest_path = testdir_path / "test_manifest_invalid.json"
|
|
with self.assertRaises(dabdatasync.DataSyncException_InvalidManifest):
|
|
dabdatasync.DataSync_Factory.get_DataSync()
|
|
|
|
def test_load_nextcloud_invalid(self):
|
|
dabdatasync.I_DataSync.manifest_path = testdir_path / "test_manifest_nextcloud_invalid.json"
|
|
with self.assertRaises(dabdatasync.DataSyncException_InvalidManifest):
|
|
dabdatasync.DataSync_Factory.get_DataSync()
|