deblan-mirror/abraunegg-onedrive
deblan-mirror/abraunegg-onedrive
1
0
Fork 0
mirror of https://github.com/abraunegg/onedrive synced 2026-03-14 14:35:46 +01:00
Code Issues Projects Releases Wiki Activity

Update PR - Add Test Cases 0017 to 0024

Browse source 
* Add Test Cases 0017 to 0024
This commit is contained in:
abraunegg 2026-03-14 12:24:46 +11:00
commit 3e363778c7
20 changed files with 1095 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

16
ci/e2e/run.py
View file

@ -25,6 +25,14 @@ from testcases.tc0013_skip_dotfiles_validation import TestCase0013SkipDotfilesVa
from testcases.tc0014_skip_size_validation import TestCase0014SkipSizeValidation
from testcases.tc0015_skip_symlinks_validation import TestCase0015SkipSymlinksValidation
from testcases.tc0016_check_nosync_validation import TestCase0016CheckNosyncValidation
from testcases.tc0017_check_nomount_validation import TestCase0017CheckNomountValidation
from testcases.tc0018_recycle_bin_validation import TestCase0018RecycleBinValidation
from testcases.tc0019_logging_and_running_config import TestCase0019LoggingAndRunningConfig
from testcases.tc0020_monitor_mode_validation import TestCase0020MonitorModeValidation
from testcases.tc0021_resumable_transfers_validation import TestCase0021ResumableTransfersValidation
from testcases.tc0022_local_first_validation import TestCase0022LocalFirstValidation
from testcases.tc0023_bypass_data_preservation_validation import TestCase0023BypassDataPreservationValidation
from testcases.tc0024_big_delete_safeguard_validation import TestCase0024BigDeleteSafeguardValidation
def build_test_suite() -> list:
@ -50,6 +58,14 @@ def build_test_suite() -> list:
TestCase0014SkipSizeValidation(),
TestCase0015SkipSymlinksValidation(),
TestCase0016CheckNosyncValidation(),
TestCase0017CheckNomountValidation(),
TestCase0018RecycleBinValidation(),
TestCase0019LoggingAndRunningConfig(),
TestCase0020MonitorModeValidation(),
TestCase0021ResumableTransfersValidation(),
TestCase0022LocalFirstValidation(),
TestCase0023BypassDataPreservationValidation(),
TestCase0024BigDeleteSafeguardValidation(),
]

4
ci/e2e/testcases/tc0006_download_only.py
View file

@ -26,10 +26,10 @@ class TestCase0006DownloadOnly(E2ETestCase):
context.bootstrap_config_dir(seed_conf); self._write_config(seed_conf / "config")
context.bootstrap_config_dir(download_conf); self._write_config(download_conf / "config")
seed_stdout = case_log_dir / "seed_stdout.log"; seed_stderr = case_log_dir / "seed_stderr.log"; dl_stdout = case_log_dir / "download_stdout.log"; dl_stderr = case_log_dir / "download_stderr.log"; local_manifest_file = state_dir / "download_manifest.txt"; metadata_file = state_dir / "seed_metadata.txt"
seed_command = [context.onedrive_bin, "--sync", "--verbose", "--resync", "--resync-auth", "--syncdir", str(seed_root), "--confdir", str(seed_conf)]
seed_command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--resync", "--resync-auth", "--syncdir", str(seed_root), "--confdir", str(seed_conf)]
seed_result = run_command(seed_command, cwd=context.repo_root)
write_text_file(seed_stdout, seed_result.stdout); write_text_file(seed_stderr, seed_result.stderr)
download_command = [context.onedrive_bin, "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--syncdir", str(download_root), "--confdir", str(download_conf)]
download_command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--syncdir", str(download_root), "--confdir", str(download_conf)]
download_result = run_command(download_command, cwd=context.repo_root)
write_text_file(dl_stdout, download_result.stdout); write_text_file(dl_stderr, download_result.stderr); local_manifest = build_manifest(download_root); write_manifest(local_manifest_file, local_manifest)
write_text_file(metadata_file, "\n".join([f"root_name={root_name}", f"seed_command={command_to_string(seed_command)}", f"seed_returncode={seed_result.returncode}", f"download_command={command_to_string(download_command)}", f"download_returncode={download_result.returncode}"]) + "\n")

4
ci/e2e/testcases/tc0007_download_only_cleanup_local_files.py
View file

@ -26,11 +26,11 @@ class TestCase0007DownloadOnlyCleanupLocalFiles(E2ETestCase):
context.bootstrap_config_dir(seed_conf); self._write_config(seed_conf / "config")
context.bootstrap_config_dir(cleanup_conf); self._write_config(cleanup_conf / "config")
seed_stdout = case_log_dir / "seed_stdout.log"; seed_stderr = case_log_dir / "seed_stderr.log"; cleanup_stdout = case_log_dir / "cleanup_stdout.log"; cleanup_stderr = case_log_dir / "cleanup_stderr.log"; post_manifest_file = state_dir / "post_cleanup_manifest.txt"; metadata_file = state_dir / "seed_metadata.txt"
seed_command = [context.onedrive_bin, "--sync", "--verbose", "--resync", "--resync-auth", "--syncdir", str(sync_root), "--confdir", str(seed_conf)]
seed_command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--resync", "--resync-auth", "--syncdir", str(sync_root), "--confdir", str(seed_conf)]
seed_result = run_command(seed_command, cwd=context.repo_root)
write_text_file(seed_stdout, seed_result.stdout); write_text_file(seed_stderr, seed_result.stderr)
stale = sync_root / root_name / "stale-local.txt"; write_text_file(stale, "stale\n")
cleanup_command = [context.onedrive_bin, "--sync", "--verbose", "--download-only", "--cleanup-local-files", "--resync", "--resync-auth", "--syncdir", str(sync_root), "--confdir", str(cleanup_conf)]
cleanup_command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--download-only", "--cleanup-local-files", "--resync", "--resync-auth", "--syncdir", str(sync_root), "--confdir", str(cleanup_conf)]
cleanup_result = run_command(cleanup_command, cwd=context.repo_root)
write_text_file(cleanup_stdout, cleanup_result.stdout); write_text_file(cleanup_stderr, cleanup_result.stderr); post_manifest = build_manifest(sync_root); write_manifest(post_manifest_file, post_manifest)
write_text_file(metadata_file, "\n".join([f"root_name={root_name}", f"seed_returncode={seed_result.returncode}", f"cleanup_returncode={cleanup_result.returncode}"]) + "\n")

4
ci/e2e/testcases/tc0008_upload_only.py
View file

@ -26,10 +26,10 @@ class TestCase0008UploadOnly(E2ETestCase):
context.bootstrap_config_dir(upload_conf); self._write_config(upload_conf / "config")
context.bootstrap_config_dir(verify_conf); self._write_config(verify_conf / "config")
stdout_file = case_log_dir / "upload_only_stdout.log"; stderr_file = case_log_dir / "upload_only_stderr.log"; verify_stdout = case_log_dir / "verify_stdout.log"; verify_stderr = case_log_dir / "verify_stderr.log"; remote_manifest_file = state_dir / "remote_verify_manifest.txt"; metadata_file = state_dir / "upload_metadata.txt"
command = [context.onedrive_bin, "--sync", "--verbose", "--upload-only", "--resync", "--resync-auth", "--syncdir", str(upload_root), "--confdir", str(upload_conf)]
command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--upload-only", "--resync", "--resync-auth", "--syncdir", str(upload_root), "--confdir", str(upload_conf)]
result = run_command(command, cwd=context.repo_root)
write_text_file(stdout_file, result.stdout); write_text_file(stderr_file, result.stderr)
verify_command = [context.onedrive_bin, "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--syncdir", str(verify_root), "--confdir", str(verify_conf)]
verify_command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--syncdir", str(verify_root), "--confdir", str(verify_conf)]
verify_result = run_command(verify_command, cwd=context.repo_root)
write_text_file(verify_stdout, verify_result.stdout); write_text_file(verify_stderr, verify_result.stderr); remote_manifest = build_manifest(verify_root); write_manifest(remote_manifest_file, remote_manifest)
write_text_file(metadata_file, "\n".join([f"root_name={root_name}", f"returncode={result.returncode}", f"verify_returncode={verify_result.returncode}"]) + "\n")

6
ci/e2e/testcases/tc0009_upload_only_no_remote_delete.py
View file

@ -27,14 +27,14 @@ class TestCase0009UploadOnlyNoRemoteDelete(E2ETestCase):
context.bootstrap_config_dir(upload_conf); self._write_config(upload_conf / "config")
context.bootstrap_config_dir(verify_conf); self._write_config(verify_conf / "config")
seed_stdout = case_log_dir / "seed_stdout.log"; seed_stderr = case_log_dir / "seed_stderr.log"; upload_stdout = case_log_dir / "upload_only_stdout.log"; upload_stderr = case_log_dir / "upload_only_stderr.log"; verify_stdout = case_log_dir / "verify_stdout.log"; verify_stderr = case_log_dir / "verify_stderr.log"; remote_manifest_file = state_dir / "remote_verify_manifest.txt"; metadata_file = state_dir / "seed_metadata.txt"
seed_command = [context.onedrive_bin, "--sync", "--verbose", "--resync", "--resync-auth", "--syncdir", str(sync_root), "--confdir", str(seed_conf)]
seed_command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--resync", "--resync-auth", "--syncdir", str(sync_root), "--confdir", str(seed_conf)]
seed_result = run_command(seed_command, cwd=context.repo_root)
write_text_file(seed_stdout, seed_result.stdout); write_text_file(seed_stderr, seed_result.stderr)
if keep_file.exists(): keep_file.unlink()
upload_command = [context.onedrive_bin, "--sync", "--verbose", "--upload-only", "--no-remote-delete", "--resync", "--resync-auth", "--syncdir", str(sync_root), "--confdir", str(upload_conf)]
upload_command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--upload-only", "--no-remote-delete", "--resync", "--resync-auth", "--syncdir", str(sync_root), "--confdir", str(upload_conf)]
upload_result = run_command(upload_command, cwd=context.repo_root)
write_text_file(upload_stdout, upload_result.stdout); write_text_file(upload_stderr, upload_result.stderr)
verify_command = [context.onedrive_bin, "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--syncdir", str(verify_root), "--confdir", str(verify_conf)]
verify_command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--syncdir", str(verify_root), "--confdir", str(verify_conf)]
verify_result = run_command(verify_command, cwd=context.repo_root)
write_text_file(verify_stdout, verify_result.stdout); write_text_file(verify_stderr, verify_result.stderr); remote_manifest = build_manifest(verify_root); write_manifest(remote_manifest_file, remote_manifest)
write_text_file(metadata_file, "\n".join([f"root_name={root_name}", f"seed_returncode={seed_result.returncode}", f"upload_returncode={upload_result.returncode}", f"verify_returncode={verify_result.returncode}"]) + "\n")

4
ci/e2e/testcases/tc0010_upload_only_remove_source_files.py
View file

@ -26,10 +26,10 @@ class TestCase0010UploadOnlyRemoveSourceFiles(E2ETestCase):
context.bootstrap_config_dir(upload_conf); self._write_config(upload_conf / "config")
context.bootstrap_config_dir(verify_conf); self._write_config(verify_conf / "config")
stdout_file = case_log_dir / "upload_only_remove_source_stdout.log"; stderr_file = case_log_dir / "upload_only_remove_source_stderr.log"; verify_stdout = case_log_dir / "verify_stdout.log"; verify_stderr = case_log_dir / "verify_stderr.log"; post_manifest_file = state_dir / "post_upload_manifest.txt"; remote_manifest_file = state_dir / "remote_verify_manifest.txt"; metadata_file = state_dir / "upload_metadata.txt"
command = [context.onedrive_bin, "--sync", "--verbose", "--upload-only", "--remove-source-files", "--resync", "--resync-auth", "--syncdir", str(sync_root), "--confdir", str(upload_conf)]
command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--upload-only", "--remove-source-files", "--resync", "--resync-auth", "--syncdir", str(sync_root), "--confdir", str(upload_conf)]
result = run_command(command, cwd=context.repo_root)
write_text_file(stdout_file, result.stdout); write_text_file(stderr_file, result.stderr); post_manifest = build_manifest(sync_root); write_manifest(post_manifest_file, post_manifest)
verify_command = [context.onedrive_bin, "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--syncdir", str(verify_root), "--confdir", str(verify_conf)]
verify_command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--syncdir", str(verify_root), "--confdir", str(verify_conf)]
verify_result = run_command(verify_command, cwd=context.repo_root)
write_text_file(verify_stdout, verify_result.stdout); write_text_file(verify_stderr, verify_result.stderr); remote_manifest = build_manifest(verify_root); write_manifest(remote_manifest_file, remote_manifest)
write_text_file(metadata_file, "\n".join([f"root_name={root_name}", f"returncode={result.returncode}", f"verify_returncode={verify_result.returncode}"]) + "\n")

4
ci/e2e/testcases/tc0011_skip_file_validation.py
View file

@ -26,10 +26,10 @@ class TestCase0011SkipFileValidation(E2ETestCase):
context.bootstrap_config_dir(confdir); self._write_config(confdir / "config")
context.bootstrap_config_dir(verify_conf); write_text_file(verify_conf / "config", "# tc0011 verify\nbypass_data_preservation = \"true\"\n")
stdout_file = case_log_dir / "skip_file_stdout.log"; stderr_file = case_log_dir / "skip_file_stderr.log"; verify_stdout = case_log_dir / "verify_stdout.log"; verify_stderr = case_log_dir / "verify_stderr.log"; remote_manifest_file = state_dir / "remote_verify_manifest.txt"; metadata_file = state_dir / "metadata.txt"
command = [context.onedrive_bin, "--sync", "--verbose", "--resync", "--resync-auth", "--syncdir", str(sync_root), "--confdir", str(confdir)]
command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--resync", "--resync-auth", "--syncdir", str(sync_root), "--confdir", str(confdir)]
result = run_command(command, cwd=context.repo_root)
write_text_file(stdout_file, result.stdout); write_text_file(stderr_file, result.stderr)
verify_command = [context.onedrive_bin, "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--syncdir", str(verify_root), "--confdir", str(verify_conf)]
verify_command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--syncdir", str(verify_root), "--confdir", str(verify_conf)]
verify_result = run_command(verify_command, cwd=context.repo_root)
write_text_file(verify_stdout, verify_result.stdout); write_text_file(verify_stderr, verify_result.stderr); remote_manifest = build_manifest(verify_root); write_manifest(remote_manifest_file, remote_manifest)
write_text_file(metadata_file, f"root_name={root_name}\nreturncode={result.returncode}\nverify_returncode={verify_result.returncode}\n")

8
ci/e2e/testcases/tc0012_skip_dir_validation.py
View file

@ -32,9 +32,9 @@ class TestCase0012SkipDirValidation(E2ETestCase):
context.bootstrap_config_dir(confdir); self._write_config(confdir / "config", "Cache", False)
context.bootstrap_config_dir(verify_conf); write_text_file(verify_conf / "config", "# verify\nbypass_data_preservation = \"true\"\n")
stdout_file = case_log_dir / "loose_match_stdout.log"; stderr_file = case_log_dir / "loose_match_stderr.log"; verify_stdout = case_log_dir / "loose_match_verify_stdout.log"; verify_stderr = case_log_dir / "loose_match_verify_stderr.log"; manifest_file = scenario_state / "remote_verify_manifest.txt"
result = run_command([context.onedrive_bin, "--sync", "--verbose", "--resync", "--resync-auth", "--syncdir", str(sync_root), "--confdir", str(confdir)], cwd=context.repo_root)
result = run_command([context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--resync", "--resync-auth", "--syncdir", str(sync_root), "--confdir", str(confdir)], cwd=context.repo_root)
write_text_file(stdout_file, result.stdout); write_text_file(stderr_file, result.stderr)
verify_result = run_command([context.onedrive_bin, "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--syncdir", str(verify_root), "--confdir", str(verify_conf)], cwd=context.repo_root)
verify_result = run_command([context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--syncdir", str(verify_root), "--confdir", str(verify_conf)], cwd=context.repo_root)
write_text_file(verify_stdout, verify_result.stdout); write_text_file(verify_stderr, verify_result.stderr); manifest = build_manifest(verify_root); write_manifest(manifest_file, manifest)
all_artifacts.extend([str(stdout_file), str(stderr_file), str(verify_stdout), str(verify_stderr), str(manifest_file)])
if result.returncode != 0: failures.append(f"Loose skip_dir scenario failed with status {result.returncode}"); return
@ -54,9 +54,9 @@ class TestCase0012SkipDirValidation(E2ETestCase):
context.bootstrap_config_dir(confdir); self._write_config(confdir / "config", f"{root}/App/Cache", True)
context.bootstrap_config_dir(verify_conf); write_text_file(verify_conf / "config", "# verify\nbypass_data_preservation = \"true\"\n")
stdout_file = case_log_dir / "strict_match_stdout.log"; stderr_file = case_log_dir / "strict_match_stderr.log"; verify_stdout = case_log_dir / "strict_match_verify_stdout.log"; verify_stderr = case_log_dir / "strict_match_verify_stderr.log"; manifest_file = scenario_state / "remote_verify_manifest.txt"
result = run_command([context.onedrive_bin, "--sync", "--verbose", "--resync", "--resync-auth", "--syncdir", str(sync_root), "--confdir", str(confdir)], cwd=context.repo_root)
result = run_command([context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--resync", "--resync-auth", "--syncdir", str(sync_root), "--confdir", str(confdir)], cwd=context.repo_root)
write_text_file(stdout_file, result.stdout); write_text_file(stderr_file, result.stderr)
verify_result = run_command([context.onedrive_bin, "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--syncdir", str(verify_root), "--confdir", str(verify_conf)], cwd=context.repo_root)
verify_result = run_command([context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--syncdir", str(verify_root), "--confdir", str(verify_conf)], cwd=context.repo_root)
write_text_file(verify_stdout, verify_result.stdout); write_text_file(verify_stderr, verify_result.stderr); manifest = build_manifest(verify_root); write_manifest(manifest_file, manifest)
all_artifacts.extend([str(stdout_file), str(stderr_file), str(verify_stdout), str(verify_stderr), str(manifest_file)])
if result.returncode != 0: failures.append(f"Strict skip_dir scenario failed with status {result.returncode}"); return

4
ci/e2e/testcases/tc0013_skip_dotfiles_validation.py
View file

@ -26,10 +26,10 @@ class TestCase0013SkipDotfilesValidation(E2ETestCase):
context.bootstrap_config_dir(confdir); self._write_config(confdir / "config")
context.bootstrap_config_dir(verify_conf); write_text_file(verify_conf / "config", "# verify\nbypass_data_preservation = \"true\"\n")
stdout_file = case_log_dir / "skip_dotfiles_stdout.log"; stderr_file = case_log_dir / "skip_dotfiles_stderr.log"; verify_stdout = case_log_dir / "verify_stdout.log"; verify_stderr = case_log_dir / "verify_stderr.log"; remote_manifest_file = state_dir / "remote_verify_manifest.txt"; metadata_file = state_dir / "metadata.txt"
command = [context.onedrive_bin, "--sync", "--verbose", "--resync", "--resync-auth", "--syncdir", str(sync_root), "--confdir", str(confdir)]
command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--resync", "--resync-auth", "--syncdir", str(sync_root), "--confdir", str(confdir)]
result = run_command(command, cwd=context.repo_root)
write_text_file(stdout_file, result.stdout); write_text_file(stderr_file, result.stderr)
verify_command = [context.onedrive_bin, "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--syncdir", str(verify_root), "--confdir", str(verify_conf)]
verify_command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--syncdir", str(verify_root), "--confdir", str(verify_conf)]
verify_result = run_command(verify_command, cwd=context.repo_root)
write_text_file(verify_stdout, verify_result.stdout); write_text_file(verify_stderr, verify_result.stderr); remote_manifest = build_manifest(verify_root); write_manifest(remote_manifest_file, remote_manifest)
write_text_file(metadata_file, f"root_name={root_name}\nreturncode={result.returncode}\nverify_returncode={verify_result.returncode}\n")

23
ci/e2e/testcases/tc0014_skip_size_validation.py
View file

@ -16,28 +16,33 @@ class TestCase0014SkipSizeValidation(E2ETestCase):
description = "Validate that skip_size prevents oversized files from synchronising"
def _write_config(self, config_path: Path) -> None:
write_text_file(config_path, "# tc0014 config\nbypass_data_preservation = \"true\"\nskip_size = \"1\"\n")
write_text_file(config_path, "# tc0014 config\nbypass_data_preservation = \"true\"\ndebug_logging = \"true\"\nenable_logging = \"true\"\nskip_size = \"1\"\n")
def run(self, context: E2EContext) -> TestResult:
case_work_dir = context.work_root / "tc0014"; case_log_dir = context.logs_dir / "tc0014"; state_dir = context.state_dir / "tc0014"
reset_directory(case_work_dir); reset_directory(case_log_dir); reset_directory(state_dir); context.ensure_refresh_token_available()
sync_root = case_work_dir / "syncroot"; confdir = case_work_dir / "conf-main"; verify_root = case_work_dir / "verifyroot"; verify_conf = case_work_dir / "conf-verify"; root_name = f"ZZ_E2E_TC0014_{context.run_id}_{os.getpid()}"
sync_root = case_work_dir / "syncroot"; confdir = case_work_dir / "conf-main"; verify_root = case_work_dir / "verifyroot"; verify_conf = case_work_dir / "conf-verify"; root_name = f"ZZ_E2E_TC0014_{context.run_id}_{os.getpid()}"; app_log_dir = case_log_dir / "app-logs"
write_text_file(sync_root / root_name / "small.bin", "a" * 16384)
big_path = sync_root / root_name / "large.bin"; big_path.parent.mkdir(parents=True, exist_ok=True); big_path.write_bytes(b"B" * (2 * 1024 * 1024))
context.bootstrap_config_dir(confdir); self._write_config(confdir / "config")
write_text_file(confdir / "config", (confdir / "config").read_text(encoding="utf-8") + f'log_dir = "{app_log_dir}"\n')
context.bootstrap_config_dir(verify_conf); write_text_file(verify_conf / "config", "# verify\nbypass_data_preservation = \"true\"\n")
stdout_file = case_log_dir / "skip_size_stdout.log"; stderr_file = case_log_dir / "skip_size_stderr.log"; verify_stdout = case_log_dir / "verify_stdout.log"; verify_stderr = case_log_dir / "verify_stderr.log"; remote_manifest_file = state_dir / "remote_verify_manifest.txt"; metadata_file = state_dir / "metadata.txt"
command = [context.onedrive_bin, "--sync", "--verbose", "--resync", "--resync-auth", "--syncdir", str(sync_root), "--confdir", str(confdir)]
stdout_file = case_log_dir / "skip_size_stdout.log"; stderr_file = case_log_dir / "skip_size_stderr.log"; verify_stdout = case_log_dir / "verify_stdout.log"; verify_stderr = case_log_dir / "verify_stderr.log"; remote_manifest_file = state_dir / "remote_verify_manifest.txt"; metadata_file = state_dir / "metadata.txt"; config_copy = state_dir / "config_used.txt"; verify_config_copy = state_dir / "verify_config_used.txt"
command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--resync", "--resync-auth", "--syncdir", str(sync_root), "--confdir", str(confdir)]
result = run_command(command, cwd=context.repo_root)
write_text_file(stdout_file, result.stdout); write_text_file(stderr_file, result.stderr)
verify_command = [context.onedrive_bin, "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--syncdir", str(verify_root), "--confdir", str(verify_conf)]
verify_command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--syncdir", str(verify_root), "--confdir", str(verify_conf)]
verify_result = run_command(verify_command, cwd=context.repo_root)
write_text_file(verify_stdout, verify_result.stdout); write_text_file(verify_stderr, verify_result.stderr); remote_manifest = build_manifest(verify_root); write_manifest(remote_manifest_file, remote_manifest)
write_text_file(metadata_file, f"root_name={root_name}\nlarge_size={big_path.stat().st_size}\nreturncode={result.returncode}\nverify_returncode={verify_result.returncode}\n")
artifacts = [str(stdout_file), str(stderr_file), str(verify_stdout), str(verify_stderr), str(remote_manifest_file), str(metadata_file)]
details = {"returncode": result.returncode, "verify_returncode": verify_result.returncode, "root_name": root_name, "large_size": big_path.stat().st_size}
write_text_file(config_copy, (confdir / "config").read_text(encoding="utf-8"))
write_text_file(verify_config_copy, (verify_conf / "config").read_text(encoding="utf-8"))
write_text_file(metadata_file, f"root_name={root_name}\nlarge_size={big_path.stat().st_size}\nlarge_size_mb_decimal={big_path.stat().st_size / 1000 / 1000:.3f}\nlarge_size_mib_binary={big_path.stat().st_size / 1024 / 1024:.3f}\nreturncode={result.returncode}\nverify_returncode={verify_result.returncode}\n")
artifacts = [str(stdout_file), str(stderr_file), str(verify_stdout), str(verify_stderr), str(remote_manifest_file), str(metadata_file), str(config_copy), str(verify_config_copy)]
if app_log_dir.exists():
artifacts.append(str(app_log_dir))
details = {"returncode": result.returncode, "verify_returncode": verify_result.returncode, "root_name": root_name, "large_size": big_path.stat().st_size, "large_size_mb_decimal": round(big_path.stat().st_size / 1000 / 1000, 3), "large_size_mib_binary": round(big_path.stat().st_size / 1024 / 1024, 3), "skip_size": 1}
if result.returncode != 0: return TestResult.fail_result(self.case_id, self.name, f"skip_size validation failed with status {result.returncode}", artifacts, details)
if verify_result.returncode != 0: return TestResult.fail_result(self.case_id, self.name, f"Remote verification failed with status {verify_result.returncode}", artifacts, details)
if f"{root_name}/small.bin" not in remote_manifest: return TestResult.fail_result(self.case_id, self.name, "Small file missing after skip_size processing", artifacts, details)
if f"{root_name}/large.bin" in remote_manifest: return TestResult.fail_result(self.case_id, self.name, "Large file exceeded skip_size threshold but was synchronised", artifacts, details)
if f"{root_name}/large.bin" in remote_manifest: return TestResult.fail_result(self.case_id, self.name, "Large file exceeded configured skip_size threshold but was synchronised; review display-running-config output and debug logs", artifacts, details)
return TestResult.pass_result(self.case_id, self.name, artifacts, details)

4
ci/e2e/testcases/tc0015_skip_symlinks_validation.py
View file

@ -26,10 +26,10 @@ class TestCase0015SkipSymlinksValidation(E2ETestCase):
context.bootstrap_config_dir(confdir); self._write_config(confdir / "config")
context.bootstrap_config_dir(verify_conf); write_text_file(verify_conf / "config", "# verify\nbypass_data_preservation = \"true\"\n")
stdout_file = case_log_dir / "skip_symlinks_stdout.log"; stderr_file = case_log_dir / "skip_symlinks_stderr.log"; verify_stdout = case_log_dir / "verify_stdout.log"; verify_stderr = case_log_dir / "verify_stderr.log"; remote_manifest_file = state_dir / "remote_verify_manifest.txt"; metadata_file = state_dir / "metadata.txt"
command = [context.onedrive_bin, "--sync", "--verbose", "--resync", "--resync-auth", "--syncdir", str(sync_root), "--confdir", str(confdir)]
command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--resync", "--resync-auth", "--syncdir", str(sync_root), "--confdir", str(confdir)]
result = run_command(command, cwd=context.repo_root)
write_text_file(stdout_file, result.stdout); write_text_file(stderr_file, result.stderr)
verify_command = [context.onedrive_bin, "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--syncdir", str(verify_root), "--confdir", str(verify_conf)]
verify_command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--syncdir", str(verify_root), "--confdir", str(verify_conf)]
verify_result = run_command(verify_command, cwd=context.repo_root)
write_text_file(verify_stdout, verify_result.stdout); write_text_file(verify_stderr, verify_result.stderr); remote_manifest = build_manifest(verify_root); write_manifest(remote_manifest_file, remote_manifest)
write_text_file(metadata_file, f"root_name={root_name}\nreturncode={result.returncode}\nverify_returncode={verify_result.returncode}\n")

4
ci/e2e/testcases/tc0016_check_nosync_validation.py
View file

@ -26,10 +26,10 @@ class TestCase0016CheckNosyncValidation(E2ETestCase):
context.bootstrap_config_dir(confdir); self._write_config(confdir / "config")
context.bootstrap_config_dir(verify_conf); write_text_file(verify_conf / "config", "# verify\nbypass_data_preservation = \"true\"\n")
stdout_file = case_log_dir / "check_nosync_stdout.log"; stderr_file = case_log_dir / "check_nosync_stderr.log"; verify_stdout = case_log_dir / "verify_stdout.log"; verify_stderr = case_log_dir / "verify_stderr.log"; remote_manifest_file = state_dir / "remote_verify_manifest.txt"; metadata_file = state_dir / "metadata.txt"
command = [context.onedrive_bin, "--sync", "--verbose", "--resync", "--resync-auth", "--syncdir", str(sync_root), "--confdir", str(confdir)]
command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--resync", "--resync-auth", "--syncdir", str(sync_root), "--confdir", str(confdir)]
result = run_command(command, cwd=context.repo_root)
write_text_file(stdout_file, result.stdout); write_text_file(stderr_file, result.stderr)
verify_command = [context.onedrive_bin, "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--syncdir", str(verify_root), "--confdir", str(verify_conf)]
verify_command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--syncdir", str(verify_root), "--confdir", str(verify_conf)]
verify_result = run_command(verify_command, cwd=context.repo_root)
write_text_file(verify_stdout, verify_result.stdout); write_text_file(verify_stderr, verify_result.stderr); remote_manifest = build_manifest(verify_root); write_manifest(remote_manifest_file, remote_manifest)
write_text_file(metadata_file, f"root_name={root_name}\nreturncode={result.returncode}\nverify_returncode={verify_result.returncode}\n")

105
ci/e2e/testcases/tc0017_check_nomount_validation.py Normal file
View file

@ -0,0 +1,105 @@
from __future__ import annotations
import os
from pathlib import Path
from framework.base import E2ETestCase
from framework.context import E2EContext
from framework.result import TestResult
from framework.utils import command_to_string, reset_directory, run_command, write_text_file
class TestCase0017CheckNomountValidation(E2ETestCase):
case_id = "0017"
name = "check_nomount validation"
description = "Validate that check_nomount aborts synchronisation when .nosync exists in the sync_dir mount point"
def _write_config(self, config_path: Path) -> None:
write_text_file(
config_path,
"# tc0017 config\n"
'bypass_data_preservation = "true"\n'
'check_nomount = "true"\n',
)
def run(self, context: E2EContext) -> TestResult:
case_work_dir = context.work_root / "tc0017"
case_log_dir = context.logs_dir / "tc0017"
state_dir = context.state_dir / "tc0017"
reset_directory(case_work_dir)
reset_directory(case_log_dir)
reset_directory(state_dir)
context.ensure_refresh_token_available()
sync_root = case_work_dir / "syncroot"
confdir = case_work_dir / "conf-main"
root_name = f"ZZ_E2E_TC0017_{context.run_id}_{os.getpid()}"
write_text_file(sync_root / ".nosync", "")
write_text_file(sync_root / root_name / "should_not_upload.txt", "blocked by check_nomount\n")
context.bootstrap_config_dir(confdir)
self._write_config(confdir / "config")
stdout_file = case_log_dir / "check_nomount_stdout.log"
stderr_file = case_log_dir / "check_nomount_stderr.log"
metadata_file = state_dir / "metadata.txt"
command = [
context.onedrive_bin,
"--display-running-config",
"--sync",
"--verbose",
"--resync",
"--resync-auth",
"--syncdir",
str(sync_root),
"--confdir",
str(confdir),
]
context.log(f"Executing Test Case {self.case_id}: {command_to_string(command)}")
result = run_command(command, cwd=context.repo_root)
write_text_file(stdout_file, result.stdout)
write_text_file(stderr_file, result.stderr)
write_text_file(
metadata_file,
"\n".join(
[
f"case_id={self.case_id}",
f"root_name={root_name}",
f"command={command_to_string(command)}",
f"returncode={result.returncode}",
]
)
+ "\n",
)
artifacts = [str(stdout_file), str(stderr_file), str(metadata_file)]
details = {
"command": command,
"returncode": result.returncode,
"root_name": root_name,
}
combined_output = (result.stdout + "\n" + result.stderr).lower()
if result.returncode == 0:
return TestResult.fail_result(
self.case_id,
self.name,
"check_nomount did not abort synchronisation when .nosync existed in the sync_dir mount point",
artifacts,
details,
)
if ".nosync file found" not in combined_output and "aborting synchronization process to safeguard data" not in combined_output:
return TestResult.fail_result(
self.case_id,
self.name,
"check_nomount did not emit the expected .nosync safeguard message",
artifacts,
details,
)
return TestResult.pass_result(self.case_id, self.name, artifacts, details)

213
ci/e2e/testcases/tc0018_recycle_bin_validation.py Normal file
View file

@ -0,0 +1,213 @@
from __future__ import annotations
import os
from pathlib import Path
from framework.base import E2ETestCase
from framework.context import E2EContext
from framework.manifest import build_manifest, write_manifest
from framework.result import TestResult
from framework.utils import command_to_string, reset_directory, run_command, write_text_file
class TestCase0018RecycleBinValidation(E2ETestCase):
case_id = "0018"
name = "recycle bin validation"
description = "Validate that online deletions are moved into a FreeDesktop-compliant recycle bin when enabled"
def _write_seed_config(self, config_path: Path) -> None:
write_text_file(config_path, "# tc0018 seed config\n" 'bypass_data_preservation = "true"\n')
def _write_cleanup_config(self, config_path: Path, recycle_bin_path: Path) -> None:
write_text_file(
config_path,
"# tc0018 cleanup config\n"
'bypass_data_preservation = "true"\n'
'cleanup_local_files = "true"\n'
'download_only = "true"\n'
'use_recycle_bin = "true"\n'
f'recycle_bin_path = "{recycle_bin_path}"\n',
)
def run(self, context: E2EContext) -> TestResult:
case_work_dir = context.work_root / "tc0018"
case_log_dir = context.logs_dir / "tc0018"
state_dir = context.state_dir / "tc0018"
reset_directory(case_work_dir)
reset_directory(case_log_dir)
reset_directory(state_dir)
context.ensure_refresh_token_available()
sync_root = case_work_dir / "syncroot"
conf_seed = case_work_dir / "conf-seed"
conf_cleanup = case_work_dir / "conf-cleanup"
verify_root = case_work_dir / "verifyroot"
conf_verify = case_work_dir / "conf-verify"
recycle_bin_root = case_work_dir / "RecycleBin"
root_name = f"ZZ_E2E_TC0018_{context.run_id}_{os.getpid()}"
write_text_file(sync_root / root_name / "Keep" / "keep.txt", "keep\n")
write_text_file(sync_root / root_name / "OldData" / "old.txt", "old\n")
context.bootstrap_config_dir(conf_seed)
self._write_seed_config(conf_seed / "config")
context.bootstrap_config_dir(conf_cleanup)
self._write_cleanup_config(conf_cleanup / "config", recycle_bin_root)
context.bootstrap_config_dir(conf_verify)
self._write_seed_config(conf_verify / "config")
seed_stdout = case_log_dir / "seed_stdout.log"
seed_stderr = case_log_dir / "seed_stderr.log"
remove_stdout = case_log_dir / "remove_stdout.log"
remove_stderr = case_log_dir / "remove_stderr.log"
cleanup_stdout = case_log_dir / "cleanup_stdout.log"
cleanup_stderr = case_log_dir / "cleanup_stderr.log"
verify_stdout = case_log_dir / "verify_stdout.log"
verify_stderr = case_log_dir / "verify_stderr.log"
recycle_manifest_file = state_dir / "recycle_manifest.txt"
remote_manifest_file = state_dir / "remote_verify_manifest.txt"
local_manifest_file = state_dir / "local_manifest_after_cleanup.txt"
metadata_file = state_dir / "metadata.txt"
seed_command = [
context.onedrive_bin,
"--display-running-config",
"--upload-only",
"--verbose",
"--resync",
"--resync-auth",
"--single-directory",
root_name,
"--syncdir",
str(sync_root),
"--confdir",
str(conf_seed),
]
context.log(f"Executing Test Case {self.case_id} seed: {command_to_string(seed_command)}")
seed_result = run_command(seed_command, cwd=context.repo_root)
write_text_file(seed_stdout, seed_result.stdout)
write_text_file(seed_stderr, seed_result.stderr)
remove_command = [
context.onedrive_bin,
"--display-running-config",
"--verbose",
"--remove-directory",
f"{root_name}/OldData",
"--syncdir",
str(sync_root),
"--confdir",
str(conf_seed),
]
remove_result = run_command(remove_command, cwd=context.repo_root)
write_text_file(remove_stdout, remove_result.stdout)
write_text_file(remove_stderr, remove_result.stderr)
cleanup_command = [
context.onedrive_bin,
"--display-running-config",
"--sync",
"--verbose",
"--download-only",
"--cleanup-local-files",
"--single-directory",
root_name,
"--syncdir",
str(sync_root),
"--confdir",
str(conf_cleanup),
]
cleanup_result = run_command(cleanup_command, cwd=context.repo_root)
write_text_file(cleanup_stdout, cleanup_result.stdout)
write_text_file(cleanup_stderr, cleanup_result.stderr)
verify_command = [
context.onedrive_bin,
"--display-running-config",
"--sync",
"--verbose",
"--download-only",
"--resync",
"--resync-auth",
"--single-directory",
root_name,
"--syncdir",
str(verify_root),
"--confdir",
str(conf_verify),
]
verify_result = run_command(verify_command, cwd=context.repo_root)
write_text_file(verify_stdout, verify_result.stdout)
write_text_file(verify_stderr, verify_result.stderr)
recycle_manifest = build_manifest(recycle_bin_root)
remote_manifest = build_manifest(verify_root)
local_manifest = build_manifest(sync_root)
write_manifest(recycle_manifest_file, recycle_manifest)
write_manifest(remote_manifest_file, remote_manifest)
write_manifest(local_manifest_file, local_manifest)
write_text_file(
metadata_file,
"\n".join(
[
f"case_id={self.case_id}",
f"root_name={root_name}",
f"seed_returncode={seed_result.returncode}",
f"remove_returncode={remove_result.returncode}",
f"cleanup_returncode={cleanup_result.returncode}",
f"verify_returncode={verify_result.returncode}",
]
)
+ "\n",
)
artifacts = [
str(seed_stdout),
str(seed_stderr),
str(remove_stdout),
str(remove_stderr),
str(cleanup_stdout),
str(cleanup_stderr),
str(verify_stdout),
str(verify_stderr),
str(recycle_manifest_file),
str(remote_manifest_file),
str(local_manifest_file),
str(metadata_file),
]
details = {
"seed_returncode": seed_result.returncode,
"remove_returncode": remove_result.returncode,
"cleanup_returncode": cleanup_result.returncode,
"verify_returncode": verify_result.returncode,
"root_name": root_name,
}
if seed_result.returncode != 0:
return TestResult.fail_result(self.case_id, self.name, f"Remote seed failed with status {seed_result.returncode}", artifacts, details)
if remove_result.returncode != 0:
return TestResult.fail_result(self.case_id, self.name, f"Online directory removal failed with status {remove_result.returncode}", artifacts, details)
if cleanup_result.returncode != 0:
return TestResult.fail_result(self.case_id, self.name, f"Recycle bin cleanup sync failed with status {cleanup_result.returncode}", artifacts, details)
if verify_result.returncode != 0:
return TestResult.fail_result(self.case_id, self.name, f"Remote verification failed with status {verify_result.returncode}", artifacts, details)
if (sync_root / root_name / "OldData").exists():
return TestResult.fail_result(self.case_id, self.name, "OldData still exists locally after online deletion cleanup", artifacts, details)
if not (sync_root / root_name / "Keep" / "keep.txt").is_file():
return TestResult.fail_result(self.case_id, self.name, "Keep file is missing locally after recycle bin processing", artifacts, details)
recycle_has_file = any(path.endswith("old.txt") for path in recycle_manifest)
recycle_has_info = any(path.endswith(".trashinfo") for path in recycle_manifest)
if not recycle_has_file:
return TestResult.fail_result(self.case_id, self.name, "Deleted content was not moved into the configured recycle bin", artifacts, details)
if not recycle_has_info:
return TestResult.fail_result(self.case_id, self.name, "Recycle bin metadata .trashinfo file was not created", artifacts, details)
if f"{root_name}/Keep/keep.txt" not in remote_manifest:
return TestResult.fail_result(self.case_id, self.name, "Keep file is missing online after recycle bin processing", artifacts, details)
if any(entry == f"{root_name}/OldData" or entry.startswith(f"{root_name}/OldData/") for entry in remote_manifest):
return TestResult.fail_result(self.case_id, self.name, "OldData still exists online after explicit online removal", artifacts, details)
return TestResult.pass_result(self.case_id, self.name, artifacts, details)

98
ci/e2e/testcases/tc0019_logging_and_running_config.py Normal file
View file

@ -0,0 +1,98 @@
from __future__ import annotations
import os
from pathlib import Path
from framework.base import E2ETestCase
from framework.context import E2EContext
from framework.result import TestResult
from framework.utils import command_to_string, reset_directory, run_command, write_text_file
class TestCase0019LoggingAndRunningConfig(E2ETestCase):
case_id = "0019"
name = "logging and running config validation"
description = "Validate custom log_dir output and display-running-config visibility"
def _write_config(self, config_path: Path, app_log_dir: Path) -> None:
write_text_file(
config_path,
"# tc0019 config\n"
'bypass_data_preservation = "true"\n'
'enable_logging = "true"\n'
f'log_dir = "{app_log_dir}"\n',
)
def run(self, context: E2EContext) -> TestResult:
case_work_dir = context.work_root / "tc0019"
case_log_dir = context.logs_dir / "tc0019"
state_dir = context.state_dir / "tc0019"
reset_directory(case_work_dir)
reset_directory(case_log_dir)
reset_directory(state_dir)
context.ensure_refresh_token_available()
sync_root = case_work_dir / "syncroot"
confdir = case_work_dir / "conf-main"
root_name = f"ZZ_E2E_TC0019_{context.run_id}_{os.getpid()}"
app_log_dir = case_log_dir / "app-logs"
write_text_file(sync_root / root_name / "logging.txt", "log me\n")
context.bootstrap_config_dir(confdir)
self._write_config(confdir / "config", app_log_dir)
stdout_file = case_log_dir / "logging_stdout.log"
stderr_file = case_log_dir / "logging_stderr.log"
metadata_file = state_dir / "metadata.txt"
command = [
context.onedrive_bin,
"--display-running-config",
"--sync",
"--verbose",
"--resync",
"--resync-auth",
"--single-directory",
root_name,
"--syncdir",
str(sync_root),
"--confdir",
str(confdir),
]
context.log(f"Executing Test Case {self.case_id}: {command_to_string(command)}")
result = run_command(command, cwd=context.repo_root)
write_text_file(stdout_file, result.stdout)
write_text_file(stderr_file, result.stderr)
log_entries = sorted(str(p.relative_to(app_log_dir)) for p in app_log_dir.rglob("*") if p.is_file()) if app_log_dir.exists() else []
write_text_file(
metadata_file,
"\n".join(
[
f"case_id={self.case_id}",
f"root_name={root_name}",
f"returncode={result.returncode}",
] + [f"log_file={entry}" for entry in log_entries]
) + "\n",
)
artifacts = [str(stdout_file), str(stderr_file), str(metadata_file)]
if app_log_dir.exists():
artifacts.append(str(app_log_dir))
details = {
"returncode": result.returncode,
"root_name": root_name,
"log_file_count": len(log_entries),
}
if result.returncode != 0:
return TestResult.fail_result(self.case_id, self.name, f"Logging validation failed with status {result.returncode}", artifacts, details)
if not log_entries:
return TestResult.fail_result(self.case_id, self.name, "No application log files were created in the configured log_dir", artifacts, details)
stdout_lower = result.stdout.lower()
if "display_running_config" not in stdout_lower and "log_dir" not in stdout_lower:
return TestResult.fail_result(self.case_id, self.name, "display-running-config output did not expose the active runtime configuration", artifacts, details)
return TestResult.pass_result(self.case_id, self.name, artifacts, details)

144
ci/e2e/testcases/tc0020_monitor_mode_validation.py Normal file
View file

@ -0,0 +1,144 @@
from __future__ import annotations
import os
import signal
import subprocess
import time
from pathlib import Path
from framework.base import E2ETestCase
from framework.context import E2EContext
from framework.manifest import build_manifest, write_manifest
from framework.result import TestResult
from framework.utils import command_to_string, reset_directory, run_command, write_text_file
class TestCase0020MonitorModeValidation(E2ETestCase):
case_id = "0020"
name = "monitor mode validation"
description = "Validate that monitor mode uploads local changes without manually re-running --sync"
def _write_config(self, config_path: Path, app_log_dir: Path) -> None:
write_text_file(
config_path,
"# tc0020 config\n"
'bypass_data_preservation = "true"\n'
'enable_logging = "true"\n'
f'log_dir = "{app_log_dir}"\n'
'monitor_interval = "5"\n'
'monitor_fullscan_frequency = "1"\n',
)
def run(self, context: E2EContext) -> TestResult:
case_work_dir = context.work_root / "tc0020"
case_log_dir = context.logs_dir / "tc0020"
state_dir = context.state_dir / "tc0020"
reset_directory(case_work_dir)
reset_directory(case_log_dir)
reset_directory(state_dir)
context.ensure_refresh_token_available()
sync_root = case_work_dir / "syncroot"
confdir = case_work_dir / "conf-main"
verify_root = case_work_dir / "verifyroot"
verify_conf = case_work_dir / "conf-verify"
root_name = f"ZZ_E2E_TC0020_{context.run_id}_{os.getpid()}"
app_log_dir = case_log_dir / "app-logs"
write_text_file(sync_root / root_name / "baseline.txt", "baseline\n")
context.bootstrap_config_dir(confdir)
self._write_config(confdir / "config", app_log_dir)
context.bootstrap_config_dir(verify_conf)
write_text_file(verify_conf / "config", "# tc0020 verify\n" 'bypass_data_preservation = "true"\n')
stdout_file = case_log_dir / "monitor_stdout.log"
stderr_file = case_log_dir / "monitor_stderr.log"
verify_stdout = case_log_dir / "verify_stdout.log"
verify_stderr = case_log_dir / "verify_stderr.log"
remote_manifest_file = state_dir / "remote_verify_manifest.txt"
metadata_file = state_dir / "metadata.txt"
command = [
context.onedrive_bin,
"--display-running-config",
"--monitor",
"--verbose",
"--resync",
"--resync-auth",
"--single-directory",
root_name,
"--syncdir",
str(sync_root),
"--confdir",
str(confdir),
]
context.log(f"Executing Test Case {self.case_id}: {command_to_string(command)}")
with stdout_file.open("w", encoding="utf-8") as stdout_fp, stderr_file.open("w", encoding="utf-8") as stderr_fp:
process = subprocess.Popen(
command,
cwd=str(context.repo_root),
stdout=stdout_fp,
stderr=stderr_fp,
text=True,
)
time.sleep(8)
write_text_file(sync_root / root_name / "monitor-added.txt", "added while monitor mode was running\n")
time.sleep(12)
process.send_signal(signal.SIGINT)
try:
process.wait(timeout=30)
except subprocess.TimeoutExpired:
process.kill()
process.wait(timeout=30)
verify_command = [
context.onedrive_bin,
"--display-running-config",
"--sync",
"--verbose",
"--download-only",
"--resync",
"--resync-auth",
"--single-directory",
root_name,
"--syncdir",
str(verify_root),
"--confdir",
str(verify_conf),
]
verify_result = run_command(verify_command, cwd=context.repo_root)
write_text_file(verify_stdout, verify_result.stdout)
write_text_file(verify_stderr, verify_result.stderr)
remote_manifest = build_manifest(verify_root)
write_manifest(remote_manifest_file, remote_manifest)
write_text_file(
metadata_file,
"\n".join(
[
f"case_id={self.case_id}",
f"root_name={root_name}",
f"monitor_returncode={process.returncode}",
f"verify_returncode={verify_result.returncode}",
]
) + "\n",
)
artifacts = [str(stdout_file), str(stderr_file), str(verify_stdout), str(verify_stderr), str(remote_manifest_file), str(metadata_file)]
if app_log_dir.exists():
artifacts.append(str(app_log_dir))
details = {
"monitor_returncode": process.returncode,
"verify_returncode": verify_result.returncode,
"root_name": root_name,
}
if verify_result.returncode != 0:
return TestResult.fail_result(self.case_id, self.name, f"Remote verification failed with status {verify_result.returncode}", artifacts, details)
if f"{root_name}/monitor-added.txt" not in remote_manifest:
return TestResult.fail_result(self.case_id, self.name, "Monitor mode did not upload the file created while the process was running", artifacts, details)
return TestResult.pass_result(self.case_id, self.name, artifacts, details)

155
ci/e2e/testcases/tc0021_resumable_transfers_validation.py Normal file
View file

@ -0,0 +1,155 @@
from __future__ import annotations
import os
import signal
import subprocess
import time
from pathlib import Path
from framework.base import E2ETestCase
from framework.context import E2EContext
from framework.manifest import build_manifest, write_manifest
from framework.result import TestResult
from framework.utils import command_to_string, reset_directory, run_command, write_text_file
class TestCase0021ResumableTransfersValidation(E2ETestCase):
case_id = "0021"
name = "resumable transfers validation"
description = "Validate interrupted upload recovery for a resumable session upload"
def _write_config(self, config_path: Path, app_log_dir: Path) -> None:
write_text_file(
config_path,
"# tc0021 config\n"
'bypass_data_preservation = "true"\n'
'enable_logging = "true"\n'
f'log_dir = "{app_log_dir}"\n'
'force_session_upload = "true"\n'
'rate_limit = "262144"\n',
)
def run(self, context: E2EContext) -> TestResult:
case_work_dir = context.work_root / "tc0021"
case_log_dir = context.logs_dir / "tc0021"
state_dir = context.state_dir / "tc0021"
reset_directory(case_work_dir)
reset_directory(case_log_dir)
reset_directory(state_dir)
context.ensure_refresh_token_available()
sync_root = case_work_dir / "syncroot"
confdir = case_work_dir / "conf-main"
verify_root = case_work_dir / "verifyroot"
verify_conf = case_work_dir / "conf-verify"
root_name = f"ZZ_E2E_TC0021_{context.run_id}_{os.getpid()}"
app_log_dir = case_log_dir / "app-logs"
large_file = sync_root / root_name / "session-large.bin"
large_file.parent.mkdir(parents=True, exist_ok=True)
large_file.write_bytes(b"R" * (5 * 1024 * 1024))
context.bootstrap_config_dir(confdir)
self._write_config(confdir / "config", app_log_dir)
context.bootstrap_config_dir(verify_conf)
write_text_file(verify_conf / "config", "# tc0021 verify\n" 'bypass_data_preservation = "true"\n')
phase1_stdout = case_log_dir / "phase1_stdout.log"
phase1_stderr = case_log_dir / "phase1_stderr.log"
phase2_stdout = case_log_dir / "phase2_stdout.log"
phase2_stderr = case_log_dir / "phase2_stderr.log"
verify_stdout = case_log_dir / "verify_stdout.log"
verify_stderr = case_log_dir / "verify_stderr.log"
remote_manifest_file = state_dir / "remote_verify_manifest.txt"
metadata_file = state_dir / "metadata.txt"
command = [
context.onedrive_bin,
"--display-running-config",
"--upload-only",
"--verbose",
"--resync",
"--resync-auth",
"--single-directory",
root_name,
"--syncdir",
str(sync_root),
"--confdir",
str(confdir),
]
context.log(f"Executing Test Case {self.case_id} phase 1: {command_to_string(command)}")
with phase1_stdout.open("w", encoding="utf-8") as stdout_fp, phase1_stderr.open("w", encoding="utf-8") as stderr_fp:
process = subprocess.Popen(
command,
cwd=str(context.repo_root),
stdout=stdout_fp,
stderr=stderr_fp,
text=True,
)
time.sleep(5)
process.send_signal(signal.SIGINT)
try:
process.wait(timeout=30)
except subprocess.TimeoutExpired:
process.kill()
process.wait(timeout=30)
context.log(f"Executing Test Case {self.case_id} phase 2: {command_to_string(command)}")
phase2_result = run_command(command, cwd=context.repo_root)
write_text_file(phase2_stdout, phase2_result.stdout)
write_text_file(phase2_stderr, phase2_result.stderr)
verify_command = [
context.onedrive_bin,
"--display-running-config",
"--sync",
"--verbose",
"--download-only",
"--resync",
"--resync-auth",
"--single-directory",
root_name,
"--syncdir",
str(verify_root),
"--confdir",
str(verify_conf),
]
verify_result = run_command(verify_command, cwd=context.repo_root)
write_text_file(verify_stdout, verify_result.stdout)
write_text_file(verify_stderr, verify_result.stderr)
remote_manifest = build_manifest(verify_root)
write_manifest(remote_manifest_file, remote_manifest)
write_text_file(
metadata_file,
"\n".join(
[
f"case_id={self.case_id}",
f"root_name={root_name}",
f"phase1_returncode={process.returncode}",
f"phase2_returncode={phase2_result.returncode}",
f"verify_returncode={verify_result.returncode}",
f"large_size={large_file.stat().st_size}",
]
) + "\n",
)
artifacts = [str(phase1_stdout), str(phase1_stderr), str(phase2_stdout), str(phase2_stderr), str(verify_stdout), str(verify_stderr), str(remote_manifest_file), str(metadata_file)]
if app_log_dir.exists():
artifacts.append(str(app_log_dir))
details = {
"phase1_returncode": process.returncode,
"phase2_returncode": phase2_result.returncode,
"verify_returncode": verify_result.returncode,
"root_name": root_name,
"large_size": large_file.stat().st_size,
}
if phase2_result.returncode != 0:
return TestResult.fail_result(self.case_id, self.name, f"Resumable upload recovery phase failed with status {phase2_result.returncode}", artifacts, details)
if verify_result.returncode != 0:
return TestResult.fail_result(self.case_id, self.name, f"Remote verification failed with status {verify_result.returncode}", artifacts, details)
if f"{root_name}/session-large.bin" not in remote_manifest:
return TestResult.fail_result(self.case_id, self.name, "Interrupted resumable upload did not complete successfully on the subsequent run", artifacts, details)
return TestResult.pass_result(self.case_id, self.name, artifacts, details)

115
ci/e2e/testcases/tc0022_local_first_validation.py Normal file
View file

@ -0,0 +1,115 @@
from __future__ import annotations
import os
from pathlib import Path
from framework.base import E2ETestCase
from framework.context import E2EContext
from framework.manifest import build_manifest, write_manifest
from framework.result import TestResult
from framework.utils import command_to_string, reset_directory, run_command, write_text_file
class TestCase0022LocalFirstValidation(E2ETestCase):
case_id = "0022"
name = "local_first validation"
description = "Validate that local_first treats local content as the source of truth during a conflict"
def _write_default_config(self, config_path: Path) -> None:
write_text_file(config_path, "# tc0022 config\n" 'bypass_data_preservation = "true"\n')
def _write_local_first_config(self, config_path: Path) -> None:
write_text_file(config_path, "# tc0022 local first config\n" 'bypass_data_preservation = "true"\n' 'local_first = "true"\n')
def run(self, context: E2EContext) -> TestResult:
case_work_dir = context.work_root / "tc0022"
case_log_dir = context.logs_dir / "tc0022"
state_dir = context.state_dir / "tc0022"
reset_directory(case_work_dir)
reset_directory(case_log_dir)
reset_directory(state_dir)
context.ensure_refresh_token_available()
seed_root = case_work_dir / "seedroot"
local_root = case_work_dir / "localroot"
remote_update_root = case_work_dir / "remoteupdateroot"
verify_root = case_work_dir / "verifyroot"
conf_seed = case_work_dir / "conf-seed"
conf_local = case_work_dir / "conf-local"
conf_remote = case_work_dir / "conf-remote"
conf_verify = case_work_dir / "conf-verify"
root_name = f"ZZ_E2E_TC0022_{context.run_id}_{os.getpid()}"
relative_file = f"{root_name}/conflict.txt"
write_text_file(seed_root / relative_file, "base\n")
write_text_file(remote_update_root / relative_file, "remote wins unless local_first applies\n")
context.bootstrap_config_dir(conf_seed)
self._write_default_config(conf_seed / "config")
context.bootstrap_config_dir(conf_local)
self._write_local_first_config(conf_local / "config")
context.bootstrap_config_dir(conf_remote)
self._write_default_config(conf_remote / "config")
context.bootstrap_config_dir(conf_verify)
self._write_default_config(conf_verify / "config")
seed_stdout = case_log_dir / "seed_stdout.log"
seed_stderr = case_log_dir / "seed_stderr.log"
download_stdout = case_log_dir / "download_stdout.log"
download_stderr = case_log_dir / "download_stderr.log"
remote_stdout = case_log_dir / "remote_update_stdout.log"
remote_stderr = case_log_dir / "remote_update_stderr.log"
final_stdout = case_log_dir / "final_sync_stdout.log"
final_stderr = case_log_dir / "final_sync_stderr.log"
verify_stdout = case_log_dir / "verify_stdout.log"
verify_stderr = case_log_dir / "verify_stderr.log"
remote_manifest_file = state_dir / "remote_verify_manifest.txt"
metadata_file = state_dir / "metadata.txt"
seed_command = [context.onedrive_bin, "--display-running-config", "--upload-only", "--verbose", "--resync", "--resync-auth", "--single-directory", root_name, "--syncdir", str(seed_root), "--confdir", str(conf_seed)]
seed_result = run_command(seed_command, cwd=context.repo_root)
write_text_file(seed_stdout, seed_result.stdout)
write_text_file(seed_stderr, seed_result.stderr)
download_command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--single-directory", root_name, "--syncdir", str(local_root), "--confdir", str(conf_local)]
download_result = run_command(download_command, cwd=context.repo_root)
write_text_file(download_stdout, download_result.stdout)
write_text_file(download_stderr, download_result.stderr)
write_text_file(local_root / relative_file, "local wins because local_first is enabled\n")
remote_command = [context.onedrive_bin, "--display-running-config", "--upload-only", "--verbose", "--single-directory", root_name, "--syncdir", str(remote_update_root), "--confdir", str(conf_remote)]
remote_result = run_command(remote_command, cwd=context.repo_root)
write_text_file(remote_stdout, remote_result.stdout)
write_text_file(remote_stderr, remote_result.stderr)
final_command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--single-directory", root_name, "--syncdir", str(local_root), "--confdir", str(conf_local)]
final_result = run_command(final_command, cwd=context.repo_root)
write_text_file(final_stdout, final_result.stdout)
write_text_file(final_stderr, final_result.stderr)
verify_command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--single-directory", root_name, "--syncdir", str(verify_root), "--confdir", str(conf_verify)]
verify_result = run_command(verify_command, cwd=context.repo_root)
write_text_file(verify_stdout, verify_result.stdout)
write_text_file(verify_stderr, verify_result.stderr)
remote_manifest = build_manifest(verify_root)
write_manifest(remote_manifest_file, remote_manifest)
local_content = (local_root / relative_file).read_text(encoding="utf-8") if (local_root / relative_file).is_file() else ""
remote_content = (verify_root / relative_file).read_text(encoding="utf-8") if (verify_root / relative_file).is_file() else ""
write_text_file(metadata_file, f"case_id={self.case_id}\nroot_name={root_name}\nseed_returncode={seed_result.returncode}\ndownload_returncode={download_result.returncode}\nremote_returncode={remote_result.returncode}\nfinal_returncode={final_result.returncode}\nverify_returncode={verify_result.returncode}\nlocal_content={local_content!r}\nremote_content={remote_content!r}\n")
artifacts = [str(seed_stdout), str(seed_stderr), str(download_stdout), str(download_stderr), str(remote_stdout), str(remote_stderr), str(final_stdout), str(final_stderr), str(verify_stdout), str(verify_stderr), str(remote_manifest_file), str(metadata_file)]
details = {"seed_returncode": seed_result.returncode, "download_returncode": download_result.returncode, "remote_returncode": remote_result.returncode, "final_returncode": final_result.returncode, "verify_returncode": verify_result.returncode, "root_name": root_name}
for label, rc in [("seed", seed_result.returncode), ("download", download_result.returncode), ("remote update", remote_result.returncode), ("final sync", final_result.returncode), ("verify", verify_result.returncode)]:
if rc != 0:
return TestResult.fail_result(self.case_id, self.name, f"{label} phase failed with status {rc}", artifacts, details)
expected = "local wins because local_first is enabled\n"
if local_content != expected:
return TestResult.fail_result(self.case_id, self.name, "Local content was not retained after conflict resolution with local_first enabled", artifacts, details)
if remote_content != expected:
return TestResult.fail_result(self.case_id, self.name, "Remote content did not converge to the local source-of-truth content when local_first was enabled", artifacts, details)
return TestResult.pass_result(self.case_id, self.name, artifacts, details)

101
ci/e2e/testcases/tc0023_bypass_data_preservation_validation.py Normal file
View file

@ -0,0 +1,101 @@
from __future__ import annotations
import os
from pathlib import Path
from framework.base import E2ETestCase
from framework.context import E2EContext
from framework.result import TestResult
from framework.utils import reset_directory, run_command, write_text_file
class TestCase0023BypassDataPreservationValidation(E2ETestCase):
case_id = "0023"
name = "bypass_data_preservation validation"
description = "Validate that bypass_data_preservation overwrites local conflict data instead of creating safeBackup files"
def _write_default_config(self, config_path: Path) -> None:
write_text_file(config_path, "# tc0023 config\n" 'bypass_data_preservation = "false"\n')
def _write_bypass_config(self, config_path: Path) -> None:
write_text_file(config_path, "# tc0023 bypass config\n" 'bypass_data_preservation = "true"\n')
def run(self, context: E2EContext) -> TestResult:
case_work_dir = context.work_root / "tc0023"
case_log_dir = context.logs_dir / "tc0023"
state_dir = context.state_dir / "tc0023"
reset_directory(case_work_dir)
reset_directory(case_log_dir)
reset_directory(state_dir)
context.ensure_refresh_token_available()
seed_root = case_work_dir / "seedroot"
local_root = case_work_dir / "localroot"
remote_update_root = case_work_dir / "remoteupdateroot"
conf_seed = case_work_dir / "conf-seed"
conf_local = case_work_dir / "conf-local"
conf_remote = case_work_dir / "conf-remote"
root_name = f"ZZ_E2E_TC0023_{context.run_id}_{os.getpid()}"
relative_file = f"{root_name}/conflict.txt"
write_text_file(seed_root / relative_file, "base\n")
write_text_file(remote_update_root / relative_file, "remote authoritative content\n")
context.bootstrap_config_dir(conf_seed)
self._write_default_config(conf_seed / "config")
context.bootstrap_config_dir(conf_local)
self._write_bypass_config(conf_local / "config")
context.bootstrap_config_dir(conf_remote)
self._write_default_config(conf_remote / "config")
seed_stdout = case_log_dir / "seed_stdout.log"
seed_stderr = case_log_dir / "seed_stderr.log"
download_stdout = case_log_dir / "download_stdout.log"
download_stderr = case_log_dir / "download_stderr.log"
remote_stdout = case_log_dir / "remote_update_stdout.log"
remote_stderr = case_log_dir / "remote_update_stderr.log"
final_stdout = case_log_dir / "final_sync_stdout.log"
final_stderr = case_log_dir / "final_sync_stderr.log"
metadata_file = state_dir / "metadata.txt"
seed_command = [context.onedrive_bin, "--display-running-config", "--upload-only", "--verbose", "--resync", "--resync-auth", "--single-directory", root_name, "--syncdir", str(seed_root), "--confdir", str(conf_seed)]
seed_result = run_command(seed_command, cwd=context.repo_root)
write_text_file(seed_stdout, seed_result.stdout)
write_text_file(seed_stderr, seed_result.stderr)
download_command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--single-directory", root_name, "--syncdir", str(local_root), "--confdir", str(conf_local)]
download_result = run_command(download_command, cwd=context.repo_root)
write_text_file(download_stdout, download_result.stdout)
write_text_file(download_stderr, download_result.stderr)
write_text_file(local_root / relative_file, "local conflicting content\n")
remote_command = [context.onedrive_bin, "--display-running-config", "--upload-only", "--verbose", "--single-directory", root_name, "--syncdir", str(remote_update_root), "--confdir", str(conf_remote)]
remote_result = run_command(remote_command, cwd=context.repo_root)
write_text_file(remote_stdout, remote_result.stdout)
write_text_file(remote_stderr, remote_result.stderr)
final_command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--single-directory", root_name, "--syncdir", str(local_root), "--confdir", str(conf_local)]
final_result = run_command(final_command, cwd=context.repo_root)
write_text_file(final_stdout, final_result.stdout)
write_text_file(final_stderr, final_result.stderr)
local_file = local_root / relative_file
local_content = local_file.read_text(encoding="utf-8") if local_file.is_file() else ""
safe_backup_files = [p.name for p in local_file.parent.glob("*safeBackup*")]
write_text_file(metadata_file, f"case_id={self.case_id}\nroot_name={root_name}\nseed_returncode={seed_result.returncode}\ndownload_returncode={download_result.returncode}\nremote_returncode={remote_result.returncode}\nfinal_returncode={final_result.returncode}\nlocal_content={local_content!r}\nsafe_backup_files={safe_backup_files!r}\n")
artifacts = [str(seed_stdout), str(seed_stderr), str(download_stdout), str(download_stderr), str(remote_stdout), str(remote_stderr), str(final_stdout), str(final_stderr), str(metadata_file)]
details = {"seed_returncode": seed_result.returncode, "download_returncode": download_result.returncode, "remote_returncode": remote_result.returncode, "final_returncode": final_result.returncode, "root_name": root_name, "safe_backup_count": len(safe_backup_files)}
for label, rc in [("seed", seed_result.returncode), ("download", download_result.returncode), ("remote update", remote_result.returncode), ("final sync", final_result.returncode)]:
if rc != 0:
return TestResult.fail_result(self.case_id, self.name, f"{label} phase failed with status {rc}", artifacts, details)
expected = "remote authoritative content\n"
if local_content != expected:
return TestResult.fail_result(self.case_id, self.name, "Local conflict content was not overwritten by the remote version when bypass_data_preservation was enabled", artifacts, details)
if safe_backup_files:
return TestResult.fail_result(self.case_id, self.name, "safeBackup files were created despite bypass_data_preservation being enabled", artifacts, details)
return TestResult.pass_result(self.case_id, self.name, artifacts, details)

111
ci/e2e/testcases/tc0024_big_delete_safeguard_validation.py Normal file
View file

@ -0,0 +1,111 @@
from __future__ import annotations
import os
import shutil
from pathlib import Path
from framework.base import E2ETestCase
from framework.context import E2EContext
from framework.manifest import build_manifest, write_manifest
from framework.result import TestResult
from framework.utils import reset_directory, run_command, write_text_file
class TestCase0024BigDeleteSafeguardValidation(E2ETestCase):
case_id = "0024"
name = "big delete safeguard validation"
description = "Validate classify_as_big_delete protection and forced acknowledgement via --force"
def _write_config(self, config_path: Path) -> None:
write_text_file(config_path, "# tc0024 config\n" 'bypass_data_preservation = "true"\n' 'classify_as_big_delete = "3"\n')
def run(self, context: E2EContext) -> TestResult:
case_work_dir = context.work_root / "tc0024"
case_log_dir = context.logs_dir / "tc0024"
state_dir = context.state_dir / "tc0024"
reset_directory(case_work_dir)
reset_directory(case_log_dir)
reset_directory(state_dir)
context.ensure_refresh_token_available()
seed_root = case_work_dir / "seedroot"
local_root = case_work_dir / "localroot"
verify_root = case_work_dir / "verifyroot"
conf_seed = case_work_dir / "conf-seed"
conf_local = case_work_dir / "conf-local"
conf_verify = case_work_dir / "conf-verify"
root_name = f"ZZ_E2E_TC0024_{context.run_id}_{os.getpid()}"
for idx in range(1, 6):
write_text_file(seed_root / root_name / "BigDelete" / f"file{idx}.txt", f"file {idx}\n")
write_text_file(seed_root / root_name / "Keep" / "keep.txt", "keep\n")
context.bootstrap_config_dir(conf_seed)
self._write_config(conf_seed / "config")
context.bootstrap_config_dir(conf_local)
self._write_config(conf_local / "config")
context.bootstrap_config_dir(conf_verify)
self._write_config(conf_verify / "config")
seed_stdout = case_log_dir / "seed_stdout.log"
seed_stderr = case_log_dir / "seed_stderr.log"
download_stdout = case_log_dir / "download_stdout.log"
download_stderr = case_log_dir / "download_stderr.log"
blocked_stdout = case_log_dir / "blocked_stdout.log"
blocked_stderr = case_log_dir / "blocked_stderr.log"
forced_stdout = case_log_dir / "forced_stdout.log"
forced_stderr = case_log_dir / "forced_stderr.log"
verify_stdout = case_log_dir / "verify_stdout.log"
verify_stderr = case_log_dir / "verify_stderr.log"
remote_manifest_file = state_dir / "remote_verify_manifest.txt"
metadata_file = state_dir / "metadata.txt"
seed_command = [context.onedrive_bin, "--display-running-config", "--upload-only", "--verbose", "--resync", "--resync-auth", "--single-directory", root_name, "--syncdir", str(seed_root), "--confdir", str(conf_seed)]
seed_result = run_command(seed_command, cwd=context.repo_root)
write_text_file(seed_stdout, seed_result.stdout)
write_text_file(seed_stderr, seed_result.stderr)
download_command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--single-directory", root_name, "--syncdir", str(local_root), "--confdir", str(conf_local)]
download_result = run_command(download_command, cwd=context.repo_root)
write_text_file(download_stdout, download_result.stdout)
write_text_file(download_stderr, download_result.stderr)
shutil.rmtree(local_root / root_name / "BigDelete")
blocked_command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--single-directory", root_name, "--syncdir", str(local_root), "--confdir", str(conf_local)]
blocked_result = run_command(blocked_command, cwd=context.repo_root)
write_text_file(blocked_stdout, blocked_result.stdout)
write_text_file(blocked_stderr, blocked_result.stderr)
forced_command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--force", "--single-directory", root_name, "--syncdir", str(local_root), "--confdir", str(conf_local)]
forced_result = run_command(forced_command, cwd=context.repo_root)
write_text_file(forced_stdout, forced_result.stdout)
write_text_file(forced_stderr, forced_result.stderr)
verify_command = [context.onedrive_bin, "--display-running-config", "--sync", "--verbose", "--download-only", "--resync", "--resync-auth", "--single-directory", root_name, "--syncdir", str(verify_root), "--confdir", str(conf_verify)]
verify_result = run_command(verify_command, cwd=context.repo_root)
write_text_file(verify_stdout, verify_result.stdout)
write_text_file(verify_stderr, verify_result.stderr)
remote_manifest = build_manifest(verify_root)
write_manifest(remote_manifest_file, remote_manifest)
blocked_output = (blocked_result.stdout + "\n" + blocked_result.stderr).lower()
write_text_file(metadata_file, f"case_id={self.case_id}\nroot_name={root_name}\nseed_returncode={seed_result.returncode}\ndownload_returncode={download_result.returncode}\nblocked_returncode={blocked_result.returncode}\nforced_returncode={forced_result.returncode}\nverify_returncode={verify_result.returncode}\n")
artifacts = [str(seed_stdout), str(seed_stderr), str(download_stdout), str(download_stderr), str(blocked_stdout), str(blocked_stderr), str(forced_stdout), str(forced_stderr), str(verify_stdout), str(verify_stderr), str(remote_manifest_file), str(metadata_file)]
details = {"seed_returncode": seed_result.returncode, "download_returncode": download_result.returncode, "blocked_returncode": blocked_result.returncode, "forced_returncode": forced_result.returncode, "verify_returncode": verify_result.returncode, "root_name": root_name}
for label, rc in [("seed", seed_result.returncode), ("download", download_result.returncode), ("forced sync", forced_result.returncode), ("verify", verify_result.returncode)]:
if rc != 0:
return TestResult.fail_result(self.case_id, self.name, f"{label} phase failed with status {rc}", artifacts, details)
if blocked_result.returncode == 0 and "big delete" not in blocked_output:
return TestResult.fail_result(self.case_id, self.name, "Big delete safeguard did not trigger before forced acknowledgement", artifacts, details)
if "big delete" not in blocked_output and "--force" not in blocked_output:
return TestResult.fail_result(self.case_id, self.name, "Blocked sync did not emit a big delete safeguard warning", artifacts, details)
if any(entry == f"{root_name}/BigDelete" or entry.startswith(f"{root_name}/BigDelete/") for entry in remote_manifest):
return TestResult.fail_result(self.case_id, self.name, "BigDelete content still exists online after acknowledged forced delete", artifacts, details)
if f"{root_name}/Keep/keep.txt" not in remote_manifest:
return TestResult.fail_result(self.case_id, self.name, "Keep content disappeared during big delete safeguard processing", artifacts, details)
return TestResult.pass_result(self.case_id, self.name, artifacts, details)