from pathlib import Path import re import unittest import unifydrive2nas as app ROOT = Path(__file__).resolve().parents[1] DAG_PATH = ROOT / "deploy" / "dagu" / "unifydrive2nas.yaml" PROJECT_PATH = ROOT / "PROJECT.md" INSTALL_DIRECTORY = "/volume1/homes/kazu634/bin/unifydrive2nas" SCRIPT_PATH = INSTALL_DIRECTORY + "/unifydrive2nas.py" DAG_STAGING_PATH = "/tmp/unifydrive2nas.yaml" DAGU_COMMAND = "DAGU_HOME=/var/lib/dagu /usr/local/bin/dagu" class DaguDeploymentContractTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.dag = DAG_PATH.read_text(encoding="utf-8") cls.project = PROJECT_PATH.read_text(encoding="utf-8") def dag_value(self, key): pattern = r"^\s*" + re.escape(key) + r":\s*(.*?)\s*$" values = re.findall(pattern, self.dag, flags=re.MULTILINE) self.assertEqual(1, len(values), "DAG の {0} は一意であること".format(key)) return values[0].strip('"\'') def test_schedule_and_execution_guards(self): self.assertEqual("0 4 * * *", self.dag_value("schedule")) self.assertEqual("skip", self.dag_value("overlap_policy")) self.assertEqual("7200", self.dag_value("timeout_sec")) self.assertIn("毎日 04:00（Asia/Tokyo）", self.project) self.assertIn('schedule: "0 4 * * *"', self.project) def test_ssh_target_and_single_command(self): expected = { "action": "ssh.run", "host": "192.168.10.201", "port": "22", "user": "kazu634", "key": "/home/kazu634/.ssh/id_ed25519_nas", "strict_host_key": "true", "known_host_file": "/home/kazu634/.ssh/known_hosts", "command": "python3 " + SCRIPT_PATH, } for key, value in expected.items(): with self.subTest(key=key): self.assertEqual(value, self.dag_value(key)) self.assertIn("| Synology `ds923p` | `192.168.10.201`", self.project) self.assertIn("ssh -i ~/.ssh/id_ed25519_nas kazu634@192.168.10.201", self.project) self.assertIn("`" + INSTALL_DIRECTORY + "/`", self.project) def test_ssh_settings_are_nested_in_one_step_with_block(self): match = re.search( r"^steps:\n" r" - id: copy_unifydrive_to_nas\n" r" action: ssh\.run\n" r" with:\n" r"((?: [^\n]+\n?)+)$", self.dag, flags=re.MULTILINE, ) self.assertIsNotNone(match) with_block = match.group(1) for key in ( "host", "port", "user", "key", "strict_host_key", "known_host_file", "command", ): with self.subTest(key=key): self.assertRegex(with_block, r"(?m)^ " + re.escape(key) + r":") self.assertEqual(1, self.dag.count("command:")) def test_project_documents_deployment_contract(self): required_values = ( "毎日 04:00", "/volume1/homes/kazu634/bin/unifydrive2nas/", "deploy/dagu/unifydrive2nas.yaml", "/var/lib/dagu/dags/", "kazu634/UnifyDrive2NAS", "--init", ) for value in required_values: with self.subTest(value=value): self.assertIn(value, self.project) def test_default_database_is_shared_by_implementation_project_and_dag(self): expected = Path.home() / ".local" / "state" / "unifydrive2nas" / "copied.sqlite3" self.assertEqual(expected, app.DEFAULT_DATABASE) self.assertIn("`~/.local/state/unifydrive2nas/copied.sqlite3`", self.project) self.assertNotIn("--database", self.dag_value("command")) def test_deployment_commands_use_consistent_paths_and_exclude_tests(self): section = self.project.split("### 配置手順", 1)[1].split("## 外部とのつながり", 1)[0] blocks = re.findall(r"```bash\n(.*?)\n[ \t]*```", section, flags=re.DOTALL) self.assertEqual(8, len(blocks)) ( runtime_copy, runtime_dry_run, initialize, staging_copy, validate, dag_dry_run, manual_start, schedule_enable, ) = blocks self.assertIn("--exclude='__pycache__' unifydrive2nas.py vendor", runtime_copy) self.assertNotIn("tests/", runtime_copy) self.assertIn("tar -C " + INSTALL_DIRECTORY + " -xf -", runtime_copy) self.assertIn("python3 " + SCRIPT_PATH + " --dry-run", runtime_dry_run) self.assertIn("python3 " + SCRIPT_PATH + " --init", initialize) self.assertNotIn("--database", initialize) self.assertIn("-C deploy/dagu unifydrive2nas.yaml", staging_copy) self.assertIn("tar -C /tmp -xf -", staging_copy) self.assertNotIn("tests/", staging_copy) self.assertIn(DAGU_COMMAND + " validate " + DAG_STAGING_PATH, validate) self.assertIn(DAGU_COMMAND + " dry " + DAG_STAGING_PATH, dag_dry_run) self.assertIn(DAGU_COMMAND + " start " + DAG_STAGING_PATH, manual_start) self.assertIn("-C deploy/dagu unifydrive2nas.yaml", schedule_enable) self.assertIn("tar -C /var/lib/dagu/dags -xf -", schedule_enable) self.assertIn("rm -f " + DAG_STAGING_PATH, schedule_enable) def test_verification_uses_temporary_tests_on_nas(self): section = self.project.split("## 確かめ方", 1)[1] blocks = re.findall(r"```bash\n(.*?)\n[ \t]*```", section, flags=re.DOTALL) verification = next(block for block in blocks if "unifydrive2nas.py vendor tests" in block) unpack = "tar -C /tmp/unifydrive2nas-test -xf -" run_tests = ( "cd /tmp/unifydrive2nas-test && " "python3 -W error::ResourceWarning -m unittest discover -s tests -v" ) cleanup = "rm -rf /tmp/unifydrive2nas-test" production_dry_run = "python3 " + SCRIPT_PATH + " --dry-run" self.assertIn("--exclude='__pycache__' unifydrive2nas.py vendor tests", verification) self.assertIn(unpack, verification) self.assertIn(run_tests, verification) self.assertEqual(2, verification.count(cleanup)) self.assertIn(production_dry_run, verification) self.assertLess(verification.index(unpack), verification.index(run_tests)) self.assertLess(verification.index(run_tests), verification.rindex(cleanup)) self.assertLess(verification.rindex(cleanup), verification.index(production_dry_run)) def test_all_documented_dagu_commands_use_service_home(self): commands = re.findall(r"/usr/local/bin/dagu (?:validate|dry|start) [^\s\"']+", self.project) self.assertEqual(5, len(commands)) self.assertEqual(5, self.project.count(DAGU_COMMAND)) self.assertNotIn("/usr/local/bin/dagu", self.project.replace(DAGU_COMMAND, "")) if __name__ == "__main__": unittest.main()